dns-sync/tests/test_persistence.py

import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path

from dns_sync.adapters import (
    CERTIFICATES,
    NormalizedObject,
    Observation,
)
from dns_sync.persistence import SQLiteStateStore


ENDPOINT = "npm://192.168.1.3:81"
T1 = datetime(2026, 9, 19, 20, 0, tzinfo=timezone.utc)
T2 = datetime(2026, 9, 19, 20, 5, tzinfo=timezone.utc)
T3 = datetime(2026, 9, 19, 20, 10, tzinfo=timezone.utc)


def certificate(
    local_key: str,
    expires_on: str,
    observed_at: datetime = T1,
    payload=None,
) -> Observation:
    return Observation(
        endpoint_id=ENDPOINT,
        object=NormalizedObject(
            object_type=CERTIFICATES,
            logical_key="app.example.com",
            local_key=local_key,
            payload=payload
            or {
                "domain_names": ["app.example.com"],
                "expires_on": expires_on,
            },
        ),
        observed_at=observed_at,
    )


class ObservationPersistenceTests(unittest.TestCase):
    def test_snapshot_preserves_duplicate_logical_identity(self):
        rows = [
            certificate("63", "2026-12-01T00:00:00+00:00"),
            certificate("67", "2026-12-15T00:00:00+00:00"),
        ]

        with SQLiteStateStore(":memory:") as store:
            store.replace_observations(
                ENDPOINT,
                CERTIFICATES,
                rows,
                T1,
            )
            loaded = store.load_observations(CERTIFICATES)

        self.assertEqual(
            tuple(item.object.local_key for item in loaded),
            ("63", "67"),
        )
        self.assertEqual(
            {item.object.logical_key for item in loaded},
            {"app.example.com"},
        )

    def test_live_snapshot_atomically_replaces_previous_rows(self):
        with SQLiteStateStore(":memory:") as store:
            store.replace_observations(
                ENDPOINT,
                CERTIFICATES,
                [
                    certificate("63", "old"),
                    certificate("67", "newer"),
                ],
                T1,
            )
            store.replace_observations(
                ENDPOINT,
                CERTIFICATES,
                [
                    certificate(
                        "67",
                        "newest",
                        observed_at=T2,
                    )
                ],
                T2,
            )
            loaded = store.load_observations(CERTIFICATES)
            status = store.observation_status(
                ENDPOINT,
                CERTIFICATES,
            )

        self.assertEqual(len(loaded), 1)
        self.assertEqual(loaded[0].object.local_key, "67")
        self.assertEqual(
            loaded[0].object.payload["expires_on"],
            "newest",
        )
        self.assertTrue(status.is_live)
        self.assertEqual(status.last_success_at, T2)
        self.assertIsNone(status.last_error)

    def test_empty_live_snapshot_removes_previous_rows(self):
        with SQLiteStateStore(":memory:") as store:
            store.replace_observations(
                ENDPOINT,
                CERTIFICATES,
                [certificate("63", "old")],
                T1,
            )
            store.replace_observations(
                ENDPOINT,
                CERTIFICATES,
                [],
                T2,
            )

            self.assertEqual(
                store.load_observations(CERTIFICATES),
                [],
            )
            self.assertTrue(
                store.observation_status(
                    ENDPOINT,
                    CERTIFICATES,
                ).is_live
            )

    def test_failed_discovery_preserves_cache_until_live_recovery(self):
        with SQLiteStateStore(":memory:") as store:
            store.replace_observations(
                ENDPOINT,
                CERTIFICATES,
                [certificate("63", "cached")],
                T1,
            )
            store.record_discovery_failure(
                ENDPOINT,
                CERTIFICATES,
                "connection refused",
                T2,
            )

            cached = store.load_observations(CERTIFICATES)
            failed = store.observation_status(
                ENDPOINT,
                CERTIFICATES,
            )

            store.replace_observations(
                ENDPOINT,
                CERTIFICATES,
                [
                    certificate(
                        "67",
                        "live",
                        observed_at=T3,
                    )
                ],
                T3,
            )
            recovered = store.observation_status(
                ENDPOINT,
                CERTIFICATES,
            )

        self.assertEqual(len(cached), 1)
        self.assertEqual(cached[0].object.local_key, "63")
        self.assertFalse(failed.is_live)
        self.assertEqual(failed.last_success_at, T1)
        self.assertEqual(failed.last_attempt_at, T2)
        self.assertEqual(failed.last_error, "connection refused")
        self.assertTrue(recovered.is_live)
        self.assertEqual(recovered.last_success_at, T3)

    def test_sensitive_payload_is_rejected_before_cache_changes(self):
        with SQLiteStateStore(":memory:") as store:
            store.replace_observations(
                ENDPOINT,
                CERTIFICATES,
                [certificate("63", "safe")],
                T1,
            )
            unsafe = certificate(
                "67",
                "unsafe",
                observed_at=T2,
                payload={
                    "domain_names": ["app.example.com"],
                    "nested": {
                        "private_key": "must-never-persist",
                    },
                },
            )

            with self.assertRaisesRegex(
                RuntimeError,
                "sensitive field",
            ):
                store.replace_observations(
                    ENDPOINT,
                    CERTIFICATES,
                    [unsafe],
                    T2,
                )

            cached = store.load_observations(CERTIFICATES)

        self.assertEqual(len(cached), 1)
        self.assertEqual(cached[0].object.local_key, "63")

    def test_duplicate_local_keys_are_rejected(self):
        duplicate = certificate("63", "duplicate")
        with SQLiteStateStore(":memory:") as store:
            with self.assertRaisesRegex(
                RuntimeError,
                "Duplicate endpoint-local key",
            ):
                store.replace_observations(
                    ENDPOINT,
                    CERTIFICATES,
                    [duplicate, duplicate],
                    T1,
                )

    def test_file_database_survives_reopen(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "state.sqlite3"

            with SQLiteStateStore(path) as store:
                store.replace_observations(
                    ENDPOINT,
                    CERTIFICATES,
                    [certificate("63", "durable")],
                    T1,
                )

            with SQLiteStateStore(path) as reopened:
                loaded = reopened.load_observations(CERTIFICATES)
                status = reopened.observation_status(
                    ENDPOINT,
                    CERTIFICATES,
                )

        self.assertEqual(len(loaded), 1)
        self.assertEqual(
            loaded[0].object.payload["expires_on"],
            "durable",
        )
        self.assertEqual(status.last_success_at, T1)


if __name__ == "__main__":
    unittest.main()