dns-sync/tests/test_adapter_models.py

import unittest

from dns_sync.adapters import (
    CERTIFICATES,
    DNS_RECORDS,
    PROXY_HOSTS,
    ApplyResult,
    Capability,
    CapabilityOperation,
    Endpoint,
    EndpointPlan,
    NormalizedObject,
    ObjectReference,
    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 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(
            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,))