Add generic adapter capability model

2026-09-19 15:49:42 -04:00
parent 9d571ada24
commit e32260fc50
5 changed files with 275 additions and 10 deletions
@@ -0,0 +1,103 @@
+import unittest
+
+from dns_sync.adapters import (
+    CERTIFICATES,
+    DNS_RECORDS,
+    PROXY_HOSTS,
+    Capability,
+    CapabilityOperation,
+    Endpoint,
+    EndpointPlan,
+    NormalizedObject,
+    Observation,
+    PlanAction,
+    PlanOperation,
+)
+
+
+class EndpointTests(unittest.TestCase):
+    def test_identity_uses_plugin_and_address_not_label(self):
+        eve = Endpoint("NPM", "192.168.1.3:81/", label="EVE")
+        renamed = Endpoint("npm", "192.168.1.3:81", label="anything")
+
+        self.assertEqual(eve.identity, "npm://192.168.1.3:81")
+        self.assertEqual(eve.identity, renamed.identity)
+
+    def test_empty_endpoint_fields_are_rejected(self):
+        with self.assertRaises(ValueError):
+            Endpoint("", "192.168.1.3:81")
+
+        with self.assertRaises(ValueError):
+            Endpoint("npm", "")
+
+
+class CapabilityTests(unittest.TestCase):
+    def test_capabilities_are_object_type_specific(self):
+        capabilities = frozenset(
+            {
+                Capability(PROXY_HOSTS, CapabilityOperation.DISCOVER),
+                Capability(PROXY_HOSTS, CapabilityOperation.APPLY),
+                Capability(CERTIFICATES, CapabilityOperation.DISCOVER),
+                Capability(CERTIFICATES, CapabilityOperation.APPLY),
+                Capability(DNS_RECORDS, CapabilityOperation.DISCOVER),
+            }
+        )
+
+        self.assertIn(
+            Capability(CERTIFICATES, CapabilityOperation.APPLY),
+            capabilities,
+        )
+        self.assertNotIn(
+            Capability(DNS_RECORDS, CapabilityOperation.APPLY),
+            capabilities,
+        )
+
+
+class ObservationAndPlanTests(unittest.TestCase):
+    def test_local_objects_can_share_one_logical_identity(self):
+        first = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="chaptarr.loxthelion.com",
+            local_key="63",
+            payload={"expires_on": "2026-10-01T00:00:00+00:00"},
+        )
+        second = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="chaptarr.loxthelion.com",
+            local_key="67",
+            payload={"expires_on": "2026-11-01T00:00:00+00:00"},
+        )
+
+        observations = [
+            Observation("npm://192.168.1.3:81", first),
+            Observation("npm://192.168.1.3:81", second),
+        ]
+
+        self.assertEqual(
+            observations[0].object.logical_key,
+            observations[1].object.logical_key,
+        )
+        self.assertNotEqual(
+            observations[0].object.local_key,
+            observations[1].object.local_key,
+        )
+
+    def test_plan_is_scoped_to_one_endpoint(self):
+        desired = NormalizedObject(
+            object_type=DNS_RECORDS,
+            logical_key="example.loxthelion.com",
+            payload={
+                "hostname": "example.loxthelion.com",
+                "ip": "192.168.1.101",
+                "record_type": "A",
+            },
+        )
+        action = PlanAction(
+            endpoint_id="verizon://192.168.1.1",
+            operation=PlanOperation.CREATE,
+            desired=desired,
+        )
+        plan = EndpointPlan(action.endpoint_id, (action,))
+
+        self.assertEqual(plan.endpoint_id, "verizon://192.168.1.1")
+        self.assertEqual(plan.actions, (action,))