Compare commits

..

4 Commits

Author SHA1 Message Date
Arthur
7b9ab87e38 cleanup 2026-04-10 12:04:48 -07:00
Arthur
90255a268f cleanup 2026-04-10 11:57:49 -07:00
Arthur
8418809344 #21879 - Add post_raw_create signal hook 2026-04-10 11:31:56 -07:00
Arthur
b39e1c73c1 #21879 - Add post_raw_create signal hook 2026-04-10 11:28:10 -07:00
7 changed files with 67 additions and 103 deletions

View File

@@ -56,9 +56,9 @@ jobs:
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
- name: Check Python linting & PEP8 compliance - name: Check Python linting & PEP8 compliance
uses: astral-sh/ruff-action@0ce1b0bf8b818ef400413f810f8a11cdbda0034b # v4.0.0 uses: astral-sh/ruff-action@4919ec5cf1f49eff0871dbcea0da843445b837e6 # v3.6.1
with: with:
version: "0.15.10" version: "0.15.2"
args: "check --output-format=github" args: "check --output-format=github"
src: "netbox/" src: "netbox/"

View File

@@ -20,7 +20,7 @@ jobs:
steps: steps:
- name: Create app token - name: Create app token
uses: actions/create-github-app-token@1b10c78c7865c340bc4f6099eb2f838309f1e8c3 # v3.1.1 uses: actions/create-github-app-token@v1
id: app-token id: app-token
with: with:
app-id: 1076524 app-id: 1076524
@@ -48,7 +48,7 @@ jobs:
run: python netbox/manage.py makemessages -l ${{ env.LOCALE }} run: python netbox/manage.py makemessages -l ${{ env.LOCALE }}
- name: Commit changes - name: Commit changes
uses: EndBug/add-and-commit@290ea2c423ad77ca9c62ae0f5b224379612c0321 # v10.0.0 uses: EndBug/add-and-commit@a94899bca583c204427a224a7af87c02f9b325d5 # v9.1.4
with: with:
add: 'netbox/translations/' add: 'netbox/translations/'
default_author: github_actions default_author: github_actions

View File

