feat: discover snapshots into Django records
Add a Django-native snapshot discovery service and management command that scans backup directories, reads snapshot metadata, and idempotently upserts SnapshotRecord rows. Expose it through the pobsync command wrapper, update admin/docs, and cover discovery behavior with tests.
This commit is contained in:
72
src/pobsync_backend/tests/test_snapshot_discovery.py
Normal file
72
src/pobsync_backend/tests/test_snapshot_discovery.py
Normal file
@@ -0,0 +1,72 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from django.core.management import call_command
|
||||
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
|
||||
|
||||
|
||||
class SnapshotDiscoveryTests(TestCase):
|
||||
def test_parse_snapshot_datetime_prefers_metadata(self) -> None:
|
||||
parsed = parse_snapshot_datetime(
|
||||
"20260519-021500Z__ABCDEFGH",
|
||||
{"started_at": "2026-05-20T03:16:00Z"},
|
||||
"started_at",
|
||||
)
|
||||
|
||||
self.assertEqual(parsed, datetime(2026, 5, 20, 3, 16, tzinfo=timezone.utc))
|
||||
|
||||
def test_parse_snapshot_datetime_falls_back_to_dirname(self) -> None:
|
||||
parsed = parse_snapshot_datetime("20260519-021500Z__ABCDEFGH", {}, "started_at")
|
||||
|
||||
self.assertEqual(parsed, datetime(2026, 5, 19, 2, 15, tzinfo=timezone.utc))
|
||||
|
||||
def test_discovery_upserts_snapshot_records_idempotently(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")
|
||||
snapshot_dir = backup_root / host.host / "scheduled" / "20260519-021500Z__ABCDEFGH"
|
||||
meta_dir = snapshot_dir / "meta"
|
||||
meta_dir.mkdir(parents=True)
|
||||
write_yaml_atomic(
|
||||
meta_dir / "meta.yaml",
|
||||
{
|
||||
"status": "success",
|
||||
"started_at": "2026-05-19T02:15:00Z",
|
||||
"ended_at": "2026-05-19T02:16:00Z",
|
||||
},
|
||||
)
|
||||
|
||||
first = discover_snapshots(host=host)
|
||||
second = discover_snapshots(host=host)
|
||||
|
||||
self.assertEqual(first["created"], 1)
|
||||
self.assertEqual(first["updated"], 0)
|
||||
self.assertEqual(second["created"], 0)
|
||||
self.assertEqual(second["updated"], 1)
|
||||
self.assertEqual(SnapshotRecord.objects.count(), 1)
|
||||
record = SnapshotRecord.objects.get()
|
||||
self.assertEqual(record.status, "success")
|
||||
self.assertEqual(record.kind, "scheduled")
|
||||
self.assertEqual(record.started_at, datetime(2026, 5, 19, 2, 15, tzinfo=timezone.utc))
|
||||
|
||||
def test_command_discovers_snapshots_for_host(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")
|
||||
snapshot_dir = backup_root / host.host / ".incomplete" / "20260519-021500Z__ABCDEFGH"
|
||||
(snapshot_dir / "meta").mkdir(parents=True)
|
||||
|
||||
call_command("discover_pobsync_snapshots", host=host.host, kind="incomplete", stdout=StringIO())
|
||||
|
||||
self.assertEqual(SnapshotRecord.objects.count(), 1)
|
||||
self.assertEqual(SnapshotRecord.objects.get().kind, "incomplete")
|
||||
Reference in New Issue
Block a user