dns-sync/tests/test_reconciliation_persistence.py

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

from dns_sync.adapters import CERTIFICATES, NormalizedObject
from dns_sync.persistence import SQLiteStateStore


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


def certificate(
    expires_on: str,
    local_key: str | None = None,
) -> NormalizedObject:
    return NormalizedObject(
        object_type=CERTIFICATES,
        logical_key="app.example.com",
        local_key=local_key,
        payload={
            "domain_names": ["app.example.com"],
            "expires_on": expires_on,
        },
    )


class DesiredStatePersistenceTests(unittest.TestCase):
    def test_snapshot_round_trips_and_can_be_cleared(self):
        desired = certificate("2026-12-15T00:00:00+00:00")

        with SQLiteStateStore(":memory:") as store:
            store.replace_desired_state(
                CERTIFICATES,
                [desired],
                T1,
            )
            loaded = store.load_desired_state(CERTIFICATES)

            store.replace_desired_state(
                CERTIFICATES,
                [],
                T2,
            )
            cleared = store.load_desired_state(CERTIFICATES)

        self.assertEqual(len(loaded), 1)
        self.assertEqual(loaded[0].object, desired)
        self.assertEqual(loaded[0].selected_at, T1)
        self.assertEqual(cleared, [])

    def test_endpoint_local_identity_is_rejected(self):
        with SQLiteStateStore(":memory:") as store:
            with self.assertRaisesRegex(
                RuntimeError,
                "endpoint-local key",
            ):
                store.replace_desired_state(
                    CERTIFICATES,
                    [certificate("new", local_key="67")],
                    T1,
                )

    def test_sensitive_payload_is_rejected_before_snapshot_changes(self):
        safe = certificate("safe")
        unsafe = NormalizedObject(
            object_type=CERTIFICATES,
            logical_key="app.example.com",
            payload={
                "domain_names": ["app.example.com"],
                "nested": {
                    "private_key": "must-never-persist",
                },
            },
        )

        with SQLiteStateStore(":memory:") as store:
            store.replace_desired_state(
                CERTIFICATES,
                [safe],
                T1,
            )

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

            loaded = store.load_desired_state(CERTIFICATES)

        self.assertEqual(len(loaded), 1)
        self.assertEqual(loaded[0].object, safe)
        self.assertEqual(loaded[0].selected_at, T1)


class AppliedStatePersistenceTests(unittest.TestCase):
    def test_apply_requires_successful_verification(self):
        first = certificate("first", local_key="91")
        second = certificate("second", local_key="91")

        with SQLiteStateStore(":memory:") as store:
            store.record_applied(ENDPOINT, first, T1)
            before = store.load_applied_state(
                ENDPOINT,
                CERTIFICATES,
            )[0]

            store.record_verification_success(
                ENDPOINT,
                first,
                T2,
            )
            verified = store.load_applied_state(
                ENDPOINT,
                CERTIFICATES,
            )[0]

            store.record_applied(ENDPOINT, second, T3)
            reapplied = store.load_applied_state(
                ENDPOINT,
                CERTIFICATES,
            )[0]

        self.assertFalse(before.is_verified)
        self.assertIsNone(before.verified_at)

        self.assertTrue(verified.is_verified)
        self.assertEqual(verified.verified_at, T2)
        self.assertEqual(verified.verification_attempted_at, T2)

        self.assertEqual(
            reapplied.object.payload["expires_on"],
            "second",
        )
        self.assertEqual(reapplied.applied_at, T3)
        self.assertFalse(reapplied.is_verified)
        self.assertIsNone(reapplied.verification_attempted_at)
        self.assertIsNone(reapplied.verified_at)

    def test_stale_payload_cannot_verify_current_applied_state(self):
        old = certificate("old", local_key="91")
        current = certificate("current", local_key="91")

        with SQLiteStateStore(":memory:") as store:
            store.record_applied(ENDPOINT, current, T1)

            with self.assertRaisesRegex(
                RuntimeError,
                "does not match",
            ):
                store.record_verification_success(
                    ENDPOINT,
                    old,
                    T2,
                )

            state = store.load_applied_state(
                ENDPOINT,
                CERTIFICATES,
            )[0]

        self.assertFalse(state.is_verified)
        self.assertIsNone(state.verification_attempted_at)
        self.assertIsNone(state.verified_at)

    def test_failed_verification_is_not_reported_as_verified(self):
        applied = certificate("current", local_key="91")

        with SQLiteStateStore(":memory:") as store:
            store.record_applied(ENDPOINT, applied, T1)
            store.record_verification_success(
                ENDPOINT,
                applied,
                T2,
            )
            store.record_verification_failure(
                ENDPOINT,
                applied,
                "target no longer matches",
                T3,
            )
            state = store.load_applied_state(
                ENDPOINT,
                CERTIFICATES,
            )[0]

        self.assertFalse(state.is_verified)
        self.assertEqual(state.verified_at, T2)
        self.assertEqual(state.verification_attempted_at, T3)
        self.assertEqual(
            state.verification_error,
            "target no longer matches",
        )


class SchemaMigrationTests(unittest.TestCase):
    def test_version_one_database_is_migrated_to_version_three(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "state.sqlite3"
            connection = sqlite3.connect(path)
            connection.execute(
                "CREATE TABLE schema_meta (version INTEGER NOT NULL)"
            )
            connection.execute(
                "INSERT INTO schema_meta(version) VALUES (1)"
            )
            connection.commit()
            connection.close()

            with SQLiteStateStore(path) as store:
                version = store.connection.execute(
                    "SELECT version FROM schema_meta"
                ).fetchone()[0]
                tables = {
                    row[0]
                    for row in store.connection.execute(
                        """
                        SELECT name
                        FROM sqlite_master
                        WHERE type = 'table'
                        """
                    )
                }
                reference_columns = {
                    table: {
                        row[1]
                        for row in store.connection.execute(
                            f"PRAGMA table_info({table})"
                        )
                    }
                    for table in (
                        "observations",
                        "desired_state",
                        "applied_state",
                    )
                }

        self.assertEqual(version, 3)
        self.assertIn("desired_state", tables)
        self.assertIn("applied_state", tables)
        self.assertTrue(
            all(
                "references_json" in columns
                for columns in reference_columns.values()
            )
        )


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