Upgrade models to full NetboxModel to allow UI/API interaction:
modified: netbox_dns_bridge/models.py deleted: netbox_dns_bridge/migrations/0003_catalogzone_and_per_zone_identifier.py new file: netbox_dns_bridge/migrations/0003_full_netbox_models.py Adjust model access and function calls to match new models: modified: netbox_dns_bridge/signals/netbox_dns_zones.py modified: netbox_dns_bridge/management/commands/dns_transfer_endpoint/catalog_zone_manager.py modified: netbox_dns_bridge/management/commands/dns_transfer_endpoint/request_handler.py Update UI elements and permission interactions: modified: netbox_dns_bridge/filtersets.py modified: netbox_dns_bridge/forms.py modified: netbox_dns_bridge/jobs/notify.py modified: netbox_dns_bridge/navigation.py modified: netbox_dns_bridge/tables.py deleted: netbox_dns_bridge/templates/netbox_dns_bridge/catalog_zone_members.html new file: netbox_dns_bridge/templates/netbox_dns_bridge/catalogzone.html new file: netbox_dns_bridge/templates/netbox_dns_bridge/catalogzonemember.html deleted: netbox_dns_bridge/templates/netbox_dns_bridge/notify_overview.html new file: netbox_dns_bridge/templates/netbox_dns_bridge/seentransferclient.html modified: netbox_dns_bridge/urls.py modified: netbox_dns_bridge/views.py AI/LLM updated tests (Not verified): modified: netbox_dns_bridge/tests/conftest.py 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_request_handler_integration.py modified: netbox_dns_bridge/tests/test_signals.py modified: netbox_dns_bridge/tests/test_smoke.py modified: netbox_dns_bridge/tests/test_views.py
This commit is contained in:
1 parent
14805ed32b
commit
1c51206196
26 files changed
+812
-450
No files matched your search
@@ -3,32 +3,8 @@ from django.db.models import Q
|
||||
|
||||
import netbox_dns.models
|
||||
from netbox.filtersets import NetBoxModelFilterSet
|
||||
from netbox_dns.choices import ZoneStatusChoices
|
||||
|
||||
from .models import CatalogZone
|
||||
|
||||
|
||||
class ZoneFilterSet(NetBoxModelFilterSet):
|
||||
view = django_filters.ModelMultipleChoiceFilter(
|
||||
field_name="view",
|
||||
queryset=netbox_dns.models.View.objects.all(),
|
||||
)
|
||||
status = django_filters.MultipleChoiceFilter(
|
||||
choices=ZoneStatusChoices,
|
||||
null_value=None,
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = netbox_dns.models.Zone
|
||||
fields = ("id", "name", "view", "status", "catz_identifier")
|
||||
|
||||
def search(self, queryset, name, value):
|
||||
if not value.strip():
|
||||
return queryset
|
||||
return queryset.filter(
|
||||
Q(name__icontains=value)
|
||||
| Q(catz_identifier__name__icontains=value)
|
||||
)
|
||||
from .models import CatalogZone, CatalogZoneMember, SeenTransferClient
|
||||
|
||||
|
||||
class CatalogZoneFilterSet(NetBoxModelFilterSet):
|
||||
@@ -45,3 +21,39 @@ class CatalogZoneFilterSet(NetBoxModelFilterSet):
|
||||
if not value.strip():
|
||||
return queryset
|
||||
return queryset.filter(Q(view__name__icontains=value))
|
||||
|
||||
|
||||
class CatalogZoneMemberFilterSet(NetBoxModelFilterSet):
|
||||
catalog_zone = django_filters.ModelMultipleChoiceFilter(
|
||||
field_name="catalog_zone",
|
||||
queryset=CatalogZone.objects.all(),
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = CatalogZoneMember
|
||||
fields = ("id", "catalog_zone", "zone")
|
||||
|
||||
def search(self, queryset, name, value):
|
||||
if not value.strip():
|
||||
return queryset
|
||||
return queryset.filter(
|
||||
Q(name__icontains=value) | Q(zone__name__icontains=value)
|
||||
)
|
||||
|
||||
|
||||
class SeenTransferClientFilterSet(NetBoxModelFilterSet):
|
||||
view = django_filters.ModelMultipleChoiceFilter(
|
||||
field_name="view",
|
||||
queryset=netbox_dns.models.View.objects.all(),
|
||||
)
|
||||
|
||||
class Meta:
|
||||
model = SeenTransferClient
|
||||
fields = ("id", "view", "source_ip")
|
||||
|
||||
def search(self, queryset, name, value):
|
||||
if not value.strip():
|
||||
return queryset
|
||||
return queryset.filter(
|
||||
Q(source_ip__icontains=value) | Q(view__name__icontains=value)
|
||||
)
|
||||
+21
-16
@@ -3,10 +3,9 @@ from django.utils.translation import gettext_lazy as _
|
||||
|
||||
import netbox_dns.models
|
||||
from netbox.forms import NetBoxModelFilterSetForm, NetBoxModelForm
|
||||
from netbox_dns.choices import ZoneStatusChoices
|
||||
from utilities.forms.fields import DynamicModelMultipleChoiceField
|
||||
|
||||
from .models import CatalogZone
|
||||
from .models import CatalogZone, CatalogZoneMember, SeenTransferClient
|
||||
|
||||
|
||||
class CatalogZoneForm(NetBoxModelForm):
|
||||
@@ -35,20 +34,6 @@ class CatalogZoneForm(NetBoxModelForm):
|
||||
)
|
||||
|
||||
|
||||
class CatalogZoneMemberFilterForm(NetBoxModelFilterSetForm):
|
||||
model = netbox_dns.models.Zone
|
||||
|
||||
q = forms.CharField(required=False, label=_("Search"))
|
||||
view = DynamicModelMultipleChoiceField(
|
||||
queryset=netbox_dns.models.View.objects.all(),
|
||||
required=False,
|
||||
)
|
||||
status = forms.MultipleChoiceField(
|
||||
choices=ZoneStatusChoices,
|
||||
required=False,
|
||||
)
|
||||
|
||||
|
||||
class CatalogZoneFilterForm(NetBoxModelFilterSetForm):
|
||||
model = CatalogZone
|
||||
|
||||
@@ -57,3 +42,23 @@ class CatalogZoneFilterForm(NetBoxModelFilterSetForm):
|
||||
queryset=netbox_dns.models.View.objects.all(),
|
||||
required=False,
|
||||
)
|
||||
|
||||
|
||||
class CatalogZoneMemberFilterForm(NetBoxModelFilterSetForm):
|
||||
model = CatalogZoneMember
|
||||
|
||||
q = forms.CharField(required=False, label=_("Search"))
|
||||
catalog_zone = forms.ModelMultipleChoiceField(
|
||||
queryset=CatalogZone.objects.all(),
|
||||
required=False,
|
||||
)
|
||||
|
||||
|
||||
class SeenTransferClientFilterForm(NetBoxModelFilterSetForm):
|
||||
model = SeenTransferClient
|
||||
|
||||
q = forms.CharField(required=False, label=_("Search"))
|
||||
view = forms.ModelMultipleChoiceField(
|
||||
queryset=netbox_dns.models.View.objects.all(),
|
||||
required=False,
|
||||
)
|
||||
@@ -12,7 +12,7 @@ from django.db import close_old_connections, OperationalError
|
||||
from netbox.jobs import JobRunner
|
||||
from netbox_dns.models import View, Zone, Record
|
||||
from netbox_dns_bridge.utils import get_logger
|
||||
from netbox_dns_bridge.models import SeenTransferClients, CatalogZone
|
||||
from netbox_dns_bridge.models import SeenTransferClient, CatalogZone
|
||||
|
||||
LOGGER = get_logger(__name__)
|
||||
SETTINGS = settings.PLUGINS_CONFIG.get("netbox_dns_bridge", {})
|
||||
@@ -111,7 +111,7 @@ class SendClientNotify(JobRunner):
|
||||
seen_cutoff = timezone.now() - timezone.timedelta(hours=cutoff_hours)
|
||||
|
||||
clients = list(
|
||||
SeenTransferClients.objects.filter(
|
||||
SeenTransferClient.objects.filter(
|
||||
view=view, last_seen__gte=seen_cutoff
|
||||
).values_list("source_ip", flat=True)
|
||||
)
|
||||
|
||||
+13
-13
@@ -4,7 +4,7 @@ import dns.rdatatype
|
||||
import dns.rdataclass
|
||||
from netbox_dns_bridge.utils import get_logger
|
||||
from netbox_dns.models import Zone as NBZone, View as NBView
|
||||
from netbox_dns_bridge.models import CatalogZone, CatalogZoneMemberIdentifier
|
||||
from netbox_dns_bridge.models import CatalogZone, CatalogZoneMember
|
||||
from netbox_dns_bridge.jobs import notify
|
||||
from uuid import uuid4
|
||||
from base64 import b32encode
|
||||
@@ -63,7 +63,7 @@ def create_zone(name, view_name) -> dns.zone.Zone:
|
||||
# get zones from netbox
|
||||
nb_zones = NBZone.objects.filter(
|
||||
view__name=view_name, active=True
|
||||
).select_related("catz_identifier")
|
||||
).select_related("catalog_zone_member")
|
||||
|
||||
ptr_base = dns.name.from_text("zones", origin)
|
||||
|
||||
@@ -73,21 +73,21 @@ def create_zone(name, view_name) -> dns.zone.Zone:
|
||||
|
||||
# Create PTR record
|
||||
try:
|
||||
catz_identifier = nb_zone.catz_identifier
|
||||
except NBZone.catz_identifier.RelatedObjectDoesNotExist:
|
||||
member = nb_zone.catalog_zone_member
|
||||
except NBZone.catalog_zone_member.RelatedObjectDoesNotExist:
|
||||
catalog_zone = _get_or_create_catalog_zone(nb_zone.view)
|
||||
catz_identifier = CatalogZoneMemberIdentifier.objects.create(
|
||||
member = CatalogZoneMember.objects.create(
|
||||
zone=nb_zone,
|
||||
name=_generate_member_identifier(),
|
||||
name=_generate_member_name(),
|
||||
catalog_zone=catalog_zone,
|
||||
)
|
||||
|
||||
p_name = catz_identifier.name
|
||||
p_name = member.name
|
||||
|
||||
ptr_name = dns.name.from_text(p_name, ptr_base)
|
||||
if not ptr_name.is_subdomain(origin):
|
||||
raise ValueError(
|
||||
f"Catalog zone member identifier {ptr_name.to_text()} not a subdomain"
|
||||
f"Catalog zone member {ptr_name.to_text()} not a subdomain"
|
||||
)
|
||||
rdata = dns.rdata.from_text(
|
||||
dns.rdataclass.IN, dns.rdatatype.PTR, qname.to_text()
|
||||
@@ -132,7 +132,7 @@ def create_zone(name, view_name) -> dns.zone.Zone:
|
||||
return None
|
||||
|
||||
|
||||
def _generate_member_identifier() -> str:
|
||||
def _generate_member_name() -> str:
|
||||
return b32encode(uuid4().bytes)[0:26].lower().decode("UTF-8")
|
||||
|
||||
|
||||
@@ -143,18 +143,18 @@ def _get_or_create_catalog_zone(view: NBView) -> CatalogZone:
|
||||
return catalog_zone
|
||||
|
||||
|
||||
def update_member_identifier(zone: NBZone) -> None:
|
||||
def update_member(zone: NBZone) -> None:
|
||||
try:
|
||||
catalog_zone = _get_or_create_catalog_zone(zone.view)
|
||||
CatalogZoneMemberIdentifier.objects.update_or_create(
|
||||
CatalogZoneMember.objects.update_or_create(
|
||||
zone=zone,
|
||||
defaults={
|
||||
"name": _generate_member_identifier(),
|
||||
"name": _generate_member_name(),
|
||||
"catalog_zone": catalog_zone,
|
||||
},
|
||||
)
|
||||
except OperationalError as e:
|
||||
LOGGER.error(f"ERROR: Failed to update Catalog Zone member identifier: {e}")
|
||||
LOGGER.error(f"ERROR: Failed to update Catalog Zone member: {e}")
|
||||
close_old_connections()
|
||||
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@ from django.db import close_old_connections, OperationalError
|
||||
from django.conf import settings
|
||||
from netbox_dns.models import Zone, View
|
||||
from . import catalog_zone_manager as catzm
|
||||
from netbox_dns_bridge.models import SeenTransferClients
|
||||
from netbox_dns_bridge.models import SeenTransferClient
|
||||
from netbox_dns_bridge.utils import build_dns_zone, get_logger
|
||||
|
||||
LOGGER = get_logger(__name__)
|
||||
@@ -77,7 +77,7 @@ class DNSBaseRequestHandler(socketserver.BaseRequestHandler):
|
||||
def _track_seen_client(self, source_ip: str, view: View) -> None:
|
||||
if SETTINGS.get("notify_clients", False):
|
||||
try:
|
||||
SeenTransferClients.objects.update_or_create(
|
||||
SeenTransferClient.objects.update_or_create(
|
||||
source_ip=source_ip,
|
||||
view=view,
|
||||
defaults={"last_seen": timezone.now()},
|
||||
|
||||
@@ -1,146 +0,0 @@
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
def copy_serials_and_link_identifiers(apps, schema_editor):
|
||||
"""Seed one CatalogZone per view from the old global serial and link identifiers."""
|
||||
IntegerKeyValueSetting = apps.get_model(
|
||||
"netbox_dns_bridge", "IntegerKeyValueSetting"
|
||||
)
|
||||
CatalogZone = apps.get_model("netbox_dns_bridge", "CatalogZone")
|
||||
CatalogZoneMemberIdentifier = apps.get_model(
|
||||
"netbox_dns_bridge", "CatalogZoneMemberIdentifier"
|
||||
)
|
||||
|
||||
# Preserve the current global serial; default to 1 when never initialized.
|
||||
seed_serial = 1
|
||||
kv = IntegerKeyValueSetting.objects.filter(key="catalog-zone-soa-serial").first()
|
||||
if kv is not None:
|
||||
seed_serial = kv.value
|
||||
|
||||
# Create a CatalogZone for every view that currently has at least one
|
||||
# member identifier, seeding each with the old global serial.
|
||||
view_ids = set(
|
||||
CatalogZoneMemberIdentifier.objects.exclude(
|
||||
zone__view__isnull=True
|
||||
).values_list("zone__view_id", flat=True)
|
||||
)
|
||||
|
||||
view_id_to_catalog_zone = {}
|
||||
for view_id in view_ids:
|
||||
cz, _ = CatalogZone.objects.get_or_create(
|
||||
view_id=view_id, defaults={"soa_serial": seed_serial}
|
||||
)
|
||||
view_id_to_catalog_zone[view_id] = cz
|
||||
|
||||
# Link each identifier to the CatalogZone matching its zone's view.
|
||||
for ident in CatalogZoneMemberIdentifier.objects.select_related("zone__view"):
|
||||
cz = view_id_to_catalog_zone.get(ident.zone.view_id)
|
||||
if cz is None:
|
||||
# Defensive: identifier existed without a resolvable view. Create
|
||||
# the CatalogZone on demand so the NOT NULL constraint can be set.
|
||||
cz, _ = CatalogZone.objects.get_or_create(
|
||||
view_id=ident.zone.view_id, defaults={"soa_serial": seed_serial}
|
||||
)
|
||||
view_id_to_catalog_zone[ident.zone.view_id] = cz
|
||||
ident.catalog_zone = cz
|
||||
ident.save(update_fields=["catalog_zone"])
|
||||
|
||||
|
||||
def reverse_migration(apps, schema_editor):
|
||||
"""Best-effort reverse: restore the KV row and null the FK."""
|
||||
IntegerKeyValueSetting = apps.get_model(
|
||||
"netbox_dns_bridge", "IntegerKeyValueSetting"
|
||||
)
|
||||
CatalogZone = apps.get_model("netbox_dns_bridge", "CatalogZone")
|
||||
CatalogZoneMemberIdentifier = apps.get_model(
|
||||
"netbox_dns_bridge", "CatalogZoneMemberIdentifier"
|
||||
)
|
||||
|
||||
# Pick a representative serial (first catalog zone) to write back.
|
||||
first = CatalogZone.objects.order_by("pk").first()
|
||||
serial = first.soa_serial if first is not None else 1
|
||||
IntegerKeyValueSetting.objects.update_or_create(
|
||||
key="catalog-zone-soa-serial", defaults={"value": serial}
|
||||
)
|
||||
|
||||
CatalogZoneMemberIdentifier.objects.update(catalog_zone=None)
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
(
|
||||
"netbox_dns",
|
||||
"0030_dnsseckeytemplate_comments_dnsseckeytemplate_owner_and_more",
|
||||
),
|
||||
("netbox_dns_bridge", "0002_alter_integerkeyvaluesetting_options_and_more"),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name="CatalogZone",
|
||||
fields=[
|
||||
(
|
||||
"id",
|
||||
models.BigAutoField(
|
||||
auto_created=True, primary_key=True, serialize=False
|
||||
),
|
||||
),
|
||||
("soa_serial", models.IntegerField(default=1)),
|
||||
("soa_refresh", models.IntegerField(default=60)),
|
||||
("soa_retry", models.IntegerField(default=10)),
|
||||
("soa_expire", models.IntegerField(default=1209600)),
|
||||
("soa_minimum", models.IntegerField(default=0)),
|
||||
(
|
||||
"view",
|
||||
models.OneToOneField(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="catalog_zone",
|
||||
to="netbox_dns.view",
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
"ordering": ("view__name",),
|
||||
},
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="catalogzonememberidentifier",
|
||||
name="catalog_zone",
|
||||
field=models.ForeignKey(
|
||||
null=True,
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="member_identifiers",
|
||||
to="netbox_dns_bridge.catalogzone",
|
||||
),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name="catalogzonememberidentifier",
|
||||
name="name",
|
||||
field=models.CharField(max_length=26),
|
||||
),
|
||||
migrations.RunPython(
|
||||
copy_serials_and_link_identifiers,
|
||||
reverse_code=reverse_migration,
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name="catalogzonememberidentifier",
|
||||
name="catalog_zone",
|
||||
field=models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="member_identifiers",
|
||||
to="netbox_dns_bridge.catalogzone",
|
||||
),
|
||||
),
|
||||
migrations.AddConstraint(
|
||||
model_name="catalogzonememberidentifier",
|
||||
constraint=models.UniqueConstraint(
|
||||
fields=("name", "catalog_zone"),
|
||||
name="unique_name_per_catalog_zone",
|
||||
),
|
||||
),
|
||||
migrations.DeleteModel(
|
||||
name="IntegerKeyValueSetting",
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,233 @@
|
||||
import taggit.managers
|
||||
import utilities.json
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
def copy_serials_and_link_members(apps, schema_editor):
|
||||
"""Seed one CatalogZone per view from the old global serial and link members."""
|
||||
IntegerKeyValueSetting = apps.get_model(
|
||||
"netbox_dns_bridge", "IntegerKeyValueSetting"
|
||||
)
|
||||
CatalogZone = apps.get_model("netbox_dns_bridge", "CatalogZone")
|
||||
CatalogZoneMember = apps.get_model("netbox_dns_bridge", "CatalogZoneMember")
|
||||
|
||||
# Preserve the current global serial; default to 1 when never initialized.
|
||||
seed_serial = 1
|
||||
kv = IntegerKeyValueSetting.objects.filter(key="catalog-zone-soa-serial").first()
|
||||
if kv is not None:
|
||||
seed_serial = kv.value
|
||||
|
||||
# Create a CatalogZone for every view that currently has at least one
|
||||
# member, seeding each with the old global serial.
|
||||
view_ids = set(
|
||||
CatalogZoneMember.objects.exclude(
|
||||
zone__view__isnull=True
|
||||
).values_list("zone__view_id", flat=True)
|
||||
)
|
||||
|
||||
view_id_to_catalog_zone = {}
|
||||
for view_id in view_ids:
|
||||
cz, _ = CatalogZone.objects.get_or_create(
|
||||
view_id=view_id, defaults={"soa_serial": seed_serial}
|
||||
)
|
||||
view_id_to_catalog_zone[view_id] = cz
|
||||
|
||||
# Link each member to the CatalogZone matching its zone's view.
|
||||
for member in CatalogZoneMember.objects.select_related("zone__view"):
|
||||
cz = view_id_to_catalog_zone.get(member.zone.view_id)
|
||||
if cz is None:
|
||||
# Defensive: member existed without a resolvable view. Create
|
||||
# the CatalogZone on demand so the NOT NULL constraint can be set.
|
||||
cz, _ = CatalogZone.objects.get_or_create(
|
||||
view_id=member.zone.view_id, defaults={"soa_serial": seed_serial}
|
||||
)
|
||||
view_id_to_catalog_zone[member.zone.view_id] = cz
|
||||
member.catalog_zone = cz
|
||||
member.save(update_fields=["catalog_zone"])
|
||||
|
||||
|
||||
def reverse_migration(apps, schema_editor):
|
||||
"""Best-effort reverse: restore the KV row and null the FK."""
|
||||
IntegerKeyValueSetting = apps.get_model(
|
||||
"netbox_dns_bridge", "IntegerKeyValueSetting"
|
||||
)
|
||||
CatalogZone = apps.get_model("netbox_dns_bridge", "CatalogZone")
|
||||
CatalogZoneMember = apps.get_model("netbox_dns_bridge", "CatalogZoneMember")
|
||||
|
||||
# Pick a representative serial (first catalog zone) to write back.
|
||||
first = CatalogZone.objects.order_by("pk").first()
|
||||
serial = first.soa_serial if first is not None else 1
|
||||
IntegerKeyValueSetting.objects.update_or_create(
|
||||
key="catalog-zone-soa-serial", defaults={"value": serial}
|
||||
)
|
||||
|
||||
CatalogZoneMember.objects.update(catalog_zone=None)
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
("extras", "0140_imageattachment_image_size"),
|
||||
(
|
||||
"netbox_dns",
|
||||
"0032_record_expiration_date",
|
||||
),
|
||||
("netbox_dns_bridge", "0002_alter_integerkeyvaluesetting_options_and_more"),
|
||||
]
|
||||
|
||||
operations = [
|
||||
# --- CatalogZone (full NetBoxModel, born with all fields) ---
|
||||
migrations.CreateModel(
|
||||
name="CatalogZone",
|
||||
fields=[
|
||||
(
|
||||
"id",
|
||||
models.BigAutoField(
|
||||
auto_created=True, primary_key=True, serialize=False
|
||||
),
|
||||
),
|
||||
("soa_serial", models.IntegerField(default=1)),
|
||||
("soa_refresh", models.IntegerField(default=60)),
|
||||
("soa_retry", models.IntegerField(default=10)),
|
||||
("soa_expire", models.IntegerField(default=1209600)),
|
||||
("soa_minimum", models.IntegerField(default=0)),
|
||||
("created", models.DateTimeField(auto_now_add=True, null=True)),
|
||||
("last_updated", models.DateTimeField(auto_now=True, null=True)),
|
||||
(
|
||||
"custom_field_data",
|
||||
models.JSONField(
|
||||
blank=True,
|
||||
default=dict,
|
||||
encoder=utilities.json.CustomFieldJSONEncoder,
|
||||
),
|
||||
),
|
||||
(
|
||||
"tags",
|
||||
taggit.managers.TaggableManager(
|
||||
through="extras.TaggedItem", to="extras.Tag"
|
||||
),
|
||||
),
|
||||
(
|
||||
"view",
|
||||
models.OneToOneField(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="catalog_zone",
|
||||
to="netbox_dns.view",
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
"ordering": ("view__name",),
|
||||
},
|
||||
),
|
||||
# --- CatalogZoneMemberIdentifier → CatalogZoneMember (rename + full NetBoxModel) ---
|
||||
migrations.RenameModel(
|
||||
old_name="CatalogZoneMemberIdentifier",
|
||||
new_name="CatalogZoneMember",
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="catalogzonemember",
|
||||
name="created",
|
||||
field=models.DateTimeField(auto_now_add=True, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="catalogzonemember",
|
||||
name="last_updated",
|
||||
field=models.DateTimeField(auto_now=True, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="catalogzonemember",
|
||||
name="custom_field_data",
|
||||
field=models.JSONField(
|
||||
blank=True,
|
||||
default=dict,
|
||||
encoder=utilities.json.CustomFieldJSONEncoder,
|
||||
),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="catalogzonemember",
|
||||
name="tags",
|
||||
field=taggit.managers.TaggableManager(
|
||||
through="extras.TaggedItem", to="extras.Tag"
|
||||
),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name="catalogzonemember",
|
||||
name="name",
|
||||
field=models.CharField(max_length=26),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name="catalogzonemember",
|
||||
name="zone",
|
||||
field=models.OneToOneField(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="catalog_zone_member",
|
||||
to="netbox_dns.zone",
|
||||
),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="catalogzonemember",
|
||||
name="catalog_zone",
|
||||
field=models.ForeignKey(
|
||||
null=True,
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="members",
|
||||
to="netbox_dns_bridge.catalogzone",
|
||||
),
|
||||
),
|
||||
migrations.RunPython(
|
||||
copy_serials_and_link_members,
|
||||
reverse_code=reverse_migration,
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name="catalogzonemember",
|
||||
name="catalog_zone",
|
||||
field=models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name="members",
|
||||
to="netbox_dns_bridge.catalogzone",
|
||||
),
|
||||
),
|
||||
migrations.AddConstraint(
|
||||
model_name="catalogzonemember",
|
||||
constraint=models.UniqueConstraint(
|
||||
fields=("name", "catalog_zone"),
|
||||
name="unique_name_per_catalog_zone",
|
||||
),
|
||||
),
|
||||
migrations.DeleteModel(
|
||||
name="IntegerKeyValueSetting",
|
||||
),
|
||||
# --- SeenTransferClients → SeenTransferClient (rename + full NetBoxModel) ---
|
||||
migrations.RenameModel(
|
||||
old_name="SeenTransferClients",
|
||||
new_name="SeenTransferClient",
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="seentransferclient",
|
||||
name="created",
|
||||
field=models.DateTimeField(auto_now_add=True, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="seentransferclient",
|
||||
name="last_updated",
|
||||
field=models.DateTimeField(auto_now=True, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="seentransferclient",
|
||||
name="custom_field_data",
|
||||
field=models.JSONField(
|
||||
blank=True,
|
||||
default=dict,
|
||||
encoder=utilities.json.CustomFieldJSONEncoder,
|
||||
),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name="seentransferclient",
|
||||
name="tags",
|
||||
field=taggit.managers.TaggableManager(
|
||||
through="extras.TaggedItem", to="extras.Tag"
|
||||
),
|
||||
),
|
||||
]
|
||||
+12
-15
@@ -1,13 +1,11 @@
|
||||
from django.db import models
|
||||
from django.urls import reverse
|
||||
|
||||
# import netbox.models
|
||||
import netbox_dns.models
|
||||
from netbox.models import NetBoxModel
|
||||
|
||||
|
||||
class CatalogZone(models.Model):
|
||||
_netbox_private = True
|
||||
|
||||
class CatalogZone(NetBoxModel):
|
||||
view = models.OneToOneField(
|
||||
to=netbox_dns.models.View,
|
||||
on_delete=models.CASCADE,
|
||||
@@ -21,17 +19,14 @@ class CatalogZone(models.Model):
|
||||
|
||||
class Meta:
|
||||
ordering = ("view__name",)
|
||||
verbose_name = "Catalog Zone"
|
||||
verbose_name_plural = "Catalog Zones"
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.view.name}: {self.soa_serial}"
|
||||
|
||||
def get_absolute_url(self):
|
||||
return reverse("plugins:netbox_dns_bridge:catalogzone_edit", args=[self.pk])
|
||||
|
||||
|
||||
class CatalogZoneMemberIdentifier(models.Model):
|
||||
_netbox_private = True
|
||||
|
||||
class CatalogZoneMember(NetBoxModel):
|
||||
name = models.CharField(
|
||||
max_length=26,
|
||||
)
|
||||
@@ -39,17 +34,19 @@ class CatalogZoneMemberIdentifier(models.Model):
|
||||
zone = models.OneToOneField(
|
||||
to=netbox_dns.models.Zone,
|
||||
on_delete=models.CASCADE,
|
||||
related_name="catz_identifier",
|
||||
related_name="catalog_zone_member",
|
||||
)
|
||||
|
||||
catalog_zone = models.ForeignKey(
|
||||
to=CatalogZone,
|
||||
on_delete=models.CASCADE,
|
||||
related_name="member_identifiers",
|
||||
related_name="members",
|
||||
)
|
||||
|
||||
class Meta:
|
||||
ordering = ("name",)
|
||||
verbose_name = "Catalog Zone Member"
|
||||
verbose_name_plural = "Catalog Zone Members"
|
||||
constraints = [
|
||||
models.UniqueConstraint(
|
||||
fields=["name", "catalog_zone"],
|
||||
@@ -61,9 +58,7 @@ class CatalogZoneMemberIdentifier(models.Model):
|
||||
return self.name
|
||||
|
||||
|
||||
class SeenTransferClients(models.Model):
|
||||
_netbox_private = True
|
||||
|
||||
class SeenTransferClient(NetBoxModel):
|
||||
source_ip = models.GenericIPAddressField()
|
||||
last_seen = models.DateTimeField()
|
||||
view = models.ForeignKey(
|
||||
@@ -77,6 +72,8 @@ class SeenTransferClients(models.Model):
|
||||
|
||||
class Meta:
|
||||
ordering = ("source_ip",)
|
||||
verbose_name = "Seen Transfer Client"
|
||||
verbose_name_plural = "Seen Transfer Clients"
|
||||
constraints = [
|
||||
models.UniqueConstraint(
|
||||
fields=["source_ip", "view"],
|
||||
|
||||
@@ -2,32 +2,33 @@ from django.utils.translation import gettext_lazy as _
|
||||
from netbox.plugins import PluginMenu, PluginMenuItem
|
||||
|
||||
catalog_zones_menu_item = PluginMenuItem(
|
||||
link="plugins:netbox_dns_bridge:catalog_zones",
|
||||
link="plugins:netbox_dns_bridge:catalogzone_list",
|
||||
link_text=_("Catalog Zones"),
|
||||
permissions=[],
|
||||
staff_only=True,
|
||||
permissions=["netbox_dns_bridge.view_catalogzone"],
|
||||
)
|
||||
|
||||
catalog_zone_members_menu_item = PluginMenuItem(
|
||||
link="plugins:netbox_dns_bridge:catalog_zone_members",
|
||||
link="plugins:netbox_dns_bridge:catalogzonemember_list",
|
||||
link_text=_("Catalog Zone Members"),
|
||||
permissions=[],
|
||||
staff_only=True,
|
||||
permissions=["netbox_dns_bridge.view_catalogzonemember"],
|
||||
)
|
||||
|
||||
notify_menu_item = PluginMenuItem(
|
||||
link="plugins:netbox_dns_bridge:notify",
|
||||
link_text=_("Notify"),
|
||||
permissions=[],
|
||||
staff_only=True,
|
||||
seen_transfer_clients_menu_item = PluginMenuItem(
|
||||
link="plugins:netbox_dns_bridge:seentransferclient_list",
|
||||
link_text=_("Clients"),
|
||||
permissions=["netbox_dns_bridge.view_seentransferclient"],
|
||||
)
|
||||
|
||||
menu = PluginMenu(
|
||||
label=_("DNS Bridge"),
|
||||
groups=(
|
||||
(
|
||||
_("Information"),
|
||||
(catalog_zones_menu_item, catalog_zone_members_menu_item, notify_menu_item),
|
||||
_("Catalog Zone"),
|
||||
(catalog_zones_menu_item, catalog_zone_members_menu_item),
|
||||
),
|
||||
(
|
||||
_("Notify"),
|
||||
(seen_transfer_clients_menu_item,),
|
||||
),
|
||||
),
|
||||
icon_class="mdi mdi-bridge",
|
||||
|
||||
@@ -77,15 +77,15 @@ def zone_post_save(sender, instance, created, **kwargs):
|
||||
transaction.on_commit(_on_commit)
|
||||
|
||||
if created:
|
||||
# If zone was created, create catz member identifier record for this zone.
|
||||
catzm.update_member_identifier(zone)
|
||||
# If zone was created, create the catalog zone member record for this zone.
|
||||
catzm.update_member(zone)
|
||||
else:
|
||||
# Re-create the catz member identifier when the zone name or its view
|
||||
# changed. A view move leaves the identifier pointing at the wrong
|
||||
# Re-create the catalog zone member when the zone name or its view
|
||||
# changed. A view move leaves the member pointing at the wrong
|
||||
# catalog zone, so it must be refreshed.
|
||||
old_name = getattr(zone, "_old_name", None)
|
||||
if (old_name and zone.name != old_name) or view_changed:
|
||||
catzm.update_member_identifier(zone)
|
||||
catzm.update_member(zone)
|
||||
|
||||
|
||||
@receiver(post_delete, sender=Zone)
|
||||
|
||||
+32
-12
@@ -1,29 +1,30 @@
|
||||
import django_tables2 as tables
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
import netbox_dns.models
|
||||
from netbox.tables import NetBoxTable
|
||||
from netbox.tables.columns import ActionsColumn
|
||||
|
||||
from .models import CatalogZone
|
||||
from .models import CatalogZone, CatalogZoneMember, SeenTransferClient
|
||||
|
||||
|
||||
class CatalogZoneMemberTable(NetBoxTable):
|
||||
name = tables.Column(linkify=True)
|
||||
zone = tables.Column(
|
||||
accessor="zone__name",
|
||||
verbose_name=_("Zone"),
|
||||
linkify=True,
|
||||
)
|
||||
view = tables.Column(
|
||||
accessor="view__name",
|
||||
accessor="catalog_zone__view__name",
|
||||
verbose_name=_("View"),
|
||||
)
|
||||
status = tables.Column(verbose_name=_("Status"))
|
||||
catz_identifier = tables.Column(
|
||||
accessor="catz_identifier__name",
|
||||
verbose_name=_("Catalog Zone Identifier"),
|
||||
default="—",
|
||||
)
|
||||
|
||||
actions = ActionsColumn(actions=())
|
||||
|
||||
class Meta(NetBoxTable.Meta):
|
||||
model = netbox_dns.models.Zone
|
||||
fields = ("name", "view", "status", "catz_identifier")
|
||||
default_columns = ("name", "view", "status", "catz_identifier")
|
||||
model = CatalogZoneMember
|
||||
fields = ("name", "zone", "view")
|
||||
default_columns = ("name", "zone", "view")
|
||||
|
||||
|
||||
class CatalogZoneTable(NetBoxTable):
|
||||
@@ -38,6 +39,8 @@ class CatalogZoneTable(NetBoxTable):
|
||||
soa_expire = tables.Column(verbose_name=_("Expire"))
|
||||
soa_minimum = tables.Column(verbose_name=_("Minimum"))
|
||||
|
||||
actions = ActionsColumn(actions=("edit", "changelog"))
|
||||
|
||||
class Meta(NetBoxTable.Meta):
|
||||
model = CatalogZone
|
||||
fields = (
|
||||
@@ -56,3 +59,20 @@ class CatalogZoneTable(NetBoxTable):
|
||||
"soa_expire",
|
||||
"soa_minimum",
|
||||
)
|
||||
|
||||
|
||||
class SeenTransferClientTable(NetBoxTable):
|
||||
source_ip = tables.Column(linkify=True)
|
||||
view = tables.Column(
|
||||
accessor="view__name",
|
||||
verbose_name=_("View"),
|
||||
linkify=True,
|
||||
)
|
||||
last_seen = tables.DateTimeColumn(verbose_name=_("Last Seen"))
|
||||
|
||||
actions = ActionsColumn(actions=())
|
||||
|
||||
class Meta(NetBoxTable.Meta):
|
||||
model = SeenTransferClient
|
||||
fields = ("source_ip", "view", "last_seen")
|
||||
default_columns = ("source_ip", "view", "last_seen")
|
||||
@@ -1,4 +0,0 @@
|
||||
{% extends 'generic/object_list.html' %}
|
||||
{% load i18n %}
|
||||
|
||||
{% block title %}{% trans "Catalog Zone Members" %}{% endblock %}
|
||||
@@ -0,0 +1,50 @@
|
||||
{% extends 'generic/object.html' %}
|
||||
{% load plugins %}
|
||||
{% load i18n %}
|
||||
|
||||
{% block content %}
|
||||
<div class="row">
|
||||
<div class="col col-md-6">
|
||||
<div class="card">
|
||||
<h5 class="card-header">{% trans "Catalog Zone SOA" %}</h5>
|
||||
<table class="table table-hover attr-table">
|
||||
<tr>
|
||||
<th scope="row">{% trans "View" %}</th>
|
||||
<td>{{ object.view|linkify }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "SOA Serial" %}</th>
|
||||
<td>{{ object.soa_serial }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "Refresh" %}</th>
|
||||
<td>{{ object.soa_refresh }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "Retry" %}</th>
|
||||
<td>{{ object.soa_retry }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "Expire" %}</th>
|
||||
<td>{{ object.soa_expire }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "Minimum" %}</th>
|
||||
<td>{{ object.soa_minimum }}</td>
|
||||
</tr>
|
||||
</table>
|
||||
</div>
|
||||
{% include 'inc/panels/custom_fields.html' %}
|
||||
{% plugin_left_page object %}
|
||||
</div>
|
||||
<div class="col col-md-6">
|
||||
{% include 'inc/panels/tags.html' %}
|
||||
{% plugin_right_page object %}
|
||||
</div>
|
||||
</div>
|
||||
<div class="row">
|
||||
<div class="col col-md-12">
|
||||
{% plugin_full_width_page object %}
|
||||
</div>
|
||||
</div>
|
||||
{% endblock %}
|
||||
@@ -0,0 +1,42 @@
|
||||
{% extends 'generic/object.html' %}
|
||||
{% load plugins %}
|
||||
{% load i18n %}
|
||||
|
||||
{% block content %}
|
||||
<div class="row">
|
||||
<div class="col col-md-6">
|
||||
<div class="card">
|
||||
<h5 class="card-header">{% trans "Catalog Zone Member" %}</h5>
|
||||
<table class="table table-hover attr-table">
|
||||
<tr>
|
||||
<th scope="row">{% trans "Name" %}</th>
|
||||
<td>{{ object.name }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "Zone" %}</th>
|
||||
<td>{{ object.zone|linkify }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "Catalog Zone" %}</th>
|
||||
<td>{{ object.catalog_zone|linkify }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "View" %}</th>
|
||||
<td>{{ object.catalog_zone.view|linkify }}</td>
|
||||
</tr>
|
||||
</table>
|
||||
</div>
|
||||
{% include 'inc/panels/custom_fields.html' %}
|
||||
{% plugin_left_page object %}
|
||||
</div>
|
||||
<div class="col col-md-6">
|
||||
{% include 'inc/panels/tags.html' %}
|
||||
{% plugin_right_page object %}
|
||||
</div>
|
||||
</div>
|
||||
<div class="row">
|
||||
<div class="col col-md-12">
|
||||
{% plugin_full_width_page object %}
|
||||
</div>
|
||||
</div>
|
||||
{% endblock %}
|
||||
@@ -1,37 +0,0 @@
|
||||
{% extends 'base/layout.html' %}
|
||||
|
||||
{% block title %}DNS Bridge Notify{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<div class="row">
|
||||
<div class="col-md-12">
|
||||
|
||||
<div class="card mb-3">
|
||||
<h5 class="card-header">Seen Transfer Clients</h5>
|
||||
<div class="card-body">
|
||||
<table class="table table-striped">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Source IP</th>
|
||||
<th>View</th>
|
||||
<th>Last Seen</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for c in seen_clients %}
|
||||
<tr>
|
||||
<td>{{ c.source_ip }}</td>
|
||||
<td>{{ c.view.name }}</td>
|
||||
<td>{{ c.last_seen|date:"Y-m-d H:i:s" }}</td>
|
||||
</tr>
|
||||
{% empty %}
|
||||
<tr><td colspan="3" class="text-muted">No seen transfer clients.</td></tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
</div>
|
||||
</div>
|
||||
{% endblock %}
|
||||
@@ -0,0 +1,38 @@
|
||||
{% extends 'generic/object.html' %}
|
||||
{% load plugins %}
|
||||
{% load i18n %}
|
||||
|
||||
{% block content %}
|
||||
<div class="row">
|
||||
<div class="col col-md-6">
|
||||
<div class="card">
|
||||
<h5 class="card-header">{% trans "Seen Transfer Client" %}</h5>
|
||||
<table class="table table-hover attr-table">
|
||||
<tr>
|
||||
<th scope="row">{% trans "Source IP" %}</th>
|
||||
<td>{{ object.source_ip }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "View" %}</th>
|
||||
<td>{{ object.view|linkify }}</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<th scope="row">{% trans "Last Seen" %}</th>
|
||||
<td>{{ object.last_seen|date:"Y-m-d H:i:s" }}</td>
|
||||
</tr>
|
||||
</table>
|
||||
</div>
|
||||
{% include 'inc/panels/custom_fields.html' %}
|
||||
{% plugin_left_page object %}
|
||||
</div>
|
||||
<div class="col col-md-6">
|
||||
{% include 'inc/panels/tags.html' %}
|
||||
{% plugin_right_page object %}
|
||||
</div>
|
||||
</div>
|
||||
<div class="row">
|
||||
<div class="col col-md-12">
|
||||
{% plugin_full_width_page object %}
|
||||
</div>
|
||||
</div>
|
||||
{% endblock %}
|
||||
@@ -182,7 +182,7 @@ def catz_serial(db):
|
||||
"""Initial SOA serial value (1) for freshly created catalog zones.
|
||||
|
||||
A ``CatalogZone`` is auto-created with ``soa_serial=1`` by the zone
|
||||
post_save signal (via ``update_member_identifier``) whenever a zone is
|
||||
post_save signal (via ``update_member``) whenever a zone is
|
||||
created. Tests compare against this initial value.
|
||||
"""
|
||||
return 1
|
||||
|
||||
@@ -9,7 +9,7 @@ from netbox_dns_bridge.management.commands.dns_transfer_endpoint import (
|
||||
)
|
||||
from netbox_dns_bridge.models import (
|
||||
CatalogZone,
|
||||
CatalogZoneMemberIdentifier,
|
||||
CatalogZoneMember,
|
||||
)
|
||||
|
||||
|
||||
@@ -72,50 +72,50 @@ class TestIncrementSoaSerial:
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestGenerateMemberIdentifier:
|
||||
class TestGenerateMemberName:
|
||||
def test_returns_26_char_string(self):
|
||||
name = catzm._generate_member_identifier()
|
||||
name = catzm._generate_member_name()
|
||||
assert len(name) == 26
|
||||
|
||||
def test_is_lowercase(self):
|
||||
name = catzm._generate_member_identifier()
|
||||
name = catzm._generate_member_name()
|
||||
assert name == name.lower()
|
||||
|
||||
def test_is_base32(self):
|
||||
import string
|
||||
|
||||
name = catzm._generate_member_identifier()
|
||||
name = catzm._generate_member_name()
|
||||
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)}
|
||||
names = {catzm._generate_member_name() for _ in range(100)}
|
||||
assert len(names) == 100
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestUpdateMemberIdentifier:
|
||||
class TestUpdateMember:
|
||||
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()
|
||||
CatalogZoneMember.objects.filter(zone=zone).delete()
|
||||
assert not CatalogZoneMember.objects.filter(zone=zone).exists()
|
||||
catzm.update_member(zone)
|
||||
assert CatalogZoneMember.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)
|
||||
catzm.update_member(zone)
|
||||
old = CatalogZoneMember.objects.get(zone=zone)
|
||||
catzm.update_member(zone)
|
||||
new = CatalogZoneMember.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
|
||||
catzm.update_member(zone)
|
||||
catzm.update_member(zone)
|
||||
assert CatalogZoneMember.objects.filter(zone=zone).count() == 1
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@@ -195,11 +195,11 @@ class TestCreateZone:
|
||||
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()
|
||||
CatalogZoneMember.objects.filter(zone=zone_obj).delete()
|
||||
assert not CatalogZoneMember.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()
|
||||
assert CatalogZoneMember.objects.filter(zone=zone_obj).exists()
|
||||
|
||||
def test_rdclass_is_in(self, make_zone, make_view):
|
||||
view = make_view(name="cz-class-view")
|
||||
|
||||
@@ -3,8 +3,8 @@ import pytest
|
||||
|
||||
from netbox_dns_bridge.models import (
|
||||
CatalogZone,
|
||||
CatalogZoneMemberIdentifier,
|
||||
SeenTransferClients,
|
||||
CatalogZoneMember,
|
||||
SeenTransferClient,
|
||||
)
|
||||
|
||||
|
||||
@@ -42,6 +42,15 @@ class TestCatalogZone:
|
||||
assert cz.soa_expire == 86400
|
||||
assert cz.soa_minimum == 3600
|
||||
|
||||
def test_is_full_netbox_model(self, make_view):
|
||||
view = make_view(name="cz-netbox-model-view")
|
||||
cz = CatalogZone.objects.create(view=view)
|
||||
assert cz.created is not None
|
||||
assert cz.last_updated is not None
|
||||
assert cz.custom_field_data == {}
|
||||
assert hasattr(cz, "tags")
|
||||
assert cz.get_absolute_url() == f"/plugins/dns-bridge/catalog-zones/{cz.pk}/"
|
||||
|
||||
def test_one_to_one_relationship(self, make_view):
|
||||
view = make_view(name="cz-oto-view")
|
||||
cz = CatalogZone.objects.create(view=view, soa_serial=7)
|
||||
@@ -67,10 +76,10 @@ class TestCatalogZone:
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestCatalogZoneMemberIdentifier:
|
||||
class TestCatalogZoneMember:
|
||||
def test_str_representation(self, make_zone):
|
||||
zone = make_zone(name="str.test")
|
||||
m = CatalogZoneMemberIdentifier.objects.get(zone=zone)
|
||||
m = CatalogZoneMember.objects.get(zone=zone)
|
||||
assert str(m) == m.name
|
||||
|
||||
def test_unique_name_per_catalog_zone(self, make_zone, make_view):
|
||||
@@ -78,11 +87,11 @@ class TestCatalogZoneMemberIdentifier:
|
||||
zone1 = make_zone(name="uniq1.test", view=view)
|
||||
zone2 = make_zone(name="uniq2.test", view=view)
|
||||
|
||||
ident1 = CatalogZoneMemberIdentifier.objects.get(zone=zone1)
|
||||
ident1 = CatalogZoneMember.objects.get(zone=zone1)
|
||||
ident1.name = "shared-name"
|
||||
ident1.save()
|
||||
|
||||
ident2 = CatalogZoneMemberIdentifier.objects.get(zone=zone2)
|
||||
ident2 = CatalogZoneMember.objects.get(zone=zone2)
|
||||
from django.db import IntegrityError
|
||||
|
||||
with pytest.raises(IntegrityError):
|
||||
@@ -95,36 +104,36 @@ class TestCatalogZoneMemberIdentifier:
|
||||
zone1 = make_zone(name="diff1.test", view=view1)
|
||||
zone2 = make_zone(name="diff2.test", view=view2)
|
||||
|
||||
ident1 = CatalogZoneMemberIdentifier.objects.get(zone=zone1)
|
||||
ident1 = CatalogZoneMember.objects.get(zone=zone1)
|
||||
ident1.name = "shared-name"
|
||||
ident1.save()
|
||||
|
||||
ident2 = CatalogZoneMemberIdentifier.objects.get(zone=zone2)
|
||||
ident2 = CatalogZoneMember.objects.get(zone=zone2)
|
||||
ident2.name = "shared-name"
|
||||
ident2.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()
|
||||
m = CatalogZoneMember.objects.get(zone=zone)
|
||||
assert CatalogZoneMember.objects.filter(pk=m.pk).exists()
|
||||
zone.delete()
|
||||
assert not CatalogZoneMemberIdentifier.objects.filter(pk=m.pk).exists()
|
||||
assert not CatalogZoneMember.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
|
||||
m = CatalogZoneMember.objects.get(zone=zone)
|
||||
assert zone.catalog_zone_member == m
|
||||
|
||||
def test_ordering_by_name(self, make_zone, make_view):
|
||||
view = make_view(name="ord-view")
|
||||
z1 = make_zone(name="z1.test", view=view)
|
||||
z2 = make_zone(name="z2.test", view=view)
|
||||
z3 = make_zone(name="z3.test", view=view)
|
||||
CatalogZoneMemberIdentifier.objects.filter(zone=z3).update(name="gamma")
|
||||
CatalogZoneMemberIdentifier.objects.filter(zone=z1).update(name="alpha")
|
||||
CatalogZoneMemberIdentifier.objects.filter(zone=z2).update(name="beta")
|
||||
CatalogZoneMember.objects.filter(zone=z3).update(name="gamma")
|
||||
CatalogZoneMember.objects.filter(zone=z1).update(name="alpha")
|
||||
CatalogZoneMember.objects.filter(zone=z2).update(name="beta")
|
||||
names = list(
|
||||
CatalogZoneMemberIdentifier.objects.filter(
|
||||
CatalogZoneMember.objects.filter(
|
||||
zone__in=[z1, z2, z3]
|
||||
).values_list("name", flat=True)
|
||||
)
|
||||
@@ -132,10 +141,10 @@ class TestCatalogZoneMemberIdentifier:
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestSeenTransferClients:
|
||||
class TestSeenTransferClient:
|
||||
def test_str_representation(self, make_view):
|
||||
view = make_view(name="str-view")
|
||||
client = SeenTransferClients.objects.create(
|
||||
client = SeenTransferClient.objects.create(
|
||||
source_ip="192.0.2.1",
|
||||
view=view,
|
||||
last_seen="2026-01-01T00:00:00Z",
|
||||
@@ -146,13 +155,13 @@ class TestSeenTransferClients:
|
||||
from django.db import IntegrityError
|
||||
|
||||
view = make_view(name="uniq-view")
|
||||
SeenTransferClients.objects.create(
|
||||
SeenTransferClient.objects.create(
|
||||
source_ip="10.0.0.1",
|
||||
view=view,
|
||||
last_seen="2026-01-01T00:00:00Z",
|
||||
)
|
||||
with pytest.raises(IntegrityError):
|
||||
SeenTransferClients.objects.create(
|
||||
SeenTransferClient.objects.create(
|
||||
source_ip="10.0.0.1",
|
||||
view=view,
|
||||
last_seen="2026-01-02T00:00:00Z",
|
||||
@@ -161,19 +170,19 @@ class TestSeenTransferClients:
|
||||
def test_same_ip_different_views_allowed(self, make_view):
|
||||
v1 = make_view(name="v1")
|
||||
v2 = make_view(name="v2")
|
||||
SeenTransferClients.objects.create(
|
||||
SeenTransferClient.objects.create(
|
||||
source_ip="10.0.0.5", view=v1, last_seen="2026-01-01T00:00:00Z"
|
||||
)
|
||||
SeenTransferClients.objects.create(
|
||||
SeenTransferClient.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
|
||||
assert SeenTransferClient.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(
|
||||
SeenTransferClient.objects.create(
|
||||
source_ip="10.0.0.9",
|
||||
view=view,
|
||||
last_seen="2026-01-01T00:00:00Z",
|
||||
@@ -187,6 +196,19 @@ class TestSeenTransferClients:
|
||||
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))
|
||||
SeenTransferClient.objects.create(source_ip=ip, view=v, last_seen=now)
|
||||
ips = list(SeenTransferClient.objects.values_list("source_ip", flat=True))
|
||||
assert ips == ["10.0.0.1", "10.0.0.2", "10.0.0.3"]
|
||||
|
||||
def test_is_full_netbox_model(self, make_view):
|
||||
from django.utils import timezone
|
||||
|
||||
view = make_view(name="stc-netbox-model-view")
|
||||
c = SeenTransferClient.objects.create(
|
||||
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
|
||||
)
|
||||
assert c.created is not None
|
||||
assert c.last_updated is not None
|
||||
assert c.custom_field_data == {}
|
||||
assert hasattr(c, "tags")
|
||||
assert c.get_absolute_url() == f"/plugins/dns-bridge/seen-transfer-clients/{c.pk}/"
|
||||
@@ -269,11 +269,11 @@ class TestScheduleCatalogZoneNotify:
|
||||
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 netbox_dns_bridge.models import SeenTransferClient
|
||||
from django.utils import timezone
|
||||
|
||||
view = make_view(name="notify-run-view")
|
||||
SeenTransferClients.objects.create(
|
||||
SeenTransferClient.objects.create(
|
||||
source_ip="192.0.2.99", view=view, last_seen=timezone.now()
|
||||
)
|
||||
|
||||
@@ -297,11 +297,11 @@ class TestSendClientNotifyRun:
|
||||
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 netbox_dns_bridge.models import SeenTransferClient
|
||||
from django.utils import timezone
|
||||
|
||||
view = make_view(name="no-catz-view")
|
||||
SeenTransferClients.objects.create(
|
||||
SeenTransferClient.objects.create(
|
||||
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
|
||||
)
|
||||
|
||||
|
||||
@@ -148,14 +148,14 @@ class TestCatzIntegration:
|
||||
|
||||
class TestSeenClientTracking:
|
||||
def _wait_for_seen_client(self, view, timeout=3.0):
|
||||
"""Poll for SeenTransferClients record (race: response is sent before
|
||||
"""Poll for SeenTransferClient record (race: response is sent before
|
||||
_track_seen_client completes in the handler thread)."""
|
||||
import time
|
||||
from netbox_dns_bridge.models import SeenTransferClients
|
||||
from netbox_dns_bridge.models import SeenTransferClient
|
||||
|
||||
deadline = time.monotonic() + timeout
|
||||
while time.monotonic() < deadline:
|
||||
if SeenTransferClients.objects.filter(
|
||||
if SeenTransferClient.objects.filter(
|
||||
source_ip="127.0.0.1", view=view
|
||||
).exists():
|
||||
return True
|
||||
@@ -186,5 +186,5 @@ class TestSeenClientTracking:
|
||||
|
||||
import time
|
||||
time.sleep(0.3)
|
||||
from netbox_dns_bridge.models import SeenTransferClients
|
||||
assert not SeenTransferClients.objects.filter(view=view).exists()
|
||||
from netbox_dns_bridge.models import SeenTransferClient
|
||||
assert not SeenTransferClient.objects.filter(view=view).exists()
|
||||
@@ -13,7 +13,7 @@ captureOnCommitCallbacks = _DjangoTestCase.captureOnCommitCallbacks
|
||||
|
||||
from netbox_dns_bridge.models import (
|
||||
CatalogZone,
|
||||
CatalogZoneMemberIdentifier,
|
||||
CatalogZoneMember,
|
||||
)
|
||||
|
||||
|
||||
@@ -35,9 +35,9 @@ class TestZonePreSave:
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestZonePostSaveCreate:
|
||||
def test_creates_catz_member_identifier_on_new_zone(self, make_zone):
|
||||
def test_creates_catalog_zone_member_on_new_zone(self, make_zone):
|
||||
zone = make_zone(name="postcreate.test")
|
||||
assert CatalogZoneMemberIdentifier.objects.filter(zone=zone).exists()
|
||||
assert CatalogZoneMember.objects.filter(zone=zone).exists()
|
||||
|
||||
def test_increments_soa_serial_on_commit(self, make_zone, catz_serial):
|
||||
with captureOnCommitCallbacks(execute=True):
|
||||
@@ -103,19 +103,19 @@ class TestZonePostSaveCreate:
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestZonePostSaveUpdate:
|
||||
def test_updates_catz_identifier_on_name_change(self, make_zone):
|
||||
def test_updates_member_on_name_change(self, make_zone):
|
||||
zone = make_zone(name="rename-old.test")
|
||||
old_name = CatalogZoneMemberIdentifier.objects.get(zone=zone).name
|
||||
old_name = CatalogZoneMember.objects.get(zone=zone).name
|
||||
|
||||
zone.name = "rename-new.test"
|
||||
zone.save()
|
||||
|
||||
new_name = CatalogZoneMemberIdentifier.objects.get(zone=zone).name
|
||||
new_name = CatalogZoneMember.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)
|
||||
ident = CatalogZoneMember.objects.get(zone=zone)
|
||||
original_name = ident.name
|
||||
|
||||
zone.soa_serial = zone.soa_serial + 1
|
||||
@@ -177,7 +177,7 @@ class TestZoneViewChange:
|
||||
view_b = make_view(name="fk-view-b")
|
||||
zone = make_zone(name="fk-move.test", view=view_a)
|
||||
|
||||
ident = CatalogZoneMemberIdentifier.objects.get(zone=zone)
|
||||
ident = CatalogZoneMember.objects.get(zone=zone)
|
||||
assert ident.catalog_zone == CatalogZone.objects.get(view=view_a)
|
||||
|
||||
zone.view = view_b
|
||||
|
||||
@@ -14,12 +14,12 @@ def test_plugin_importable():
|
||||
def test_models_importable():
|
||||
from netbox_dns_bridge.models import (
|
||||
CatalogZone,
|
||||
CatalogZoneMemberIdentifier,
|
||||
SeenTransferClients,
|
||||
CatalogZoneMember,
|
||||
SeenTransferClient,
|
||||
)
|
||||
assert CatalogZone._meta.model_name
|
||||
assert CatalogZoneMemberIdentifier._meta.model_name
|
||||
assert SeenTransferClients._meta.model_name
|
||||
assert CatalogZoneMember._meta.model_name
|
||||
assert SeenTransferClient._meta.model_name
|
||||
|
||||
|
||||
def test_request_handler_importable():
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Tests for views: CatalogZoneMembersView, CatalogZoneListView, CatalogZoneEditView, NotifyOverviewView — auth gate and context."""
|
||||
"""Tests for views: CatalogZone list/detail/edit, CatalogZoneMember list/detail, SeenTransferClient list/detail — auth gate and context."""
|
||||
import pytest
|
||||
from django.test import Client
|
||||
from django.contrib.auth import get_user_model
|
||||
@@ -28,7 +28,7 @@ def regular_user(db):
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestCatalogZoneMembersView:
|
||||
class TestCatalogZoneMemberListView:
|
||||
def test_unauthenticated_redirects_to_login(self, client):
|
||||
resp = client.get("/plugins/dns-bridge/catalog-zone-members/")
|
||||
assert resp.status_code in (302, 403)
|
||||
@@ -38,11 +38,48 @@ class TestCatalogZoneMembersView:
|
||||
resp = client.get("/plugins/dns-bridge/catalog-zone-members/")
|
||||
assert resp.status_code in (302, 403)
|
||||
|
||||
def test_superuser_can_access(self, client, superuser):
|
||||
def test_superuser_can_access(self, client, superuser, make_zone):
|
||||
make_zone(name="member-list.test")
|
||||
client.login(username="admin", password="testpass")
|
||||
resp = client.get("/plugins/dns-bridge/catalog-zone-members/")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_list_contains_members(self, client, superuser, make_zone):
|
||||
zone = make_zone(name="member-row.test")
|
||||
client.login(username="admin", password="testpass")
|
||||
resp = client.get("/plugins/dns-bridge/catalog-zone-members/")
|
||||
member_list = list(resp.context["table"].data)
|
||||
assert len(member_list) >= 1
|
||||
m = next(m for m in member_list if m.zone_id == zone.id)
|
||||
assert m.name
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestCatalogZoneMemberDetailView:
|
||||
def test_unauthenticated_redirects_to_login(self, client, make_zone):
|
||||
zone = make_zone(name="member-detail.test")
|
||||
from netbox_dns_bridge.models import CatalogZoneMember
|
||||
m = CatalogZoneMember.objects.get(zone=zone)
|
||||
resp = client.get(f"/plugins/dns-bridge/catalog-zone-members/{m.pk}/")
|
||||
assert resp.status_code in (302, 403)
|
||||
|
||||
def test_regular_user_forbidden(self, client, regular_user, make_zone):
|
||||
zone = make_zone(name="member-detail-forbidden.test")
|
||||
from netbox_dns_bridge.models import CatalogZoneMember
|
||||
m = CatalogZoneMember.objects.get(zone=zone)
|
||||
client.login(username="regular", password="testpass")
|
||||
resp = client.get(f"/plugins/dns-bridge/catalog-zone-members/{m.pk}/")
|
||||
assert resp.status_code in (302, 403)
|
||||
|
||||
def test_superuser_can_access_detail(self, client, superuser, make_zone):
|
||||
zone = make_zone(name="member-detail-get.test")
|
||||
from netbox_dns_bridge.models import CatalogZoneMember
|
||||
m = CatalogZoneMember.objects.get(zone=zone)
|
||||
client.login(username="admin", password="testpass")
|
||||
resp = client.get(f"/plugins/dns-bridge/catalog-zone-members/{m.pk}/")
|
||||
assert resp.status_code == 200
|
||||
assert resp.context["object"].name == m.name
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestCatalogZoneListView:
|
||||
@@ -138,45 +175,104 @@ class TestCatalogZoneEditView:
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestNotifyOverviewView:
|
||||
class TestCatalogZoneDetailView:
|
||||
def test_unauthenticated_redirects_to_login(self, client, make_view):
|
||||
view = make_view(name="detail-cz-view")
|
||||
from netbox_dns_bridge.models import CatalogZone
|
||||
cz = CatalogZone.objects.create(view=view)
|
||||
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/")
|
||||
assert resp.status_code in (302, 403)
|
||||
|
||||
def test_regular_user_forbidden(self, client, regular_user, make_view):
|
||||
view = make_view(name="detail-cz-view-forbidden")
|
||||
from netbox_dns_bridge.models import CatalogZone
|
||||
cz = CatalogZone.objects.create(view=view)
|
||||
client.login(username="regular", password="testpass")
|
||||
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/")
|
||||
assert resp.status_code in (302, 403)
|
||||
|
||||
def test_superuser_can_access_detail(self, client, superuser, make_view):
|
||||
view = make_view(name="detail-cz-view-get")
|
||||
from netbox_dns_bridge.models import CatalogZone
|
||||
cz = CatalogZone.objects.create(view=view, soa_serial=42)
|
||||
client.login(username="admin", password="testpass")
|
||||
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/")
|
||||
assert resp.status_code == 200
|
||||
assert resp.context["object"].soa_serial == 42
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestSeenTransferClientListView:
|
||||
def test_unauthenticated_redirects_to_login(self, client):
|
||||
resp = client.get("/plugins/dns-bridge/notify/")
|
||||
resp = client.get("/plugins/dns-bridge/seen-transfer-clients/")
|
||||
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/")
|
||||
resp = client.get("/plugins/dns-bridge/seen-transfer-clients/")
|
||||
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
|
||||
def test_superuser_can_access(self, client, superuser, make_view):
|
||||
from netbox_dns_bridge.models import SeenTransferClient
|
||||
from django.utils import timezone
|
||||
|
||||
view = make_view(name="notify-view")
|
||||
SeenTransferClients.objects.create(
|
||||
view = make_view(name="stc-list-view")
|
||||
SeenTransferClient.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"
|
||||
resp = client.get("/plugins/dns-bridge/seen-transfer-clients/")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_seen_clients_ordered_by_ip(self, client, superuser, make_view):
|
||||
from netbox_dns_bridge.models import SeenTransferClients
|
||||
def test_list_contains_clients(self, client, superuser, make_view):
|
||||
from netbox_dns_bridge.models import SeenTransferClient
|
||||
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)
|
||||
view = make_view(name="stc-list-row-view")
|
||||
SeenTransferClient.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/")
|
||||
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"]
|
||||
resp = client.get("/plugins/dns-bridge/seen-transfer-clients/")
|
||||
client_list = list(resp.context["table"].data)
|
||||
assert len(client_list) >= 1
|
||||
assert client_list[0].source_ip == "192.0.2.1"
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestSeenTransferClientDetailView:
|
||||
def test_unauthenticated_redirects_to_login(self, client, make_view):
|
||||
from netbox_dns_bridge.models import SeenTransferClient
|
||||
from django.utils import timezone
|
||||
|
||||
view = make_view(name="stc-detail-view")
|
||||
c = SeenTransferClient.objects.create(
|
||||
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
|
||||
)
|
||||
resp = client.get(f"/plugins/dns-bridge/seen-transfer-clients/{c.pk}/")
|
||||
assert resp.status_code in (302, 403)
|
||||
|
||||
def test_regular_user_forbidden(self, client, regular_user, make_view):
|
||||
from netbox_dns_bridge.models import SeenTransferClient
|
||||
from django.utils import timezone
|
||||
|
||||
view = make_view(name="stc-detail-forbidden-view")
|
||||
c = SeenTransferClient.objects.create(
|
||||
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
|
||||
)
|
||||
client.login(username="regular", password="testpass")
|
||||
resp = client.get(f"/plugins/dns-bridge/seen-transfer-clients/{c.pk}/")
|
||||
assert resp.status_code in (302, 403)
|
||||
|
||||
def test_superuser_can_access_detail(self, client, superuser, make_view):
|
||||
from netbox_dns_bridge.models import SeenTransferClient
|
||||
from django.utils import timezone
|
||||
|
||||
view = make_view(name="stc-detail-get-view")
|
||||
c = SeenTransferClient.objects.create(
|
||||
source_ip="192.0.2.42", view=view, last_seen=timezone.now()
|
||||
)
|
||||
client.login(username="admin", password="testpass")
|
||||
resp = client.get(f"/plugins/dns-bridge/seen-transfer-clients/{c.pk}/")
|
||||
assert resp.status_code == 200
|
||||
assert resp.context["object"].source_ip == "192.0.2.42"
|
||||
+35
-11
@@ -1,23 +1,47 @@
|
||||
from django.urls import path
|
||||
from django.urls import include, path
|
||||
|
||||
from .views import (
|
||||
from utilities.urls import get_model_urls
|
||||
|
||||
# Import views so register_model_view decorators run.
|
||||
from .views import ( # noqa: F401
|
||||
CatalogZoneEditView,
|
||||
CatalogZoneListView,
|
||||
CatalogZoneMembersView,
|
||||
NotifyOverviewView,
|
||||
CatalogZoneMemberListView,
|
||||
CatalogZoneMemberView,
|
||||
CatalogZoneView,
|
||||
SeenTransferClientListView,
|
||||
SeenTransferClientView,
|
||||
)
|
||||
|
||||
urlpatterns = [
|
||||
path("catalog-zones/", CatalogZoneListView.as_view(), name="catalog_zones"),
|
||||
path(
|
||||
"catalog-zones/<int:pk>/edit/",
|
||||
CatalogZoneEditView.as_view(),
|
||||
name="catalogzone_edit",
|
||||
"catalog-zones/",
|
||||
include(
|
||||
get_model_urls("netbox_dns_bridge", "catalogzone", detail=False)
|
||||
),
|
||||
),
|
||||
path(
|
||||
"catalog-zones/<int:pk>/",
|
||||
include(get_model_urls("netbox_dns_bridge", "catalogzone")),
|
||||
),
|
||||
path(
|
||||
"catalog-zone-members/",
|
||||
CatalogZoneMembersView.as_view(),
|
||||
name="catalog_zone_members",
|
||||
include(
|
||||
get_model_urls("netbox_dns_bridge", "catalogzonemember", detail=False)
|
||||
),
|
||||
),
|
||||
path(
|
||||
"catalog-zone-members/<int:pk>/",
|
||||
include(get_model_urls("netbox_dns_bridge", "catalogzonemember")),
|
||||
),
|
||||
path(
|
||||
"seen-transfer-clients/",
|
||||
include(
|
||||
get_model_urls("netbox_dns_bridge", "seentransferclient", detail=False)
|
||||
),
|
||||
),
|
||||
path(
|
||||
"seen-transfer-clients/<int:pk>/",
|
||||
include(get_model_urls("netbox_dns_bridge", "seentransferclient")),
|
||||
),
|
||||
path("notify/", NotifyOverviewView.as_view(), name="notify"),
|
||||
]
|
||||
+59
-50
@@ -1,67 +1,76 @@
|
||||
from django.contrib.auth.decorators import user_passes_test
|
||||
from django.shortcuts import render
|
||||
from django.utils.decorators import method_decorator
|
||||
from django.views import View
|
||||
from netbox.object_actions import EditObject
|
||||
from netbox.views import generic
|
||||
from utilities.views import register_model_view
|
||||
|
||||
import netbox_dns.models
|
||||
from netbox.views.generic import ObjectEditView, ObjectListView
|
||||
|
||||
from .filtersets import CatalogZoneFilterSet, ZoneFilterSet
|
||||
from .filtersets import (
|
||||
CatalogZoneFilterSet,
|
||||
CatalogZoneMemberFilterSet,
|
||||
SeenTransferClientFilterSet,
|
||||
)
|
||||
from .forms import (
|
||||
CatalogZoneFilterForm,
|
||||
CatalogZoneForm,
|
||||
CatalogZoneMemberFilterForm,
|
||||
SeenTransferClientFilterForm,
|
||||
)
|
||||
from .models import CatalogZone, CatalogZoneMember, SeenTransferClient
|
||||
from .tables import (
|
||||
CatalogZoneMemberTable,
|
||||
CatalogZoneTable,
|
||||
SeenTransferClientTable,
|
||||
)
|
||||
from .models import CatalogZone, SeenTransferClients
|
||||
from .tables import CatalogZoneMemberTable, CatalogZoneTable
|
||||
|
||||
|
||||
@method_decorator(
|
||||
user_passes_test(lambda u: u.is_superuser), name="dispatch"
|
||||
)
|
||||
class CatalogZoneListView(ObjectListView):
|
||||
@register_model_view(CatalogZone, "list", path="", detail=False)
|
||||
class CatalogZoneListView(generic.ObjectListView):
|
||||
queryset = CatalogZone.objects.select_related("view").order_by("view__name")
|
||||
table = CatalogZoneTable
|
||||
filterset = CatalogZoneFilterSet
|
||||
filterset_form = CatalogZoneFilterForm
|
||||
template_name = "netbox_dns_bridge/catalog_zone_list.html"
|
||||
actions = ("edit", "export")
|
||||
|
||||
|
||||
@method_decorator(
|
||||
user_passes_test(lambda u: u.is_superuser), name="dispatch"
|
||||
)
|
||||
class CatalogZoneEditView(ObjectEditView):
|
||||
queryset = CatalogZone.objects.select_related("view")
|
||||
form = CatalogZoneForm
|
||||
template_name = "netbox_dns_bridge/catalog_zone_edit.html"
|
||||
|
||||
|
||||
@method_decorator(
|
||||
user_passes_test(lambda u: u.is_superuser), name="dispatch"
|
||||
)
|
||||
class CatalogZoneMembersView(ObjectListView):
|
||||
queryset = netbox_dns.models.Zone.objects.select_related(
|
||||
"view", "catz_identifier"
|
||||
).order_by("view__name", "name")
|
||||
table = CatalogZoneMemberTable
|
||||
filterset = ZoneFilterSet
|
||||
filterset_form = CatalogZoneMemberFilterForm
|
||||
template_name = "netbox_dns_bridge/catalog_zone_members.html"
|
||||
actions = ()
|
||||
|
||||
|
||||
@method_decorator(
|
||||
user_passes_test(lambda u: u.is_superuser), name="dispatch"
|
||||
)
|
||||
class NotifyOverviewView(View):
|
||||
template_name = "netbox_dns_bridge/notify_overview.html"
|
||||
@register_model_view(CatalogZone)
|
||||
class CatalogZoneView(generic.ObjectView):
|
||||
queryset = CatalogZone.objects.select_related("view")
|
||||
actions = (EditObject,)
|
||||
|
||||
def get(self, request):
|
||||
seen_clients = (
|
||||
SeenTransferClients.objects.select_related("view")
|
||||
.all()
|
||||
.order_by("source_ip")
|
||||
|
||||
@register_model_view(CatalogZone, "edit")
|
||||
class CatalogZoneEditView(generic.ObjectEditView):
|
||||
queryset = CatalogZone.objects.select_related("view")
|
||||
form = CatalogZoneForm
|
||||
|
||||
|
||||
@register_model_view(CatalogZoneMember, "list", path="", detail=False)
|
||||
class CatalogZoneMemberListView(generic.ObjectListView):
|
||||
queryset = CatalogZoneMember.objects.select_related(
|
||||
"zone", "catalog_zone__view"
|
||||
).order_by("catalog_zone__view__name", "name")
|
||||
table = CatalogZoneMemberTable
|
||||
filterset = CatalogZoneMemberFilterSet
|
||||
filterset_form = CatalogZoneMemberFilterForm
|
||||
actions = ()
|
||||
|
||||
|
||||
@register_model_view(CatalogZoneMember)
|
||||
class CatalogZoneMemberView(generic.ObjectView):
|
||||
queryset = CatalogZoneMember.objects.select_related(
|
||||
"zone", "catalog_zone__view"
|
||||
)
|
||||
context = {"seen_clients": seen_clients}
|
||||
return render(request, self.template_name, context)
|
||||
actions = ()
|
||||
|
||||
|
||||
@register_model_view(SeenTransferClient, "list", path="", detail=False)
|
||||
class SeenTransferClientListView(generic.ObjectListView):
|
||||
queryset = SeenTransferClient.objects.select_related("view").order_by("source_ip")
|
||||
table = SeenTransferClientTable
|
||||
filterset = SeenTransferClientFilterSet
|
||||
filterset_form = SeenTransferClientFilterForm
|
||||
actions = ()
|
||||
|
||||
|
||||
@register_model_view(SeenTransferClient)
|
||||
class SeenTransferClientView(generic.ObjectView):
|
||||
queryset = SeenTransferClient.objects.select_related("view")
|
||||
actions = ()
|
||||
Reference in new issue
Block a user