73 lines
3.1 KiB
Python
73 lines
3.1 KiB
Python
|
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")
|