fix(dcim): Rebuild paths when replaying Cable Termination changes
Treat explicit assignments to uncached cable ends as modifications, including empty assignments. Branching may replay CableTermination rows before applying the parent cable update, so matching stored terminations do not guarantee that the derived paths are current. Reconcile unassigned ends against stored rows rather than stale prefetched data. Preserve no-op handling for unchanged cached ends and form submissions, and retain path invalidation when profile or connector changes force termination recreation. Add regression coverage for serialized and changed-end-only replay, clearing ends, stale prefetches, profile changes, connector ordering, and partial API updates. Verify that unchanged assignments, rejected updates, and changes to unrelated cables preserve existing topology. Refs #23097
This commit is contained in:
Martin Hauser
parent
eaf30a6fb0
commit
ec52a04f29
|
|
@ -229,16 +229,6 @@ class Cable(PrimaryModel):
|
|||
ct.termination for ct in self.terminations.all() if ct.cable_end == side
|
||||
]
|
||||
|
||||
def _cache_stored_terminations(self):
|
||||
"""
|
||||
Fill each cold termination cache from the CableTermination rows, in their stored order.
|
||||
"""
|
||||
a_terminations, b_terminations = self.get_terminations()
|
||||
if not hasattr(self, '_a_terminations'):
|
||||
self._a_terminations = list(a_terminations.keys())
|
||||
if not hasattr(self, '_b_terminations'):
|
||||
self._b_terminations = list(b_terminations.keys())
|
||||
|
||||
def _set_x_terminations(self, side, value):
|
||||
"""
|
||||
Set the terminating objects for the given cable end (A or B).
|
||||
|
|
@ -254,11 +244,8 @@ class Cable(PrimaryModel):
|
|||
ct.termination for ct in CableTermination.objects.filter(pk__in=value).prefetch_related('termination')
|
||||
]
|
||||
|
||||
# Compare a saved cable against its stored rows, not against a possibly stale prefetch of self.terminations
|
||||
if self.pk and not hasattr(self, _attr):
|
||||
self._cache_stored_terminations()
|
||||
|
||||
if not self.pk or getattr(self, _attr) != list(value):
|
||||
# A cold end always counts as assigned: replay writes the rows first, so equal rows do not imply current paths
|
||||
if not self.pk or not hasattr(self, _attr) or getattr(self, _attr) != list(value):
|
||||
self._terminations_modified = True
|
||||
|
||||
setattr(self, _attr, value)
|
||||
|
|
@ -517,6 +504,12 @@ class Cable(PrimaryModel):
|
|||
"""
|
||||
a_terminations, b_terminations = self.get_terminations()
|
||||
|
||||
# An end the caller did not assign is reconciled against its stored rows, not a possibly stale prefetch
|
||||
if not hasattr(self, '_a_terminations'):
|
||||
self._a_terminations = list(a_terminations.keys())
|
||||
if not hasattr(self, '_b_terminations'):
|
||||
self._b_terminations = list(b_terminations.keys())
|
||||
|
||||
# A CableTermination's connector is derived from its position within its end's list of terminating
|
||||
# objects, so reordering that list (or removing an object from the middle of it) rewires the Cable
|
||||
# without changing which objects it connects. Recreate the affected end's CableTerminations so that
|
||||
|
|
@ -528,14 +521,6 @@ class Cable(PrimaryModel):
|
|||
if force_a or force_b:
|
||||
self._terminations_modified = True
|
||||
|
||||
# When force-recreating terminations (e.g. after a profile change), cache the termination objects
|
||||
# from the database before deleting, so they are available for recreation. Without this, the
|
||||
# a_terminations/b_terminations properties would query the DB after deletion and return empty lists.
|
||||
if force_a and not hasattr(self, '_a_terminations'):
|
||||
self._a_terminations = list(a_terminations.keys())
|
||||
if force_b and not hasattr(self, '_b_terminations'):
|
||||
self._b_terminations = list(b_terminations.keys())
|
||||
|
||||
# Delete any stale CableTerminations
|
||||
for termination, ct in a_terminations.items():
|
||||
if force_a or (termination.pk and termination not in self.a_terminations):
|
||||
|
|
|
|||
|
|
@ -4719,6 +4719,108 @@ class CableTestCase(APIViewTestCases.APIViewTestCase):
|
|||
},
|
||||
]
|
||||
|
||||
def _cable_topology(self, cable):
|
||||
"""
|
||||
Return the termination rows, path rows and endpoint references of a cable, for equality checks.
|
||||
"""
|
||||
paths = CablePath.objects.filter(_nodes__contains=cable).order_by('pk')
|
||||
return (
|
||||
list(CableTermination.objects.filter(cable=cable).values_list('pk', 'cable_end', 'termination_id')),
|
||||
list(paths.values_list('pk', 'path', 'is_complete', 'is_active')),
|
||||
list(Interface.objects.filter(cable=cable).values_list('pk', 'cable_end', '_path_id')),
|
||||
)
|
||||
|
||||
def test_patch_without_terminations_leaves_the_topology_alone(self):
|
||||
"""
|
||||
A PATCH that omits both termination lists must not touch the cable's rows or its endpoints.
|
||||
"""
|
||||
self.add_permissions('dcim.change_cable')
|
||||
cable = Cable.objects.get(label='Cable 1')
|
||||
topology = self._cable_topology(cable)
|
||||
|
||||
response = self.client.patch(self._get_detail_url(cable), {'label': 'Renamed'}, format='json', **self.header)
|
||||
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
self.assertEqual(Cable.objects.get(pk=cable.pk).label, 'Renamed')
|
||||
self.assertEqual(self._cable_topology(cable), topology)
|
||||
|
||||
def test_patch_clearing_an_end_keeps_the_other_end(self):
|
||||
"""
|
||||
A PATCH with an empty termination list must detach that end only and keep the other end's row.
|
||||
"""
|
||||
self.add_permissions('dcim.change_cable')
|
||||
for label, attr, cleared_side, kept_side in (
|
||||
('Cable 1', 'a_terminations', CableEndChoices.SIDE_A, CableEndChoices.SIDE_B),
|
||||
('Cable 2', 'b_terminations', CableEndChoices.SIDE_B, CableEndChoices.SIDE_A),
|
||||
):
|
||||
with self.subTest(attr=attr):
|
||||
cable = Cable.objects.get(label=label)
|
||||
cleared = Interface.objects.get(cable=cable, cable_end=cleared_side)
|
||||
kept = Interface.objects.get(cable=cable, cable_end=kept_side)
|
||||
kept_row_pk = CableTermination.objects.get(cable=cable, cable_end=kept_side).pk
|
||||
|
||||
response = self.client.patch(self._get_detail_url(cable), {attr: []}, format='json', **self.header)
|
||||
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
self.assertEqual(
|
||||
list(
|
||||
CableTermination.objects.filter(cable=cable)
|
||||
.values_list('pk', 'cable_end', 'termination_id')
|
||||
),
|
||||
[(kept_row_pk, kept_side, kept.pk)]
|
||||
)
|
||||
cleared.refresh_from_db()
|
||||
self.assertIsNone(cleared.cable)
|
||||
self.assertIsNone(cleared._path_id)
|
||||
kept.refresh_from_db()
|
||||
self.assertEqual(kept.cable, cable)
|
||||
self.assertFalse(kept._path.is_complete)
|
||||
|
||||
def test_patch_replacing_one_end_keeps_the_other_end_row(self):
|
||||
"""
|
||||
A PATCH that replaces one end must rewire it, keep the other end's row and detach the old endpoint.
|
||||
"""
|
||||
self.add_permissions('dcim.change_cable')
|
||||
cable = Cable.objects.get(label='Cable 1')
|
||||
interface_a = Interface.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_A)
|
||||
old_b = Interface.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_B)
|
||||
new_b = Interface.objects.get(device__name='Device 2', name='eth3')
|
||||
a_row_pk = CableTermination.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_A).pk
|
||||
data = {'b_terminations': [{'object_type': 'dcim.interface', 'object_id': new_b.pk}]}
|
||||
|
||||
response = self.client.patch(self._get_detail_url(cable), data, format='json', **self.header)
|
||||
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
self.assertEqual(CableTermination.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_A).pk, a_row_pk)
|
||||
for interface, peer in ((interface_a, new_b), (new_b, interface_a)):
|
||||
interface = Interface.objects.get(pk=interface.pk)
|
||||
self.assertEqual(interface.cable, cable)
|
||||
self.assertTrue(interface._path.is_complete)
|
||||
self.assertEqual(interface.connected_endpoints, [peer])
|
||||
old_b.refresh_from_db()
|
||||
self.assertIsNone(old_b.cable)
|
||||
self.assertIsNone(old_b._path_id)
|
||||
|
||||
def test_rejected_patch_leaves_both_cables_alone(self):
|
||||
"""
|
||||
A PATCH rejected for an occupied endpoint must persist nothing, not even the accompanying label.
|
||||
"""
|
||||
self.add_permissions('dcim.change_cable')
|
||||
cable = Cable.objects.get(label='Cable 1')
|
||||
other = Cable.objects.get(label='Cable 2')
|
||||
occupied = Interface.objects.get(cable=other, cable_end=CableEndChoices.SIDE_B)
|
||||
topology = (self._cable_topology(cable), self._cable_topology(other))
|
||||
data = {
|
||||
'label': 'Must not persist',
|
||||
'b_terminations': [{'object_type': 'dcim.interface', 'object_id': occupied.pk}],
|
||||
}
|
||||
|
||||
response = self.client.patch(self._get_detail_url(cable), data, format='json', **self.header)
|
||||
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertEqual(Cable.objects.get(pk=cable.pk).label, 'Cable 1')
|
||||
self.assertEqual((self._cable_topology(cable), self._cable_topology(other)), topology)
|
||||
|
||||
def test_graphql_cable_termination_cached_filters(self):
|
||||
"""
|
||||
Validate filtering cables by cached CableTermination relations via GraphQL:
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
from circuits.models import *
|
||||
from dcim.choices import LinkStatusChoices
|
||||
from dcim.choices import CableEndChoices, LinkStatusChoices
|
||||
from dcim.models import *
|
||||
from dcim.svg import CableTraceSVG
|
||||
from dcim.tests.utils import BaseCablePathTestCase
|
||||
|
|
@ -2892,9 +2892,52 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
|
|||
interface3.refresh_from_db()
|
||||
self.assertPathIsNotSet(interface3)
|
||||
|
||||
def test_304_resave_cable_with_unchanged_terminations(self):
|
||||
def test_304_replayed_termination_move_rebuilds_paths(self):
|
||||
"""
|
||||
[IF1] --C1-- [IF2] becomes [IF1] --C1-- [IF3]
|
||||
|
||||
Assigning terminations whose rows were already replaced must rebuild the paths from those rows.
|
||||
"""
|
||||
interface1 = Interface.objects.create(device=self.device, name='Interface 1')
|
||||
interface2 = Interface.objects.create(device=self.device, name='Interface 2')
|
||||
interface3 = Interface.objects.create(device=self.device, name='Interface 3')
|
||||
|
||||
cable1 = Cable(
|
||||
a_terminations=[interface1],
|
||||
b_terminations=[interface2]
|
||||
)
|
||||
cable1.save()
|
||||
|
||||
# Replace the B end's row directly, as change replay does before it saves the cable itself
|
||||
CableTermination.objects.get(cable=cable1, cable_end=CableEndChoices.SIDE_B).delete()
|
||||
CableTermination(cable=cable1, cable_end=CableEndChoices.SIDE_B, termination=interface3).save()
|
||||
termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True))
|
||||
interface1.refresh_from_db()
|
||||
interface3.refresh_from_db()
|
||||
self.assertFalse(interface1._path.is_complete)
|
||||
self.assertPathIsNotSet(interface3)
|
||||
|
||||
data = cable1.serialize_object()
|
||||
cable1 = Cable.objects.get(pk=cable1.pk)
|
||||
cable1.a_terminations = data['a_terminations']
|
||||
cable1.b_terminations = data['b_terminations']
|
||||
cable1.save()
|
||||
|
||||
self.assertCurrentPathExists((interface1, cable1, interface3), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interface3, cable1, interface1), is_complete=True, is_active=True)
|
||||
self.assertEqual(
|
||||
set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
|
||||
termination_pks
|
||||
)
|
||||
interface2.refresh_from_db()
|
||||
self.assertIsNone(interface2.cable)
|
||||
self.assertPathIsNotSet(interface2)
|
||||
|
||||
def test_305_reassigning_terminations_restores_missing_paths(self):
|
||||
"""
|
||||
[IF1] --C1-- [IF2]
|
||||
|
||||
Assigning a fresh cable's own terminations, as objects or as serialized IDs, must recreate removed paths.
|
||||
"""
|
||||
interface1 = Interface.objects.create(device=self.device, name='Interface 1')
|
||||
interface2 = Interface.objects.create(device=self.device, name='Interface 2')
|
||||
|
|
@ -2904,39 +2947,109 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
|
|||
b_terminations=[interface2]
|
||||
)
|
||||
cable1.save()
|
||||
|
||||
path_pks = set(CablePath.objects.values_list('pk', flat=True))
|
||||
termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True))
|
||||
self.assertEqual(len(path_pks), 2)
|
||||
self.assertEqual(len(termination_pks), 2)
|
||||
serialized = cable1.serialize_object()
|
||||
|
||||
# Reassign the same terminations on a freshly loaded instance
|
||||
for value in ({'a_terminations': [interface1], 'b_terminations': [interface2]}, serialized):
|
||||
with self.subTest(value_type=type(value['a_terminations'][0]).__name__):
|
||||
# Delete each path on its own so the origins' _path references are cleared with it
|
||||
for path in list(CablePath.objects.all()):
|
||||
path.delete()
|
||||
interface1.refresh_from_db()
|
||||
interface2.refresh_from_db()
|
||||
self.assertPathIsNotSet(interface1)
|
||||
self.assertPathIsNotSet(interface2)
|
||||
|
||||
cable1 = Cable.objects.get(pk=cable1.pk)
|
||||
cable1.a_terminations = value['a_terminations']
|
||||
cable1.b_terminations = value['b_terminations']
|
||||
cable1.save()
|
||||
|
||||
self.assertCurrentPathExists((interface1, cable1, interface2), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interface2, cable1, interface1), is_complete=True, is_active=True)
|
||||
self.assertEqual(CablePath.objects.count(), 2)
|
||||
self.assertEqual(
|
||||
set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
|
||||
termination_pks
|
||||
)
|
||||
|
||||
def test_306_replaying_one_end_rebuilds_paths(self):
|
||||
"""
|
||||
[IF1] --C1-- [IF2] becomes [IF1] --C1-- [IF3] or [IF3] --C1-- [IF2]
|
||||
|
||||
Assigning only the replaced end's serialized terminations must rebuild both paths and keep the other end's row.
|
||||
"""
|
||||
for side in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B):
|
||||
with self.subTest(side=side):
|
||||
interface1 = Interface.objects.create(device=self.device, name=f'Interface {side}1')
|
||||
interface2 = Interface.objects.create(device=self.device, name=f'Interface {side}2')
|
||||
interface3 = Interface.objects.create(device=self.device, name=f'Interface {side}3')
|
||||
cable1 = Cable(a_terminations=[interface1], b_terminations=[interface2])
|
||||
cable1.save()
|
||||
old, kept = (interface1, interface2) if side == CableEndChoices.SIDE_A else (interface2, interface1)
|
||||
|
||||
# Replace one end's row directly, as change replay does before it saves the cable itself
|
||||
CableTermination.objects.get(cable=cable1, cable_end=side).delete()
|
||||
CableTermination(cable=cable1, cable_end=side, termination=interface3).save()
|
||||
termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True))
|
||||
kept.refresh_from_db()
|
||||
interface3.refresh_from_db()
|
||||
self.assertFalse(kept._path.is_complete)
|
||||
self.assertPathIsNotSet(interface3)
|
||||
|
||||
attr = f'{side.lower()}_terminations'
|
||||
data = cable1.serialize_object()
|
||||
cable1 = Cable.objects.get(pk=cable1.pk)
|
||||
setattr(cable1, attr, data[attr])
|
||||
cable1.save()
|
||||
|
||||
self.assertCurrentPathExists((kept, cable1, interface3), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interface3, cable1, kept), is_complete=True, is_active=True)
|
||||
self.assertEqual(
|
||||
set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
|
||||
termination_pks
|
||||
)
|
||||
old.refresh_from_db()
|
||||
self.assertIsNone(old.cable)
|
||||
self.assertPathIsNotSet(old)
|
||||
|
||||
def test_307_replaying_a_termination_move_leaves_other_cables_alone(self):
|
||||
"""
|
||||
[IF1] --C1-- [IF2] becomes [IF1] --C1-- [IF3], while [IF4] --C2-- [IF5] stays as it is
|
||||
|
||||
Rebuilding one cable's paths must not touch another cable's termination or path rows.
|
||||
"""
|
||||
interfaces = [Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 6)]
|
||||
cable1 = Cable(a_terminations=[interfaces[0]], b_terminations=[interfaces[1]])
|
||||
cable1.save()
|
||||
cable2 = Cable(a_terminations=[interfaces[3]], b_terminations=[interfaces[4]])
|
||||
cable2.save()
|
||||
|
||||
def other_rows():
|
||||
return (
|
||||
list(CableTermination.objects.filter(cable=cable2).values_list('pk', 'cable_end', 'termination_id')),
|
||||
list(
|
||||
CablePath.objects.filter(_nodes__contains=cable2)
|
||||
.order_by('pk')
|
||||
.values_list('pk', 'path', 'is_complete', 'is_active')
|
||||
),
|
||||
)
|
||||
|
||||
expected = other_rows()
|
||||
self.assertEqual(len(expected[1]), 2)
|
||||
|
||||
CableTermination.objects.get(cable=cable1, cable_end=CableEndChoices.SIDE_B).delete()
|
||||
CableTermination(cable=cable1, cable_end=CableEndChoices.SIDE_B, termination=interfaces[2]).save()
|
||||
data = cable1.serialize_object()
|
||||
cable1 = Cable.objects.get(pk=cable1.pk)
|
||||
cable1.a_terminations = [interface1]
|
||||
cable1.b_terminations = [interface2]
|
||||
cable1.label = 'Renamed'
|
||||
cable1.b_terminations = data['b_terminations']
|
||||
cable1.save()
|
||||
|
||||
self.assertEqual(set(CablePath.objects.values_list('pk', flat=True)), path_pks)
|
||||
self.assertEqual(
|
||||
set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
|
||||
termination_pks
|
||||
)
|
||||
|
||||
path1 = self.assertPathExists(
|
||||
(interface1, cable1, interface2),
|
||||
is_complete=True,
|
||||
is_active=True
|
||||
)
|
||||
path2 = self.assertPathExists(
|
||||
(interface2, cable1, interface1),
|
||||
is_complete=True,
|
||||
is_active=True
|
||||
)
|
||||
interface1.refresh_from_db()
|
||||
interface2.refresh_from_db()
|
||||
self.assertPathIsSet(interface1, path1)
|
||||
self.assertPathIsSet(interface2, path2)
|
||||
self.assertCurrentPathExists((interfaces[0], cable1, interfaces[2]), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interfaces[2], cable1, interfaces[0]), is_complete=True, is_active=True)
|
||||
self.assertEqual(other_rows(), expected)
|
||||
self.assertCurrentPathExists((interfaces[3], cable2, interfaces[4]), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interfaces[4], cable2, interfaces[3]), is_complete=True, is_active=True)
|
||||
|
||||
def test_401_exclude_midspan_devices(self):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
from unittest import skip
|
||||
|
||||
from circuits.models import Circuit, CircuitTermination, ProviderNetwork
|
||||
from dcim.choices import CableLengthUnitChoices, CableProfileChoices
|
||||
from dcim.choices import CableEndChoices, CableLengthUnitChoices, CableProfileChoices
|
||||
from dcim.models import *
|
||||
from dcim.svg import CableTraceSVG
|
||||
from dcim.tests.utils import BaseCablePathTestCase
|
||||
|
|
@ -2786,48 +2786,159 @@ class CablePathTestCase(BaseCablePathTestCase):
|
|||
termination_pks
|
||||
)
|
||||
|
||||
def test_311_change_cable_profile_after_reassigning_unchanged_terminations(self):
|
||||
def test_311_change_cable_profile_on_a_warm_instance_rebuilds_paths(self):
|
||||
"""
|
||||
[IF1] --C1-- [IF2]
|
||||
|
||||
Applying a profile after both termination caches have been populated must still rebuild the paths.
|
||||
Applying a profile to an instance whose termination caches are warm must recreate the rows and the paths.
|
||||
"""
|
||||
interfaces = [
|
||||
Interface.objects.create(device=self.device, name='Interface 1'),
|
||||
Interface.objects.create(device=self.device, name='Interface 2'),
|
||||
]
|
||||
|
||||
# Create cable 1 without a profile
|
||||
# Creating the cable warms both caches and saving it resets the flag
|
||||
cable1 = Cable(
|
||||
a_terminations=[interfaces[0]],
|
||||
b_terminations=[interfaces[1]],
|
||||
)
|
||||
cable1.clean()
|
||||
cable1.save()
|
||||
termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True))
|
||||
self.assertEqual(CablePath.objects.count(), 2)
|
||||
|
||||
# Reload and populate both termination caches by reassigning their stored values
|
||||
cable1 = Cable.objects.get(pk=cable1.pk)
|
||||
cable1.a_terminations = [interfaces[0]]
|
||||
cable1.b_terminations = [interfaces[1]]
|
||||
self.assertTrue(hasattr(cable1, '_a_terminations') and hasattr(cable1, '_b_terminations'))
|
||||
self.assertFalse(cable1._terminations_modified)
|
||||
|
||||
cable1.profile = CableProfileChoices.SINGLE_1C1P
|
||||
cable1.full_clean()
|
||||
cable1.save()
|
||||
|
||||
path1 = self.assertPathExists(
|
||||
(interfaces[0], cable1, interfaces[1]),
|
||||
is_complete=True,
|
||||
is_active=True
|
||||
)
|
||||
path2 = self.assertPathExists(
|
||||
(interfaces[1], cable1, interfaces[0]),
|
||||
is_complete=True,
|
||||
is_active=True
|
||||
)
|
||||
self.assertCurrentPathExists((interfaces[0], cable1, interfaces[1]), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interfaces[1], cable1, interfaces[0]), is_complete=True, is_active=True)
|
||||
self.assertEqual(CablePath.objects.count(), 2)
|
||||
interfaces[0].refresh_from_db()
|
||||
interfaces[1].refresh_from_db()
|
||||
self.assertPathIsSet(interfaces[0], path1)
|
||||
self.assertPathIsSet(interfaces[1], path2)
|
||||
new_termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True))
|
||||
self.assertTrue(termination_pks.isdisjoint(new_termination_pks))
|
||||
self.assertEqual(
|
||||
sorted(CableTermination.objects.filter(cable=cable1).values_list('cable_end', 'connector', 'positions')),
|
||||
[(CableEndChoices.SIDE_A, 1, [1]), (CableEndChoices.SIDE_B, 1, [1])]
|
||||
)
|
||||
data = cable1.serialize_object()
|
||||
self.assertEqual(set(data['a_terminations'] + data['b_terminations']), new_termination_pks)
|
||||
|
||||
def test_312_replayed_midspan_terminations_complete_existing_paths(self):
|
||||
"""
|
||||
[IF1] --C1-- [FP1][RP1] --C3-- [RP2][FP2] --C2-- [IF2]
|
||||
|
||||
Assigning a mid-span cable's stored terminations must complete the paths that stop at its rear ports.
|
||||
"""
|
||||
interfaces = [
|
||||
Interface.objects.create(device=self.device, name='Interface 1'),
|
||||
Interface.objects.create(device=self.device, name='Interface 2'),
|
||||
]
|
||||
rear_ports = [
|
||||
RearPort.objects.create(device=self.device, name='Rear Port 1'),
|
||||
RearPort.objects.create(device=self.device, name='Rear Port 2'),
|
||||
]
|
||||
front_ports = [
|
||||
FrontPort.objects.create(device=self.device, name='Front Port 1'),
|
||||
FrontPort.objects.create(device=self.device, name='Front Port 2'),
|
||||
]
|
||||
for front_port, rear_port in zip(front_ports, rear_ports):
|
||||
PortMapping.objects.create(
|
||||
device=self.device,
|
||||
front_port=front_port,
|
||||
front_port_position=1,
|
||||
rear_port=rear_port,
|
||||
rear_port_position=1
|
||||
)
|
||||
|
||||
cable1 = Cable(a_terminations=[interfaces[0]], b_terminations=[front_ports[0]])
|
||||
cable1.clean()
|
||||
cable1.save()
|
||||
cable2 = Cable(a_terminations=[front_ports[1]], b_terminations=[interfaces[1]])
|
||||
cable2.clean()
|
||||
cable2.save()
|
||||
cable3 = Cable(a_terminations=[rear_ports[0]], b_terminations=[rear_ports[1]])
|
||||
cable3.clean()
|
||||
cable3.save()
|
||||
|
||||
nodes_a_to_b = (
|
||||
interfaces[0], cable1, front_ports[0], rear_ports[0], cable3, rear_ports[1], front_ports[1], cable2,
|
||||
interfaces[1],
|
||||
)
|
||||
nodes_b_to_a = tuple(reversed(nodes_a_to_b))
|
||||
self.assertCurrentPathExists(nodes_a_to_b, is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists(nodes_b_to_a, is_complete=True, is_active=True)
|
||||
|
||||
# Recreate the B end's row directly, which leaves both paths incomplete at the rear ports
|
||||
CableTermination.objects.get(cable=cable3, cable_end=CableEndChoices.SIDE_B).delete()
|
||||
CableTermination(cable=cable3, cable_end=CableEndChoices.SIDE_B, termination=rear_ports[1]).save()
|
||||
termination_pks = set(CableTermination.objects.filter(cable=cable3).values_list('pk', flat=True))
|
||||
for interface in interfaces:
|
||||
interface.refresh_from_db()
|
||||
self.assertFalse(interface._path.is_complete)
|
||||
|
||||
data = cable3.serialize_object()
|
||||
cable3 = Cable.objects.get(pk=cable3.pk)
|
||||
cable3.a_terminations = data['a_terminations']
|
||||
cable3.b_terminations = data['b_terminations']
|
||||
cable3.save()
|
||||
|
||||
self.assertCurrentPathExists(nodes_a_to_b, is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists(nodes_b_to_a, is_complete=True, is_active=True)
|
||||
self.assertEqual(CablePath.objects.count(), 2)
|
||||
self.assertEqual(
|
||||
set(CableTermination.objects.filter(cable=cable3).values_list('pk', flat=True)),
|
||||
termination_pks
|
||||
)
|
||||
|
||||
def test_313_reordering_trunk_terminations_rewires_connectors_and_paths(self):
|
||||
"""
|
||||
[IF1] --1 C1 1-- [IF3] becomes [IF2] --1 C1 1-- [IF3]
|
||||
[IF2] --2 2-- [IF4] [IF1] --2 2-- [IF4]
|
||||
|
||||
Reordering one end's members on a fresh instance must recreate that end's rows and rebuild the paths.
|
||||
"""
|
||||
interfaces = [
|
||||
Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 5)
|
||||
]
|
||||
|
||||
cable1 = Cable(
|
||||
profile=CableProfileChoices.TRUNK_2C1P,
|
||||
a_terminations=[interfaces[0], interfaces[1]],
|
||||
b_terminations=[interfaces[2], interfaces[3]],
|
||||
)
|
||||
cable1.full_clean()
|
||||
cable1.save()
|
||||
self.assertCurrentPathExists((interfaces[0], cable1, interfaces[2]), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interfaces[1], cable1, interfaces[3]), is_complete=True, is_active=True)
|
||||
a_termination_pks = set(
|
||||
CableTermination.objects.filter(cable=cable1, cable_end=CableEndChoices.SIDE_A).values_list('pk', flat=True)
|
||||
)
|
||||
b_terminations = list(
|
||||
CableTermination.objects.filter(cable=cable1, cable_end=CableEndChoices.SIDE_B)
|
||||
.values_list('pk', 'connector', 'termination_id')
|
||||
)
|
||||
|
||||
cable1 = Cable.objects.get(pk=cable1.pk)
|
||||
cable1.a_terminations = [interfaces[1], interfaces[0]]
|
||||
cable1.full_clean()
|
||||
cable1.save()
|
||||
|
||||
self.assertCurrentPathExists((interfaces[1], cable1, interfaces[2]), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interfaces[0], cable1, interfaces[3]), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interfaces[2], cable1, interfaces[1]), is_complete=True, is_active=True)
|
||||
self.assertCurrentPathExists((interfaces[3], cable1, interfaces[0]), is_complete=True, is_active=True)
|
||||
a_terminations = CableTermination.objects.filter(cable=cable1, cable_end=CableEndChoices.SIDE_A)
|
||||
self.assertEqual(
|
||||
list(a_terminations.order_by('connector').values_list('connector', 'termination_id')),
|
||||
[(1, interfaces[1].pk), (2, interfaces[0].pk)]
|
||||
)
|
||||
self.assertTrue(a_termination_pks.isdisjoint(a_terminations.values_list('pk', flat=True)))
|
||||
self.assertEqual(
|
||||
list(
|
||||
CableTermination.objects.filter(cable=cable1, cable_end=CableEndChoices.SIDE_B)
|
||||
.values_list('pk', 'connector', 'termination_id')
|
||||
),
|
||||
b_terminations
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
from decimal import Decimal
|
||||
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.db.models import ProtectedError
|
||||
from django.db.models import F, ProtectedError
|
||||
from django.db.models.signals import post_save
|
||||
from django.test import TestCase, tag
|
||||
|
||||
|
|
@ -2413,49 +2413,185 @@ class CableTestCase(TestCase):
|
|||
with self.assertRaises(ValidationError):
|
||||
cable.clean()
|
||||
|
||||
def test_reassigning_unchanged_terminations_does_not_flag_a_change(self):
|
||||
def test_assigning_terminations_to_a_fresh_instance_flags_a_change(self):
|
||||
"""
|
||||
Assigning the stored terminations to a freshly loaded cable must leave them unflagged.
|
||||
Assigning either end of a freshly loaded cable must flag a change, even with its stored terminations.
|
||||
"""
|
||||
interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
|
||||
interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0')
|
||||
|
||||
# A cable loaded from the database has no cached terminations
|
||||
cable = Cable.objects.first()
|
||||
cable.a_terminations = [interface1]
|
||||
cable.b_terminations = [interface2]
|
||||
for attr, interface in (('a_terminations', interface1), ('b_terminations', interface2)):
|
||||
with self.subTest(attr=attr):
|
||||
cable = Cable.objects.first()
|
||||
self.assertFalse(cable._terminations_modified)
|
||||
setattr(cable, attr, [interface])
|
||||
self.assertTrue(cable._terminations_modified)
|
||||
|
||||
def test_assigning_a_different_end_to_a_warm_instance_flags_a_change(self):
|
||||
"""
|
||||
Assigning a different termination to an end already held in memory must flag a change.
|
||||
"""
|
||||
interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0')
|
||||
interface3 = Interface.objects.get(device__name='TestDevice2', name='eth1')
|
||||
|
||||
# Assigning warms the B end's cache and the save resets the flag
|
||||
cable = Cable.objects.first()
|
||||
cable.b_terminations = [interface2]
|
||||
cable.save()
|
||||
self.assertFalse(cable._terminations_modified)
|
||||
|
||||
def test_reassigning_different_terminations_flags_a_change(self):
|
||||
cable.b_terminations = [interface3]
|
||||
self.assertTrue(cable._terminations_modified)
|
||||
|
||||
def test_assigning_one_end_reconciles_the_other_against_its_stored_rows(self):
|
||||
"""
|
||||
Assigning a different termination to a freshly loaded cable must flag the change.
|
||||
Assigning one end must leave the other end's stored rows alone, even when a prefetch of them is stale.
|
||||
"""
|
||||
interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0')
|
||||
interface3 = Interface.objects.get(device__name='TestDevice2', name='eth1')
|
||||
cable = Cable.objects.prefetch_related('terminations__termination').first()
|
||||
|
||||
# Moving the A end through a second instance leaves the prefetch above stale
|
||||
moved = Cable.objects.get(pk=cable.pk)
|
||||
moved.a_terminations = [interface3]
|
||||
moved.save()
|
||||
a_rows = CableTermination.objects.filter(cable=cable, cable_end=CableEndChoices.SIDE_A)
|
||||
a_row_pk = a_rows.get().pk
|
||||
|
||||
cable.b_terminations = [interface2]
|
||||
cable.save()
|
||||
|
||||
self.assertEqual(list(a_rows.values_list('pk', 'termination_id')), [(a_row_pk, interface3.pk)])
|
||||
|
||||
def test_assigning_serialized_termination_ids_flags_a_change(self):
|
||||
"""
|
||||
Serialized CableTermination IDs assigned to a fresh instance must resolve to the endpoints and flag a change.
|
||||
"""
|
||||
interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
|
||||
interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0')
|
||||
data = Cable.objects.first().serialize_object()
|
||||
|
||||
for attr, interface in (('a_terminations', interface1), ('b_terminations', interface2)):
|
||||
with self.subTest(attr=attr):
|
||||
cable = Cable.objects.first()
|
||||
setattr(cable, attr, data[attr])
|
||||
self.assertEqual(getattr(cable, attr), [interface])
|
||||
self.assertTrue(cable._terminations_modified)
|
||||
|
||||
def test_serialize_object_includes_termination_ids(self):
|
||||
"""
|
||||
Serialization must carry each end's CableTermination IDs, not the IDs of the terminating objects.
|
||||
"""
|
||||
cable = Cable.objects.first()
|
||||
# Move the rows above every existing ID so matching endpoint IDs cannot satisfy the assertion
|
||||
offset = max(
|
||||
Interface.objects.order_by('-pk').first().pk,
|
||||
CableTermination.objects.order_by('-pk').first().pk,
|
||||
) + 1
|
||||
CableTermination.objects.filter(cable=cable).update(id=F('id') + offset)
|
||||
|
||||
data = cable.serialize_object()
|
||||
|
||||
for side, attr in ((CableEndChoices.SIDE_A, 'a_terminations'), (CableEndChoices.SIDE_B, 'b_terminations')):
|
||||
with self.subTest(side=side):
|
||||
self.assertEqual(data[attr], [CableTermination.objects.get(cable=cable, cable_end=side).pk])
|
||||
|
||||
def test_clearing_an_end_of_a_fresh_instance_removes_its_terminations(self):
|
||||
"""
|
||||
Assigning an empty list to an end of a freshly loaded cable must delete that end's terminations only.
|
||||
"""
|
||||
interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
|
||||
interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0')
|
||||
cable_pk = Cable.objects.first().pk
|
||||
|
||||
for attr, kept_side, cleared, kept in (
|
||||
('a_terminations', CableEndChoices.SIDE_B, interface1, interface2),
|
||||
('b_terminations', CableEndChoices.SIDE_A, interface2, interface1),
|
||||
):
|
||||
with self.subTest(attr=attr):
|
||||
cable = Cable.objects.get(pk=cable_pk)
|
||||
kept_row_pk = CableTermination.objects.get(cable=cable, cable_end=kept_side).pk
|
||||
setattr(cable, attr, [])
|
||||
self.assertTrue(cable._terminations_modified)
|
||||
cable.full_clean()
|
||||
cable.save()
|
||||
|
||||
self.assertEqual(
|
||||
list(CableTermination.objects.filter(cable=cable).values_list('pk', 'cable_end')),
|
||||
[(kept_row_pk, kept_side)]
|
||||
)
|
||||
cleared.refresh_from_db()
|
||||
self.assertIsNone(cleared.cable)
|
||||
self.assertIsNone(cleared._path_id)
|
||||
kept.refresh_from_db()
|
||||
self.assertEqual(kept.cable, cable)
|
||||
self.assertFalse(kept._path.is_complete)
|
||||
|
||||
# Reconnect the cleared end so the other side starts from a complete cable
|
||||
cable = Cable.objects.get(pk=cable_pk)
|
||||
setattr(cable, attr, [cleared])
|
||||
cable.save()
|
||||
|
||||
cable = Cable.objects.get(pk=cable_pk)
|
||||
cable.a_terminations = []
|
||||
cable.save()
|
||||
cable = Cable.objects.get(pk=cable_pk)
|
||||
cable.b_terminations = []
|
||||
cable.save()
|
||||
|
||||
self.assertFalse(CableTermination.objects.filter(cable=cable).exists())
|
||||
for interface in (interface1, interface2):
|
||||
interface.refresh_from_db()
|
||||
self.assertIsNone(interface.cable)
|
||||
self.assertIsNone(interface._path_id)
|
||||
|
||||
def test_reassigning_an_unchanged_end_on_a_warm_instance_does_not_flag_a_change(self):
|
||||
"""
|
||||
Assigning the value an end already holds in memory must not flag a change or clear a pending one.
|
||||
"""
|
||||
interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
|
||||
interface3 = Interface.objects.get(device__name='TestDevice2', name='eth1')
|
||||
|
||||
# Assigning warms the A end's cache and the save resets the flag
|
||||
cable = Cable.objects.first()
|
||||
cable.a_terminations = [interface1]
|
||||
cable.save()
|
||||
self.assertFalse(cable._terminations_modified)
|
||||
termination_pks = set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True))
|
||||
path_pks = set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True))
|
||||
|
||||
cable.a_terminations = [interface1]
|
||||
self.assertFalse(cable._terminations_modified)
|
||||
cable.save()
|
||||
self.assertEqual(
|
||||
set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)),
|
||||
termination_pks
|
||||
)
|
||||
self.assertEqual(set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)), path_pks)
|
||||
|
||||
cable.b_terminations = [interface3]
|
||||
|
||||
cable.a_terminations = [interface1]
|
||||
self.assertTrue(cable._terminations_modified)
|
||||
|
||||
def test_reassigning_stale_prefetched_terminations_flags_a_change(self):
|
||||
def test_saving_a_fresh_instance_without_an_assignment_leaves_terminations_and_paths(self):
|
||||
"""
|
||||
A stale prefetched relation must not hide a real termination change.
|
||||
A save on a freshly loaded cable that assigns no end must keep its termination and path rows.
|
||||
"""
|
||||
cable = Cable.objects.prefetch_related('terminations__termination').first()
|
||||
stale_termination = cable.b_terminations[0]
|
||||
current_termination = Interface.objects.get(device__name='TestDevice2', name='eth1')
|
||||
cable = Cable.objects.first()
|
||||
termination_pks = set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True))
|
||||
path_pks = set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True))
|
||||
self.assertEqual(len(path_pks), 2)
|
||||
|
||||
# Moving the B end through a second instance leaves the prefetch above stale
|
||||
moved = Cable.objects.get(pk=cable.pk)
|
||||
moved.b_terminations = [current_termination]
|
||||
moved.save()
|
||||
cable = Cable.objects.get(pk=cable.pk)
|
||||
cable.label = 'Renamed'
|
||||
cable.full_clean()
|
||||
cable.save()
|
||||
|
||||
# The value matches the stale prefetch but not the stored row
|
||||
cable.b_terminations = [stale_termination]
|
||||
self.assertTrue(cable._terminations_modified)
|
||||
self.assertEqual(
|
||||
set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)),
|
||||
termination_pks
|
||||
)
|
||||
self.assertEqual(set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)), path_pks)
|
||||
|
||||
def test_partial_save_does_not_apply_an_unwritten_profile(self):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -64,6 +64,18 @@ class BaseCablePathTestCase(TestCase):
|
|||
cablepath = self._get_cablepath(nodes, **kwargs)
|
||||
self.assertIsNone(cablepath, msg='Unexpected CablePath found')
|
||||
|
||||
def assertCurrentPathExists(self, nodes, **kwargs):
|
||||
"""
|
||||
Assert that the first node references a CablePath with the given route via _path, and return it.
|
||||
|
||||
:param nodes: Iterable of steps, the first being the originating path endpoint object
|
||||
"""
|
||||
origin = type(nodes[0]).objects.get(pk=nodes[0].pk)
|
||||
self.assertIsNotNone(origin._path_id, msg=f'No path set on originating endpoint {origin}')
|
||||
cablepath = self._get_cablepath(nodes, pk=origin._path_id, **kwargs)
|
||||
self.assertIsNotNone(cablepath, msg=f'Path #{origin._path_id} on {origin} does not match the expected route')
|
||||
return cablepath
|
||||
|
||||
def assertPathIsSet(self, origin, cablepath, msg=None):
|
||||
"""
|
||||
Assert that a specific CablePath instance is set as the path on the origin.
|
||||
|
|
|
|||
Loading…
Reference in New Issue