"""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, ) from netbox_dns_bridge.jobs import notify return [notify, 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()