Add generic object reference models

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

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(