netbox-plugin-dns-bridge/netbox_dns_bridge/tests/conftest.py

"""Pytest configuration and shared fixtures.

Tests run inside the NetBox container (``DJANGO_SETTINGS_MODULE=netbox.configuration``)
where ``netbox`` and ``netbox_dns`` are importable. The plugin's module-level
``SETTINGS`` bindings are patched per-test via the ``patch_settings`` fixture.

Heavy imports are deferred into fixtures so this conftest imports cleanly before
the Django app registry is guaranteed ready (the root conftest calls
``django.setup()``).
"""
from __future__ import annotations

import pytest


@pytest.fixture
def default_view(db):
    """Ensure NetBox DNS has a default view (required before creating zones)."""
    from netbox_dns.models import View

    view, _ = View.objects.get_or_create(
        name="default", defaults={"default_view": True}
    )
    return view


def _settings_modules():
    from netbox_dns_bridge.management.commands.dns_transfer_endpoint import (
        request_handler,
        catalog_zone_manager as catzm,  # noqa: F401
    )
    from netbox_dns_bridge.signals import netbox_dns_zones, notify_handler

    return [netbox_dns_zones, notify_handler, request_handler]


@pytest.fixture
def patch_settings(monkeypatch):
    """Patch the module-level ``SETTINGS`` dict on every consumer module."""
    mods = _settings_modules()
    originals = {mod: mod.SETTINGS for mod in mods}

    def _apply(new_settings):
        for mod in mods:
            monkeypatch.setattr(mod, "SETTINGS", new_settings)
        return new_settings

    yield _apply

    for mod, original in originals.items():
        mod.SETTINGS = original


@pytest.fixture
def tsig_key():
    from netbox_dns_bridge.tests._helpers import make_tsig_key
    return make_tsig_key("testkey.")


@pytest.fixture
def capture_handler():
    """Factory: build a capture handler for unit tests."""
    from netbox_dns_bridge.tests import _helpers

    def _make(handler_cls, keyring, tsig_view_map, client_ip="127.0.0.1"):
        return _helpers.make_capture_handler(
            handler_cls, keyring, tsig_view_map, client_ip
        )
    return _make


# ---------------------------------------------------------------------------
# Database factories
# ---------------------------------------------------------------------------

@pytest.fixture
def make_view(db, default_view):
    from netbox_dns.models import View

    counter = [0]

    def _make(name=None, default_view=False):
        counter[0] += 1
        name = name or f"test-view-{counter[0]}"
        view, _ = View.objects.get_or_create(
            name=name, defaults={"default_view": default_view}
        )
        return view
    return _make


@pytest.fixture
def make_nameserver(db):
    from netbox_dns.models import NameServer

    counter = [0]

    def _make(name=None):
        counter[0] += 1
        name = name or f"ns{counter[0]}.example.test"
        ns, _ = NameServer.objects.get_or_create(name=name)
        return ns
    return _make


@pytest.fixture
def make_zone(db, make_view, make_nameserver):
    from netbox_dns.models import Zone

    counter = [0]

    def _make(name=None, view=None, soa_serial=2024010101, **kwargs):
        counter[0] += 1
        name = name or f"zone-{counter[0]}.example.test"
        view = view or make_view()
        ns = make_nameserver(f"ns1.{name}")
        defaults = dict(
            view=view,
            name=name,
            soa_mname=ns,
            soa_rname=f"hostmaster.{name}",
            soa_ttl=300,
            soa_serial=soa_serial,
            soa_serial_auto=False,
            soa_refresh=300,
            soa_retry=100,
            soa_expire=1209600,
            soa_minimum=3600,
            default_ttl=300,
            status="active",
        )
        defaults.update(kwargs)
        zone = Zone.objects.create(**defaults)
        # Ensure an active SOA record exists (netbox_dns usually creates one,
        # but be deterministic so build_dns_zone / _get_soa_serial always find it).
        from netbox_dns.models import Record
        from netbox_dns.choices import RecordStatusChoices
        if not Record.objects.filter(zone=zone, type="SOA", active=True).exists():
            fqdn = zone.name
            Record.objects.create(
                zone=zone,
                name="",
                fqdn=fqdn,
                type="SOA",
                value=(
                    f"ns1.{zone.name}. hostmaster.{zone.name}. {soa_serial} "
                    f"300 100 1209600 3600"
                ),
                ttl=zone.soa_ttl,
                status=RecordStatusChoices.STATUS_ACTIVE,
                managed=False,
            )
        return zone
    return _make


@pytest.fixture
def make_record(db):
    from netbox_dns.models import Record
    from netbox_dns.choices import RecordStatusChoices

    counter = [0]

    def _make(zone, name="", type_="A", value="127.0.0.1", ttl=None,
              status=RecordStatusChoices.STATUS_ACTIVE, managed=False):
        counter[0] += 1
        fqdn = f"{name}.{zone.name}".lstrip(".") if name else zone.name
        return Record.objects.create(
            zone=zone,
            name=name,
            fqdn=fqdn,
            type=type_,
            value=value,
            ttl=ttl if ttl is not None else zone.default_ttl,
            status=status,
            managed=managed,
        )
    return _make


@pytest.fixture
def catz_serial(db):
    """Initial SOA serial value (1) for freshly created catalog zones.

    A ``CatalogZone`` is auto-created with ``soa_serial=1`` by the zone
    post_save signal (via ``update_member_identifier``) whenever a zone is
    created. Tests compare against this initial value.
    """
    return 1


# ---------------------------------------------------------------------------
# Real-socket integration server
# ---------------------------------------------------------------------------

@pytest.fixture
def dns_server(tsig_key):
    """Start real TCP+UDP DNS servers on 127.0.0.1:0 in background threads.

    Usage::

        srv = dns_server.start(views={"testkey.": view_obj})
        resp = dns.query.tcp(msg, srv["host"], port=srv["port"])
    """
    import threading
    import dns.name
    from netbox_dns_bridge.management.commands.dns_transfer_endpoint.dns_server import (
        TCPDNSServer,
        UDPDNSServer,
    )
    from netbox_dns_bridge.management.commands.dns_transfer_endpoint.request_handler import (
        TCPRequestHandler,
        UDPRequestHandler,
    )

    class _Server:
        def __init__(self, key):
            self.tsig_key = key
            self._tcp = None
            self._udp = None

        def start(self, views):
            keyring = {self.tsig_key.name: self.tsig_key}
            tsig_view_map = {}
            for key_name, view in views.items():
                canon = (
                    dns.name.from_text(key_name, origin=dns.name.root)
                    .canonicalize()
                    .to_text()
                )
                tsig_view_map[canon] = view

            self._tcp = TCPDNSServer(
                ("127.0.0.1", 0), TCPRequestHandler, keyring, tsig_view_map
            )
            port = self._tcp.server_address[1]
            # Bind UDP to the same port number (TCP/UDP sockets may share it).
            self._udp = UDPDNSServer(
                ("127.0.0.1", port), UDPRequestHandler, keyring, tsig_view_map
            )
            threading.Thread(target=self._tcp.serve_forever, daemon=True).start()
            threading.Thread(target=self._udp.serve_forever, daemon=True).start()
            return {
                "host": "127.0.0.1",
                "port": port,
                "tsig_key": self.tsig_key,
            }

        def shutdown(self):
            for srv in (self._tcp, self._udp):
                if srv is not None:
                    srv.shutdown()
                    srv.server_close()

    server = _Server(tsig_key)
    try:
        yield server
    finally:
        server.shutdown()