sven/netbox-plugin-dns-bridge
0
0 Fork 0
Code Actions Packages Releases Wiki Activity
Files
14805ed32b73c91e1c3015ff6fb034b6c44a5c2a
netbox-plugin-dns-bridge/netbox_dns_bridge/tests/test_notify.py
T
sven 14805ed32b Move from hard-coded CatalogZone SOA values (refresh, retry, expire and minimum) 
to DB driven to permit user modification:

modified:   netbox_dns_bridge/models.py
  Add 4 cols for values.

modified:   netbox_dns_bridge/migrations/0003_catalogzone_and_per_zone_identifier.py
  Adjust migration. This expects that the last migration the user has is 0002 (<=v1.5.8)

modified:   netbox_dns_bridge/jobs/notify.py
modified:   netbox_dns_bridge/management/commands/dns_transfer_endpoint/catalog_zone_manager.py
  Update to use new db values for SOA creation.

Change UI to permit editing for SOA values and disable member identifier editing.
modified:   netbox_dns_bridge/forms.py
modified:   netbox_dns_bridge/navigation.py
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/catalog_zone_edit.html
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/catalog_zone_list.html
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/catalog_zone_members.html
deleted:    netbox_dns_bridge/templates/netbox_dns_bridge/catz_overview.html
modified:   netbox_dns_bridge/tables.py
modified:   netbox_dns_bridge/urls.py
modified:   netbox_dns_bridge/views.py
modified:   netbox_dns_bridge/filtersets.py
  AI created/modified: Left the UI update to the LLM.

modified:   netbox_dns_bridge/tests/test_catalog_zone_manager.py
modified:   netbox_dns_bridge/tests/test_models.py
modified:   netbox_dns_bridge/tests/test_notify.py
modified:   netbox_dns_bridge/tests/test_views.py
  AI created/modified: Updating tests
2026-08-17 13:20:17 +02:00

427 lines
17 KiB
Python
Raw Blame History

"""Tests for notify: _load_tsig_key, _get_soa_serial, _resolve_zone_nameserver_ips,
schedule_client_notify, schedule_ns_notify, schedule_catalog_zone_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.jobs.notify 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_enabled(self, make_zone, monkeypatch):
zone = make_zone(name="sched-client.test", soa_serial=77)
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": True})
with patch.object(nh.SendClientNotify, "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, monkeypatch):
from netbox_dns.models import Record
zone = make_zone(name="sched-client-nosoa.test")
Record.objects.filter(zone=zone, type="SOA").delete()
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": True})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_client_notify(zone)
mock_enq.assert_not_called()
def test_does_not_enqueue_when_notify_clients_disabled(self, make_zone, monkeypatch):
zone = make_zone(name="sched-client-disabled.test", soa_serial=77)
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": False})
with patch.object(nh.SendClientNotify, "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_enabled(self, make_zone, monkeypatch):
zone = make_zone(name="sched-ns.test", soa_serial=88)
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_all_zones": True})
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, monkeypatch):
from netbox_dns.models import Record
zone = make_zone(name="sched-ns-nosoa.test")
Record.objects.filter(zone=zone, type="SOA").delete()
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_all_zones": True})
with patch.object(nh.SendNSNotify, "enqueue") as mock_enq:
nh.schedule_ns_notify(zone)
mock_enq.assert_not_called()
def test_does_not_enqueue_when_ns_notify_disabled(self, make_zone, monkeypatch):
zone = make_zone(name="sched-ns-disabled.test", soa_serial=88)
monkeypatch.setattr(nh, "SETTINGS", {})
with patch.object(nh.SendNSNotify, "enqueue") as mock_enq:
nh.schedule_ns_notify(zone)
mock_enq.assert_not_called()
@pytest.mark.django_db
class TestScheduleCatalogZoneNotify:
def test_enqueues_two_jobs_when_enabled(self, make_view, monkeypatch):
from netbox_dns_bridge.models import CatalogZone
view = make_view(name="catz-notify-view")
CatalogZone.objects.create(view=view, soa_serial=42)
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": True})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_catalog_zone_notify(view)
assert mock_enq.call_count == 2
mock_enq.assert_any_call(
zone_name="catz",
soa_serial=42,
view_name="catz-notify-view",
soa_refresh=60,
soa_retry=10,
soa_expire=1209600,
soa_minimum=0,
)
mock_enq.assert_any_call(
zone_name="catz-notify-view.catz",
soa_serial=42,
view_name="catz-notify-view",
soa_refresh=60,
soa_retry=10,
soa_expire=1209600,
soa_minimum=0,
)
def test_does_not_enqueue_when_notify_clients_disabled(self, make_view, monkeypatch):
from netbox_dns_bridge.models import CatalogZone
view = make_view(name="catz-notify-disabled")
CatalogZone.objects.create(view=view, soa_serial=42)
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": False})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_catalog_zone_notify(view)
mock_enq.assert_not_called()
def test_does_not_enqueue_when_no_catalog_zone(self, make_view, monkeypatch):
view = make_view(name="catz-notify-no-cz")
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": True})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_catalog_zone_notify(view)
mock_enq.assert_not_called()
@pytest.mark.django_db
class TestSendClientNotifyRun:
def test_sends_single_notify_to_seen_clients(self, make_zone, make_view, monkeypatch):
import netbox_dns_bridge.jobs.notify as nh
from netbox_dns_bridge.models import SeenTransferClients
from django.utils import timezone
view = make_view(name="notify-run-view")
SeenTransferClients.objects.create(
source_ip="192.0.2.99", view=view, last_seen=timezone.now()
)
monkeypatch.setattr(nh, "SETTINGS", {
"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()
with patch("dns.query.udp") as mock_udp:
nh.SendClientNotify.run(job,
zone_name="z.test", soa_serial=10, view_name="notify-run-view")
assert mock_udp.call_count == 1
def test_sends_notify_without_catalog_zone(self, monkeypatch, make_view):
from netbox_dns_bridge.models import SeenTransferClients
from django.utils import timezone
view = make_view(name="no-catz-view")
SeenTransferClients.objects.create(
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
)
monkeypatch.setattr(nh, "SETTINGS", {
"tsig_keys": {
"no-catz-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.SendClientNotify.run(job,
zone_name="z.test", soa_serial=1, view_name="no-catz-view")
assert mock_udp.call_count == 1
@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.jobs.notify 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()
@pytest.mark.django_db
class TestNotifyNsEnabled:
def test_enabled_by_all_zones_flag(self, make_zone, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_all_zones": True})
from netbox_dns_bridge.jobs.notify 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, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {})
from netbox_dns_bridge.jobs.notify 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, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_custom_field_name": "enable_notify"})
from netbox_dns_bridge.jobs.notify 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, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_custom_field_name": "enable_notify"})
from netbox_dns_bridge.jobs.notify 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, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_custom_field_name": "enable_notify"})
from netbox_dns_bridge.jobs.notify import _notify_ns_enabled
zone = make_zone(name="nscf-missing.test")
zone.custom_field_data = {}
assert _notify_ns_enabled(zone) is False
View Git Blame Copy Permalink