@@ -1409,16 +1409,8 @@ class CableImportForm(PrimaryModelImportForm):
side_a_device = CSVModelChoiceField( side_a_device = CSVModelChoiceField(
label=_('Side A device'), label=_('Side A device'),
queryset=Device.objects.all(), queryset=Device.objects.all(),
required=False,
to_field_name='name', to_field_name='name',
help_text=_('Device name (for device component terminations)') help_text=_('Device name')
)
side_a_power_panel = CSVModelChoiceField(
label=_('Side A power panel'),
queryset=PowerPanel.objects.all(),
required=False,
to_field_name='name',
help_text=_('Power panel name (for power feed terminations)')
) )
side_a_type = CSVContentTypeField( side_a_type = CSVContentTypeField(
label=_('Side A type'), label=_('Side A type'),
@@ -1442,16 +1434,8 @@ class CableImportForm(PrimaryModelImportForm):
side_b_device = CSVModelChoiceField( side_b_device = CSVModelChoiceField(
label=_('Side B device'), label=_('Side B device'),
queryset=Device.objects.all(), queryset=Device.objects.all(),
required=False,
to_field_name='name', to_field_name='name',
help_text=_('Device name (for device component terminations)') help_text=_('Device name')
)
side_b_power_panel = CSVModelChoiceField(
label=_('Side B power panel'),
queryset=PowerPanel.objects.all(),
required=False,
to_field_name='name',
help_text=_('Power panel name (for power feed terminations)')
) )
side_b_type = CSVContentTypeField( side_b_type = CSVContentTypeField(
label=_('Side B type'), label=_('Side B type'),
@@ -1506,9 +1490,8 @@ class CableImportForm(PrimaryModelImportForm):
class Meta: class Meta:
model = Cable model = Cable
fields = [ fields = [
'side_a_site', 'side_a_device', 'side_a_power_panel', 'side_a_type', 'side_a_name', 'side_a_site', 'side_a_device', 'side_a_type', 'side_a_name', 'side_b_site', 'side_b_device', 'side_b_type',
'side_b_site', 'side_b_device', 'side_b_power_panel', 'side_b_type', 'side_b_name', 'side_b_name', 'type', 'status', 'profile', 'tenant', 'label', 'color', 'length', 'length_unit',
'type', 'status', 'profile', 'tenant', 'label', 'color', 'length', 'length_unit',
'description', 'owner', 'comments', 'tags', 'description', 'owner', 'comments', 'tags',
] ]
@@ -1518,22 +1501,16 @@ class CableImportForm(PrimaryModelImportForm):
if data: if data:
# Limit choices for side_a_device to the assigned side_a_site # Limit choices for side_a_device to the assigned side_a_site
if side_a_site := data.get('side_a_site'): if side_a_site := data.get('side_a_site'):
side_a_parent_params = {f'site__{self.fields['side_a_site'].to_field_name}': side_a_site} side_a_device_params = {f'site__{self.fields["side_a_site"].to_field_name}': side_a_site}
self.fields['side_a_device'].queryset = self.fields['side_a_device'].queryset.filter( self.fields['side_a_device'].queryset = self.fields['side_a_device'].queryset.filter(
**side_a_parent_params **side_a_device_params
)
self.fields['side_a_power_panel'].queryset = self.fields['side_a_power_panel'].queryset.filter(
**side_a_parent_params
) )
# Limit choices for side_b_device to the assigned side_b_site # Limit choices for side_b_device to the assigned side_b_site
if side_b_site := data.get('side_b_site'): if side_b_site := data.get('side_b_site'):
side_b_parent_params = {f'site__{self.fields['side_b_site'].to_field_name}': side_b_site} side_b_device_params = {f'site__{self.fields["side_b_site"].to_field_name}': side_b_site}
self.fields['side_b_device'].queryset = self.fields['side_b_device'].queryset.filter( self.fields['side_b_device'].queryset = self.fields['side_b_device'].queryset.filter(
**side_b_parent_params **side_b_device_params
)
self.fields['side_b_power_panel'].queryset = self.fields['side_b_power_panel'].queryset.filter(
**side_b_parent_params
) )
def _clean_side(self, side): def _clean_side(self, side):
@@ -1545,57 +1522,33 @@ class CableImportForm(PrimaryModelImportForm):
assert side in 'ab', f"Invalid side designation: {side}" assert side in 'ab', f"Invalid side designation: {side}"
device = self.cleaned_data.get(f'side_{side}_device') device = self.cleaned_data.get(f'side_{side}_device')
power_panel = self.cleaned_data.get(f'side_{side}_power_panel')
content_type = self.cleaned_data.get(f'side_{side}_type') content_type = self.cleaned_data.get(f'side_{side}_type')
name = self.cleaned_data.get(f'side_{side}_name') name = self.cleaned_data.get(f'side_{side}_name')
if not content_type or not name: if not device or not content_type or not name:
return None return None
model = content_type.model_class() model = content_type.model_class()
try:
# PowerFeed terminations reference a PowerPanel, not a Device if (
if content_type.model == 'powerfeed': device.virtual_chassis and
if not power_panel: device.virtual_chassis.master == device and
return None not model.objects.filter(device=device, name=name).exists()
try: ):
termination_object = model.objects.get(power_panel=power_panel, name=name) termination_object = model.objects.get(device__in=device.virtual_chassis.members.all(), name=name)
if termination_object.cable is not None and termination_object.cable != self.instance: else:
raise forms.ValidationError( termination_object = model.objects.get(device=device, name=name)
_("Side {side_upper}: {power_panel} {termination_object} is already connected").format( if termination_object.cable is not None and termination_object.cable != self.instance:
side_upper=side.upper(), power_panel=power_panel, termination_object=termination_object
)
)
except ObjectDoesNotExist:
raise forms.ValidationError( raise forms.ValidationError(
_("{side_upper} side termination not found: {power_panel} {name}").format( _("Side {side_upper}: {device} {termination_object} is already connected").format(
side_upper=side.upper(), power_panel=power_panel, name=name side_upper=side.upper(), device=device, termination_object=termination_object
) )
) )
else: except ObjectDoesNotExist:
if not device: raise forms.ValidationError(
return None _("{side_upper} side termination not found: {device} {name}").format(
try: side_upper=side.upper(), device=device, name=name
if (
device.virtual_chassis and
device.virtual_chassis.master == device and
not model.objects.filter(device=device, name=name).exists()
):
termination_object = model.objects.get(device__in=device.virtual_chassis.members.all(), name=name)
else:
termination_object = model.objects.get(device=device, name=name)
if termination_object.cable is not None and termination_object.cable != self.instance:
raise forms.ValidationError(
_("Side {side_upper}: {device} {termination_object} is already connected").format(
side_upper=side.upper(), device=device, termination_object=termination_object
)
)
except ObjectDoesNotExist:
raise forms.ValidationError(
_("{side_upper} side termination not found: {device} {name}").format(
side_upper=side.upper(), device=device, name=name
)
) )
)
setattr(self.instance, f'{side}_terminations', [termination_object]) setattr(self.instance, f'{side}_terminations', [termination_object])
return termination_object return termination_object

View File

@@ -6,6 +6,7 @@ from django.dispatch import receiver
from dcim.choices import CableEndChoices, LinkStatusChoices from dcim.choices import CableEndChoices, LinkStatusChoices
from ipam.models import Prefix from ipam.models import Prefix
from netbox.signals import post_raw_create
from virtualization.models import Cluster, VMInterface from virtualization.models import Cluster, VMInterface
from wireless.models import WirelessLAN from wireless.models import WirelessLAN
@@ -166,6 +167,27 @@ def retrace_cable_paths(instance, **kwargs):
cablepath.retrace() cablepath.retrace()
@receiver(post_raw_create, sender=Cable)
def retrace_cable_paths_after_raw_create(sender, pks, **kwargs):
"""
When Cables are created via a raw save, the normal Cable.save() path is bypassed,
so trace_paths is never sent. Retrace paths for all newly created cables.
Callers must only send this signal after all CableTerminations for the given cables
have been applied. If a cable has no terminations, update_connected_endpoints will
find empty termination lists and skip path creation — so this is safe to call even
if terminations are absent, but path tracing will have no effect.
Note: raw=False (the default) is intentional here — we explicitly want
update_connected_endpoints to run, unlike during fixture loading (raw=True).
"""
logger = logging.getLogger('netbox.dcim.cable')
for cable in Cable.objects.filter(pk__in=pks):
cable._terminations_modified = True
trace_paths.send(Cable, instance=cable, created=True)
logger.debug(f"Retraced cable paths for Cable {cable.pk}")
@receiver((post_delete, post_save), sender=PortMapping) @receiver((post_delete, post_save), sender=PortMapping)
def update_passthrough_port_paths(instance, **kwargs): def update_passthrough_port_paths(instance, **kwargs):
""" """

View File

@@ -3603,21 +3603,6 @@ class CableTestCase(
cable3 = Cable(a_terminations=[interfaces[2]], b_terminations=[interfaces[5]], type=CableTypeChoices.TYPE_CAT6) cable3 = Cable(a_terminations=[interfaces[2]], b_terminations=[interfaces[5]], type=CableTypeChoices.TYPE_CAT6)
cable3.save() cable3.save()
# Power panel, power feeds, and power ports for powerfeed-to-powerport cable import tests
power_panel = PowerPanel.objects.create(site=sites[0], name='Power Panel 1')
power_feeds = (
PowerFeed(name='Power Feed 1', power_panel=power_panel),
PowerFeed(name='Power Feed 2', power_panel=power_panel),
PowerFeed(name='Power Feed 3', power_panel=power_panel),
)
PowerFeed.objects.bulk_create(power_feeds)
power_ports = (
PowerPort(device=devices[3], name='Power Port 1'),
PowerPort(device=devices[3], name='Power Port 2'),
PowerPort(device=devices[3], name='Power Port 3'),
)
PowerPort.objects.bulk_create(power_ports)
tags = create_tags('Alpha', 'Bravo', 'Charlie') tags = create_tags('Alpha', 'Bravo', 'Charlie')
cls.form_data = { cls.form_data = {
@@ -3655,14 +3640,7 @@ class CableTestCase(
"Site 1,Device 3,dcim.interface,Interface 3,Site 2,Device 1,dcim.interface,Interface 3", "Site 1,Device 3,dcim.interface,Interface 3,Site 2,Device 1,dcim.interface,Interface 3",
"Site 1,Device 1,dcim.interface,Device 2 Interface,Site 2,Device 1,dcim.interface,Interface 4", "Site 1,Device 1,dcim.interface,Device 2 Interface,Site 2,Device 1,dcim.interface,Interface 4",
"Site 1,Device 1,dcim.interface,Device 3 Interface,Site 2,Device 1,dcim.interface,Interface 5", "Site 1,Device 1,dcim.interface,Device 3 Interface,Site 2,Device 1,dcim.interface,Interface 5",
), )
'powerfeed-to-powerport': (
# Ensure that powerfeed-to-powerport cables can be imported via CSV using side_a_power_panel
"side_a_power_panel,side_a_type,side_a_name,side_b_device,side_b_type,side_b_name",
"Power Panel 1,dcim.powerfeed,Power Feed 1,Device 4,dcim.powerport,Power Port 1",
"Power Panel 1,dcim.powerfeed,Power Feed 2,Device 4,dcim.powerport,Power Port 2",
"Power Panel 1,dcim.powerfeed,Power Feed 3,Device 4,dcim.powerport,Power Port 3",
),
} }
cls.csv_update_data = ( cls.csv_update_data = (

View File

@@ -5,6 +5,8 @@ from functools import cached_property
from django.contrib.contenttypes.fields import GenericRelation from django.contrib.contenttypes.fields import GenericRelation
from django.db import models from django.db import models
from django.db.models import Q from django.db.models import Q
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.urls import reverse from django.urls import reverse
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
@@ -186,7 +188,9 @@ class ScriptModule(PythonModuleMixin, JobsMixin, ManagedFile):
def save(self, *args, **kwargs): def save(self, *args, **kwargs):
self.file_root = ManagedFileRootPathChoices.SCRIPTS self.file_root = ManagedFileRootPathChoices.SCRIPTS
super().save(*args, **kwargs) super().save(*args, **kwargs)
# Sync script classes after the module has been saved. This is the
# single intended synchronization path for ScriptModule saves.
self.sync_classes() self.sync_classes()
@receiver(post_save, sender=ScriptModule)
def script_module_post_save_handler(instance, created, **kwargs):
instance.sync_classes()

View File

@@ -2,3 +2,10 @@ from django.dispatch import Signal
# Signals that a model has completed its clean() method # Signals that a model has completed its clean() method
post_clean = Signal() post_clean = Signal()
# Sent after objects of a given model are created via raw save.
# Expected call signature: post_raw_create.send(sender=MyModel, pks=[...])
# Provides: pks (list) - PKs of the newly created objects.
# Callers must ensure all related objects (e.g. M2M, dependent rows) are in place
# before sending, as receivers may query related data to perform post-create work.
post_raw_create = Signal()