From 8caac79a1e28ad3356dc9ce802a35d542b7413cf Mon Sep 17 00:00:00 2001 From: Lyncoln Date: Sat, 19 Sep 2026 21:34:23 -0400 Subject: [PATCH] Resolve endpoint-local object dependencies --- dns_sync/dependencies.py | 98 +++++ dns_sync/executor.py | 294 +++++++++++-- dns_sync/interfaces.py | 3 +- dns_sync/plugins/adapters/verizon.py | 46 +- dns_sync/reconciliation.py | 3 +- tests/test_executor.py | 627 ++++++++++++++++++++++++++- tests/test_planning.py | 116 +++++ tests/test_verizon_adapter.py | 29 +- 8 files changed, 1174 insertions(+), 42 deletions(-) create mode 100644 dns_sync/dependencies.py diff --git a/dns_sync/dependencies.py b/dns_sync/dependencies.py new file mode 100644 index 0000000..54ebbe5 --- /dev/null +++ b/dns_sync/dependencies.py @@ -0,0 +1,98 @@ +from collections.abc import Iterable + +from .adapters import PlanAction + + +def _action_identity( + action: PlanAction, +) -> tuple[str, str, str]: + desired = action.desired + if desired is None: + raise RuntimeError( + "Dependency-ordered plan action requires desired state" + ) + + return ( + action.endpoint_id, + desired.object_type, + desired.logical_key, + ) + + +def order_plan_actions( + actions: Iterable[PlanAction], +) -> tuple[PlanAction, ...]: + action_list = list(actions) + by_identity: dict[ + tuple[str, str, str], + PlanAction, + ] = {} + positions: dict[ + tuple[str, str, str], + int, + ] = {} + + for position, action in enumerate(action_list): + identity = _action_identity(action) + if identity in by_identity: + raise RuntimeError( + "Duplicate plan action identity: " + f"{identity[0]} {identity[1]} {identity[2]!r}" + ) + by_identity[identity] = action + positions[identity] = position + + dependencies: dict[ + tuple[str, str, str], + set[tuple[str, str, str]], + ] = { + identity: set() + for identity in by_identity + } + + for identity, action in by_identity.items(): + desired = action.desired + for reference in desired.references: + if reference.logical_key is None: + continue + + dependency = ( + action.endpoint_id, + reference.object_type, + reference.logical_key, + ) + if dependency in by_identity: + dependencies[identity].add(dependency) + + remaining = set(by_identity) + ordered: list[PlanAction] = [] + + while remaining: + ready = [ + identity + for identity in remaining + if not dependencies[identity].intersection(remaining) + ] + + if not ready: + cycle = ", ".join( + f"{identity[1]} {identity[2]!r}" + for identity in sorted( + remaining, + key=lambda item: positions[item], + ) + ) + raise RuntimeError( + "Plan action dependency cycle: " + f"{cycle}" + ) + + ready.sort( + key=lambda identity: positions[identity] + ) + + for identity in ready: + ordered.append(by_identity[identity]) + remaining.remove(identity) + + return tuple(ordered) diff --git a/dns_sync/executor.py b/dns_sync/executor.py index 3a7b2c8..35c50eb 100644 --- a/dns_sync/executor.py +++ b/dns_sync/executor.py @@ -1,35 +1,28 @@ from collections.abc import Iterable -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timezone from typing import Protocol from .adapters import ( - CERTIFICATES, - DNS_RECORDS, - PROXY_HOSTS, + ApplyResult, Capability, CapabilityOperation, Endpoint, EndpointPlan, NormalizedObject, + ObjectReference, PlanAction, PlanOperation, ) +from .dependencies import order_plan_actions from .persistence import SQLiteStateStore -_OBJECT_TYPE_ORDER = { - CERTIFICATES: 0, - PROXY_HOSTS: 1, - DNS_RECORDS: 2, -} - - class ExecutionAdapter(Protocol): endpoint: Endpoint capabilities: frozenset[Capability] - def apply(self, action: PlanAction) -> str | None: ... + def apply(self, action: PlanAction) -> ApplyResult | None: ... def verify(self, object: NormalizedObject) -> bool: ... @@ -48,19 +41,238 @@ class ExecutionResult: skipped: tuple[PlanAction, ...] -def _action_sort_key( +@dataclass +class _LocalReferenceIndex: + by_logical: dict[ + tuple[str, str], + set[str], + ] + by_local: set[tuple[str, str]] + + +def _remember_local_object( + index: _LocalReferenceIndex, + object: NormalizedObject, + *, + authoritative: bool = False, +) -> None: + local_key = object.local_key + if local_key is None: + return + + identity = ( + object.object_type, + object.logical_key, + ) + + if authoritative: + previous = index.by_logical.get(identity, set()) + for previous_local_key in previous: + index.by_local.discard( + ( + object.object_type, + previous_local_key, + ) + ) + index.by_logical[identity] = {local_key} + else: + index.by_logical.setdefault( + identity, + set(), + ).add(local_key) + + index.by_local.add( + ( + object.object_type, + local_key, + ) + ) + + +def _local_reference_index( + store: SQLiteStateStore, + endpoint_id: str, +) -> _LocalReferenceIndex: + index = _LocalReferenceIndex( + by_logical={}, + by_local=set(), + ) + observed_identities: set[ + tuple[str, str] + ] = set() + + for observation in store.load_observations(): + if observation.endpoint_id != endpoint_id: + continue + + object = observation.object + if object.local_key is None: + continue + + identity = ( + object.object_type, + object.logical_key, + ) + observed_identities.add(identity) + _remember_local_object(index, object) + + for applied in store.load_applied_state( + endpoint_id=endpoint_id, + ): + if not applied.is_verified: + continue + + object = applied.object + if object.local_key is None: + continue + + identity = ( + object.object_type, + object.logical_key, + ) + + # Endpoint observations are authoritative when present. + # Verified applied state is only a fallback when this + # logical identity is absent from the observation snapshot. + if identity in observed_identities: + continue + + _remember_local_object(index, object) + + return index + + +def _resolved_reference( + endpoint_id: str, + reference: ObjectReference, + local_references: _LocalReferenceIndex, +) -> ObjectReference: + if reference.local_key is not None: + local_identity = ( + reference.object_type, + reference.local_key, + ) + + if local_identity not in local_references.by_local: + raise RuntimeError( + "Unresolved endpoint-local reference for " + f"{endpoint_id}: " + f"{reference.relation} -> " + f"{reference.object_type} " + f"local key {reference.local_key!r}" + ) + + return reference + + identity = ( + reference.object_type, + reference.logical_key, + ) + candidates = local_references.by_logical.get( + identity, + set(), + ) + + if not candidates: + raise RuntimeError( + "Unresolved endpoint-local reference for " + f"{endpoint_id}: " + f"{reference.relation} -> " + f"{reference.object_type} " + f"{reference.logical_key!r}" + ) + + if len(candidates) > 1: + raise RuntimeError( + "Ambiguous endpoint-local reference for " + f"{endpoint_id}: " + f"{reference.relation} -> " + f"{reference.object_type} " + f"{reference.logical_key!r}: " + f"{', '.join(sorted(candidates))}" + ) + + local_key = next(iter(candidates)) + + return ObjectReference( + relation=reference.relation, + object_type=reference.object_type, + local_key=local_key, + ) + + +def _resolve_action_references( action: PlanAction, -) -> tuple[int, str, str]: + local_references: _LocalReferenceIndex, +) -> PlanAction: desired = action.desired if desired is None: - return (999, action.endpoint_id, "") + raise RuntimeError( + "Executable plan action requires desired state" + ) - return ( - _OBJECT_TYPE_ORDER.get(desired.object_type, 100), - action.endpoint_id, - desired.logical_key, + if not desired.references: + return action + + references = tuple( + _resolved_reference( + action.endpoint_id, + reference, + local_references, + ) + for reference in desired.references ) + resolved_desired = replace( + desired, + references=references, + ) + + return replace( + action, + desired=resolved_desired, + ) + + +def _validated_apply_result( + action: PlanAction, + result: ApplyResult | None, +) -> ApplyResult: + desired = action.desired + if desired is None: + raise RuntimeError( + "Executable plan action requires desired state" + ) + + if not isinstance(result, ApplyResult): + raise RuntimeError( + f"{action.endpoint_id} " + f"{desired.object_type} " + f"{desired.logical_key}: " + "adapter apply did not return ApplyResult" + ) + + applied = result.object + + if applied.object_type != desired.object_type: + raise RuntimeError( + "Applied object type does not match desired object type" + ) + + if applied.logical_key != desired.logical_key: + raise RuntimeError( + "Applied object logical key does not match desired " + "logical key" + ) + + if applied.references != desired.references: + raise RuntimeError( + "Applied object references do not match resolved " + "desired references" + ) + + return result + def _required_capability( adapter: ExecutionAdapter, @@ -164,25 +376,46 @@ def execute_endpoint_plans( plans, ) applied: list[ExecutedAction] = [] + local_references: dict[ + str, + _LocalReferenceIndex, + ] = {} - for action in sorted(executable, key=_action_sort_key): + for endpoint_id in adapters_by_endpoint: + local_references[endpoint_id] = ( + _local_reference_index( + store, + endpoint_id, + ) + ) + + for action in order_plan_actions(executable): adapter = adapters_by_endpoint[action.endpoint_id] - desired = action.desired + resolved_action = _resolve_action_references( + action, + local_references[action.endpoint_id], + ) + desired = resolved_action.desired + + result = _validated_apply_result( + resolved_action, + adapter.apply(resolved_action), + ) + applied_object = result.object - adapter.apply(action) store.record_applied( action.endpoint_id, - desired, + applied_object, timestamp, ) try: - verified = adapter.verify(desired) + verified = adapter.verify(applied_object) except Exception as exc: error = str(exc) store.record_verification_failure( action.endpoint_id, - desired, + applied_object, error, timestamp, ) @@ -196,7 +429,7 @@ def execute_endpoint_plans( error = "adapter returned false" store.record_verification_failure( action.endpoint_id, - desired, + applied_object, error, timestamp, ) @@ -208,9 +441,16 @@ def execute_endpoint_plans( store.record_verification_success( action.endpoint_id, - desired, + applied_object, timestamp, ) + + _remember_local_object( + local_references[action.endpoint_id], + applied_object, + authoritative=True, + ) + applied.append( ExecutedAction( endpoint_id=action.endpoint_id, diff --git a/dns_sync/interfaces.py b/dns_sync/interfaces.py index 001c350..eb08ea2 100644 --- a/dns_sync/interfaces.py +++ b/dns_sync/interfaces.py @@ -1,6 +1,7 @@ from typing import Protocol from .adapters import ( + ApplyResult, Capability, DiscoveredObject, Endpoint, @@ -17,7 +18,7 @@ class AdapterPlugin(Protocol): def connect(self) -> None: ... def discover(self, object_type: str) -> list[DiscoveredObject]: ... - def apply(self, action: PlanAction) -> str | None: ... + def apply(self, action: PlanAction) -> ApplyResult | None: ... def verify(self, object: NormalizedObject) -> bool: ... diff --git a/dns_sync/plugins/adapters/verizon.py b/dns_sync/plugins/adapters/verizon.py index 24e6613..25ebcf4 100644 --- a/dns_sync/plugins/adapters/verizon.py +++ b/dns_sync/plugins/adapters/verizon.py @@ -1,5 +1,6 @@ from ...adapters import ( DNS_RECORDS, + ApplyResult, Capability, CapabilityOperation, DiscoveredObject, @@ -119,7 +120,7 @@ class Plugin: wildcard=False, ) - def apply(self, action: PlanAction) -> str | None: + def apply(self, action: PlanAction) -> ApplyResult | None: if action.endpoint_id != self.endpoint.identity: raise RuntimeError( "Plan action belongs to a different endpoint" @@ -139,13 +140,44 @@ class Plugin: record = self._desired_record(action.desired) if action.operation == PlanOperation.CREATE: - return self.provider.add_record(record) - if action.operation == PlanOperation.UPDATE: - return self.provider.update_record(record) + detail = self.provider.add_record(record) + elif action.operation == PlanOperation.UPDATE: + detail = self.provider.update_record(record) + else: + raise RuntimeError( + f"Verizon adapter does not support " + f"{action.operation.value} actions" + ) - raise RuntimeError( - f"Verizon adapter does not support " - f"{action.operation.value} actions" + records = self.provider.list_records() + applied_record = records.get(record.hostname.lower()) + if applied_record is None: + applied_record = records.get(record.hostname) + + if applied_record is None: + raise RuntimeError( + "Verizon mutation succeeded but the applied DNS " + f"record {record.hostname!r} could not be rediscovered" + ) + + slot = applied_record.metadata.get("slot") + if isinstance(slot, bool) or not isinstance(slot, int): + raise RuntimeError( + f"Verizon DNS record {record.hostname!r} " + "has no valid endpoint-local slot after apply" + ) + + applied = NormalizedObject( + object_type=action.desired.object_type, + logical_key=action.desired.logical_key, + local_key=str(slot), + payload=dict(action.desired.payload), + references=action.desired.references, + ) + + return ApplyResult( + object=applied, + detail=detail, ) def verify(self, object: NormalizedObject) -> bool: diff --git a/dns_sync/reconciliation.py b/dns_sync/reconciliation.py index d4e1c5f..4c6bfec 100644 --- a/dns_sync/reconciliation.py +++ b/dns_sync/reconciliation.py @@ -13,6 +13,7 @@ from .adapters import ( PlanAction, PlanOperation, ) +from .dependencies import order_plan_actions ObservationGroupKey = tuple[str, str] @@ -255,7 +256,7 @@ def build_endpoint_plans( plans.append( EndpointPlan( endpoint_id=endpoint_id, - actions=tuple(actions), + actions=order_plan_actions(actions), ) ) diff --git a/tests/test_executor.py b/tests/test_executor.py index 9900f8c..ed9d210 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -4,6 +4,7 @@ from datetime import datetime, timezone from dns_sync.adapters import ( CERTIFICATES, + ApplyResult, DNS_RECORDS, PROXY_HOSTS, Capability, @@ -11,6 +12,8 @@ from dns_sync.adapters import ( Endpoint, EndpointPlan, NormalizedObject, + ObjectReference, + Observation, PlanAction, PlanOperation, ) @@ -24,6 +27,8 @@ NOW = datetime(2026, 9, 19, 23, 15, tzinfo=timezone.utc) def object_for( object_type: str, logical_key: str, + *, + references=(), ) -> NormalizedObject: if object_type == DNS_RECORDS: payload = { @@ -50,6 +55,7 @@ def object_for( object_type=object_type, logical_key=logical_key, payload=payload, + references=references, ) @@ -58,11 +64,17 @@ def action( operation: PlanOperation, object_type: str = DNS_RECORDS, logical_key: str = "app.example.com", + *, + references=(), ) -> PlanAction: return PlanAction( endpoint_id=endpoint_id, operation=operation, - desired=object_for(object_type, logical_key), + desired=object_for( + object_type, + logical_key, + references=references, + ), ) @@ -95,6 +107,13 @@ class FakeAdapter: verify_result: bool = True apply_error: Exception | None = None verify_error: Exception | None = None + apply_local_keys: dict[ + tuple[str, str], + str, + ] = field(default_factory=dict) + applied_actions: list[PlanAction] = field( + default_factory=list + ) events: list[tuple[str, str, str]] = field( default_factory=list ) @@ -110,7 +129,27 @@ class FakeAdapter: ) if self.apply_error is not None: raise self.apply_error - return "test detail" + + self.applied_actions.append(plan_action) + + local_key = self.apply_local_keys.get( + ( + desired.object_type, + desired.logical_key, + ), + desired.local_key or "test-local", + ) + + return ApplyResult( + object=NormalizedObject( + object_type=desired.object_type, + logical_key=desired.logical_key, + local_key=local_key, + payload=dict(desired.payload), + references=desired.references, + ), + detail="test detail", + ) def verify(self, object): self.events.append( @@ -287,6 +326,13 @@ class ExecutionTests(unittest.TestCase): PlanOperation.CREATE, PROXY_HOSTS, "app.example.com", + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), ), action( endpoint.identity, @@ -315,6 +361,583 @@ class ExecutionTests(unittest.TestCase): ), ) + def test_existing_reference_resolves_from_observation(self): + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(PROXY_HOSTS), + ) + reference = ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ) + proxy_action = action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "app.example.com", + references=(reference,), + ) + + observed_certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + "expires_on": "2026-12-15T00:00:00+00:00", + }, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + endpoint.identity, + CERTIFICATES, + [ + Observation( + endpoint.identity, + observed_certificate, + NOW, + ), + ], + NOW, + ) + + execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + (proxy_action,), + ) + ], + store, + executed_at=NOW, + ) + + applied_reference = ( + adapter.applied_actions[0] + .desired + .references[0] + ) + self.assertEqual(applied_reference.local_key, "67") + self.assertIsNone(applied_reference.logical_key) + + + def test_created_reference_uses_apply_result_local_key(self): + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(CERTIFICATES, PROXY_HOSTS), + apply_local_keys={ + ( + CERTIFICATES, + "app.example.com", + ): "68", + ( + PROXY_HOSTS, + "app.example.com", + ): "12", + }, + ) + + certificate_action = action( + endpoint.identity, + PlanOperation.CREATE, + CERTIFICATES, + "app.example.com", + ) + proxy_action = action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "app.example.com", + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + ( + proxy_action, + certificate_action, + ), + ) + ], + store, + executed_at=NOW, + ) + persisted = store.load_applied_state( + endpoint.identity, + ) + + self.assertEqual( + tuple( + item.desired.object_type + for item in adapter.applied_actions + ), + (CERTIFICATES, PROXY_HOSTS), + ) + + resolved = ( + adapter.applied_actions[1] + .desired + .references[0] + ) + self.assertEqual(resolved.local_key, "68") + self.assertIsNone(resolved.logical_key) + + persisted_by_type = { + row.object.object_type: row.object + for row in persisted + } + self.assertEqual( + persisted_by_type[CERTIFICATES].local_key, + "68", + ) + self.assertEqual( + persisted_by_type[PROXY_HOSTS] + .references[0] + .local_key, + "68", + ) + + + def test_unresolved_reference_fails_before_dependent_apply(self): + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(PROXY_HOSTS), + ) + proxy_action = action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "app.example.com", + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="missing.example.com", + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "Unresolved endpoint-local reference", + ): + execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + (proxy_action,), + ) + ], + store, + executed_at=NOW, + ) + + self.assertEqual( + store.load_applied_state(), + [], + ) + + self.assertEqual(adapter.events, []) + self.assertEqual(adapter.applied_actions, []) + + + def test_existing_reference_falls_back_to_verified_applied_state(self): + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(PROXY_HOSTS), + ) + + certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="71", + payload={ + "domain_names": ["app.example.com"], + }, + ) + + proxy_action = action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "app.example.com", + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + store.record_applied( + endpoint.identity, + certificate, + NOW, + ) + store.record_verification_success( + endpoint.identity, + certificate, + NOW, + ) + + execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + (proxy_action,), + ) + ], + store, + executed_at=NOW, + ) + + resolved = ( + adapter.applied_actions[0] + .desired + .references[0] + ) + self.assertEqual(resolved.local_key, "71") + + + def test_ambiguous_logical_reference_fails_closed(self): + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(PROXY_HOSTS), + ) + + proxy_action = action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "app.example.com", + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + first = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="63", + payload={ + "domain_names": ["app.example.com"], + }, + ) + second = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + }, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + endpoint.identity, + CERTIFICATES, + [ + Observation( + endpoint.identity, + first, + NOW, + ), + Observation( + endpoint.identity, + second, + NOW, + ), + ], + NOW, + ) + + with self.assertRaisesRegex( + RuntimeError, + "Ambiguous endpoint-local reference", + ): + execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + (proxy_action,), + ) + ], + store, + executed_at=NOW, + ) + + self.assertEqual(adapter.events, []) + self.assertEqual(adapter.applied_actions, []) + + + def test_unrelated_duplicate_identity_does_not_block_apply(self): + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(PROXY_HOSTS), + ) + + proxy_action = action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "unrelated.example.com", + ) + + first = NormalizedObject( + object_type=CERTIFICATES, + logical_key="duplicate.example.com", + local_key="63", + payload={ + "domain_names": ["duplicate.example.com"], + }, + ) + second = NormalizedObject( + object_type=CERTIFICATES, + logical_key="duplicate.example.com", + local_key="67", + payload={ + "domain_names": ["duplicate.example.com"], + }, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + endpoint.identity, + CERTIFICATES, + [ + Observation( + endpoint.identity, + first, + NOW, + ), + Observation( + endpoint.identity, + second, + NOW, + ), + ], + NOW, + ) + + result = execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + (proxy_action,), + ) + ], + store, + executed_at=NOW, + ) + + self.assertEqual(len(result.applied), 1) + self.assertEqual(len(adapter.applied_actions), 1) + + + def test_unknown_local_reference_is_not_trusted(self): + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(PROXY_HOSTS), + ) + + proxy_action = action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "app.example.com", + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="999", + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "local key '999'", + ): + execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + (proxy_action,), + ) + ], + store, + executed_at=NOW, + ) + + self.assertEqual(adapter.events, []) + self.assertEqual(adapter.applied_actions, []) + + + def test_known_local_reference_is_preserved(self): + endpoint = Endpoint("npm", "192.168.1.4:81") + adapter = FakeAdapter( + endpoint, + capabilities(PROXY_HOSTS), + ) + + certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={ + "domain_names": ["app.example.com"], + }, + ) + local_reference = ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="67", + ) + proxy_action = action( + endpoint.identity, + PlanOperation.CREATE, + PROXY_HOSTS, + "app.example.com", + references=(local_reference,), + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + endpoint.identity, + CERTIFICATES, + [ + Observation( + endpoint.identity, + certificate, + NOW, + ), + ], + NOW, + ) + + execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + (proxy_action,), + ) + ], + store, + executed_at=NOW, + ) + + self.assertEqual( + adapter.applied_actions[0].desired.references, + (local_reference,), + ) + + + def test_dependency_cycle_fails_before_any_mutation(self): + endpoint = Endpoint("generic", "node.example:443") + object_a = "object_a" + object_b = "object_b" + + adapter = FakeAdapter( + endpoint, + capabilities(object_a, object_b), + ) + + action_a = PlanAction( + endpoint_id=endpoint.identity, + operation=PlanOperation.CREATE, + desired=NormalizedObject( + object_type=object_a, + logical_key="a", + payload={}, + references=( + ObjectReference( + relation="b", + object_type=object_b, + logical_key="b", + ), + ), + ), + ) + action_b = PlanAction( + endpoint_id=endpoint.identity, + operation=PlanOperation.CREATE, + desired=NormalizedObject( + object_type=object_b, + logical_key="b", + payload={}, + references=( + ObjectReference( + relation="a", + object_type=object_a, + logical_key="a", + ), + ), + ), + ) + + with SQLiteStateStore(":memory:") as store: + with self.assertRaisesRegex( + RuntimeError, + "dependency cycle", + ): + execute_endpoint_plans( + [adapter], + [ + EndpointPlan( + endpoint.identity, + ( + action_a, + action_b, + ), + ) + ], + store, + executed_at=NOW, + ) + + self.assertEqual( + store.load_applied_state(), + [], + ) + + self.assertEqual(adapter.events, []) + self.assertEqual(adapter.applied_actions, []) + + def test_invalid_plan_fails_preflight_without_mutation(self): endpoint = Endpoint("verizon", "192.168.1.1:443") adapter = FakeAdapter( diff --git a/tests/test_planning.py b/tests/test_planning.py index 3e0c43c..e92f68a 100644 --- a/tests/test_planning.py +++ b/tests/test_planning.py @@ -5,10 +5,12 @@ from datetime import datetime, timezone from dns_sync.adapters import ( CERTIFICATES, DNS_RECORDS, + PROXY_HOSTS, Capability, CapabilityOperation, Endpoint, NormalizedObject, + ObjectReference, Observation, PlanOperation, ) @@ -173,6 +175,120 @@ class EndpointPlanningTests(unittest.TestCase): self.assertNotIn("app.example.com", planned_keys) self.assertNotIn("unmanaged.example.com", planned_keys) + def test_plan_orders_referenced_object_before_dependent(self): + endpoint = Endpoint("generic", "node.example:443") + generic_adapter = FakeAdapter( + endpoint=endpoint, + capabilities=frozenset( + { + Capability( + CERTIFICATES, + CapabilityOperation.APPLY, + ), + Capability( + PROXY_HOSTS, + CapabilityOperation.APPLY, + ), + } + ), + ) + + certificate = NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + }, + ) + proxy_host = NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "192.168.1.50", + "forward_port": 8080, + "enabled": True, + }, + references=( + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="app.example.com", + ), + ), + ) + + plans = build_endpoint_plans( + [generic_adapter], + [ + proxy_host, + certificate, + ], + [], + ) + + self.assertEqual(len(plans), 1) + self.assertEqual( + tuple( + action.desired.object_type + for action in plans[0].actions + ), + ( + CERTIFICATES, + PROXY_HOSTS, + ), + ) + + + def test_plan_dependency_cycle_fails_before_plan_is_returned(self): + endpoint = Endpoint("generic", "node.example:443") + object_a = "object_a" + object_b = "object_b" + + generic_adapter = adapter( + "generic", + "node.example:443", + object_a, + object_b, + ) + + first = NormalizedObject( + object_type=object_a, + logical_key="a", + payload={}, + references=( + ObjectReference( + relation="b", + object_type=object_b, + logical_key="b", + ), + ), + ) + second = NormalizedObject( + object_type=object_b, + logical_key="b", + payload={}, + references=( + ObjectReference( + relation="a", + object_type=object_a, + logical_key="a", + ), + ), + ) + + with self.assertRaisesRegex( + RuntimeError, + "dependency cycle", + ): + build_endpoint_plans( + [generic_adapter], + [first, second], + [], + ) + + def test_duplicate_endpoint_objects_fail_without_payload_output(self): desired = [ dns_object("app.example.com", "192.168.1.101") diff --git a/tests/test_verizon_adapter.py b/tests/test_verizon_adapter.py index 3ee9869..eec94c7 100644 --- a/tests/test_verizon_adapter.py +++ b/tests/test_verizon_adapter.py @@ -147,8 +147,15 @@ class VerizonAdapterTests(unittest.TestCase): def test_create_and_update_delegate_without_backend_ip(self): provider = Mock() - provider.add_record.return_value = "slot 41" - provider.update_record.return_value = "slot 42" + provider.add_record.return_value = "created" + provider.update_record.return_value = "updated" + provider.list_records.return_value = { + "app.example.com": ProviderRecord( + hostname="app.example.com", + ip="192.168.1.101", + metadata={"slot": 41}, + ), + } adapter = self.adapter(provider) desired = self.dns_object(ip="192.168.1.101") @@ -167,8 +174,22 @@ class VerizonAdapterTests(unittest.TestCase): ) ) - self.assertEqual(create_result, "slot 41") - self.assertEqual(update_result, "slot 42") + self.assertEqual( + create_result.object.local_key, + "41", + ) + self.assertEqual( + create_result.detail, + "created", + ) + self.assertEqual( + update_result.object.local_key, + "41", + ) + self.assertEqual( + update_result.detail, + "updated", + ) self.assertEqual( provider.add_record.call_args.args[0].ip, "192.168.1.101",