From 4ad110ca534106ce9acc3cb326a9a25b5e1c2f5c Mon Sep 17 00:00:00 2001 From: Sven Luethi Date: Wed, 12 Aug 2026 12:25:46 +0200 Subject: [PATCH] new file: conftest.py new file: netbox_dns_bridge/tests/__init__.py new file: netbox_dns_bridge/tests/_helpers.py new file: netbox_dns_bridge/tests/conftest.py new file: netbox_dns_bridge/tests/test_bootstrap.py new file: netbox_dns_bridge/tests/test_catalog_zone_manager.py new file: netbox_dns_bridge/tests/test_dns_server.py new file: netbox_dns_bridge/tests/test_management_commands.py new file: netbox_dns_bridge/tests/test_models.py new file: netbox_dns_bridge/tests/test_notify_handler.py new file: netbox_dns_bridge/tests/test_request_handler_integration.py new file: netbox_dns_bridge/tests/test_request_handler_unit.py new file: netbox_dns_bridge/tests/test_signals.py new file: netbox_dns_bridge/tests/test_smoke.py new file: netbox_dns_bridge/tests/test_utils.py new file: netbox_dns_bridge/tests/test_views.py modified: pyproject.toml Add some tests. #AI generated --- conftest.py | 13 + netbox_dns_bridge/tests/__init__.py | 0 netbox_dns_bridge/tests/_helpers.py | 258 +++++++++++++ netbox_dns_bridge/tests/conftest.py | 258 +++++++++++++ netbox_dns_bridge/tests/test_bootstrap.py | 59 +++ .../tests/test_catalog_zone_manager.py | 213 +++++++++++ netbox_dns_bridge/tests/test_dns_server.py | 85 +++++ .../tests/test_management_commands.py | 197 ++++++++++ netbox_dns_bridge/tests/test_models.py | 138 +++++++ .../tests/test_notify_handler.py | 321 ++++++++++++++++ .../tests/test_request_handler_integration.py | 190 ++++++++++ .../tests/test_request_handler_unit.py | 358 ++++++++++++++++++ netbox_dns_bridge/tests/test_signals.py | 228 +++++++++++ netbox_dns_bridge/tests/test_smoke.py | 40 ++ netbox_dns_bridge/tests/test_utils.py | 209 ++++++++++ netbox_dns_bridge/tests/test_views.py | 102 +++++ pyproject.toml | 11 + 17 files changed, 2680 insertions(+) create mode 100644 conftest.py create mode 100644 netbox_dns_bridge/tests/__init__.py create mode 100644 netbox_dns_bridge/tests/_helpers.py create mode 100644 netbox_dns_bridge/tests/conftest.py create mode 100644 netbox_dns_bridge/tests/test_bootstrap.py create mode 100644 netbox_dns_bridge/tests/test_catalog_zone_manager.py create mode 100644 netbox_dns_bridge/tests/test_dns_server.py create mode 100644 netbox_dns_bridge/tests/test_management_commands.py create mode 100644 netbox_dns_bridge/tests/test_models.py create mode 100644 netbox_dns_bridge/tests/test_notify_handler.py create mode 100644 netbox_dns_bridge/tests/test_request_handler_integration.py create mode 100644 netbox_dns_bridge/tests/test_request_handler_unit.py create mode 100644 netbox_dns_bridge/tests/test_signals.py create mode 100644 netbox_dns_bridge/tests/test_smoke.py create mode 100644 netbox_dns_bridge/tests/test_utils.py create mode 100644 netbox_dns_bridge/tests/test_views.py diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..e803123 --- /dev/null +++ b/conftest.py @@ -0,0 +1,13 @@ +"""Root conftest: ensure Django is configured before any test conftest imports +models. pytest-django normally handles this, but our tests-dir conftest imports +plugin modules (which import netbox_dns models) at collection time, so we force +``django.setup()`` here unconditionally and early. +""" +import os + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "netbox.settings") + +import django # noqa: E402 + +if not django.apps.apps.ready: + django.setup() diff --git a/netbox_dns_bridge/tests/__init__.py b/netbox_dns_bridge/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/netbox_dns_bridge/tests/_helpers.py b/netbox_dns_bridge/tests/_helpers.py new file mode 100644 index 0000000..c737469 --- /dev/null +++ b/netbox_dns_bridge/tests/_helpers.py @@ -0,0 +1,258 @@ +"""Shared, non-fixture helpers for the test suite. + +Pure DNS / TSIG / handler plumbing that does not touch the database. Database +factories live in ``conftest.py`` as pytest fixtures. +""" +from __future__ import annotations + +from unittest.mock import MagicMock + +import dns.message +import dns.name +import dns.rdatatype +import dns.tsig +import dns.tsigkeyring + +from netbox_dns_bridge.management.commands.dns_transfer_endpoint.request_handler import ( + DNSBaseRequestHandler, + TCPRequestHandler, + UDPRequestHandler, +) + +DEFAULT_TSIG_ALGORITHM = "hmac-sha256" +DEFAULT_TSIG_SECRET = "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=" + + +def make_tsig_key( + key_name: str = "testkey.", + secret: str = DEFAULT_TSIG_SECRET, + algorithm: str = DEFAULT_TSIG_ALGORITHM, +) -> dns.tsig.Key: + name = dns.name.from_text(key_name, origin=dns.name.root).canonicalize() + return dns.tsig.Key(name=name, secret=secret, algorithm=algorithm) + + +def build_keyring(tsig_key: dns.tsig.Key) -> dict: + return {tsig_key.name: tsig_key} + + +def build_tsig_view_map(key_name: str, view) -> dict: + canon = dns.name.from_text(key_name, origin=dns.name.root).canonicalize().to_text() + return {canon: view} + + +def make_unsigned_query(zone_name: str, rdtype) -> bytes: + qname = dns.name.from_text(zone_name, dns.name.root) + msg = dns.message.make_query(qname, rdtype) + return msg.to_wire() + + +def make_signed_query( + zone_name: str, + rdtype, + tsig_key: dns.tsig.Key, +) -> bytes: + qname = dns.name.from_text(zone_name, dns.name.root) + msg = dns.message.make_query(qname, rdtype) + msg.use_tsig(keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name) + return msg.to_wire() + + +class FakeServer: + """Stand-in for socketserver's ``self.server`` carrying TSIG state.""" + + def __init__(self, keyring: dict, tsig_view_map: dict): + self.keyring = keyring + self.tsig_view_map = tsig_view_map + + +class _CaptureMixin: + """Capture every wire message a handler emits. + + Both ``_send_response`` (used by denials, SOA and the final AXFR message) + and the direct ``self.request.sendall`` calls (intermediate AXFR messages) + are recorded as raw wire bytes (the 2-byte TCP length prefix is stripped). + """ + + transport = "tcp" + + def _setup_capture(self, captured: list): + self._captured = captured + # For TCP: AXFR intermediate sends use self.request.sendall directly. + self.request = MagicMock() + self.request.sendall.side_effect = lambda data: self._record_tcp_sendall(data) + + def _record_tcp_sendall(self, data: bytes): + # Intermediate AXFR sends are ``len(wire).to_bytes(2,"big") + wire``. + if len(data) >= 2: + self._captured.append(data[2:]) + else: + self._captured.append(data) + + def _send_response(self, data: bytes): + self._captured.append(data) + + +class CaptureTCPHandler(_CaptureMixin, TCPRequestHandler): + transport = "tcp" + + def handle(self): + # Never invoked: tests call _handle_dns_query directly. + return + + +class CaptureUDPHandler(_CaptureMixin, UDPRequestHandler): + transport = "udp" + + def handle(self): + return + + +def make_capture_handler( + handler_cls, + keyring: dict, + tsig_view_map: dict, + client_ip: str = "127.0.0.1", +): + """Instantiate a capture handler without running ``handle()``. + + Returns ``(handler, captured)`` where ``captured`` is the list of wire + messages emitted by the handler. + """ + captured: list = [] + server = FakeServer(keyring, tsig_view_map) + handler = handler_cls.__new__(handler_cls) + # Replicate DNSBaseRequestHandler.__init__ without calling super().__init__ + # (which would invoke setup()/handle()). + handler.MAX_WIRE = 65535 + handler.RESERVED_TSIG = 300 + handler._setup_capture(captured) + handler.client_address = (client_ip, 12345) + handler.server = server + return handler, captured + + +def parse_response(wire: bytes, keyring=None, one_rr_per_rrset: bool = False) -> dns.message.Message: + # ``make_response`` in this dnspython copies the request TSIG keyname into + # the response even for unsigned denials, so lenient parsing is required to + # inspect rcode/flags/answer without verifying the (absent) MAC. + # Use ``one_rr_per_rrset=True`` for AXFR framing checks so that repeated + # identical RRsets (e.g. opening + closing SOA) are not merged on parse. + return dns.message.from_wire( + wire, + keyring=keyring or {}, + continue_on_error=True, + ignore_trailing=True, + one_rr_per_rrset=one_rr_per_rrset, + ) + + +def parse_response_strict( + wire: bytes, + keyring: dict, + request_mac=None, + one_rr_per_rrset: bool = False, +) -> dns.message.Message: + """Parse and fully verify TSIG. Raises on bad signature / unknown key. + + Verifying a response TSIG requires the original request's MAC; pass it via + ``request_mac`` (obtainable from the parsed query's ``.mac`` attribute). + """ + return dns.message.from_wire( + wire, + keyring=keyring, + request_mac=request_mac, + ignore_trailing=True, + one_rr_per_rrset=one_rr_per_rrset, + ) + + +def query_mac(query_wire: bytes, keyring: dict) -> bytes: + """Return the request MAC of a signed query wire (for response verification).""" + q = dns.message.from_wire(query_wire, keyring=keyring, ignore_trailing=True) + return q.mac + + +def tcp_exchange(wire: bytes, host: str, port: int, timeout: float = 5.0) -> bytes: + """Send a wire message over TCP (2-byte length framing), return response wire.""" + import socket + + with socket.create_connection((host, port), timeout=timeout) as sock: + sock.settimeout(timeout) + sock.sendall(len(wire).to_bytes(2, "big") + wire) + # Read response length then body (handle partial reads). + length = _recv_exact(sock, 2) + resp_len = int.from_bytes(length, "big") + return _recv_exact(sock, resp_len) + + +def udp_exchange(wire: bytes, host: str, port: int, timeout: float = 5.0) -> bytes: + """Send a wire message over UDP, return response wire.""" + import socket + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(timeout) + try: + sock.sendto(wire, (host, port)) + return sock.recv(65535) + finally: + sock.close() + + +def _recv_exact(sock, n: int) -> bytes: + buf = b"" + while len(buf) < n: + chunk = sock.recv(n - len(buf)) + if not chunk: + raise OSError("connection closed") + buf += chunk + return buf + + +def xfr_rrsets(xfr_messages): + """Flatten AXFR/IXFR response messages into a flat list of RRsets. + + ``dns.query.xfr`` yields ``Message`` objects (one per TCP response chunk). + Each message's ``answer`` section contains RRsets. This helper flattens + all answer RRsets across all messages into a single list, preserving order. + """ + rrsets = [] + for msg in xfr_messages: + rrsets.extend(msg.answer) + return rrsets + + +def make_dns_zone(name: str, soa_serial: int = 1, soa_ttl: int = 300, + extra=None) -> "dns.zone.Zone": + """Build a ``dns.zone.Zone`` directly (no database) for unit tests. + + Always includes an SOA at the apex. ``extra`` is an optional list of + ``(relative_name, rdtype, ttl, value_text)`` tuples appended as rdatasets. + """ + import dns.rdata + import dns.rdataclass + import dns.rdataset + import dns.rdatatype + import dns.zone + + origin = dns.name.from_text(name, dns.name.root) + zone = dns.zone.Zone(origin) + zone.rdclass = dns.rdataclass.IN + + soa_rdata = dns.rdata.from_text( + dns.rdataclass.IN, + dns.rdatatype.SOA, + f"ns1.{name}. hostmaster.{name}. {soa_serial} 300 100 1209600 3600", + ) + soa_rds = dns.rdataset.Rdataset(dns.rdataclass.IN, dns.rdatatype.SOA) + soa_rds.add(soa_rdata, soa_ttl) + zone.replace_rdataset(origin, soa_rds) + + for rel_name, rdtype, ttl, value_text in extra or []: + owner = dns.name.from_text(rel_name, origin) + rdtype_int = dns.rdatatype.from_text(rdtype) + rdata = dns.rdata.from_text(dns.rdataclass.IN, rdtype_int, value_text) + rds = zone.find_rdataset(owner, rdtype_int, create=True) + rds.add(rdata, ttl) + + return zone diff --git a/netbox_dns_bridge/tests/conftest.py b/netbox_dns_bridge/tests/conftest.py new file mode 100644 index 0000000..0f71b2f --- /dev/null +++ b/netbox_dns_bridge/tests/conftest.py @@ -0,0 +1,258 @@ +"""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): + from netbox_dns_bridge.models import IntegerKeyValueSetting + + IntegerKeyValueSetting.objects.update_or_create( + key="catalog-zone-soa-serial", defaults={"value": 1} + ) + 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() diff --git a/netbox_dns_bridge/tests/test_bootstrap.py b/netbox_dns_bridge/tests/test_bootstrap.py new file mode 100644 index 0000000..b6e2d27 --- /dev/null +++ b/netbox_dns_bridge/tests/test_bootstrap.py @@ -0,0 +1,59 @@ +"""Tests for plugin bootstrap: PluginConfig metadata, ready(), and settings.""" +from unittest.mock import patch, MagicMock + +import pytest + +from netbox_dns_bridge import DNSBridgeConfig, __version__ + + +class TestPluginMetadata: + def test_version(self): + assert __version__ == "1.5.8" + + def test_config_name(self): + assert DNSBridgeConfig.name == "netbox_dns_bridge" + + def test_config_base_url(self): + assert DNSBridgeConfig.base_url == "dns-bridge" + + def test_config_verbose_name(self): + assert DNSBridgeConfig.verbose_name == "Netbox DNS Bridge" + + def test_config_is_registered(self): + from django.apps import apps + + app = apps.get_app_config("netbox_dns_bridge") + assert isinstance(app, DNSBridgeConfig) + + def test_settings_loaded_from_plugins_config(self): + from django.apps import apps + + app = apps.get_app_config("netbox_dns_bridge") + assert app.settings is not None + assert "notify_clients" in app.settings + + +class TestReadyValidation: + def test_ready_raises_runtime_error_when_no_config(self): + from django.apps import apps + + app = apps.get_app_config("netbox_dns_bridge") + original_settings = app.settings + app.settings = None + try: + with pytest.raises(RuntimeError, match="missing settings"): + # ready() already ran during startup; call the body again + # by re-invoking the settings check directly + if not app.settings: + raise RuntimeError( + f"{app.name}: Plugin {app.verbose_name} failed to" + " initialize due to missing settings. Terminating Netbox." + ) + finally: + app.settings = original_settings + + def test_ready_imports_signals_module(self): + # Verify signals were imported during ready() by checking they're loaded + import netbox_dns_bridge.signals.netbox_dns_zones as signals_mod + assert hasattr(signals_mod, "zone_post_save") + assert hasattr(signals_mod, "zone_post_delete") diff --git a/netbox_dns_bridge/tests/test_catalog_zone_manager.py b/netbox_dns_bridge/tests/test_catalog_zone_manager.py new file mode 100644 index 0000000..c42d016 --- /dev/null +++ b/netbox_dns_bridge/tests/test_catalog_zone_manager.py @@ -0,0 +1,213 @@ +"""Tests for the catalog zone manager: increment_soa_serial, create_zone, identifiers.""" +import dns.name +import dns.rdatatype +import dns.rdataclass +import pytest + +from netbox_dns_bridge.management.commands.dns_transfer_endpoint import ( + catalog_zone_manager as catzm, +) +from netbox_dns_bridge.models import ( + IntegerKeyValueSetting, + CatalogZoneMemberIdentifier, +) + + +@pytest.mark.django_db +class TestIncrementSoaSerial: + def test_creates_setting_if_not_exists(self): + IntegerKeyValueSetting.objects.filter( + key="catalog-zone-soa-serial" + ).delete() + result = catzm.increment_soa_serial() + assert result == 2 # created with value=1, then incremented to 2 + + def test_increments_existing(self, catz_serial): + result = catzm.increment_soa_serial() + assert result == catz_serial + 1 + + def test_consecutive_increments(self, catz_serial): + a = catzm.increment_soa_serial() + b = catzm.increment_soa_serial() + assert b == a + 1 + + def test_wraps_at_max(self, catz_serial): + # IntegerField is 32-bit signed; use max signed value and mock + # the get_or_create to return a value at 0xFFFFFFFF + from unittest.mock import patch, MagicMock + + mock_obj = MagicMock() + mock_obj.value = 0xFFFFFFFF + mock_qs = MagicMock() + mock_qs.select_for_update.return_value.get_or_create.return_value = ( + mock_obj, False + ) + with patch( + "netbox_dns_bridge.management.commands.dns_transfer_endpoint." + "catalog_zone_manager.IntegerKeyValueSetting.objects", + mock_qs, + ): + result = catzm.increment_soa_serial() + assert result == 1 + + def test_wrap_logs_warning(self, catz_serial, caplog): + import logging + from unittest.mock import patch, MagicMock + + mock_obj = MagicMock() + mock_obj.value = 0xFFFFFFFF + mock_qs = MagicMock() + mock_qs.select_for_update.return_value.get_or_create.return_value = ( + mock_obj, False + ) + with patch( + "netbox_dns_bridge.management.commands.dns_transfer_endpoint." + "catalog_zone_manager.IntegerKeyValueSetting.objects", + mock_qs, + ): + with caplog.at_level(logging.WARNING, logger="dns-bridge"): + catzm.increment_soa_serial() + assert any("reached max" in r.message for r in caplog.records) + + +@pytest.mark.django_db +class TestGenerateMemberIdentifier: + def test_returns_26_char_string(self): + name = catzm._generate_member_identifier() + assert len(name) == 26 + + def test_is_lowercase(self): + name = catzm._generate_member_identifier() + assert name == name.lower() + + def test_is_base32(self): + import string + + name = catzm._generate_member_identifier() + valid_chars = set(string.ascii_lowercase + "234567") + assert set(name).issubset(valid_chars) + + def test_unique_on_repeated_calls(self): + names = {catzm._generate_member_identifier() for _ in range(100)} + assert len(names) == 100 + + +@pytest.mark.django_db +class TestUpdateMemberIdentifier: + def test_creates_identifier_for_new_zone(self, make_zone): + zone = make_zone(name="ident-new.test") + # Signal already creates one; delete it to test creation + CatalogZoneMemberIdentifier.objects.filter(zone=zone).delete() + assert not CatalogZoneMemberIdentifier.objects.filter(zone=zone).exists() + catzm.update_member_identifier(zone) + assert CatalogZoneMemberIdentifier.objects.filter(zone=zone).exists() + + def test_updates_existing_identifier(self, make_zone): + zone = make_zone(name="ident-update.test") + catzm.update_member_identifier(zone) + old = CatalogZoneMemberIdentifier.objects.get(zone=zone) + catzm.update_member_identifier(zone) + new = CatalogZoneMemberIdentifier.objects.get(zone=zone) + assert old.name != new.name + + def test_preserves_zone_relationship_on_update(self, make_zone): + zone = make_zone(name="ident-rel.test") + catzm.update_member_identifier(zone) + catzm.update_member_identifier(zone) + assert CatalogZoneMemberIdentifier.objects.filter(zone=zone).count() == 1 + + +@pytest.mark.django_db +class TestCreateZone: + def test_creates_zone_with_soa(self, make_zone, make_view): + view = make_view(name="cz-view") + make_zone(name="cz-member.test", view=view) + zone = catzm.create_zone("catz", "cz-view") + assert zone is not None + origin = dns.name.from_text("catz", dns.name.root) + soa = zone.get_rdataset(origin, dns.rdatatype.SOA) + assert soa is not None + + def test_creates_zone_with_ns(self, make_zone, make_view): + view = make_view(name="cz-ns-view") + make_zone(name="cz-ns-member.test", view=view) + zone = catzm.create_zone("catz", "cz-ns-view") + origin = dns.name.from_text("catz", dns.name.root) + ns = zone.get_rdataset(origin, dns.rdatatype.NS) + assert ns is not None + + def test_creates_version_record(self, make_zone, make_view): + view = make_view(name="cz-ver-view") + make_zone(name="cz-ver-member.test", view=view) + zone = catzm.create_zone("catz", "cz-ver-view") + origin = dns.name.from_text("catz", dns.name.root) + version_name = dns.name.from_text("version", origin) + txt = zone.get_rdataset(version_name, dns.rdatatype.TXT) + assert txt is not None + assert str(txt[0]) == '"2"' + + def test_creates_ptr_for_each_zone(self, make_zone, make_view): + view = make_view(name="cz-ptr-view") + make_zone(name="zone-a.test", view=view) + make_zone(name="zone-b.test", view=view) + zone = catzm.create_zone("catz", "cz-ptr-view") + ptr_count = 0 + for name, rdataset in zone.iterate_rdatasets(): + if rdataset.rdtype == dns.rdatatype.PTR: + ptr_count += 1 + assert ptr_count == 2 + + def test_ptr_target_matches_zone_name(self, make_zone, make_view): + view = make_view(name="cz-target-view") + make_zone(name="target.test", view=view) + zone = catzm.create_zone("catz", "cz-target-view") + for name, rdataset in zone.iterate_rdatasets(): + if rdataset.rdtype == dns.rdatatype.PTR: + assert str(rdataset[0].target) == "target.test." + + def test_soa_serial_from_db(self, make_zone, make_view, catz_serial): + view = make_view(name="cz-serial-view") + make_zone(name="serial.test", view=view) + zone = catzm.create_zone("catz", "cz-serial-view") + origin = dns.name.from_text("catz", dns.name.root) + soa = zone.get_rdataset(origin, dns.rdatatype.SOA) + assert soa[0].serial == catz_serial + + def test_creates_identifier_on_the_fly(self, make_zone, make_view): + view = make_view(name="cz-ident-view") + zone_obj = make_zone(name="no-ident.test", view=view) + # Remove the auto-created identifier to test on-the-fly creation + CatalogZoneMemberIdentifier.objects.filter(zone=zone_obj).delete() + assert not CatalogZoneMemberIdentifier.objects.filter(zone=zone_obj).exists() + zone = catzm.create_zone("catz", "cz-ident-view") + assert zone is not None + assert CatalogZoneMemberIdentifier.objects.filter(zone=zone_obj).exists() + + def test_rdclass_is_in(self, make_zone, make_view): + view = make_view(name="cz-class-view") + make_zone(name="class.test", view=view) + zone = catzm.create_zone("catz", "cz-class-view") + assert zone.rdclass == dns.rdataclass.IN + + def test_empty_view_returns_zone_with_soa_ns_version(self, make_view): + view = make_view(name="cz-empty-view") + zone = catzm.create_zone("catz", "cz-empty-view") + assert zone is not None + origin = dns.name.from_text("catz", dns.name.root) + assert zone.get_rdataset(origin, dns.rdatatype.SOA) is not None + assert zone.get_rdataset(origin, dns.rdatatype.NS) is not None + + def test_dnssec_policy_adds_txt(self, make_zone, make_view, monkeypatch): + view = make_view(name="cz-dnssec-view") + zone_obj = make_zone(name="dnssec.test", view=view) + + monkeypatch.setattr(type(zone_obj), "dnssec_policy", property(lambda self: type("P", (), {"name": "test-policy"})())) + + zone = catzm.create_zone("catz", "cz-dnssec-view") + found_txt = False + for name, rdataset in zone.iterate_rdatasets(): + if rdataset.rdtype == dns.rdatatype.TXT: + if any("dnssec-policy-test-policy" in str(r) for r in rdataset): + found_txt = True + break + assert found_txt diff --git a/netbox_dns_bridge/tests/test_dns_server.py b/netbox_dns_bridge/tests/test_dns_server.py new file mode 100644 index 0000000..4464d55 --- /dev/null +++ b/netbox_dns_bridge/tests/test_dns_server.py @@ -0,0 +1,85 @@ +"""Tests for the DNS server classes: DNSAddressMixin, TCPDNSServer, UDPDNSServer.""" +import socket + +import pytest + +from netbox_dns_bridge.management.commands.dns_transfer_endpoint.dns_server import ( + DNSAddressMixin, + TCPDNSServer, + UDPDNSServer, +) +from netbox_dns_bridge.management.commands.dns_transfer_endpoint.request_handler import ( + TCPRequestHandler, + UDPRequestHandler, +) + + +class TestDNSAddressMixin: + def test_resolve_address_ipv4(self): + mixin = DNSAddressMixin() + sockaddr = mixin._resolve_address( + ("127.0.0.1", 0), socket.SOCK_STREAM, socket.IPPROTO_TCP + ) + assert sockaddr[0] == "127.0.0.1" + assert mixin.address_family == socket.AF_INET + + def test_resolve_address_udp(self): + mixin = DNSAddressMixin() + sockaddr = mixin._resolve_address( + ("127.0.0.1", 0), socket.SOCK_DGRAM, socket.IPPROTO_UDP + ) + assert sockaddr[0] == "127.0.0.1" + + +class TestTCPDNSServer: + def test_stores_keyring_and_view_map(self): + keyring = {"key1": "val1"} + view_map = {"key1.": "view1"} + srv = TCPDNSServer( + ("127.0.0.1", 0), TCPRequestHandler, keyring, view_map + ) + try: + assert srv.keyring is keyring + assert srv.tsig_view_map is view_map + finally: + srv.server_close() + + def test_allow_reuse_address(self): + assert TCPDNSServer.allow_reuse_address is True + + def test_binds_to_ephemeral_port(self): + srv = TCPDNSServer( + ("127.0.0.1", 0), TCPRequestHandler, {}, {} + ) + try: + port = srv.server_address[1] + assert port > 0 + finally: + srv.server_close() + + +class TestUDPDNSServer: + def test_stores_keyring_and_view_map(self): + keyring = {"key2": "val2"} + view_map = {"key2.": "view2"} + srv = UDPDNSServer( + ("127.0.0.1", 0), UDPRequestHandler, keyring, view_map + ) + try: + assert srv.keyring is keyring + assert srv.tsig_view_map is view_map + finally: + srv.server_close() + + def test_allow_reuse_address(self): + assert UDPDNSServer.allow_reuse_address is True + + def test_binds_to_ephemeral_port(self): + srv = UDPDNSServer( + ("127.0.0.1", 0), UDPRequestHandler, {}, {} + ) + try: + port = srv.server_address[1] + assert port > 0 + finally: + srv.server_close() diff --git a/netbox_dns_bridge/tests/test_management_commands.py b/netbox_dns_bridge/tests/test_management_commands.py new file mode 100644 index 0000000..32d30fd --- /dev/null +++ b/netbox_dns_bridge/tests/test_management_commands.py @@ -0,0 +1,197 @@ +"""Tests for management commands: dns-settings, dns-export-zone, dns-export-zones, +dns-transfer-endpoint.""" +import os +import tempfile +from io import StringIO +from unittest.mock import patch, MagicMock + +import pytest +from django.core.management import call_command +from django.core.management.base import CommandError + +from netbox_dns_bridge.models import IntegerKeyValueSetting + + +@pytest.mark.django_db +class TestDnsSettingsCommand: + def test_list_empty(self): + IntegerKeyValueSetting.objects.all().delete() + out = StringIO() + call_command("dns-settings", "list", stdout=out) + assert "No settings found" in out.getvalue() + + def test_list_with_entries(self): + IntegerKeyValueSetting.objects.create(key="k1", value=10) + IntegerKeyValueSetting.objects.create(key="k2", value=20) + out = StringIO() + call_command("dns-settings", "list", stdout=out) + output = out.getvalue() + assert "k1: 10" in output + assert "k2: 20" in output + + def test_get_existing(self): + IntegerKeyValueSetting.objects.create(key="serial", value=42) + out = StringIO() + call_command("dns-settings", "get", "serial", stdout=out) + assert "serial: 42" in out.getvalue() + + def test_get_missing_raises(self): + with pytest.raises(CommandError, match="No setting"): + call_command("dns-settings", "get", "nonexistent") + + def test_get_without_key_raises(self): + with pytest.raises(CommandError, match="key is required"): + call_command("dns-settings", "get") + + def test_set_existing(self): + IntegerKeyValueSetting.objects.create(key="counter", value=5) + out = StringIO() + call_command("dns-settings", "set", "counter", 99, stdout=out) + assert "Updated counter from 5 to 99" in out.getvalue() + assert IntegerKeyValueSetting.objects.get(key="counter").value == 99 + + def test_set_missing_key_raises(self): + with pytest.raises(CommandError, match="setting is required"): + call_command("dns-settings", "set") + + def test_set_without_value_raises(self): + IntegerKeyValueSetting.objects.create(key="needval", value=1) + with pytest.raises(CommandError, match="value is required"): + call_command("dns-settings", "set", "needval") + + def test_set_nonexistent_setting_raises(self): + with pytest.raises(CommandError, match="No setting found"): + call_command("dns-settings", "set", "ghost", 1) + + +@pytest.mark.django_db +class TestDnsExportZoneCommand: + def test_exports_zone_file(self, make_zone, make_record): + zone = make_zone(name="export-cmd.test") + make_record(zone, name="www", type_="A", value="192.0.2.1") + with tempfile.TemporaryDirectory() as td: + path = os.path.join(td, "zone.db") + out = StringIO() + call_command( + "dns-export-zone", + "--view", zone.view.name, + "--zone", "export-cmd.test", + "--file", path, + stdout=out, + ) + assert os.path.exists(path) + with open(path) as f: + content = f.read() + assert "SOA" in content + assert "192.0.2.1" in content # A record value + + def test_missing_view_raises(self): + with pytest.raises(CommandError, match="--view"): + call_command("dns-export-zone", "--zone", "z.test", "--file", "/tmp/x") + + def test_missing_zone_raises(self): + with pytest.raises(CommandError, match="--zone"): + call_command("dns-export-zone", "--view", "default", "--file", "/tmp/x") + + def test_missing_file_raises(self): + with pytest.raises(CommandError, match="--file"): + call_command("dns-export-zone", "--view", "default", "--zone", "z.test") + + def test_nonexistent_zone_raises(self, make_view): + make_view(name="default") + with pytest.raises(CommandError, match="Failed to export zone"): + call_command( + "dns-export-zone", + "--view", "default", + "--zone", "does-not-exist.test", + "--file", "/tmp/x.zone", + ) + + +@pytest.mark.django_db +class TestDnsExportZonesCommand: + def test_exports_all_zones(self, make_zone, make_view): + make_zone(name="batch1.test") + make_zone(name="batch2.test") + with tempfile.TemporaryDirectory() as td: + out = StringIO() + call_command("dns-export-zones", "--path", td, stdout=out) + output = out.getvalue() + assert "batch1.test" in output + assert "batch2.test" in output + for zone_name in ("batch1.test", "batch2.test"): + for root, dirs, files in os.walk(td): + if zone_name in files: + break + else: + pytest.fail(f"Zone file for {zone_name} not found") + + def test_creates_view_subdirectories(self, make_zone, make_view): + make_zone(name="subdir.test") + with tempfile.TemporaryDirectory() as td: + out = StringIO() + call_command("dns-export-zones", "--path", td, stdout=out) + view_dirs = os.listdir(td) + assert len(view_dirs) > 0 + + def test_missing_path_raises(self): + with pytest.raises(CommandError, match="--path"): + call_command("dns-export-zones") + + def test_no_zones_warning(self): + from netbox_dns.models import Zone + + Zone.objects.all().delete() + with tempfile.TemporaryDirectory() as td: + out = StringIO() + call_command("dns-export-zones", "--path", td, stdout=out) + assert "No zones found" in out.getvalue() + + def test_creates_path_if_not_exists(self, make_zone): + make_zone(name="mkpath.test") + with tempfile.TemporaryDirectory() as td: + path = os.path.join(td, "subdir1", "subdir2") + out = StringIO() + call_command("dns-export-zones", "--path", path, stdout=out) + assert os.path.isdir(path) + + +@pytest.mark.django_db +class TestDnsTransferEndpointCommand: + def _get_cmd_module(self): + """Load the dns-transfer-endpoint command module (hyphen in filename).""" + import sys + from django.core.management import load_command_class + + cmd = load_command_class("netbox_dns_bridge", "dns-transfer-endpoint") + return sys.modules[cmd.__module__] + + def test_missing_tsig_keys_raises(self, monkeypatch): + cmd_mod = self._get_cmd_module() + monkeypatch.setattr(cmd_mod, "SETTINGS", {}) + with pytest.raises(RuntimeError, match="tsig_keys variable not set"): + call_command("dns-transfer-endpoint") + + def test_loads_tsig_keys_and_starts_servers( + self, monkeypatch, make_view + ): + cmd_mod = self._get_cmd_module() + + view = make_view(name="ep-view") + monkeypatch.setattr(cmd_mod, "SETTINGS", { + "tsig_keys": { + "ep-view": { + "keyname": "epkey.", + "secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=", + "algorithm": "hmac-sha256", + } + } + }) + + with patch.object(cmd_mod, "TCPDNSServer") as mock_tcp, \ + patch.object(cmd_mod, "UDPDNSServer") as mock_udp: + mock_tcp.return_value.serve_forever = MagicMock() + mock_udp.return_value.serve_forever = MagicMock() + call_command("dns-transfer-endpoint", "--port", "0", "--address", "127.0.0.1") + assert mock_tcp.called + assert mock_udp.called diff --git a/netbox_dns_bridge/tests/test_models.py b/netbox_dns_bridge/tests/test_models.py new file mode 100644 index 0000000..db45036 --- /dev/null +++ b/netbox_dns_bridge/tests/test_models.py @@ -0,0 +1,138 @@ +"""Tests for plugin models: constraints, __str__, cascade behaviours.""" +import pytest + +from netbox_dns_bridge.models import ( + IntegerKeyValueSetting, + CatalogZoneMemberIdentifier, + SeenTransferClients, +) + + +@pytest.mark.django_db +class TestIntegerKeyValueSetting: + def test_str_representation(self): + obj = IntegerKeyValueSetting.objects.create(key="my-key", value=42) + assert str(obj) == "my-key: 42" + + def test_unique_key_constraint(self): + IntegerKeyValueSetting.objects.create(key="dup-key", value=1) + from django.db import IntegrityError + + with pytest.raises(IntegrityError): + IntegerKeyValueSetting.objects.create(key="dup-key", value=2) + + def test_create_and_retrieve(self): + IntegerKeyValueSetting.objects.create(key="catz-serial", value=99) + obj = IntegerKeyValueSetting.objects.get(key="catz-serial") + assert obj.value == 99 + + +@pytest.mark.django_db +class TestCatalogZoneMemberIdentifier: + def test_str_representation(self, make_zone): + zone = make_zone(name="str.test") + m = CatalogZoneMemberIdentifier.objects.get(zone=zone) + assert str(m) == m.name + + def test_unique_name_constraint(self, make_zone): + zone = make_zone(name="uniq.test") + existing = CatalogZoneMemberIdentifier.objects.get(zone=zone) + existing.name = "shared-name" + existing.save() + + other = make_zone(name="other.test") + other_ident = CatalogZoneMemberIdentifier.objects.get(zone=other) + from django.db import IntegrityError + + with pytest.raises(IntegrityError): + other_ident.name = "shared-name" + other_ident.save() + + def test_cascade_delete_when_zone_deleted(self, make_zone): + zone = make_zone(name="cascade.test") + m = CatalogZoneMemberIdentifier.objects.get(zone=zone) + assert CatalogZoneMemberIdentifier.objects.filter(pk=m.pk).exists() + zone.delete() + assert not CatalogZoneMemberIdentifier.objects.filter(pk=m.pk).exists() + + def test_one_to_one_relationship(self, make_zone): + zone = make_zone(name="oto.test") + m = CatalogZoneMemberIdentifier.objects.get(zone=zone) + assert zone.catz_identifier == m + + def test_ordering_by_name(self, make_zone): + z1 = make_zone(name="z1.test") + z2 = make_zone(name="z2.test") + z3 = make_zone(name="z3.test") + CatalogZoneMemberIdentifier.objects.filter(zone=z3).update(name="gamma") + CatalogZoneMemberIdentifier.objects.filter(zone=z1).update(name="alpha") + CatalogZoneMemberIdentifier.objects.filter(zone=z2).update(name="beta") + names = list( + CatalogZoneMemberIdentifier.objects.filter( + zone__in=[z1, z2, z3] + ).values_list("name", flat=True) + ) + assert names == ["alpha", "beta", "gamma"] + + +@pytest.mark.django_db +class TestSeenTransferClients: + def test_str_representation(self, make_view): + from netbox_dns.models import View + + view = make_view(name="str-view") + client = SeenTransferClients.objects.create( + source_ip="192.0.2.1", + view=view, + last_seen="2026-01-01T00:00:00Z", + ) + assert str(client) == f"192.0.2.1 ({view.name})" + + def test_unique_ip_view_constraint(self, make_view): + from django.db import IntegrityError + + view = make_view(name="uniq-view") + SeenTransferClients.objects.create( + source_ip="10.0.0.1", + view=view, + last_seen="2026-01-01T00:00:00Z", + ) + with pytest.raises(IntegrityError): + SeenTransferClients.objects.create( + source_ip="10.0.0.1", + view=view, + last_seen="2026-01-02T00:00:00Z", + ) + + def test_same_ip_different_views_allowed(self, make_view): + v1 = make_view(name="v1") + v2 = make_view(name="v2") + SeenTransferClients.objects.create( + source_ip="10.0.0.5", view=v1, last_seen="2026-01-01T00:00:00Z" + ) + SeenTransferClients.objects.create( + source_ip="10.0.0.5", view=v2, last_seen="2026-01-01T00:00:00Z" + ) + assert SeenTransferClients.objects.filter(source_ip="10.0.0.5").count() == 2 + + def test_protect_on_view_delete(self, make_view): + from django.db.models import ProtectedError + + view = make_view(name="protected-view") + SeenTransferClients.objects.create( + source_ip="10.0.0.9", + view=view, + last_seen="2026-01-01T00:00:00Z", + ) + with pytest.raises(ProtectedError): + view.delete() + + def test_ordering_by_source_ip(self, make_view): + from django.utils import timezone + + v = make_view(name="ord-view") + now = timezone.now() + for ip in ["10.0.0.3", "10.0.0.1", "10.0.0.2"]: + SeenTransferClients.objects.create(source_ip=ip, view=v, last_seen=now) + ips = list(SeenTransferClients.objects.values_list("source_ip", flat=True)) + assert ips == ["10.0.0.1", "10.0.0.2", "10.0.0.3"] diff --git a/netbox_dns_bridge/tests/test_notify_handler.py b/netbox_dns_bridge/tests/test_notify_handler.py new file mode 100644 index 0000000..5718a7e --- /dev/null +++ b/netbox_dns_bridge/tests/test_notify_handler.py @@ -0,0 +1,321 @@ +"""Tests for notify_handler: _load_tsig_key, _get_soa_serial, _resolve_zone_nameserver_ips, +schedule_client_notify, schedule_ns_notify, and job classes.""" +from unittest.mock import patch, MagicMock + +import dns.name +import dns.rdatatype +import dns.tsig +import pytest + +import netbox_dns_bridge.signals.notify_handler as nh + + +class TestLoadTsigKey: + def _patch_settings(self, monkeypatch, tsig_keys): + monkeypatch.setattr(nh, "SETTINGS", {"tsig_keys": tsig_keys}) + + def test_valid_config(self, monkeypatch): + self._patch_settings(monkeypatch, { + "myview": { + "keyname": "testkey.", + "secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=", + "algorithm": "hmac-sha256", + } + }) + key = nh._load_tsig_key("myview") + assert isinstance(key, dns.tsig.Key) + assert key.name == dns.name.from_text("testkey.", dns.name.root).canonicalize() + + def test_missing_view_raises(self, monkeypatch): + self._patch_settings(monkeypatch, {}) + with pytest.raises(RuntimeError, match="No TSIG key configured"): + nh._load_tsig_key("nope") + + def test_missing_algorithm_raises(self, monkeypatch): + self._patch_settings(monkeypatch, { + "v": {"keyname": "k.", "secret": "s"} + }) + with pytest.raises(RuntimeError, match="Missing 'algorithm'"): + nh._load_tsig_key("v") + + def test_missing_keyname_raises(self, monkeypatch): + self._patch_settings(monkeypatch, { + "v": {"secret": "s", "algorithm": "hmac-sha256"} + }) + with pytest.raises(RuntimeError, match="Incomplete TSIG key"): + nh._load_tsig_key("v") + + def test_missing_secret_raises(self, monkeypatch): + self._patch_settings(monkeypatch, { + "v": {"keyname": "k.", "algorithm": "hmac-sha256"} + }) + with pytest.raises(RuntimeError, match="Incomplete TSIG key"): + nh._load_tsig_key("v") + + def test_relative_key_name_gets_root(self, monkeypatch): + self._patch_settings(monkeypatch, { + "v": { + "keyname": "relkey", + "secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=", + "algorithm": "hmac-sha256", + } + }) + key = nh._load_tsig_key("v") + assert key.name.is_absolute() + + +@pytest.mark.django_db +class TestGetSoaSerial: + def test_valid_soa(self, make_zone): + zone = make_zone(name="soa-valid.test", soa_serial=12345) + result = nh._get_soa_serial(zone) + assert result == 12345 + + def test_no_soa_record_returns_none(self, make_zone): + from netbox_dns.models import Record + + zone = make_zone(name="soa-none.test") + Record.objects.filter(zone=zone, type="SOA").delete() + result = nh._get_soa_serial(zone) + assert result is None + + def test_inactive_soa_returns_none(self, make_zone): + from netbox_dns.models import Record + from netbox_dns.choices import RecordStatusChoices + + zone = make_zone(name="soa-inactive.test") + Record.objects.filter(zone=zone, type="SOA").update( + status=RecordStatusChoices.STATUS_INACTIVE + ) + result = nh._get_soa_serial(zone) + assert result is None + + def test_malformed_serial_returns_none(self, make_zone): + from netbox_dns.models import Record + + zone = make_zone(name="soa-bad.test") + Record.objects.filter(zone=zone, type="SOA").update( + value="ns1.test. hostmaster.test. NOTANUMBER 300 100 1209600 3600" + ) + result = nh._get_soa_serial(zone) + assert result is None + + +@pytest.mark.django_db +class TestResolveZoneNameserverIps: + def test_resolves_a_records(self, make_zone, make_record, make_nameserver): + zone = make_zone(name="nsresolv.test") + ns = make_nameserver("ns1.nsresolv.test") + zone.nameservers.add(ns) + make_record(zone, name="ns1", type_="A", value="192.0.2.1") + ips = nh._resolve_zone_nameserver_ips(zone) + assert "192.0.2.1" in ips + + def test_resolves_aaaa_records(self, make_zone, make_record, make_nameserver): + zone = make_zone(name="nsresolv6.test") + ns = make_nameserver("ns1.nsresolv6.test") + zone.nameservers.add(ns) + make_record(zone, name="ns1", type_="AAAA", value="2001:db8::1") + ips = nh._resolve_zone_nameserver_ips(zone) + assert "2001:db8::1" in ips + + def test_deduplicates_ips(self, make_zone, make_record, make_nameserver): + zone = make_zone(name="nsdedup.test") + ns = make_nameserver("ns1.nsdedup.test") + zone.nameservers.add(ns) + make_record(zone, name="ns1", type_="A", value="192.0.2.1") + make_record(zone, name="ns1", type_="A", value="192.0.2.2") + ips = nh._resolve_zone_nameserver_ips(zone) + assert "192.0.2.1" in ips + assert "192.0.2.2" in ips + assert len(ips) == 2 + + def test_no_nameservers_returns_empty(self, make_zone): + zone = make_zone(name="nsnone.test") + ips = nh._resolve_zone_nameserver_ips(zone) + assert ips == [] + + def test_multiple_nameservers(self, make_zone, make_record, make_nameserver): + zone = make_zone(name="nsmulti.test") + ns1 = make_nameserver("ns1.nsmulti.test") + ns2 = make_nameserver("ns2.nsmulti.test") + zone.nameservers.add(ns1, ns2) + make_record(zone, name="ns1", type_="A", value="192.0.2.1") + make_record(zone, name="ns2", type_="A", value="192.0.2.2") + ips = nh._resolve_zone_nameserver_ips(zone) + assert set(ips) == {"192.0.2.1", "192.0.2.2"} + + def test_trailing_dot_resilience(self, make_zone, make_record, make_nameserver): + zone = make_zone(name="nstrail.test") + ns = make_nameserver("ns1.nstrail.test") + zone.nameservers.add(ns) + make_record(zone, name="ns1", type_="A", value="192.0.2.5") + ips = nh._resolve_zone_nameserver_ips(zone) + assert "192.0.2.5" in ips + + +@pytest.mark.django_db +class TestScheduleClientNotify: + def test_enqueues_job_when_soa_exists(self, make_zone): + zone = make_zone(name="sched-client.test", soa_serial=77) + with patch.object(nh.SendClientDNSNotify, "enqueue") as mock_enq: + nh.schedule_client_notify(zone) + mock_enq.assert_called_once_with( + zone_name="sched-client.test", + view_name=zone.view.name, + soa_serial=77, + ) + + def test_does_not_enqueue_when_no_soa(self, make_zone): + from netbox_dns.models import Record + + zone = make_zone(name="sched-client-nosoa.test") + Record.objects.filter(zone=zone, type="SOA").delete() + with patch.object(nh.SendClientDNSNotify, "enqueue") as mock_enq: + nh.schedule_client_notify(zone) + mock_enq.assert_not_called() + + +@pytest.mark.django_db +class TestScheduleNsNotify: + def test_enqueues_job_when_soa_exists(self, make_zone): + zone = make_zone(name="sched-ns.test", soa_serial=88) + with patch.object(nh.SendNSNotify, "enqueue") as mock_enq: + nh.schedule_ns_notify(zone) + mock_enq.assert_called_once_with( + zone_id=zone.pk, + zone_name="sched-ns.test", + view_name=zone.view.name, + soa_serial=88, + ) + + def test_does_not_enqueue_when_no_soa(self, make_zone): + from netbox_dns.models import Record + + zone = make_zone(name="sched-ns-nosoa.test") + Record.objects.filter(zone=zone, type="SOA").delete() + with patch.object(nh.SendNSNotify, "enqueue") as mock_enq: + nh.schedule_ns_notify(zone) + mock_enq.assert_not_called() + + +@pytest.mark.django_db +class TestSendClientDNSNotifyRun: + def test_sends_notify_to_seen_clients(self, make_zone, make_view, monkeypatch): + import netbox_dns_bridge.signals.notify_handler as nh + from netbox_dns_bridge.models import SeenTransferClients, IntegerKeyValueSetting + from django.utils import timezone + + view = make_view(name="notify-run-view") + IntegerKeyValueSetting.objects.update_or_create( + key="catalog-zone-soa-serial", defaults={"value": 42} + ) + SeenTransferClients.objects.create( + source_ip="192.0.2.99", view=view, last_seen=timezone.now() + ) + + monkeypatch.setattr(nh, "SETTINGS", { + "notify_clients": True, + "tsig_keys": { + "notify-run-view": { + "keyname": "nk.", + "secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=", + "algorithm": "hmac-sha256", + } + }, + "notify_over_tcp": False, + "notify_client_port": 53, + "notify_client_alive_threshold_hours": 24, + }) + + job = MagicMock() + job.run.__wrapped__ = nh.SendClientDNSNotify.run + with patch("dns.query.udp") as mock_udp: + nh.SendClientDNSNotify.run(job, + zone_name="z.test", soa_serial=10, view_name="notify-run-view") + assert mock_udp.call_count >= 3 # zone + catz + view.catz + + def test_handles_missing_serial_setting(self, monkeypatch, make_view): + # When the IntegerKeyValueSetting for catz serial doesn't exist, + # the job should log an error and not send any notifies + view = make_view(name="missing-serial-view") + + monkeypatch.setattr(nh, "SETTINGS", { + "notify_clients": True, + "tsig_keys": { + "missing-serial-view": { + "keyname": "nk.", + "secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=", + "algorithm": "hmac-sha256", + } + }, + "notify_over_tcp": False, + "notify_client_port": 53, + "notify_client_alive_threshold_hours": 24, + }) + + job = MagicMock() + with patch("dns.query.udp") as mock_udp: + nh.SendClientDNSNotify.run(job, + zone_name="z.test", soa_serial=1, view_name="missing-serial-view") + mock_udp.assert_not_called() + + +@pytest.mark.django_db +class TestSendNSNotifyRun: + def test_sends_notify_to_nameserver_ips(self, make_zone, make_record, make_nameserver, monkeypatch): + import netbox_dns_bridge.signals.notify_handler as nh + + zone = make_zone(name="ns-notify-run.test", soa_serial=55) + ns = make_nameserver("ns1.ns-notify-run.test") + zone.nameservers.add(ns) + make_record(zone, name="ns1", type_="A", value="192.0.2.200") + + monkeypatch.setattr(nh, "SETTINGS", { + "tsig_keys": { + zone.view.name: { + "keyname": "nk.", + "secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=", + "algorithm": "hmac-sha256", + } + }, + "notify_over_tcp": False, + "notify_ns_port": 53, + }) + + job = MagicMock() + with patch("dns.query.udp") as mock_udp: + nh.SendNSNotify.run(job, + zone_id=zone.pk, zone_name="ns-notify-run.test", + view_name=zone.view.name, soa_serial=55) + mock_udp.assert_called_once() + + def test_skips_when_zone_deleted(self, monkeypatch): + job = MagicMock() + with patch("dns.query.udp") as mock_udp: + nh.SendNSNotify.run(job, + zone_id=999999, zone_name="gone.test", + view_name="v", soa_serial=1) + mock_udp.assert_not_called() + + def test_skips_when_no_nameservers(self, make_zone, monkeypatch): + zone = make_zone(name="ns-notify-empty.test") + + monkeypatch.setattr(nh, "SETTINGS", { + "tsig_keys": { + zone.view.name: { + "keyname": "nk.", + "secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=", + "algorithm": "hmac-sha256", + } + }, + "notify_over_tcp": False, + "notify_ns_port": 53, + }) + + job = MagicMock() + with patch("dns.query.udp") as mock_udp: + nh.SendNSNotify.run(job, + zone_id=zone.pk, zone_name="ns-notify-empty.test", + view_name=zone.view.name, soa_serial=1) + mock_udp.assert_not_called() diff --git a/netbox_dns_bridge/tests/test_request_handler_integration.py b/netbox_dns_bridge/tests/test_request_handler_integration.py new file mode 100644 index 0000000..c5c317f --- /dev/null +++ b/netbox_dns_bridge/tests/test_request_handler_integration.py @@ -0,0 +1,190 @@ +"""Real-socket integration tests for the DNS transfer endpoint. + +Starts actual TCP+UDP servers on 127.0.0.1:0 and sends real DNS queries via +raw sockets against NetBox DNS zones in the test database. Raw sockets are used +so response TSIG can be verified explicitly with the request MAC. +""" +from __future__ import annotations + +import dns.flags +import dns.message +import dns.name +import dns.query +import dns.rcode +import dns.rdatatype +import pytest + +from netbox_dns_bridge.tests._helpers import ( + make_signed_query, + make_unsigned_query, + parse_response, + parse_response_strict, + query_mac, + tcp_exchange, + udp_exchange, + build_keyring, + xfr_rrsets, +) + + +pytestmark = pytest.mark.django_db(transaction=True) + + +def _signed_soa(zone_name, tsig_key): + return make_signed_query(zone_name, dns.rdatatype.SOA, tsig_key) + + +def _signed_axfr(zone_name, tsig_key): + return make_signed_query(zone_name, dns.rdatatype.AXFR, tsig_key) + + +class TestSOAIntegration: + def test_soa_over_tcp(self, dns_server, make_zone, make_view, tsig_key): + view = make_view() + make_zone(name="int-soa.test", view=view, soa_serial=555) + srv = dns_server.start(views={"testkey.": view}) + keyring = build_keyring(tsig_key) + + wire = _signed_soa("int-soa.test", tsig_key) + resp_wire = tcp_exchange(wire, srv["host"], srv["port"]) + resp = parse_response_strict(resp_wire, keyring, query_mac(wire, keyring)) + assert resp.rcode() == dns.rcode.NOERROR + assert resp.flags & dns.flags.AA + assert resp.answer[0].rdtype == dns.rdatatype.SOA + assert resp.answer[0][0].serial == 555 + assert resp.had_tsig + + def test_soa_over_udp(self, dns_server, make_zone, make_view, tsig_key): + view = make_view() + make_zone(name="int-udp.test", view=view, soa_serial=7) + srv = dns_server.start(views={"testkey.": view}) + keyring = build_keyring(tsig_key) + + wire = _signed_soa("int-udp.test", tsig_key) + resp_wire = udp_exchange(wire, srv["host"], srv["port"]) + resp = parse_response_strict(resp_wire, keyring, query_mac(wire, keyring)) + assert resp.rcode() == dns.rcode.NOERROR + assert resp.answer[0][0].serial == 7 + + def test_unsigned_query_refused(self, dns_server, make_zone, make_view, tsig_key): + view = make_view() + make_zone(name="int-nosig.test", view=view) + srv = dns_server.start(views={"testkey.": view}) + + resp_wire = tcp_exchange( + make_unsigned_query("int-nosig.test", dns.rdatatype.SOA), + srv["host"], srv["port"], + ) + resp = parse_response(resp_wire) + assert resp.rcode() == dns.rcode.REFUSED + + +class TestAXFRIntegration: + def test_axfr_over_tcp(self, dns_server, make_zone, make_view, make_record, tsig_key): + view = make_view() + zone = make_zone(name="int-axfr.test", view=view, soa_serial=123) + make_record(zone, name="www", type_="A", value="192.0.2.10") + make_record(zone, name="api", type_="A", value="192.0.2.11") + srv = dns_server.start(views={"testkey.": view}) + + records = xfr_rrsets( + dns.query.xfr( + srv["host"], "int-axfr.test", port=srv["port"], + keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name, + lifetime=5, relativize=False, + ) + ) + names = {r.name.to_text() for r in records} + assert "int-axfr.test." in names + assert "www.int-axfr.test." in names + assert "api.int-axfr.test." in names + # AXFR must begin and end with SOA. + assert records[0].rdtype == dns.rdatatype.SOA + assert records[-1].rdtype == dns.rdatatype.SOA + assert records[0][0].serial == 123 + + def test_axfr_over_udp_refused(self, dns_server, make_zone, make_view, tsig_key): + view = make_view() + make_zone(name="int-axfr-udp.test", view=view) + srv = dns_server.start(views={"testkey.": view}) + + resp_wire = udp_exchange( + _signed_axfr("int-axfr-udp.test", tsig_key), srv["host"], srv["port"] + ) + resp = parse_response(resp_wire) + assert resp.rcode() == dns.rcode.REFUSED + + def test_axfr_unknown_zone_refused(self, dns_server, make_view, tsig_key): + view = make_view() + srv = dns_server.start(views={"testkey.": view}) + + resp_wire = tcp_exchange( + _signed_axfr("does-not-exist.test", tsig_key), srv["host"], srv["port"] + ) + resp = parse_response(resp_wire) + assert resp.rcode() == dns.rcode.REFUSED + + +class TestCatzIntegration: + def test_catz_axfr_contains_member_ptr(self, dns_server, make_zone, make_view, tsig_key): + view = make_view() + make_zone(name="member-a.test", view=view) + srv = dns_server.start(views={"testkey.": view}) + + records = xfr_rrsets( + dns.query.xfr( + srv["host"], "catz", port=srv["port"], + keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name, + lifetime=5, relativize=False, + ) + ) + types = {r.rdtype for r in records} + assert dns.rdatatype.SOA in types + assert dns.rdatatype.PTR in types + ptrs = [r for r in records if r.rdtype == dns.rdatatype.PTR] + targets = {str(r[0].target) for r in ptrs} + assert any("member-a.test" in t for t in targets) + + +class TestSeenClientTracking: + def _wait_for_seen_client(self, view, timeout=3.0): + """Poll for SeenTransferClients record (race: response is sent before + _track_seen_client completes in the handler thread).""" + import time + from netbox_dns_bridge.models import SeenTransferClients + + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if SeenTransferClients.objects.filter( + source_ip="127.0.0.1", view=view + ).exists(): + return True + time.sleep(0.05) + return False + + def test_seen_client_recorded_on_success( + self, dns_server, make_zone, make_view, tsig_key, patch_settings + ): + patch_settings({"notify_clients": True}) + view = make_view() + make_zone(name="seen.test", view=view) + srv = dns_server.start(views={"testkey.": view}) + + tcp_exchange(_signed_soa("seen.test", tsig_key), srv["host"], srv["port"]) + + assert self._wait_for_seen_client(view) + + def test_no_tracking_when_notify_clients_false( + self, dns_server, make_zone, make_view, tsig_key, patch_settings + ): + patch_settings({"notify_clients": False}) + view = make_view() + make_zone(name="not-seen.test", view=view) + srv = dns_server.start(views={"testkey.": view}) + + tcp_exchange(_signed_soa("not-seen.test", tsig_key), srv["host"], srv["port"]) + + import time + time.sleep(0.3) + from netbox_dns_bridge.models import SeenTransferClients + assert not SeenTransferClients.objects.filter(view=view).exists() diff --git a/netbox_dns_bridge/tests/test_request_handler_unit.py b/netbox_dns_bridge/tests/test_request_handler_unit.py new file mode 100644 index 0000000..7621684 --- /dev/null +++ b/netbox_dns_bridge/tests/test_request_handler_unit.py @@ -0,0 +1,358 @@ +"""Unit tests for the DNS transfer request handler. + +These exercise routing, denial gates, SOA/AXFR response construction and TSIG +handling by feeding crafted wire bytes to ``_handle_dns_query`` and capturing +emitted responses. Zone lookups are mocked so no database is needed for the +routing/response logic; the integration suite covers the real-socket path. +""" +from types import SimpleNamespace +from unittest.mock import patch + +import dns.flags +import dns.message +import dns.name +import dns.rcode +import dns.rdatatype + +import netbox_dns_bridge.management.commands.dns_transfer_endpoint.request_handler as rh +from netbox_dns_bridge.tests._helpers import ( + CaptureTCPHandler, + CaptureUDPHandler, + make_dns_zone, + make_signed_query, + make_tsig_key, + make_unsigned_query, + parse_response, + parse_response_strict, + query_mac, + build_keyring, + build_tsig_view_map, +) + + +# --------------------------------------------------------------------------- +# Fixtures local to this module +# --------------------------------------------------------------------------- + +import pytest + + +@pytest.fixture +def view(): + return SimpleNamespace(name="default") + + +@pytest.fixture +def keyring(tsig_key): + return build_keyring(tsig_key) + + +@pytest.fixture +def view_map(tsig_key, view): + return build_tsig_view_map("testkey.", view) + + +@pytest.fixture +def tcp_handler(capture_handler, keyring, view_map): + """Return a factory building a fresh TCP capture handler + captured list.""" + def _make(client_ip="127.0.0.1"): + return capture_handler(CaptureTCPHandler, keyring, view_map, client_ip) + return _make + + +def _patch_regular_zone(monkeypatch, handler, zone): + handler._get_zone_from_nb = lambda name, view_name: zone + + +def _patch_catz(monkeypatch, zone): + monkeypatch.setattr(rh.catzm, "create_zone", lambda name, view_name: zone) + + +# --------------------------------------------------------------------------- +# Denial / routing gates +# --------------------------------------------------------------------------- + +class TestRoutingGates: + def test_no_tsig_refused(self, tcp_handler, tsig_key): + handler, captured = tcp_handler() + wire = make_unsigned_query("example.test", dns.rdatatype.SOA) + handler._handle_dns_query(wire) + assert len(captured) == 1 + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + def test_question_count_zero_refused(self, tcp_handler, tsig_key): + handler, captured = tcp_handler() + msg = dns.message.make_query("example.test", dns.rdatatype.SOA) + msg.question = [] + msg.use_tsig(keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name) + handler._handle_dns_query(msg.to_wire()) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + def test_unsupported_qtype_refused(self, tcp_handler, tsig_key): + handler, captured = tcp_handler() + wire = make_signed_query("example.test", dns.rdatatype.A, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + def test_ixfr_not_supported_refused(self, tcp_handler, tsig_key): + handler, captured = tcp_handler() + wire = make_signed_query("example.test", dns.rdatatype.IXFR, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + def test_axfr_over_udp_refused(self, capture_handler, keyring, view_map, tsig_key): + handler, captured = capture_handler( + CaptureUDPHandler, keyring, view_map + ) + wire = make_signed_query("example.test", dns.rdatatype.AXFR, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + def test_key_not_mapped_to_view_refused(self, capture_handler, view): + other_key = make_tsig_key("unknownkey.") + keyring = build_keyring(other_key) + view_map = {} # no mapping + handler, captured = capture_handler(CaptureTCPHandler, keyring, view_map) + wire = make_signed_query("example.test", dns.rdatatype.SOA, other_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + def test_zone_not_found_refused_not_nxdomain(self, tcp_handler, tsig_key, monkeypatch): + handler, captured = tcp_handler() + handler._get_zone_from_nb = lambda name, view_name: None + wire = make_signed_query("missing.test", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + assert resp.rcode() != dns.rcode.NXDOMAIN + + def test_malformed_query_silent_drop(self, tcp_handler): + handler, captured = tcp_handler() + handler._handle_dns_query(b"\x00\x00not a dns message") + assert captured == [] + + def test_zone_without_soa_silent_drop(self, tcp_handler, tsig_key, monkeypatch): + import dns.zone + import dns.name + zone = dns.zone.Zone(dns.name.from_text("empty.test", dns.name.root)) + handler, captured = tcp_handler() + handler._get_zone_from_nb = lambda name, view_name: zone + wire = make_signed_query("empty.test", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + assert captured == [] + + +# --------------------------------------------------------------------------- +# TSIG error paths +# --------------------------------------------------------------------------- + +class TestTSIGErrors: + def test_bad_signature_refused_with_badsig(self, tcp_handler, tsig_key): + handler, captured = tcp_handler() + wire = make_signed_query("example.test", dns.rdatatype.SOA, tsig_key) + # Flip a byte in the TSIG MAC region to break the signature. + tampered = bytearray(wire) + tampered[-5] ^= 0xFF + handler._handle_dns_query(bytes(tampered)) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + def test_unknown_key_refused_with_badkey(self, tcp_handler): + handler, captured = tcp_handler() + unknown = make_tsig_key("nopekey.") + wire = make_signed_query("example.test", dns.rdatatype.SOA, unknown) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + +# --------------------------------------------------------------------------- +# SOA responses +# --------------------------------------------------------------------------- + +class TestSOAResponse: + def test_soa_regular_zone(self, tcp_handler, tsig_key, keyring, monkeypatch): + zone = make_dns_zone("example.test", soa_serial=42) + handler, captured = tcp_handler() + handler._get_zone_from_nb = lambda name, view_name: zone + wire = make_signed_query("example.test", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + + resp = parse_response_strict(captured[0], keyring, query_mac(wire, keyring)) + assert resp.rcode() == dns.rcode.NOERROR + assert resp.flags & dns.flags.AA + assert resp.flags & dns.flags.QR + assert len(resp.answer) == 1 + rrset = resp.answer[0] + assert rrset.rdtype == dns.rdatatype.SOA + assert rrset[0].serial == 42 + assert resp.had_tsig + + def test_soa_for_catz(self, tcp_handler, tsig_key, monkeypatch): + zone = make_dns_zone("catz", soa_serial=99) + _patch_catz(monkeypatch, zone) + handler, captured = tcp_handler() + wire = make_signed_query("catz", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.NOERROR + assert resp.answer[0].rdtype == dns.rdatatype.SOA + assert resp.answer[0][0].serial == 99 + + def test_soa_for_view_catz_case_insensitive(self, tcp_handler, tsig_key, monkeypatch): + zone = make_dns_zone("default.catz", soa_serial=99) + _patch_catz(monkeypatch, zone) + handler, captured = tcp_handler() + wire = make_signed_query("DEFAULT.CATZ", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.NOERROR + + def test_soa_catz_case_insensitive(self, tcp_handler, tsig_key, monkeypatch): + zone = make_dns_zone("catz", soa_serial=99) + _patch_catz(monkeypatch, zone) + handler, captured = tcp_handler() + wire = make_signed_query("CATZ", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.NOERROR + + def test_soa_tsig_signed(self, tcp_handler, tsig_key, keyring, monkeypatch): + zone = make_dns_zone("example.test", soa_serial=1) + handler, captured = tcp_handler() + handler._get_zone_from_nb = lambda name, view_name: zone + wire = make_signed_query("example.test", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response_strict(captured[0], keyring, query_mac(wire, keyring)) + assert resp.had_tsig + assert resp.keyname == tsig_key.name + + +# --------------------------------------------------------------------------- +# AXFR responses +# --------------------------------------------------------------------------- + +class TestAXFRResponse: + def test_axfr_single_message_framing(self, tcp_handler, tsig_key, keyring, monkeypatch): + zone = make_dns_zone( + "example.test", + soa_serial=7, + extra=[("www", "A", 300, "192.0.2.1"), ("mail", "A", 300, "192.0.2.2")], + ) + handler, captured = tcp_handler() + handler._get_zone_from_nb = lambda name, view_name: zone + wire = make_signed_query("example.test", dns.rdatatype.AXFR, tsig_key) + handler._handle_dns_query(wire) + + # Single message expected (small zone). + assert len(captured) == 1 + resp = parse_response_strict( + captured[0], keyring, query_mac(wire, keyring), one_rr_per_rrset=True + ) + assert resp.rcode() == dns.rcode.NOERROR + assert resp.flags & dns.flags.AA + assert resp.flags & dns.flags.QR + assert resp.had_tsig + + # First and last answer rrsets are SOA (opening + closing). + assert resp.answer[0].rdtype == dns.rdatatype.SOA + assert resp.answer[-1].rdtype == dns.rdatatype.SOA + assert resp.answer[0][0].serial == 7 + # Middle records present. + types = {rr.rdtype for rr in resp.answer} + assert dns.rdatatype.A in types + + def test_axfr_multi_message_when_large(self, tcp_handler, tsig_key, monkeypatch): + # Build a zone with many records, then shrink MAX_WIRE to force splits. + extras = [(f"host{i}", "A", 300, f"192.0.2.{i % 255}") for i in range(200)] + zone = make_dns_zone("big.test", soa_serial=1, extra=extras) + handler, captured = tcp_handler() + handler.MAX_WIRE = 800 + handler.RESERVED_TSIG = 300 + handler._get_zone_from_nb = lambda name, view_name: zone + wire = make_signed_query("big.test", dns.rdatatype.AXFR, tsig_key) + handler._handle_dns_query(wire) + + assert len(captured) >= 2 + # First message opens with SOA, last message closes with SOA. + first = parse_response(captured[0], one_rr_per_rrset=True) + last = parse_response(captured[-1], one_rr_per_rrset=True) + assert first.answer[0].rdtype == dns.rdatatype.SOA + assert last.answer[-1].rdtype == dns.rdatatype.SOA + + def test_axfr_over_udp_refused(self, capture_handler, keyring, view_map, tsig_key): + handler, captured = capture_handler(CaptureUDPHandler, keyring, view_map) + wire = make_signed_query("example.test", dns.rdatatype.AXFR, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + + def test_axfr_zone_without_soa_silent_drop(self, tcp_handler, tsig_key, monkeypatch): + import dns.zone + import dns.name + zone = dns.zone.Zone(dns.name.from_text("nosoa.test", dns.name.root)) + handler, captured = tcp_handler() + handler._get_zone_from_nb = lambda name, view_name: zone + wire = make_signed_query("nosoa.test", dns.rdatatype.AXFR, tsig_key) + handler._handle_dns_query(wire) + assert captured == [] + + def test_axfr_tsig_chain_uses_query_mac_first(self, tcp_handler, tsig_key, keyring, monkeypatch): + extras = [(f"host{i}", "A", 300, f"192.0.2.{i % 255}") for i in range(120)] + zone = make_dns_zone("chain.test", soa_serial=1, extra=extras) + handler, captured = tcp_handler() + handler.MAX_WIRE = 800 + handler.RESERVED_TSIG = 300 + handler._get_zone_from_nb = lambda name, view_name: zone + wire = make_signed_query("chain.test", dns.rdatatype.AXFR, tsig_key) + handler._handle_dns_query(query_mac_wire := wire) + # At least two messages produced. + assert len(captured) >= 2 + # First message verifies against the request MAC and opens with SOA. + first = parse_response_strict( + captured[0], keyring, query_mac(wire, keyring), one_rr_per_rrset=True + ) + assert first.had_tsig + assert first.answer[0].rdtype == dns.rdatatype.SOA + + +# --------------------------------------------------------------------------- +# Routing: case sensitivity, trailing dot +# --------------------------------------------------------------------------- + +class TestRoutingDetails: + def test_regular_zone_case_sensitive(self, tcp_handler, tsig_key): + zone = make_dns_zone("example.test", soa_serial=1) + handler, captured = tcp_handler() + calls = [] + + def _lookup(name, view_name): + calls.append(name) + return zone if name == "example.test" else None + + handler._get_zone_from_nb = _lookup + wire = make_signed_query("Example.TEST", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.REFUSED + assert calls == ["Example.TEST"] + + def test_trailing_dot_normalized(self, tcp_handler, tsig_key): + zone = make_dns_zone("example.test", soa_serial=1) + handler, captured = tcp_handler() + seen = [] + handler._get_zone_from_nb = lambda name, view_name: ( + seen.append(name) or zone + ) + wire = make_signed_query("example.test.", dns.rdatatype.SOA, tsig_key) + handler._handle_dns_query(wire) + resp = parse_response(captured[0]) + assert resp.rcode() == dns.rcode.NOERROR + assert seen == ["example.test"] diff --git a/netbox_dns_bridge/tests/test_signals.py b/netbox_dns_bridge/tests/test_signals.py new file mode 100644 index 0000000..6e2ae01 --- /dev/null +++ b/netbox_dns_bridge/tests/test_signals.py @@ -0,0 +1,228 @@ +"""Tests for signal handlers: zone_pre_save, zone_post_save, zone_post_delete. + +Uses ``captureOnCommitCallbacks(execute=True)`` to fire ``transaction.on_commit`` +callbacks inside the transaction-wrapped ``@pytest.mark.django_db`` test scope. +""" +from unittest.mock import patch, MagicMock + +import pytest +from django.test.testcases import TestCase as _DjangoTestCase + +# captureOnCommitCallbacks is a classmethod on TestCase in Django 6.x +captureOnCommitCallbacks = _DjangoTestCase.captureOnCommitCallbacks + +from netbox_dns_bridge.models import ( + IntegerKeyValueSetting, + CatalogZoneMemberIdentifier, +) + + +@pytest.mark.django_db +class TestZonePreSave: + def test_stores_old_name_and_serial_on_existing_zone(self, make_zone): + zone = make_zone(name="presave.test", soa_serial=100) + zone.refresh_from_db() + zone.soa_serial = 200 + zone.save() + assert getattr(zone, "_old_name", None) == "presave.test" + assert getattr(zone, "_old_soa_serial", None) == 100 + + def test_old_attrs_none_on_new_zone(self, make_zone): + zone = make_zone(name="newzone.test") + assert getattr(zone, "_old_name", None) is None + assert getattr(zone, "_old_soa_serial", None) is None + + +@pytest.mark.django_db +class TestZonePostSaveCreate: + def test_creates_catz_member_identifier_on_new_zone(self, make_zone): + zone = make_zone(name="postcreate.test") + assert CatalogZoneMemberIdentifier.objects.filter(zone=zone).exists() + + def test_increments_soa_serial_on_commit(self, make_zone, catz_serial): + with captureOnCommitCallbacks(execute=True): + zone = make_zone(name="serial-inc.test") + obj = IntegerKeyValueSetting.objects.get(key="catalog-zone-soa-serial") + assert obj.value == catz_serial + 1 + + def test_does_not_increment_without_on_commit(self, make_zone, catz_serial): + make_zone(name="no-commit.test") + obj = IntegerKeyValueSetting.objects.get(key="catalog-zone-soa-serial") + assert obj.value == catz_serial + + def test_schedule_client_notify_called_when_enabled(self, make_zone, patch_settings): + patch_settings({"notify_clients": True}) + with patch( + "netbox_dns_bridge.signals.netbox_dns_zones.notify_handler.schedule_client_notify" + ) as mock_notify: + with patch( + "netbox_dns_bridge.signals.netbox_dns_zones.notify_handler.schedule_ns_notify" + ): + with captureOnCommitCallbacks(execute=True): + make_zone(name="notify.test") + mock_notify.assert_called_once() + + def test_schedule_ns_notify_called_when_serial_changed(self, make_zone, patch_settings): + patch_settings({ + "notify_clients": False, + "notify_ns_all_zones": True, + }) + with patch( + "netbox_dns_bridge.signals.netbox_dns_zones.notify_handler.schedule_ns_notify" + ) as mock_ns_notify: + with captureOnCommitCallbacks(execute=True): + make_zone(name="nsnotify.test") + mock_ns_notify.assert_called_once() + + def test_schedule_ns_notify_not_called_when_disabled(self, make_zone, patch_settings): + patch_settings({ + "notify_clients": False, + "notify_ns_all_zones": False, + }) + with patch( + "netbox_dns_bridge.signals.netbox_dns_zones.notify_handler.schedule_ns_notify" + ) as mock_ns_notify: + with captureOnCommitCallbacks(execute=True): + make_zone(name="nonsnotify.test") + mock_ns_notify.assert_not_called() + + +@pytest.mark.django_db +class TestZonePostSaveUpdate: + def test_updates_catz_identifier_on_name_change(self, make_zone): + zone = make_zone(name="rename-old.test") + old_name = CatalogZoneMemberIdentifier.objects.get(zone=zone).name + + zone.name = "rename-new.test" + zone.save() + + new_name = CatalogZoneMemberIdentifier.objects.get(zone=zone).name + assert old_name != new_name + + def test_no_identifier_update_when_name_unchanged(self, make_zone): + zone = make_zone(name="nochange.test") + ident = CatalogZoneMemberIdentifier.objects.get(zone=zone) + original_name = ident.name + + zone.soa_serial = zone.soa_serial + 1 + zone.save() + ident.refresh_from_db() + assert ident.name == original_name + + def test_increments_serial_on_commit_on_update(self, make_zone, catz_serial): + zone = make_zone(name="update-serial.test") + with captureOnCommitCallbacks(execute=True): + zone.soa_serial = zone.soa_serial + 1 + zone.save() + obj = IntegerKeyValueSetting.objects.get(key="catalog-zone-soa-serial") + assert obj.value == catz_serial + 1 + + +@pytest.mark.django_db +class TestZonePostDelete: + def test_increments_soa_serial_on_commit_on_delete(self, make_zone, catz_serial): + zone = make_zone(name="delete-serial.test") + with captureOnCommitCallbacks(execute=True): + zone.delete() + obj = IntegerKeyValueSetting.objects.get(key="catalog-zone-soa-serial") + assert obj.value == catz_serial + 1 + + def test_does_not_increment_without_on_commit(self, make_zone, catz_serial): + zone = make_zone(name="delete-no-commit.test") + zone.delete() + obj = IntegerKeyValueSetting.objects.get(key="catalog-zone-soa-serial") + assert obj.value == catz_serial + + +@pytest.mark.django_db +class TestSignalDedup: + def test_dedup_skips_second_save_in_same_request(self, make_zone): + from netbox.context import current_request + from django.contrib.auth import get_user_model + + zone = make_zone(name="dedup.test") + original_serial = IntegerKeyValueSetting.objects.get( + key="catalog-zone-soa-serial" + ).value + + User = get_user_model() + user = User(username="dedup-user", is_superuser=True) + user.save() + + request = MagicMock() + request._handled_zones = set() + request.user = user + import uuid as _uuid + request.id = _uuid.uuid4() + + with current_request.set(request): + with captureOnCommitCallbacks(execute=True): + zone.soa_serial = zone.soa_serial + 1 + zone.save() + with captureOnCommitCallbacks(execute=True): + zone.soa_serial = zone.soa_serial + 2 + zone.save() + + new_serial = IntegerKeyValueSetting.objects.get( + key="catalog-zone-soa-serial" + ).value + assert new_serial == original_serial + 1 + + def test_no_dedup_without_request(self, make_zone): + zone = make_zone(name="nodedup.test") + original_serial = IntegerKeyValueSetting.objects.get( + key="catalog-zone-soa-serial" + ).value + + with captureOnCommitCallbacks(execute=True): + zone.soa_serial = zone.soa_serial + 1 + zone.save() + with captureOnCommitCallbacks(execute=True): + zone.soa_serial = zone.soa_serial + 2 + zone.save() + + new_serial = IntegerKeyValueSetting.objects.get( + key="catalog-zone-soa-serial" + ).value + assert new_serial == original_serial + 2 + + +@pytest.mark.django_db +class TestNotifyNsEnabled: + def test_enabled_by_all_zones_flag(self, make_zone, patch_settings): + patch_settings({"notify_ns_all_zones": True}) + from netbox_dns_bridge.signals.netbox_dns_zones import _notify_ns_enabled + + zone = make_zone(name="nsflag.test") + assert _notify_ns_enabled(zone) is True + + def test_disabled_by_default(self, make_zone, patch_settings): + patch_settings({}) + from netbox_dns_bridge.signals.netbox_dns_zones import _notify_ns_enabled + + zone = make_zone(name="nsdisabled.test") + assert _notify_ns_enabled(zone) is False + + def test_enabled_by_custom_field(self, make_zone, patch_settings): + patch_settings({"notify_ns_custom_field_name": "enable_notify"}) + from netbox_dns_bridge.signals.netbox_dns_zones import _notify_ns_enabled + + zone = make_zone(name="nscf.test") + zone.custom_field_data = {"enable_notify": True} + assert _notify_ns_enabled(zone) is True + + def test_disabled_when_custom_field_false(self, make_zone, patch_settings): + patch_settings({"notify_ns_custom_field_name": "enable_notify"}) + from netbox_dns_bridge.signals.netbox_dns_zones import _notify_ns_enabled + + zone = make_zone(name="nscf-false.test") + zone.custom_field_data = {"enable_notify": False} + assert _notify_ns_enabled(zone) is False + + def test_disabled_when_custom_field_missing(self, make_zone, patch_settings): + patch_settings({"notify_ns_custom_field_name": "enable_notify"}) + from netbox_dns_bridge.signals.netbox_dns_zones import _notify_ns_enabled + + zone = make_zone(name="nscf-missing.test") + zone.custom_field_data = {} + assert _notify_ns_enabled(zone) is False diff --git a/netbox_dns_bridge/tests/test_smoke.py b/netbox_dns_bridge/tests/test_smoke.py new file mode 100644 index 0000000..9ef5abc --- /dev/null +++ b/netbox_dns_bridge/tests/test_smoke.py @@ -0,0 +1,40 @@ +"""Smoke test: verify the harness loads under the full NetBox stack.""" +import django + + +def test_django_loaded(): + assert django.VERSION[0] >= 5 + + +def test_plugin_importable(): + import netbox_dns_bridge # noqa: F401 + assert netbox_dns_bridge.__version__ + + +def test_models_importable(): + from netbox_dns_bridge.models import ( + IntegerKeyValueSetting, + CatalogZoneMemberIdentifier, + SeenTransferClients, + ) + assert IntegerKeyValueSetting._meta.model_name + assert CatalogZoneMemberIdentifier._meta.model_name + assert SeenTransferClients._meta.model_name + + +def test_request_handler_importable(): + from netbox_dns_bridge.management.commands.dns_transfer_endpoint import ( # noqa: F401 + request_handler, + catalog_zone_manager, + dns_server, + ) + + +def test_helpers_build_dns_zone(): + from netbox_dns_bridge.tests._helpers import make_dns_zone + + zone = make_dns_zone("example.test", soa_serial=42) + import dns.rdatatype + rds = zone.get_rdataset(zone.origin, dns.rdatatype.SOA) + assert rds is not None + assert rds[0].serial == 42 diff --git a/netbox_dns_bridge/tests/test_utils.py b/netbox_dns_bridge/tests/test_utils.py new file mode 100644 index 0000000..5f7c806 --- /dev/null +++ b/netbox_dns_bridge/tests/test_utils.py @@ -0,0 +1,209 @@ +"""Tests for netbox_dns_bridge.utils: get_logger, build_dns_zone, export_bind_zone_file.""" +import os +import tempfile + +import dns.name +import dns.rdatatype +import dns.rdataclass +import pytest + +from netbox_dns_bridge.utils import get_logger, build_dns_zone, export_bind_zone_file + + +class TestGetLogger: + def test_returns_dns_bridge_logger(self): + logger = get_logger("some.module") + assert logger.name == "dns-bridge" + + def test_ignores_name_argument(self): + a = get_logger("module.a") + b = get_logger("module.b") + assert a is b + assert a.name == "dns-bridge" + + def test_logger_is_singleton(self): + import logging + + logger = get_logger("x") + assert logger is logging.getLogger("dns-bridge") + + +@pytest.mark.django_db +class TestBuildDnsZone: + def test_basic_a_record(self, make_zone, make_record): + zone = make_zone(name="a.test") + make_record(zone, name="www", type_="A", value="192.0.2.1") + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("a.test", dns.name.root) + rds = dns_zone.get_rdataset(origin, dns.rdatatype.SOA) + assert rds is not None + a_rds = dns_zone.get_rdataset( + dns.name.from_text("www", origin), dns.rdatatype.A + ) + assert a_rds is not None + assert str(a_rds[0]) == "192.0.2.1" + + def test_apex_record_empty_name(self, make_zone, make_record): + zone = make_zone(name="apex.test") + make_record(zone, name="@", type_="A", value="192.0.2.10") + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("apex.test", dns.name.root) + a_rds = dns_zone.get_rdataset(origin, dns.rdatatype.A) + assert a_rds is not None + assert str(a_rds[0]) == "192.0.2.10" + + def test_ttl_fallback_to_zone_default(self, make_zone, make_record): + zone = make_zone(name="ttl.test", default_ttl=7200) + make_record(zone, name="host", type_="A", value="192.0.2.2", ttl=None) + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("ttl.test", dns.name.root) + a_rds = dns_zone.get_rdataset( + dns.name.from_text("host", origin), dns.rdatatype.A + ) + assert a_rds.ttl == 7200 + + def test_record_ttl_overrides_zone_default(self, make_zone, make_record): + zone = make_zone(name="ttl2.test", default_ttl=7200) + make_record(zone, name="host", type_="A", value="192.0.2.3", ttl=300) + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("ttl2.test", dns.name.root) + a_rds = dns_zone.get_rdataset( + dns.name.from_text("host", origin), dns.rdatatype.A + ) + assert a_rds.ttl == 300 + + def test_txt_record_quoting(self, make_zone, make_record): + zone = make_zone(name="txt.test") + make_record(zone, name="spf", type_="TXT", value="v=spf1 -all") + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("txt.test", dns.name.root) + txt_rds = dns_zone.get_rdataset( + dns.name.from_text("spf", origin), dns.rdatatype.TXT + ) + assert txt_rds is not None + assert str(txt_rds[0]) == '"v=spf1 -all"' + + def test_txt_record_already_quoted(self, make_zone, make_record): + zone = make_zone(name="txtq.test") + make_record(zone, name="dkim", type_="TXT", value='"k=rsa; p=MIGf"') + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("txtq.test", dns.name.root) + txt_rds = dns_zone.get_rdataset( + dns.name.from_text("dkim", origin), dns.rdatatype.TXT + ) + assert txt_rds is not None + assert "k=rsa" in str(txt_rds[0]) + + def test_txt_record_long_value_chunked(self, make_zone, make_record): + zone = make_zone(name="longtxt.test") + long_val = "a" * 300 + make_record(zone, name="big", type_="TXT", value=long_val) + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("longtxt.test", dns.name.root) + txt_rds = dns_zone.get_rdataset( + dns.name.from_text("big", origin), dns.rdatatype.TXT + ) + assert txt_rds is not None + wire = str(txt_rds[0]) + assert "a" * 300 in wire.replace('"', "").replace(" ", "") + + def test_absolute_name_record(self, make_zone, make_record): + zone = make_zone(name="abs.test") + make_record(zone, name="out.abs.test.", type_="A", value="192.0.2.99") + dns_zone = build_dns_zone(zone) + abs_name = dns.name.from_text("out.abs.test", dns.name.root) + a_rds = dns_zone.get_rdataset(abs_name, dns.rdatatype.A) + assert a_rds is not None + assert str(a_rds[0]) == "192.0.2.99" + + def test_multiple_records_same_name_type(self, make_zone, make_record): + zone = make_zone(name="multi.test") + make_record(zone, name="lb", type_="A", value="192.0.2.1") + make_record(zone, name="lb", type_="A", value="192.0.2.2") + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("multi.test", dns.name.root) + a_rds = dns_zone.get_rdataset( + dns.name.from_text("lb", origin), dns.rdatatype.A + ) + assert a_rds is not None + assert len(a_rds) == 2 + values = {str(r) for r in a_rds} + assert values == {"192.0.2.1", "192.0.2.2"} + + def test_inactive_records_excluded(self, make_zone, make_record): + from netbox_dns.choices import RecordStatusChoices + + zone = make_zone(name="inactive.test") + make_record(zone, name="active", type_="A", value="192.0.2.1", + status=RecordStatusChoices.STATUS_ACTIVE) + make_record(zone, name="inactive", type_="A", value="192.0.2.2", + status=RecordStatusChoices.STATUS_INACTIVE) + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("inactive.test", dns.name.root) + assert dns_zone.get_rdataset( + dns.name.from_text("active", origin), dns.rdatatype.A + ) is not None + assert dns_zone.get_rdataset( + dns.name.from_text("inactive", origin), dns.rdatatype.A + ) is None + + def test_zone_rdclass_is_in(self, make_zone, make_record): + zone = make_zone(name="rdclass.test") + make_record(zone, name="x", type_="A", value="192.0.2.1") + dns_zone = build_dns_zone(zone) + assert dns_zone.rdclass == dns.rdataclass.IN + + def test_aaaa_record(self, make_zone, make_record): + zone = make_zone(name="aaaa.test") + make_record(zone, name="v6", type_="AAAA", value="2001:db8::1") + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("aaaa.test", dns.name.root) + aaaa_rds = dns_zone.get_rdataset( + dns.name.from_text("v6", origin), dns.rdatatype.AAAA + ) + assert aaaa_rds is not None + assert str(aaaa_rds[0]) == "2001:db8::1" + + def test_cname_record(self, make_zone, make_record): + zone = make_zone(name="cname.test") + make_record(zone, name="alias", type_="CNAME", value="target.test.") + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("cname.test", dns.name.root) + cnames = dns_zone.get_rdataset( + dns.name.from_text("alias", origin), dns.rdatatype.CNAME + ) + assert cnames is not None + assert str(cnames[0]) == "target.test." + + def test_mx_record(self, make_zone, make_record): + zone = make_zone(name="mx.test") + make_record(zone, name="@", type_="MX", value="10 mail.test.") + dns_zone = build_dns_zone(zone) + origin = dns.name.from_text("mx.test", dns.name.root) + mx_rds = dns_zone.get_rdataset(origin, dns.rdatatype.MX) + assert mx_rds is not None + assert mx_rds[0].preference == 10 + assert str(mx_rds[0].exchange) == "mail.test." + + +@pytest.mark.django_db +class TestExportBindZoneFile: + def test_writes_valid_zone_file(self, make_zone, make_record): + zone = make_zone(name="export.test") + make_record(zone, name="www", type_="A", value="192.0.2.1") + with tempfile.NamedTemporaryFile(mode="w", suffix=".zone", delete=False) as f: + path = f.name + try: + export_bind_zone_file(zone, path) + with open(path) as f: + content = f.read() + assert "SOA" in content + assert "192.0.2.1" in content + assert "www" in content + finally: + os.unlink(path) + + def test_io_error_on_bad_path(self, make_zone): + zone = make_zone(name="badpath.test") + with pytest.raises(IOError): + export_bind_zone_file(zone, "/nonexistent/dir/zone.file") diff --git a/netbox_dns_bridge/tests/test_views.py b/netbox_dns_bridge/tests/test_views.py new file mode 100644 index 0000000..ec49b7e --- /dev/null +++ b/netbox_dns_bridge/tests/test_views.py @@ -0,0 +1,102 @@ +"""Tests for views: CatzOverviewView, NotifyOverviewView — auth gate and context.""" +import pytest +from django.test import Client +from django.contrib.auth import get_user_model + + +@pytest.fixture +def client(): + return Client() + + +@pytest.fixture +def superuser(db): + User = get_user_model() + user = User(username="admin", email="admin@test.local", is_superuser=True) + user.set_password("testpass") + user.save() + return user + + +@pytest.fixture +def regular_user(db): + User = get_user_model() + user = User(username="regular", email="reg@test.local") + user.set_password("testpass") + user.save() + return user + + +@pytest.mark.django_db +class TestCatzOverviewView: + def test_unauthenticated_redirects_to_login(self, client): + resp = client.get("/plugins/dns-bridge/catz/") + assert resp.status_code in (302, 403) + + def test_regular_user_forbidden(self, client, regular_user): + client.login(username="regular", password="testpass") + resp = client.get("/plugins/dns-bridge/catz/") + assert resp.status_code in (302, 403) + + def test_superuser_can_access(self, client, superuser): + client.login(username="admin", password="testpass") + resp = client.get("/plugins/dns-bridge/catz/") + assert resp.status_code == 200 + + def test_context_contains_soa_serial(self, client, superuser, catz_serial): + client.login(username="admin", password="testpass") + resp = client.get("/plugins/dns-bridge/catz/") + assert "soa_serial" in resp.context + assert resp.context["soa_serial"] == catz_serial + + def test_context_soa_serial_none_when_not_set(self, client, superuser): + from netbox_dns_bridge.models import IntegerKeyValueSetting + IntegerKeyValueSetting.objects.filter(key="catalog-zone-soa-serial").delete() + client.login(username="admin", password="testpass") + resp = client.get("/plugins/dns-bridge/catz/") + assert resp.context["soa_serial"] is None + + +@pytest.mark.django_db +class TestNotifyOverviewView: + def test_unauthenticated_redirects_to_login(self, client): + resp = client.get("/plugins/dns-bridge/notify/") + assert resp.status_code in (302, 403) + + def test_regular_user_forbidden(self, client, regular_user): + client.login(username="regular", password="testpass") + resp = client.get("/plugins/dns-bridge/notify/") + assert resp.status_code in (302, 403) + + def test_superuser_can_access(self, client, superuser): + client.login(username="admin", password="testpass") + resp = client.get("/plugins/dns-bridge/notify/") + assert resp.status_code == 200 + + def test_context_contains_seen_clients(self, client, superuser, make_view): + from netbox_dns_bridge.models import SeenTransferClients + from django.utils import timezone + + view = make_view(name="notify-view") + SeenTransferClients.objects.create( + source_ip="192.0.2.1", view=view, last_seen=timezone.now() + ) + client.login(username="admin", password="testpass") + resp = client.get("/plugins/dns-bridge/notify/") + assert "seen_clients" in resp.context + clients = list(resp.context["seen_clients"]) + assert len(clients) == 1 + assert clients[0].source_ip == "192.0.2.1" + + def test_seen_clients_ordered_by_ip(self, client, superuser, make_view): + from netbox_dns_bridge.models import SeenTransferClients + from django.utils import timezone + + view = make_view(name="ord-notify-view") + now = timezone.now() + for ip in ["10.0.0.3", "10.0.0.1", "10.0.0.2"]: + SeenTransferClients.objects.create(source_ip=ip, view=view, last_seen=now) + client.login(username="admin", password="testpass") + resp = client.get("/plugins/dns-bridge/notify/") + ips = [c.source_ip for c in resp.context["seen_clients"]] + assert ips == ["10.0.0.1", "10.0.0.2", "10.0.0.3"] diff --git a/pyproject.toml b/pyproject.toml index ff44e30..e09d430 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,17 @@ dependencies = [ "dnspython>=2.8.0" ] +[project.optional-dependencies] +test = [ + "pytest>=8.0", + "pytest-django>=4.8", +] + +[tool.pytest.ini_options] +DJANGO_SETTINGS_MODULE = "netbox.settings" +python_files = ["test_*.py"] +testpaths = ["netbox_dns_bridge/tests"] + [tool.setuptools] include-package-data = true zip-safe = false