hippogrief/pobsync
1
0
Fork 0
Code Issues 21 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Track snapshot base lineage in Django

Browse Source
This commit is contained in:
Peter van Arkel
2026-05-19 11:19:22 +02:00
parent 5808800981
commit 659377d894
6 changed files with 201 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
src/pobsync_backend/tests/test_snapshot_discovery.py
View File

@@ -10,7 +10,12 @@ from django.test import TestCase
from pobsync.util import write_yaml_atomic
from pobsync_backend.models import GlobalConfig, HostConfig, SnapshotRecord
from pobsync_backend.snapshot_discovery import discover_snapshots, parse_snapshot_datetime
from pobsync_backend.snapshot_discovery import (
discover_snapshots,
parse_snapshot_datetime,
resolve_base_links,
upsert_snapshot_record,
)
class SnapshotDiscoveryTests(TestCase):
@@ -58,6 +63,82 @@ class SnapshotDiscoveryTests(TestCase):
self.assertEqual(record.kind, "scheduled")
self.assertEqual(record.started_at, datetime(2026, 5, 19, 2, 15, tzinfo=timezone.utc))
def test_discovery_links_snapshot_to_base_record(self) -> None:
with TemporaryDirectory() as tmp:
backup_root = Path(tmp) / "backups"
GlobalConfig.objects.create(name="default", backup_root=str(backup_root))
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
base_dir = backup_root / host.host / "scheduled" / "20260518-021500Z__BASESNAP"
child_dir = backup_root / host.host / "scheduled" / "20260519-021500Z__CHILDSNP"
(base_dir / "meta").mkdir(parents=True)
(child_dir / "meta").mkdir(parents=True)
write_yaml_atomic(
base_dir / "meta" / "meta.yaml",
{
"id": "base-id",
"status": "success",
"started_at": "2026-05-18T02:15:00Z",
"base": None,
},
)
write_yaml_atomic(
child_dir / "meta" / "meta.yaml",
{
"id": "child-id",
"status": "success",
"started_at": "2026-05-19T02:15:00Z",
"base": {
"kind": "scheduled",
"dirname": base_dir.name,
"id": "base-id",
"path": str(base_dir / "data"),
},
},
)
result = discover_snapshots(host=host)
self.assertEqual(result["created"], 2)
child = SnapshotRecord.objects.get(dirname=child_dir.name)
base = SnapshotRecord.objects.get(dirname=base_dir.name)
self.assertEqual(child.base, base)
self.assertEqual(child.base_kind, "scheduled")
self.assertEqual(child.base_dirname, base_dir.name)
self.assertEqual(child.base_snapshot_id, "base-id")
self.assertEqual(child.base_path, str(base_dir / "data"))
def test_base_link_can_be_resolved_after_base_record_exists(self) -> None:
with TemporaryDirectory() as tmp:
backup_root = Path(tmp) / "backups"
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
base_dir = backup_root / host.host / "scheduled" / "20260518-021500Z__BASESNAP"
child_dir = backup_root / host.host / "scheduled" / "20260519-021500Z__CHILDSNP"
(base_dir / "meta").mkdir(parents=True)
(child_dir / "meta").mkdir(parents=True)
write_yaml_atomic(base_dir / "meta" / "meta.yaml", {"status": "success"})
write_yaml_atomic(
child_dir / "meta" / "meta.yaml",
{
"status": "success",
"base": {
"kind": "scheduled",
"dirname": base_dir.name,
"id": "base-id",
"path": str(base_dir / "data"),
},
},
)
child, _created = upsert_snapshot_record(host=host, kind="scheduled", snapshot_dir=child_dir)
upsert_snapshot_record(host=host, kind="scheduled", snapshot_dir=base_dir)
linked = resolve_base_links(host=host)
child.refresh_from_db()
self.assertEqual(linked, 1)
self.assertIsNotNone(child.base)
self.assertEqual(child.base.dirname, base_dir.name)
self.assertEqual(child.base_dirname, base_dir.name)
def test_command_discovers_snapshots_for_host(self) -> None:
with TemporaryDirectory() as tmp:
backup_root = Path(tmp) / "backups"