dns-sync/tests/test_planning.py

import unittest
from dataclasses import dataclass
from datetime import datetime, timezone

from dns_sync.adapters import (
    CERTIFICATES,
    DNS_RECORDS,
    Capability,
    CapabilityOperation,
    Endpoint,
    NormalizedObject,
    Observation,
    PlanOperation,
)
from dns_sync.reconciliation import (
    PlanningConflict,
    build_endpoint_plans,
)


NOW = datetime(2026, 9, 19, 22, 30, tzinfo=timezone.utc)
FIOS = "verizon://192.168.1.1"
SECONDARY = "test-dns://192.168.1.2"


@dataclass
class FakeAdapter:
    endpoint: Endpoint
    capabilities: frozenset[Capability]


def adapter(
    plugin: str,
    address: str,
    *apply_types: str,
) -> FakeAdapter:
    return FakeAdapter(
        endpoint=Endpoint(plugin, address),
        capabilities=frozenset(
            Capability(
                object_type,
                CapabilityOperation.APPLY,
            )
            for object_type in apply_types
        ),
    )


def dns_object(
    hostname: str,
    ip: str,
    local_key: str | None = None,
) -> NormalizedObject:
    return NormalizedObject(
        object_type=DNS_RECORDS,
        logical_key=hostname,
        local_key=local_key,
        payload={
            "hostname": hostname,
            "ip": ip,
            "record_type": "A",
            "wildcard": False,
        },
    )


def observed(
    endpoint_id: str,
    object: NormalizedObject,
) -> Observation:
    return Observation(
        endpoint_id=endpoint_id,
        object=object,
        observed_at=NOW,
    )


class EndpointPlanningTests(unittest.TestCase):
    def test_builds_separate_capability_filtered_plans(self):
        desired = [
            dns_object("add.example.com", "192.168.1.101"),
            dns_object("keep.example.com", "192.168.1.101"),
            dns_object("update.example.com", "192.168.1.101"),
            NormalizedObject(
                object_type=CERTIFICATES,
                logical_key="app.example.com",
                payload={
                    "domain_names": ["app.example.com"],
                    "expires_on": "2026-12-15",
                },
            ),
        ]
        current = [
            observed(
                FIOS,
                dns_object(
                    "keep.example.com",
                    "192.168.1.101",
                    local_key="41",
                ),
            ),
            observed(
                FIOS,
                dns_object(
                    "update.example.com",
                    "192.168.1.50",
                    local_key="42",
                ),
            ),
            observed(
                FIOS,
                dns_object(
                    "unmanaged.example.com",
                    "192.168.1.77",
                    local_key="43",
                ),
            ),
        ]
        adapters = [
            adapter("verizon", "192.168.1.1", DNS_RECORDS),
            adapter("test-dns", "192.168.1.2", DNS_RECORDS),
            FakeAdapter(
                endpoint=Endpoint("npm", "192.168.1.3:81"),
                capabilities=frozenset(
                    {
                        Capability(
                            CERTIFICATES,
                            CapabilityOperation.DISCOVER,
                        )
                    }
                ),
            ),
        ]

        plans = build_endpoint_plans(
            adapters,
            desired,
            current,
        )

        self.assertEqual(
            tuple(plan.endpoint_id for plan in plans),
            (SECONDARY, FIOS),
        )

        secondary_actions = {
            action.desired.logical_key: action.operation
            for action in plans[0].actions
        }
        self.assertEqual(
            set(secondary_actions.values()),
            {PlanOperation.CREATE},
        )

        fios_actions = {
            action.desired.logical_key: action.operation
            for action in plans[1].actions
        }
        self.assertEqual(
            fios_actions,
            {
                "add.example.com": PlanOperation.CREATE,
                "keep.example.com": PlanOperation.NOOP,
                "update.example.com": PlanOperation.UPDATE,
            },
        )

        planned_keys = {
            action.desired.logical_key
            for plan in plans
            for action in plan.actions
        }
        self.assertNotIn("app.example.com", planned_keys)
        self.assertNotIn("unmanaged.example.com", planned_keys)

    def test_duplicate_endpoint_objects_fail_without_payload_output(self):
        desired = [
            dns_object("app.example.com", "192.168.1.101")
        ]
        current = [
            observed(
                FIOS,
                dns_object(
                    "app.example.com",
                    "192.168.1.50",
                    local_key="41",
                ),
            ),
            observed(
                FIOS,
                dns_object(
                    "app.example.com",
                    "203.0.113.77",
                    local_key="42",
                ),
            ),
        ]

        with self.assertRaises(PlanningConflict) as raised:
            build_endpoint_plans(
                [adapter("verizon", "192.168.1.1", DNS_RECORDS)],
                desired,
                current,
            )

        message = str(raised.exception)
        self.assertIn(FIOS, message)
        self.assertIn("app.example.com", message)
        self.assertNotIn("192.168.1.50", message)
        self.assertNotIn("203.0.113.77", message)

    def test_duplicate_desired_identity_is_rejected(self):
        duplicate = dns_object(
            "app.example.com",
            "192.168.1.101",
        )

        with self.assertRaisesRegex(
            RuntimeError,
            "Duplicate desired object identity",
        ):
            build_endpoint_plans(
                [adapter("verizon", "192.168.1.1", DNS_RECORDS)],
                [duplicate, duplicate],
                [],
            )


if __name__ == "__main__":
    unittest.main()