RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(dcim): Refresh cable path for endpoints loaded before tracing 

Repair stale `_path` references when an endpoint instance is cabled but
has no path set, as occurs during cable creation before path tracing.
The `path` accessor now refreshes the denormalized FK from the database
in this case, ensuring event payloads include connected endpoints.

Fixes #21338
This commit is contained in:
Martin Hauser 2026-06-02 19:56:22 +02:00 committed by Jeremy Stretch
parent 3172e47904
commit 5561deb1e4
4 changed files with 158 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
netbox/dcim/models/device_components.py
View File

@ -384,12 +384,27 @@ class PathEndpoint(models.Model):
a stale in-memory `_path` relation while the database already points to
a different CablePath (or to no path at all).
If the cached relation points to a CablePath that has just been
deleted, refresh only the `_path` field from the database and retry.
This keeps the fix cheap and narrowly scoped to the denormalized FK.
Two stale cases are repaired by refreshing only the `_path` field
from the database:
1. The endpoint is linked (by cable or wireless link) but `_path` is
unset, because the instance was loaded before its path was traced
(e.g. while queued for event serialization during link creation).
2. The cached relation points to a CablePath row that has just been
deleted.
Repairing case 1 costs one query per access for a linked endpoint
whose path is genuinely absent in the database. That state is
transient outside of tracing failures, so no result caching is
attempted here.
"""
if self._path_id is None:
return None
has_link = self.cable_id is not None or getattr(self, 'wireless_link_id', None) is not None
if self.pk and has_link:
self.refresh_from_db(fields=['_path'])
if self._path_id is None:
return None
try:
return self._path

56
netbox/dcim/tests/test_models.py
View File

@ -1,4 +1,5 @@
from django.core.exceptions import ValidationError
from django.db.models.signals import post_save
from django.test import TestCase, tag
from circuits.models import *
@ -2132,6 +2133,61 @@ class CableTestCase(TestCase):
self.assertIsNone(data['connected_endpoints_type'])
self.assertFalse(data['connected_endpoints_reachable'])
@tag('regression') # #21338
def test_path_refreshes_unset_cablepath_reference(self):
"""
An endpoint instance saved during cable creation, before path tracing,
should resolve its path and connected endpoints.
The stale-instance preconditions rely on Cable.save() saving each
CableTermination (which re-saves the endpoint) before trace_paths
creates the CablePath records.
"""
device = Device.objects.get(name='TestDevice2')
interface_a = Interface.objects.create(device=device, name='eth2')
interface_b = Interface.objects.create(device=device, name='eth3')
# Capture the instances handed to the event machinery on save
saved_instances = []
def capture(sender, instance, **kwargs):
saved_instances.append(instance)
post_save.connect(capture, sender=Interface)
try:
Cable(a_terminations=[interface_a], b_terminations=[interface_b]).save()
finally:
post_save.disconnect(capture, sender=Interface)
self.assertEqual(len(saved_instances), 2)
captured_a = next(i for i in saved_instances if i.pk == interface_a.pk)
captured_b = next(i for i in saved_instances if i.pk == interface_b.pk)
# The captured instances predate path tracing: cabled, but no path yet
self.assertIsNotNone(captured_a.cable_id)
self.assertIsNone(captured_a._path_id)
self.assertIsNone(captured_b._path_id)
# The accessor must repair the unset denormalized reference
self.assertIsNotNone(captured_a.path)
self.assertEqual(captured_a.connected_endpoints, [interface_b])
# Serialization as performed by the event queue must see the peer
data = serialize_for_event(captured_b)
self.assertEqual([endpoint['id'] for endpoint in data['connected_endpoints']], [interface_a.pk])
self.assertEqual([peer['id'] for peer in data['link_peers']], [interface_a.pk])
self.assertTrue(data['connected_endpoints_reachable'])
def test_path_returns_none_for_unsaved_endpoint(self):
"""
An unsaved endpoint with a link assigned should report no path rather
than attempting a database refresh.
"""
device = Device.objects.get(name='TestDevice1')
cable = Cable.objects.first()
interface = Interface(device=device, name='tmp', cable=cable)
self.assertIsNone(interface.path)
class VirtualDeviceContextTestCase(TestCase):

48
netbox/extras/tests/test_event_rules.py
View File

@ -6,7 +6,7 @@ from unittest.mock import Mock, patch
import django_rq
from django.conf import settings
from django.http import HttpResponse
from django.test import RequestFactory
from django.test import RequestFactory, tag
from django.urls import reverse
from requests import Session
from rest_framework import status
@ -14,14 +14,14 @@ from rest_framework import status
from core.events import *
from core.models import ObjectType
from dcim.choices import SiteStatusChoices
from dcim.models import Site
from dcim.models import Interface, Site
from extras.choices import EventRuleActionChoices
from extras.events import enqueue_event, flush_events, serialize_for_event
from extras.models import EventRule, Script, Tag, Webhook
from extras.signals import process_job_end_event_rules
from extras.webhooks import generate_signature, send_webhook
from netbox.context_managers import event_tracking
from utilities.testing import APITestCase
from utilities.testing import APITestCase, create_test_device
class EventRuleTestCase(APITestCase):
@ -531,6 +531,48 @@ class EventRuleTestCase(APITestCase):
self.assertEqual(event['data']['name'], 'Site 1')
self.assertIsNone(event['snapshots']['postchange'])
@tag('regression') # #21338
def test_cable_creation_event_payload_includes_connected_endpoints(self):
"""
Interface update events queued during cable creation must include the
peer interface in connected_endpoints and link_peers.
"""
webhook = Webhook.objects.get(name='Webhook 1')
event_rule = EventRule.objects.create(
name='Interface Update Rule',
event_types=[OBJECT_UPDATED],
action_type=EventRuleActionChoices.WEBHOOK,
action_object_type=ObjectType.objects.get_for_model(Webhook),
action_object_id=webhook.id,
)
event_rule.object_types.set([ObjectType.objects.get_for_model(Interface)])
device = create_test_device('Device 1')
interface_a = Interface.objects.create(device=device, name='eth0')
interface_b = Interface.objects.create(device=device, name='eth1')
# Create a cable between the two interfaces via the REST API
data = {
'a_terminations': [{'object_type': 'dcim.interface', 'object_id': interface_a.pk}],
'b_terminations': [{'object_type': 'dcim.interface', 'object_id': interface_b.pk}],
}
url = reverse('dcim-api:cable-list')
self.add_permissions('dcim.add_cable')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
# One update event was queued for each interface
self.assertEqual(self.queue.count, 2)
payloads = {job.kwargs['data']['id']: job.kwargs['data'] for job in self.queue.jobs}
peers = {interface_a.pk: interface_b.pk, interface_b.pk: interface_a.pk}
self.assertEqual(set(payloads), set(peers))
for interface_id, payload in payloads.items():
peer_id = peers[interface_id]
self.assertIsNotNone(payload['connected_endpoints'])
self.assertEqual([endpoint['id'] for endpoint in payload['connected_endpoints']], [peer_id])
self.assertEqual([peer['id'] for peer in payload['link_peers']], [peer_id])
self.assertTrue(payload['connected_endpoints_reachable'])
def test_duplicate_triggers(self):
"""
Test for erroneous duplicate event triggers resulting from saving an object multiple times

