dns-sync/tests/test_object_normalizer.py

import unittest
from datetime import datetime, timedelta, timezone

from dns_sync.adapters import (
    CERTIFICATES,
    DNS_RECORDS,
    PROXY_HOSTS,
    DiscoveredObject,
    ObjectReference,
)
from dns_sync.object_normalizer import (
    normalize_discovered_object,
    normalize_observations,
)


class CertificateNormalizationTests(unittest.TestCase):
    def test_same_domains_share_identity_but_keep_local_keys(self):
        rows = [
            DiscoveredObject(
                object_type=CERTIFICATES,
                local_key="63",
                payload={
                    "domain_names": [
                        "APP.EXAMPLE.COM",
                        "example.com.",
                    ],
                    "provider": "letsencrypt",
                    "expires_on": "2026-12-01T00:00:00Z",
                },
            ),
            DiscoveredObject(
                object_type=CERTIFICATES,
                local_key="67",
                payload={
                    "domain_names": [
                        "example.com",
                        "app.example.com",
                    ],
                    "provider": "other",
                    "expires_on": "2026-12-15T00:00:00Z",
                },
            ),
        ]

        observations = normalize_observations(
            "npm://192.168.1.3:81",
            rows,
        )

        self.assertEqual(
            observations[0].object.logical_key,
            "app.example.com,example.com",
        )
        self.assertEqual(
            observations[0].object.logical_key,
            observations[1].object.logical_key,
        )
        self.assertEqual(
            tuple(item.object.local_key for item in observations),
            ("63", "67"),
        )
        self.assertEqual(
            observations[1].object.payload["expires_on"],
            "2026-12-15T00:00:00+00:00",
        )


class ProxyHostNormalizationTests(unittest.TestCase):
    def test_proxy_host_identity_and_payload_are_deterministic(self):
        discovered = DiscoveredObject(
            object_type=PROXY_HOSTS,
            local_key="12",
            payload={
                "domain_names": [
                    "WWW.EXAMPLE.COM.",
                    "example.com",
                    "www.example.com",
                ],
                "forward_scheme": "HTTP",
                "forward_host": "backend.internal",
                "forward_port": "8080",
                "certificate_id": 67,
                "ssl_forced": 1,
                "enabled": True,
            },
            references=(
                ObjectReference(
                    relation="certificate",
                    object_type=CERTIFICATES,
                    local_key="67",
                ),
            ),
        )

        normalized = normalize_discovered_object(discovered)

        self.assertEqual(
            normalized.logical_key,
            "example.com,www.example.com",
        )
        self.assertEqual(
            normalized.payload["domain_names"],
            ["example.com", "www.example.com"],
        )
        self.assertEqual(normalized.payload["forward_scheme"], "http")
        self.assertEqual(normalized.payload["forward_port"], 8080)
        self.assertNotIn(
            "certificate_local_key",
            normalized.payload,
        )
        self.assertNotIn(
            "certificate_id",
            normalized.payload,
        )
        self.assertEqual(
            normalized.references,
            (
                ObjectReference(
                    relation="certificate",
                    object_type=CERTIFICATES,
                    local_key="67",
                ),
            ),
        )
        self.assertTrue(normalized.payload["ssl_forced"])
        self.assertTrue(normalized.payload["enabled"])
        self.assertFalse(normalized.payload["hsts_enabled"])

    def test_invalid_forward_port_is_rejected(self):
        discovered = DiscoveredObject(
            object_type=PROXY_HOSTS,
            local_key="12",
            payload={
                "domain_names": ["example.com"],
                "forward_scheme": "http",
                "forward_host": "backend.internal",
                "forward_port": 70000,
            },
        )

        with self.assertRaisesRegex(RuntimeError, "between 1 and 65535"):
            normalize_discovered_object(discovered)


class DnsNormalizationTests(unittest.TestCase):
    def test_dns_record_normalizes_hostname_and_ipv4(self):
        discovered = DiscoveredObject(
            object_type=DNS_RECORDS,
            local_key="4",
            payload={
                "hostname": "APP.EXAMPLE.COM.",
                "ip": "192.168.001.101",
                "record_type": "a",
            },
        )

        with self.assertRaisesRegex(RuntimeError, "valid IP address"):
            normalize_discovered_object(discovered)

        valid = DiscoveredObject(
            object_type=DNS_RECORDS,
            local_key="4",
            payload={
                "hostname": "APP.EXAMPLE.COM.",
                "ip": "192.168.1.101",
                "record_type": "a",
            },
        )
        normalized = normalize_discovered_object(valid)

        self.assertEqual(normalized.logical_key, "app.example.com")
        self.assertEqual(normalized.payload["record_type"], "A")
        self.assertEqual(normalized.payload["ip"], "192.168.1.101")


class ObservationNormalizationTests(unittest.TestCase):
    def test_batch_uses_one_utc_timestamp(self):
        observed_at = datetime(
            2026,
            9,
            19,
            16,
            0,
            tzinfo=timezone(timedelta(hours=-4)),
        )
        row = DiscoveredObject(
            object_type=DNS_RECORDS,
            local_key="1",
            payload={
                "hostname": "example.com",
                "ip": "192.168.1.101",
            },
        )

        observations = normalize_observations(
            "verizon://192.168.1.1",
            [row, row],
            observed_at=observed_at,
        )

        self.assertEqual(
            {item.observed_at.isoformat() for item in observations},
            {"2026-09-19T20:00:00+00:00"},
        )

    def test_unsupported_type_and_naive_timestamp_are_rejected(self):
        unsupported = DiscoveredObject(
            object_type="unknown",
            local_key="1",
            payload={},
        )
        with self.assertRaisesRegex(RuntimeError, "No normalizer"):
            normalize_discovered_object(unsupported)

        with self.assertRaisesRegex(RuntimeError, "timezone-aware"):
            normalize_observations(
                "npm://192.168.1.3:81",
                [],
                observed_at=datetime(2026, 9, 19),
            )


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