From 4b1326e150c297ad5ed7947dc7e120373f6cc04f Mon Sep 17 00:00:00 2001 From: Lyncoln Date: Sat, 19 Sep 2026 19:53:23 -0400 Subject: [PATCH] Add generic object reference models --- dns_sync/adapters.py | 70 ++++++++++++++++++++++++ tests/test_adapter_models.py | 100 +++++++++++++++++++++++++++++++++++ 2 files changed, 170 insertions(+) diff --git a/dns_sync/adapters.py b/dns_sync/adapters.py index 996c677..f3b9106 100644 --- a/dns_sync/adapters.py +++ b/dns_sync/adapters.py @@ -59,6 +59,57 @@ class Endpoint: return f"{self.plugin}://{self.address}" +@dataclass(frozen=True, order=True) +class ObjectReference: + relation: str + object_type: str + local_key: str | None = None + logical_key: str | None = None + + def __post_init__(self) -> None: + relation = self.relation.strip().lower() + object_type = self.object_type.strip().lower() + + if not relation: + raise ValueError("Object reference relation cannot be empty") + if not object_type: + raise ValueError( + "Object reference object_type cannot be empty" + ) + if (self.local_key is None) == (self.logical_key is None): + raise ValueError( + "Object reference requires exactly one of " + "local_key or logical_key" + ) + + object.__setattr__(self, "relation", relation) + object.__setattr__(self, "object_type", object_type) + + if self.local_key is not None: + if not isinstance(self.local_key, str): + raise ValueError( + "Object reference local_key must be a string" + ) + local_key = self.local_key.strip() + if not local_key: + raise ValueError( + "Object reference local_key cannot be empty" + ) + object.__setattr__(self, "local_key", local_key) + + if self.logical_key is not None: + if not isinstance(self.logical_key, str): + raise ValueError( + "Object reference logical_key must be a string" + ) + logical_key = self.logical_key.strip() + if not logical_key: + raise ValueError( + "Object reference logical_key cannot be empty" + ) + object.__setattr__(self, "logical_key", logical_key) + + @dataclass(frozen=True) class DiscoveredObject: object_type: str @@ -74,6 +125,25 @@ class NormalizedObject: local_key: str | None = None +@dataclass(frozen=True) +class ApplyResult: + object: NormalizedObject + detail: str | None = None + + def __post_init__(self) -> None: + if self.object.local_key is None: + raise ValueError( + "Applied object requires an endpoint-local key" + ) + if self.detail is not None: + detail = self.detail.strip() + object.__setattr__( + self, + "detail", + detail or None, + ) + + @dataclass(frozen=True) class Observation: endpoint_id: str diff --git a/tests/test_adapter_models.py b/tests/test_adapter_models.py index 61c11d1..e8cfccd 100644 --- a/tests/test_adapter_models.py +++ b/tests/test_adapter_models.py @@ -4,11 +4,13 @@ from dns_sync.adapters import ( CERTIFICATES, DNS_RECORDS, PROXY_HOSTS, + ApplyResult, Capability, CapabilityOperation, Endpoint, EndpointPlan, NormalizedObject, + ObjectReference, Observation, PlanAction, PlanOperation, @@ -53,6 +55,104 @@ class CapabilityTests(unittest.TestCase): ) +class ObjectReferenceTests(unittest.TestCase): + def test_endpoint_local_reference_is_app_agnostic(self): + reference = ObjectReference( + relation="Certificate", + object_type="Certificates", + local_key=" 67 ", + ) + + self.assertEqual(reference.relation, "certificate") + self.assertEqual(reference.object_type, CERTIFICATES) + self.assertEqual(reference.local_key, "67") + self.assertIsNone(reference.logical_key) + + def test_logical_reference_does_not_contain_local_id(self): + reference = ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + logical_key="chaptarr.loxthelion.com", + ) + + self.assertIsNone(reference.local_key) + self.assertEqual( + reference.logical_key, + "chaptarr.loxthelion.com", + ) + + def test_reference_requires_exactly_one_identity(self): + with self.assertRaisesRegex( + ValueError, + "exactly one", + ): + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + ) + + with self.assertRaisesRegex( + ValueError, + "exactly one", + ): + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key="67", + logical_key="chaptarr.loxthelion.com", + ) + + def test_reference_rejects_empty_fields(self): + with self.assertRaisesRegex(ValueError, "relation"): + ObjectReference( + relation="", + object_type=CERTIFICATES, + local_key="67", + ) + + with self.assertRaisesRegex(ValueError, "local_key"): + ObjectReference( + relation="certificate", + object_type=CERTIFICATES, + local_key=" ", + ) + + +class ApplyResultTests(unittest.TestCase): + def test_result_carries_adapter_assigned_local_key(self): + applied = NormalizedObject( + object_type=CERTIFICATES, + logical_key="chaptarr.loxthelion.com", + local_key="68", + payload={ + "domain_names": ["chaptarr.loxthelion.com"], + }, + ) + + result = ApplyResult( + object=applied, + detail=" created ", + ) + + self.assertEqual(result.object.local_key, "68") + self.assertEqual(result.detail, "created") + + def test_result_rejects_object_without_local_key(self): + desired = NormalizedObject( + object_type=CERTIFICATES, + logical_key="chaptarr.loxthelion.com", + payload={ + "domain_names": ["chaptarr.loxthelion.com"], + }, + ) + + with self.assertRaisesRegex( + ValueError, + "endpoint-local key", + ): + ApplyResult(object=desired) + + class ObservationAndPlanTests(unittest.TestCase): def test_local_objects_can_share_one_logical_identity(self): first = NormalizedObject(