dns-sync/tests/test_orchestrator_npm_desired.py

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

from dns_sync.adapters import (
    CERTIFICATES,
    DNS_RECORDS,
    PROXY_HOSTS,
    Capability,
    CapabilityOperation,
    DiscoveredObject,
    Endpoint,
    ObjectReference,
)
from dns_sync.orchestrator import discover_reconcile_plan
from dns_sync.persistence import SQLiteStateStore


NOW = datetime(
    2026,
    9,
    20,
    9,
    30,
    tzinfo=timezone.utc,
)


@dataclass
class FakeNpmAdapter:
    endpoint: Endpoint

    @property
    def capabilities(self):
        return frozenset(
            {
                Capability(
                    CERTIFICATES,
                    CapabilityOperation.DISCOVER,
                ),
                Capability(
                    CERTIFICATES,
                    CapabilityOperation.VERIFY,
                ),
                Capability(
                    PROXY_HOSTS,
                    CapabilityOperation.DISCOVER,
                ),
                Capability(
                    PROXY_HOSTS,
                    CapabilityOperation.VERIFY,
                ),
            }
        )

    def connect(self):
        return None

    def discover(self, object_type):
        if object_type == CERTIFICATES:
            return [
                DiscoveredObject(
                    object_type=CERTIFICATES,
                    local_key="67",
                    payload={
                        "domain_names": [
                            "app.example.com",
                        ],
                        "provider": "letsencrypt",
                        "nice_name": "app",
                        "valid_from": (
                            "2026-09-01T00:00:00Z"
                        ),
                        "expires_on": (
                            "2026-12-01T00:00:00Z"
                        ),
                        "cluster_managed": True,
                        "acme_renew": True,
                        "dns_challenge": False,
                    },
                ),
            ]

        if object_type == PROXY_HOSTS:
            return [
                DiscoveredObject(
                    object_type=PROXY_HOSTS,
                    local_key="12",
                    payload={
                        "domain_names": [
                            "app.example.com",
                        ],
                        "forward_scheme": "http",
                        "forward_host": "backend.internal",
                        "forward_port": 8080,
                        "ssl_forced": True,
                        "caching_enabled": False,
                        "block_exploits": True,
                        "allow_websocket_upgrade": True,
                        "http2_support": True,
                        "hsts_enabled": True,
                        "hsts_subdomains": False,
                        "enabled": True,
                    },
                    references=(
                        ObjectReference(
                            relation="certificate",
                            object_type=CERTIFICATES,
                            local_key="67",
                        ),
                    ),
                ),
            ]

        raise RuntimeError(
            f"unsupported discovery type: {object_type}"
        )

    def apply(self, action):
        raise AssertionError(
            "NPM APPLY must remain unavailable"
        )

    def verify(self, object):
        return True


class NpmDesiredStateOrchestrationTests(
    unittest.TestCase
):
    def test_orchestrator_builds_and_persists_all_three_desired_types(
        self,
    ):
        adapter = FakeNpmAdapter(
            Endpoint(
                "npm",
                "192.168.1.3:81",
            )
        )

        with SQLiteStateStore(":memory:") as store:
            run = discover_reconcile_plan(
                [adapter],
                store,
                target_ip="192.168.1.101",
                run_at=NOW,
            )

            desired_by_type = {
                object.object_type: object
                for object in run.desired
            }

            self.assertEqual(
                set(desired_by_type),
                {
                    CERTIFICATES,
                    PROXY_HOSTS,
                    DNS_RECORDS,
                },
            )

            certificate = desired_by_type[
                CERTIFICATES
            ]
            proxy = desired_by_type[PROXY_HOSTS]
            dns = desired_by_type[DNS_RECORDS]

            self.assertIsNone(
                certificate.local_key
            )
            self.assertIsNone(
                proxy.local_key
            )
            self.assertIsNone(
                dns.local_key
            )

            self.assertEqual(
                proxy.references,
                (
                    ObjectReference(
                        relation="certificate",
                        object_type=CERTIFICATES,
                        logical_key="app.example.com",
                    ),
                ),
            )

            self.assertEqual(
                dns.payload["hostname"],
                "app.example.com",
            )
            self.assertEqual(
                dns.payload["ip"],
                "192.168.1.101",
            )

            for object_type in (
                CERTIFICATES,
                PROXY_HOSTS,
                DNS_RECORDS,
            ):
                persisted = (
                    store.load_desired_state(
                        object_type
                    )
                )

                self.assertEqual(
                    len(persisted),
                    1,
                )
                self.assertEqual(
                    persisted[0].object,
                    desired_by_type[
                        object_type
                    ],
                )

            # The NPM adapter intentionally does not advertise
            # APPLY, so this read-only orchestration must not
            # produce an NPM endpoint plan.
            self.assertEqual(
                run.plans,
                (),
            )

    def test_empty_npm_snapshot_clears_stale_npm_desired_state(
        self,
    ):
        adapter = FakeNpmAdapter(
            Endpoint(
                "npm",
                "192.168.1.3:81",
            )
        )

        with SQLiteStateStore(":memory:") as store:
            first = discover_reconcile_plan(
                [adapter],
                store,
                target_ip="192.168.1.101",
                run_at=NOW,
            )

            self.assertTrue(first.desired)
            self.assertEqual(
                len(
                    store.load_desired_state(
                        CERTIFICATES
                    )
                ),
                1,
            )
            self.assertEqual(
                len(
                    store.load_desired_state(
                        PROXY_HOSTS
                    )
                ),
                1,
            )

            class EmptyNpmAdapter(
                FakeNpmAdapter
            ):
                def discover(
                    self,
                    object_type,
                ):
                    if object_type in {
                        CERTIFICATES,
                        PROXY_HOSTS,
                    }:
                        return []

                    return super().discover(
                        object_type
                    )

            empty = EmptyNpmAdapter(
                adapter.endpoint
            )

            second = discover_reconcile_plan(
                [empty],
                store,
                target_ip="192.168.1.101",
                run_at=NOW.replace(minute=31),
            )

            self.assertEqual(
                second.desired,
                (),
            )
            self.assertEqual(
                store.load_desired_state(
                    CERTIFICATES
                ),
                [],
            )
            self.assertEqual(
                store.load_desired_state(
                    PROXY_HOSTS
                ),
                [],
            )
            self.assertEqual(
                store.load_desired_state(
                    DNS_RECORDS
                ),
                [],
            )


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