dns-sync/tests/test_verizon_adapter.py

import os
import unittest
from unittest.mock import Mock, patch

from dns_sync.adapters import (
    DNS_RECORDS,
    Capability,
    CapabilityOperation,
    NormalizedObject,
    PlanAction,
    PlanOperation,
)
from dns_sync.config import AppConfig
from dns_sync.models import ProviderRecord
from dns_sync.registry import load_adapter


class VerizonAdapterTests(unittest.TestCase):
    def config(self):
        with patch.dict(
            os.environ,
            {
                "NPM_ENDPOINTS": "192.168.1.3",
                "VERIZON_ENDPOINTS": "192.168.1.1",
                "ENABLED_PLUGINS": "npm,verizon",
                "DNS_SYNC_PLUGIN_PATH": "external_plugins",
            },
            clear=True,
        ):
            return AppConfig.from_env()

    def adapter(self, provider):
        config = self.config()

        import dns_sync.registry as registry

        with patch.dict(
            os.environ,
            {
                "ENABLED_PLUGINS": "npm,verizon",
                "DNS_SYNC_PLUGIN_PATH": "external_plugins",
            },
            clear=False,
        ):
            module = registry._load_external_adapter_module(
                "verizon"
            )

            if module is None:
                raise AssertionError(
                    "external Verizon adapter was not found"
                )

            with patch.object(
                module,
                "VerizonProvider",
                return_value=provider,
            ):
                return load_adapter(
                    "verizon",
                    config,
                    config.verizon_endpoints[0],
                )

    @staticmethod
    def dns_object(
        hostname="app.example.com",
        ip="192.168.1.101",
        wildcard=False,
    ):
        return NormalizedObject(
            object_type=DNS_RECORDS,
            logical_key=hostname,
            payload={
                "hostname": hostname,
                "ip": ip,
                "record_type": "A",
                "wildcard": wildcard,
            },
        )

    def test_capabilities_cover_dns_discovery_apply_and_verify(self):
        adapter = self.adapter(Mock())

        self.assertEqual(
            adapter.capabilities,
            frozenset(
                {
                    Capability(
                        DNS_RECORDS,
                        CapabilityOperation.DISCOVER,
                    ),
                    Capability(
                        DNS_RECORDS,
                        CapabilityOperation.APPLY,
                    ),
                    Capability(
                        DNS_RECORDS,
                        CapabilityOperation.VERIFY,
                    ),
                }
            ),
        )

    def test_generic_adapter_uses_endpoint_not_router_url(self):
        with patch.dict(
            os.environ,
            {
                "NPM_ENDPOINTS": "192.168.1.3",
                "VERIZON_ENDPOINTS": "192.168.1.1:8443",
                "ENABLED_PLUGINS": "npm,verizon",
                "DNS_SYNC_PLUGIN_PATH": "external_plugins",
                "ROUTER_URL": "https://192.168.1.254",
                "ROUTER_PASSWORD": "not-a-real-password",
            },
            clear=True,
        ):
            config = AppConfig.from_env()
            adapter = load_adapter(
                "verizon",
                config,
                config.verizon_endpoints[0],
            )

        self.assertEqual(
            adapter.endpoint.identity,
            "verizon://192.168.1.1:8443",
        )
        self.assertEqual(
            adapter.provider.router_url,
            "https://192.168.1.1:8443",
        )

    def test_connect_delegates_to_provider(self):
        provider = Mock()
        adapter = self.adapter(provider)

        adapter.connect()

        provider.connect.assert_called_once_with()

    def test_discovers_dns_records_with_slot_local_keys(self):
        provider = Mock()
        provider.list_records.return_value = {
            "second.example.com": ProviderRecord(
                hostname="second.example.com",
                ip="192.168.1.101",
                metadata={"slot": 42},
            ),
            "first.example.com": ProviderRecord(
                hostname="first.example.com",
                ip="192.168.1.101",
                metadata={"slot": 41},
            ),
        }
        adapter = self.adapter(provider)

        discovered = adapter.discover(DNS_RECORDS)

        self.assertEqual(
            tuple(item.local_key for item in discovered),
            ("41", "42"),
        )
        self.assertEqual(
            discovered[0].payload,
            {
                "hostname": "first.example.com",
                "ip": "192.168.1.101",
                "record_type": "A",
            },
        )

    def test_create_and_update_delegate_without_backend_ip(self):
        provider = Mock()
        provider.add_record.return_value = "created"
        provider.update_record.return_value = "updated"
        provider.list_records.return_value = {
            "app.example.com": ProviderRecord(
                hostname="app.example.com",
                ip="192.168.1.101",
                metadata={"slot": 41},
            ),
        }
        adapter = self.adapter(provider)
        desired = self.dns_object(ip="192.168.1.101")

        create_result = adapter.apply(
            PlanAction(
                endpoint_id=adapter.endpoint.identity,
                operation=PlanOperation.CREATE,
                desired=desired,
            )
        )
        update_result = adapter.apply(
            PlanAction(
                endpoint_id=adapter.endpoint.identity,
                operation=PlanOperation.UPDATE,
                desired=desired,
            )
        )

        self.assertEqual(
            create_result.object.local_key,
            "41",
        )
        self.assertEqual(
            create_result.detail,
            "created",
        )
        self.assertEqual(
            update_result.object.local_key,
            "41",
        )
        self.assertEqual(
            update_result.detail,
            "updated",
        )
        self.assertEqual(
            provider.add_record.call_args.args[0].ip,
            "192.168.1.101",
        )
        self.assertEqual(
            provider.update_record.call_args.args[0].ip,
            "192.168.1.101",
        )

    def test_noop_does_not_mutate_router(self):
        provider = Mock()
        adapter = self.adapter(provider)

        result = adapter.apply(
            PlanAction(
                endpoint_id=adapter.endpoint.identity,
                operation=PlanOperation.NOOP,
                desired=self.dns_object(),
            )
        )

        self.assertIsNone(result)
        provider.add_record.assert_not_called()
        provider.update_record.assert_not_called()

    def test_verify_delegates_expected_dns_record(self):
        provider = Mock()
        provider.verify_record.return_value = True
        adapter = self.adapter(provider)

        verified = adapter.verify(self.dns_object())

        self.assertTrue(verified)
        record = provider.verify_record.call_args.args[0]
        self.assertEqual(record.hostname, "app.example.com")
        self.assertEqual(record.ip, "192.168.1.101")

    def test_unsupported_objects_actions_and_wildcards_fail(self):
        provider = Mock()
        adapter = self.adapter(provider)

        with self.assertRaisesRegex(RuntimeError, "cannot discover"):
            adapter.discover("certificates")

        with self.assertRaisesRegex(RuntimeError, "different endpoint"):
            adapter.apply(
                PlanAction(
                    endpoint_id="verizon://different:443",
                    operation=PlanOperation.CREATE,
                    desired=self.dns_object(),
                )
            )

        with self.assertRaisesRegex(RuntimeError, "wildcard"):
            adapter.apply(
                PlanAction(
                    endpoint_id=adapter.endpoint.identity,
                    operation=PlanOperation.CREATE,
                    desired=self.dns_object(
                        hostname="*.example.com",
                        wildcard=True,
                    ),
                )
            )

        with self.assertRaisesRegex(RuntimeError, "delete"):
            adapter.apply(
                PlanAction(
                    endpoint_id=adapter.endpoint.identity,
                    operation=PlanOperation.DELETE,
                    desired=self.dns_object(),
                )
            )


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