diff --git a/dns_sync/persistence.py b/dns_sync/persistence.py index 507405a..031ec95 100644 --- a/dns_sync/persistence.py +++ b/dns_sync/persistence.py @@ -6,7 +6,11 @@ from datetime import datetime, timezone from os import PathLike from typing import Any -from .adapters import NormalizedObject, Observation +from .adapters import ( + NormalizedObject, + ObjectReference, + Observation, +) _FORBIDDEN_PERSISTED_KEYS = frozenset( @@ -116,6 +120,71 @@ def _deserialize_payload(value: str) -> dict[str, Any]: return payload +def _serialize_references( + references: Iterable[ObjectReference], +) -> str: + return json.dumps( + [ + { + "relation": reference.relation, + "object_type": reference.object_type, + "local_key": reference.local_key, + "logical_key": reference.logical_key, + } + for reference in references + ], + sort_keys=True, + separators=(",", ":"), + ) + + +def _deserialize_references( + value: str, +) -> tuple[ObjectReference, ...]: + try: + serialized = json.loads(value) + except json.JSONDecodeError as exc: + raise RuntimeError( + "Persisted object references are invalid JSON" + ) from exc + + if not isinstance(serialized, list): + raise RuntimeError( + "Persisted object references are not a list" + ) + + expected_fields = { + "relation", + "object_type", + "local_key", + "logical_key", + } + references: list[ObjectReference] = [] + for item in serialized: + if ( + not isinstance(item, dict) + or set(item) != expected_fields + ): + raise RuntimeError( + "Persisted object reference has invalid fields" + ) + try: + references.append( + ObjectReference( + relation=item["relation"], + object_type=item["object_type"], + local_key=item["local_key"], + logical_key=item["logical_key"], + ) + ) + except (TypeError, ValueError) as exc: + raise RuntimeError( + "Persisted object reference is invalid" + ) from exc + + return tuple(references) + + class SQLiteStateStore: def __init__(self, path: str | PathLike[str]): self.path = str(path) @@ -185,7 +254,7 @@ class SQLiteStateStore: ) version = versions[0]["version"] - if version not in (1, 2): + if version not in (1, 2, 3): raise RuntimeError( "Unsupported SQLite state schema version: " f"{version!r}" @@ -232,6 +301,31 @@ class SQLiteStateStore: SET version = 2; """ ) + version = 2 + + if version == 2: + migrations = ( + ("observations", "references_json"), + ("desired_state", "references_json"), + ("applied_state", "references_json"), + ) + for table, column in migrations: + existing_columns = { + row["name"] + for row in self.connection.execute( + f"PRAGMA table_info({table})" + ) + } + if column not in existing_columns: + self.connection.execute( + f"ALTER TABLE {table} " + f"ADD COLUMN {column} TEXT NOT NULL " + "DEFAULT '[]'" + ) + + self.connection.execute( + "UPDATE schema_meta SET version = 3" + ) def close(self) -> None: self.connection.close() @@ -254,7 +348,9 @@ class SQLiteStateStore: snapshot_at, "Observation snapshot timestamp", ) - prepared: list[tuple[str, str, str, str, str, str]] = [] + prepared: list[ + tuple[str, str, str, str, str, str, str] + ] = [] local_keys: set[str] = set() for observation in rows: @@ -284,6 +380,7 @@ class SQLiteStateStore: normalized.logical_key, normalized.local_key, _serialize_payload(normalized.payload), + _serialize_references(normalized.references), _timestamp( observation.observed_at, "Observation timestamp", @@ -307,8 +404,9 @@ class SQLiteStateStore: logical_key, local_key, payload_json, + references_json, observed_at - ) VALUES (?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?) """, prepared, ) @@ -378,6 +476,7 @@ class SQLiteStateStore: logical_key, local_key, payload_json, + references_json, observed_at FROM observations """ @@ -402,6 +501,9 @@ class SQLiteStateStore: logical_key=row["logical_key"], local_key=row["local_key"], payload=_deserialize_payload(row["payload_json"]), + references=_deserialize_references( + row["references_json"] + ), ), observed_at=_parse_timestamp(row["observed_at"]), ) @@ -419,7 +521,9 @@ class SQLiteStateStore: selected_at, "Desired-state selection timestamp", ) - prepared: list[tuple[str, str, str, str]] = [] + prepared: list[ + tuple[str, str, str, str, str] + ] = [] logical_keys: set[str] = set() for normalized in rows: @@ -444,6 +548,7 @@ class SQLiteStateStore: object_type, normalized.logical_key, _serialize_payload(normalized.payload), + _serialize_references(normalized.references), selected_timestamp, ) ) @@ -462,8 +567,9 @@ class SQLiteStateStore: object_type, logical_key, payload_json, + references_json, selected_at - ) VALUES (?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?) """, prepared, ) @@ -477,6 +583,7 @@ class SQLiteStateStore: object_type, logical_key, payload_json, + references_json, selected_at FROM desired_state """ @@ -493,6 +600,9 @@ class SQLiteStateStore: object_type=row["object_type"], logical_key=row["logical_key"], payload=_deserialize_payload(row["payload_json"]), + references=_deserialize_references( + row["references_json"] + ), ), selected_at=_parse_timestamp(row["selected_at"]), ) @@ -509,6 +619,9 @@ class SQLiteStateStore: raise RuntimeError("Applied-state endpoint_id cannot be empty") payload_json = _serialize_payload(object.payload) + references_json = _serialize_references( + object.references + ) applied_timestamp = _timestamp( applied_at, "Application timestamp", @@ -523,11 +636,12 @@ class SQLiteStateStore: logical_key, local_key, payload_json, + references_json, applied_at, verification_attempted_at, verified_at, verification_error - ) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, NULL) + ) VALUES (?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL) ON CONFLICT( endpoint_id, object_type, @@ -535,6 +649,7 @@ class SQLiteStateStore: ) DO UPDATE SET local_key = excluded.local_key, payload_json = excluded.payload_json, + references_json = excluded.references_json, applied_at = excluded.applied_at, verification_attempted_at = NULL, verified_at = NULL, @@ -546,6 +661,7 @@ class SQLiteStateStore: object.logical_key, object.local_key, payload_json, + references_json, applied_timestamp, ), ) @@ -564,6 +680,9 @@ class SQLiteStateStore: "Verification timestamp", ) payload_json = _serialize_payload(object.payload) + references_json = _serialize_references( + object.references + ) if succeeded: verified_at = attempted_timestamp @@ -587,6 +706,7 @@ class SQLiteStateStore: AND object_type = ? AND logical_key = ? AND payload_json = ? + AND references_json = ? """, ( attempted_timestamp, @@ -596,6 +716,7 @@ class SQLiteStateStore: object.object_type, object.logical_key, payload_json, + references_json, ), ) if cursor.rowcount != 1: @@ -655,6 +776,7 @@ class SQLiteStateStore: logical_key, local_key, payload_json, + references_json, applied_at, verification_attempted_at, verified_at, @@ -682,6 +804,9 @@ class SQLiteStateStore: logical_key=row["logical_key"], local_key=row["local_key"], payload=_deserialize_payload(row["payload_json"]), + references=_deserialize_references( + row["references_json"] + ), ), applied_at=_parse_timestamp(row["applied_at"]), verification_attempted_at=_parse_timestamp( diff --git a/tests/test_reconciliation_persistence.py b/tests/test_reconciliation_persistence.py index 891f44d..fa85fea 100644 --- a/tests/test_reconciliation_persistence.py +++ b/tests/test_reconciliation_persistence.py @@ -203,7 +203,7 @@ class AppliedStatePersistenceTests(unittest.TestCase): class SchemaMigrationTests(unittest.TestCase): - def test_version_one_database_is_migrated_to_version_two(self): + 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) @@ -230,10 +230,29 @@ class SchemaMigrationTests(unittest.TestCase): """ ) } + 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, 2) + 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__": diff --git a/tests/test_reference_persistence.py b/tests/test_reference_persistence.py new file mode 100644 index 0000000..d643c60 --- /dev/null +++ b/tests/test_reference_persistence.py @@ -0,0 +1,142 @@ +import unittest +from datetime import datetime, timezone + +from dns_sync.adapters import ( + CERTIFICATES, + PROXY_HOSTS, + NormalizedObject, + ObjectReference, + Observation, +) +from dns_sync.persistence import SQLiteStateStore + + +ENDPOINT = "npm://192.168.1.4:81" +NOW = datetime(2026, 9, 20, 0, 0, tzinfo=timezone.utc) + + +def local_certificate_reference() -> ObjectReference: + return ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="67", + ) + + +def logical_certificate_reference() -> ObjectReference: + return ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ) + + +def proxy_host( + *, + local_key: str | None, + reference: ObjectReference, +) -> NormalizedObject: + return NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key=local_key, + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "192.168.1.20", + "forward_port": 8080, + "enabled": True, + }, + references=(reference,), + ) + + +class ReferencePersistenceTests(unittest.TestCase): + def test_observed_local_reference_round_trips(self): + observed = Observation( + endpoint_id=ENDPOINT, + object=proxy_host( + local_key="12", + reference=local_certificate_reference(), + ), + observed_at=NOW, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + ENDPOINT, + PROXY_HOSTS, + [observed], + NOW, + ) + loaded = store.load_observations(PROXY_HOSTS) + + self.assertEqual(loaded, [observed]) + self.assertEqual( + loaded[0].object.references[0].local_key, + "67", + ) + + def test_desired_logical_reference_round_trips(self): + desired = proxy_host( + local_key=None, + reference=logical_certificate_reference(), + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_desired_state( + PROXY_HOSTS, + [desired], + NOW, + ) + loaded = store.load_desired_state(PROXY_HOSTS) + + self.assertEqual(loaded[0].object, desired) + self.assertEqual( + loaded[0].object.references[0].logical_key, + "app.example.com", + ) + + def test_applied_reference_participates_in_verification(self): + applied = proxy_host( + local_key="22", + reference=logical_certificate_reference(), + ) + different_reference = proxy_host( + local_key="22", + reference=ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="other.example.com", + ), + ) + + with SQLiteStateStore(":memory:") as store: + store.record_applied(ENDPOINT, applied, NOW) + + with self.assertRaisesRegex( + RuntimeError, + "does not match", + ): + store.record_verification_success( + ENDPOINT, + different_reference, + NOW, + ) + + store.record_verification_success( + ENDPOINT, + applied, + NOW, + ) + loaded = store.load_applied_state( + ENDPOINT, + PROXY_HOSTS, + ) + + self.assertEqual(loaded[0].object, applied) + self.assertTrue(loaded[0].is_verified) + + +if __name__ == "__main__": + unittest.main()