diff --git a/dns_sync/reconciliation.py b/dns_sync/reconciliation.py index 42c2973..d4e1c5f 100644 --- a/dns_sync/reconciliation.py +++ b/dns_sync/reconciliation.py @@ -3,9 +3,15 @@ from collections.abc import Iterable, Mapping, Sequence from typing import Protocol from .adapters import ( + Capability, + CapabilityOperation, DNS_RECORDS, + Endpoint, + EndpointPlan, NormalizedObject, Observation, + PlanAction, + PlanOperation, ) @@ -16,6 +22,15 @@ class ReconciliationConflict(RuntimeError): """Raised when observations cannot be reconciled safely.""" +class PlanningConflict(RuntimeError): + """Raised when an endpoint cannot be planned safely.""" + + +class PlanningAdapter(Protocol): + endpoint: Endpoint + capabilities: frozenset[Capability] + + class ReconciliationPolicy(Protocol): object_type: str @@ -134,3 +149,114 @@ def reconcile_observations( desired.append(policy.reconcile(logical_key, rows)) return desired + + +def build_endpoint_plans( + adapters: Iterable[PlanningAdapter], + desired_objects: Iterable[NormalizedObject], + observations: Iterable[Observation], +) -> tuple[EndpointPlan, ...]: + desired_by_identity: dict[ + tuple[str, str], + NormalizedObject, + ] = {} + + for desired in desired_objects: + if desired.local_key is not None: + raise RuntimeError( + "Desired objects in plans cannot contain " + "an endpoint-local key" + ) + + identity = ( + desired.object_type, + desired.logical_key, + ) + if identity in desired_by_identity: + raise RuntimeError( + "Duplicate desired object identity: " + f"{desired.object_type} {desired.logical_key}" + ) + desired_by_identity[identity] = desired + + current_by_identity: dict[ + tuple[str, str, str], + list[Observation], + ] = defaultdict(list) + for observation in observations: + current_by_identity[ + ( + observation.endpoint_id, + observation.object.object_type, + observation.object.logical_key, + ) + ].append(observation) + + adapters_by_endpoint: dict[str, PlanningAdapter] = {} + for adapter in adapters: + endpoint_id = adapter.endpoint.identity + if endpoint_id in adapters_by_endpoint: + raise RuntimeError( + f"Duplicate adapter endpoint: {endpoint_id}" + ) + adapters_by_endpoint[endpoint_id] = adapter + + plans: list[EndpointPlan] = [] + for endpoint_id, adapter in sorted( + adapters_by_endpoint.items() + ): + apply_types = { + capability.object_type + for capability in adapter.capabilities + if capability.operation == CapabilityOperation.APPLY + } + if not apply_types: + continue + + actions: list[PlanAction] = [] + for identity, desired in sorted( + desired_by_identity.items() + ): + object_type, logical_key = identity + if object_type not in apply_types: + continue + + current_rows = current_by_identity.get( + (endpoint_id, object_type, logical_key), + [], + ) + if len(current_rows) > 1: + raise PlanningConflict( + f"Multiple endpoint objects for {endpoint_id} " + f"{object_type} {logical_key!r}" + ) + + current = ( + current_rows[0].object + if current_rows + else None + ) + if current is None: + operation = PlanOperation.CREATE + elif current.payload == desired.payload: + operation = PlanOperation.NOOP + else: + operation = PlanOperation.UPDATE + + actions.append( + PlanAction( + endpoint_id=endpoint_id, + operation=operation, + desired=desired, + current=current, + ) + ) + + plans.append( + EndpointPlan( + endpoint_id=endpoint_id, + actions=tuple(actions), + ) + ) + + return tuple(plans) diff --git a/tests/test_planning.py b/tests/test_planning.py new file mode 100644 index 0000000..3e0c43c --- /dev/null +++ b/tests/test_planning.py @@ -0,0 +1,230 @@ +import unittest +from dataclasses import dataclass +from datetime import datetime, timezone + +from dns_sync.adapters import ( + CERTIFICATES, + DNS_RECORDS, + Capability, + CapabilityOperation, + Endpoint, + NormalizedObject, + Observation, + PlanOperation, +) +from dns_sync.reconciliation import ( + PlanningConflict, + build_endpoint_plans, +) + + +NOW = datetime(2026, 9, 19, 22, 30, tzinfo=timezone.utc) +FIOS = "verizon://192.168.1.1" +SECONDARY = "test-dns://192.168.1.2" + + +@dataclass +class FakeAdapter: + endpoint: Endpoint + capabilities: frozenset[Capability] + + +def adapter( + plugin: str, + address: str, + *apply_types: str, +) -> FakeAdapter: + return FakeAdapter( + endpoint=Endpoint(plugin, address), + capabilities=frozenset( + Capability( + object_type, + CapabilityOperation.APPLY, + ) + for object_type in apply_types + ), + ) + + +def dns_object( + hostname: str, + ip: str, + local_key: str | None = None, +) -> NormalizedObject: + return NormalizedObject( + object_type=DNS_RECORDS, + logical_key=hostname, + local_key=local_key, + payload={ + "hostname": hostname, + "ip": ip, + "record_type": "A", + "wildcard": False, + }, + ) + + +def observed( + endpoint_id: str, + object: NormalizedObject, +) -> Observation: + return Observation( + endpoint_id=endpoint_id, + object=object, + observed_at=NOW, + ) + + +class EndpointPlanningTests(unittest.TestCase): + def test_builds_separate_capability_filtered_plans(self): + desired = [ + dns_object("add.example.com", "192.168.1.101"), + dns_object("keep.example.com", "192.168.1.101"), + dns_object("update.example.com", "192.168.1.101"), + NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "expires_on": "2026-12-15", + }, + ), + ] + current = [ + observed( + FIOS, + dns_object( + "keep.example.com", + "192.168.1.101", + local_key="41", + ), + ), + observed( + FIOS, + dns_object( + "update.example.com", + "192.168.1.50", + local_key="42", + ), + ), + observed( + FIOS, + dns_object( + "unmanaged.example.com", + "192.168.1.77", + local_key="43", + ), + ), + ] + adapters = [ + adapter("verizon", "192.168.1.1", DNS_RECORDS), + adapter("test-dns", "192.168.1.2", DNS_RECORDS), + FakeAdapter( + endpoint=Endpoint("npm", "192.168.1.3:81"), + capabilities=frozenset( + { + Capability( + CERTIFICATES, + CapabilityOperation.DISCOVER, + ) + } + ), + ), + ] + + plans = build_endpoint_plans( + adapters, + desired, + current, + ) + + self.assertEqual( + tuple(plan.endpoint_id for plan in plans), + (SECONDARY, FIOS), + ) + + secondary_actions = { + action.desired.logical_key: action.operation + for action in plans[0].actions + } + self.assertEqual( + set(secondary_actions.values()), + {PlanOperation.CREATE}, + ) + + fios_actions = { + action.desired.logical_key: action.operation + for action in plans[1].actions + } + self.assertEqual( + fios_actions, + { + "add.example.com": PlanOperation.CREATE, + "keep.example.com": PlanOperation.NOOP, + "update.example.com": PlanOperation.UPDATE, + }, + ) + + planned_keys = { + action.desired.logical_key + for plan in plans + for action in plan.actions + } + self.assertNotIn("app.example.com", planned_keys) + self.assertNotIn("unmanaged.example.com", planned_keys) + + def test_duplicate_endpoint_objects_fail_without_payload_output(self): + desired = [ + dns_object("app.example.com", "192.168.1.101") + ] + current = [ + observed( + FIOS, + dns_object( + "app.example.com", + "192.168.1.50", + local_key="41", + ), + ), + observed( + FIOS, + dns_object( + "app.example.com", + "203.0.113.77", + local_key="42", + ), + ), + ] + + with self.assertRaises(PlanningConflict) as raised: + build_endpoint_plans( + [adapter("verizon", "192.168.1.1", DNS_RECORDS)], + desired, + current, + ) + + message = str(raised.exception) + self.assertIn(FIOS, message) + self.assertIn("app.example.com", message) + self.assertNotIn("192.168.1.50", message) + self.assertNotIn("203.0.113.77", message) + + def test_duplicate_desired_identity_is_rejected(self): + duplicate = dns_object( + "app.example.com", + "192.168.1.101", + ) + + with self.assertRaisesRegex( + RuntimeError, + "Duplicate desired object identity", + ): + build_endpoint_plans( + [adapter("verizon", "192.168.1.1", DNS_RECORDS)], + [duplicate, duplicate], + [], + ) + + +if __name__ == "__main__": + unittest.main()