41
netbox/wireless/tests/test_signals.py
View File

@ -1,7 +1,8 @@
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from django.test import SimpleTestCase, TestCase
from django.db.models.signals import post_save
from django.test import SimpleTestCase, TestCase, tag
from dcim.choices import InterfaceTypeChoices
from dcim.models import CablePath, Interface
@ -20,7 +21,7 @@ class WirelessLinkSignalTestCase(TestCase):
@classmethod
def setUpTestData(cls):
cls.device = create_test_device('Device 1')
# Eight interfaces — one distinct pair per test method so no test sees stale
# Ten interfaces — one distinct pair per test method so no test sees stale
# in-memory state mutated by a previous test.
cls.interfaces = [
Interface.objects.create(
@ -31,7 +32,7 @@ class WirelessLinkSignalTestCase(TestCase):
rf_channel_frequency=5160,
rf_channel_width=20,
)
for i in range(8)
for i in range(10)
]
def test_creating_link_assigns_wireless_link_to_both_interfaces(self):
@ -77,6 +78,40 @@ class WirelessLinkSignalTestCase(TestCase):
# All wireless cable paths should be gone.
self.assertEqual(CablePath.objects.count(), 0)
@tag('regression') # #21338
def test_path_refreshes_unset_cablepath_reference(self):
"""
An interface instance saved during wireless link creation, before path
tracing, should resolve its path and connected endpoints.
The stale-instance preconditions rely on update_connected_interfaces
saving both interfaces before creating cable paths.
"""
interface_a, interface_b = self.interfaces[8], self.interfaces[9]
# Capture the instances handed to the event machinery on save
saved_instances = []
def capture(sender, instance, **kwargs):
saved_instances.append(instance)
post_save.connect(capture, sender=Interface)
try:
WirelessLink(interface_a=interface_a, interface_b=interface_b, ssid='LINK1').save()
finally:
post_save.disconnect(capture, sender=Interface)
self.assertEqual(len(saved_instances), 2)
captured_a = next(i for i in saved_instances if i.pk == interface_a.pk)
# The captured instance predates path tracing: linked, but no path yet
self.assertIsNotNone(captured_a.wireless_link_id)
self.assertIsNone(captured_a._path_id)
# The accessor must repair the unset denormalized reference
self.assertIsNotNone(captured_a.path)
self.assertEqual(captured_a.connected_endpoints, [interface_b])
class UpdateConnectedInterfacesDirectHandlerTestCase(SimpleTestCase):
"""