#21879 - Add update_dependent_objects() model hook

docs/plugins/development/models.md

@@ -183,6 +183,29 @@ register_model_feature('foo', supports_foo)
 !!! tip
     Consider performing feature registration inside your PluginConfig's `ready()` method.
 
+## Dependent Objects
+
+Some models maintain dependent objects from their `save()` method: saving a cable, for example, traces and records its cable paths. Callers which write objects directly to the database bypass `save()` — replaying serialized changes, restoring deleted objects, or importing data — and those dependent objects are never created.
+
+A model can implement `update_dependent_objects()` to expose that work to such a caller:
+
+```python
+# models.py
+class MyModel(NetBoxModel):
+
+    def update_dependent_objects(self):
+        # Recreate any objects derived from this one
+```
+
+The method is optional; callers should check for its presence before calling it. NetBox never calls it during a normal save.
+
+Two constraints apply to an implementation:
+
+* It must derive its work entirely from the database. The in-memory state a normal `save()` relies on (which fields changed, for instance) is not available to a caller replaying serialized data.
+* It must be idempotent and safe to call when nothing needs to change, as a caller will generally invoke it for every object it has written.
+
+The caller is responsible for calling the method only once every related object is in place: `Cable.update_dependent_objects()` retraces the cable's paths, which requires its `CableTermination` objects to exist.
+
 ## Choice Sets
 
 For model fields which support the selection of one or more values from a predefined list of choices, NetBox provides the `ChoiceSet` utility class. This can be used in place of a regular choices tuple to provide enhanced functionality, namely dynamic configuration and colorization. (See [Django's documentation](https://docs.djangoproject.com/en/stable/ref/models/fields/#choices) on the `choices` parameter for supported model fields.)

netbox/dcim/models/cables.py

@@ -19,7 +19,7 @@ from dcim.choices import *
 from dcim.constants import *
 from dcim.exceptions import UnsupportedCablePath
 from dcim.fields import PathField
-from dcim.utils import decompile_path_node, object_to_path_node
+from dcim.utils import decompile_path_node, object_to_path_node, rebuild_cable_paths
 from netbox.choices import ColorChoices
 from netbox.models import ChangeLoggedModel, PrimaryModel
 from utilities.conversion import to_meters
@@ -508,6 +508,12 @@ class Cable(PrimaryModel):
 
         return instance
 
+    def update_dependent_objects(self):
+        """
+        Recreate the CablePaths traversing this Cable from its current terminations.
+        """
+        rebuild_cable_paths(self)
+
     def get_terminations(self):
         """
         Return two dictionaries mapping A & B side terminating objects to their corresponding CableTerminations

netbox/dcim/tests/test_cablepaths.py

@@ -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
@@ -3033,3 +3033,60 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         self.assertEqual(CablePath.objects.count(), 1)
         interface.refresh_from_db()
         self.assertPathIsSet(interface, path)
+
+
+class CableDependentObjectsTestCase(BaseCablePathTestCase):
+    """
+    Test Cable.update_dependent_objects(), which retraces the paths of a Cable written to the database
+    by a process that bypasses save() (e.g. a tool replaying serialized changes).
+    """
+    def _create_cable_raw(self, termination_a, termination_b):
+        """
+        Write a Cable and its terminations directly to the database, bypassing Cable.save().
+        """
+        cable = Cable(status=LinkStatusChoices.STATUS_CONNECTED)
+        cable.save_base(raw=True)
+
+        for termination, cable_end in (
+            (termination_a, CableEndChoices.SIDE_A),
+            (termination_b, CableEndChoices.SIDE_B),
+        ):
+            ct = CableTermination(cable=cable, cable_end=cable_end, termination=termination)
+            ct.cache_related_objects()
+            ct.save_base(raw=True)
+            termination.cable = cable
+            termination.cable_end = cable_end
+            termination.save()
+
+        return cable
+
+    def test_retrace_after_raw_create(self):
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+
+        cable = self._create_cable_raw(interface1, interface2)
+        self.assertEqual(CablePath.objects.count(), 0)
+
+        cable.update_dependent_objects()
+
+        self.assertPathExists((interface1, cable, interface2), is_complete=True, is_active=True)
+        self.assertPathExists((interface2, cable, interface1), is_complete=True, is_active=True)
+        self.assertEqual(CablePath.objects.count(), 2)
+
+    def test_retrace_is_idempotent(self):
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+
+        cable = Cable(a_terminations=[interface1], b_terminations=[interface2])
+        cable.save()
+        self.assertEqual(CablePath.objects.count(), 2)
+
+        cable.update_dependent_objects()
+
+        path1 = self.assertPathExists((interface1, cable, interface2), is_complete=True, is_active=True)
+        path2 = self.assertPathExists((interface2, cable, interface1), is_complete=True, is_active=True)
+        self.assertEqual(CablePath.objects.count(), 2)
+        interface1.refresh_from_db()
+        interface2.refresh_from_db()
+        self.assertPathIsSet(interface1, path1)
+        self.assertPathIsSet(interface2, path2)