Coverage for dcim/utils.py: 16%
147 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from collections import defaultdict
3from django.apps import apps
4from django.contrib.contenttypes.models import ContentType
5from django.db import router, transaction
6from django.utils.translation import gettext as _
8from dcim.constants import MODULE_TOKEN
11def inherit_module_token(position, parent_positions):
12 """
13 Resolve a single {module} token in a bay position by inheriting from the position
14 one level deeper in a module bay hierarchy. Returns position unchanged unless
15 parent_positions is non-empty and position contains {module}, in which case the
16 token is substituted with parent_positions[-1].
18 Used by resolve_position_chain(), the single inheritance implementation shared by
19 get_module_bay_positions() and the module move planner.
20 """
21 if parent_positions and MODULE_TOKEN in position:
22 return position.replace(MODULE_TOKEN, parent_positions[-1])
23 return position
26def get_module_bay_raw_positions(module_bay):
27 """
28 Given a module bay, traverse up the module hierarchy and return the stored
29 (unresolved) bay position strings from root to leaf.
31 Raises ValueError if the module bay hierarchy contains a cycle.
32 """
33 positions = []
34 visited = set()
35 while module_bay:
36 if module_bay.pk in visited:
37 raise ValueError(_("Module bay hierarchy contains a cycle."))
38 visited.add(module_bay.pk)
39 positions.append(module_bay.position or '')
40 module_bay = module_bay.module.module_bay if module_bay.module else None
41 positions.reverse()
42 return positions
45def resolve_position_chain(raw_positions):
46 """
47 Apply leaf-to-root {module} token inheritance over a root-to-leaf list of raw bay
48 positions: each position inherits from the resolved position one level deeper, and
49 the leaf's own token is never resolved. Shared by get_module_bay_positions() and
50 the module move planner so a planned chain always equals what a fresh walk
51 computes once the planned positions are stored.
52 """
53 resolved = []
54 for position in reversed(raw_positions):
55 resolved.append(inherit_module_token(position, resolved))
56 resolved.reverse()
57 return resolved
60def get_module_bay_positions(module_bay):
61 """
62 Given a module bay, traverse up the module hierarchy and return a list of bay
63 position strings from root to leaf, resolving any {module} tokens in each
64 position using the parent position (position inheritance).
66 Raises ValueError if the module bay hierarchy contains a cycle.
67 """
68 return resolve_position_chain(get_module_bay_raw_positions(module_bay))
71def resolve_module_placeholder(value, positions):
72 """
73 Resolve {module} placeholder tokens in a string using the given
74 list of module bay positions (ordered root to leaf).
76 A single {module} token resolves to the leaf (immediate parent) bay's position.
77 Multiple tokens must match the tree depth and resolve level-by-level.
79 Returns the resolved string.
80 Raises ValueError if token count is greater than 1 and doesn't match tree depth.
81 """
82 if MODULE_TOKEN not in value:
83 return value
85 token_count = value.count(MODULE_TOKEN)
86 if token_count == 1:
87 return value.replace(MODULE_TOKEN, positions[-1])
88 if token_count == len(positions):
89 for pos in positions:
90 value = value.replace(MODULE_TOKEN, pos, 1)
91 return value
92 raise ValueError(
93 _("Cannot install module with placeholder values in a module bay tree "
94 "{level} levels deep but {tokens} placeholders given.").format(
95 level=len(positions), tokens=token_count
96 )
97 )
100def compile_path_node(ct_id, object_id):
101 return f'{ct_id}:{object_id}'
104def decompile_path_node(repr):
105 ct_id, object_id = repr.split(':')
106 return int(ct_id), int(object_id)
109def object_to_path_node(obj):
110 """
111 Return a representation of an object suitable for inclusion in a CablePath path. Node representation is in the
112 form <ContentType ID>:<Object ID>.
113 """
114 ct = ContentType.objects.get_for_model(obj)
115 return compile_path_node(ct.pk, obj.pk)
118def path_node_to_object(repr):
119 """
120 Given the string representation of a path node, return the corresponding instance. If the object no longer
121 exists, return None.
122 """
123 ct_id, object_id = decompile_path_node(repr)
124 ct = ContentType.objects.get_for_id(ct_id)
125 return ct.model_class().objects.filter(pk=object_id).first()
128def create_cablepaths(objects):
129 """
130 Create CablePaths for all paths originating from the specified set of nodes.
132 :param objects: Iterable of cabled objects (e.g. Interfaces)
133 """
134 from dcim.models import CablePath, Interface
136 # Expand any channelized interface into its channel subinterfaces. A channelized parent originates no path of its
137 # own; instead, each channel subinterface traces independently from the single connector position it occupies.
138 # Plain (non-channelized) origins pass through unchanged, keeping this expansion re-entrant so that
139 # rebuild_paths() -> create_cablepaths(cp.origins) does not re-expand the channel subinterfaces it already holds.
140 expanded = []
141 for obj in objects:
142 if isinstance(obj, Interface) and obj.channels:
143 expanded.extend(obj.child_interfaces.filter(channel_id__isnull=False, cable__isnull=False))
144 else:
145 expanded.append(obj)
147 # Arrange objects by cable connector. All objects with a null connector are grouped together. Channel
148 # subinterfaces must each originate their own path, as sharing a connector would otherwise collapse a group of
149 # siblings into a single malformed path.
150 origins = defaultdict(list)
151 for obj in expanded:
152 if isinstance(obj, Interface) and obj.channel_id:
153 if cp := CablePath.from_origin([obj]):
154 cp.save()
155 else:
156 origins[obj.cable_connector].append(obj)
158 for connector, objects in origins.items():
159 if cp := CablePath.from_origin(objects):
160 cp.save()
163def rebuild_paths(terminations):
164 """
165 Rebuild all CablePaths which traverse the specified nodes.
166 """
167 from dcim.models import CablePath
169 for obj in terminations:
170 cable_paths = CablePath.objects.filter(_nodes__contains=obj)
172 with transaction.atomic(using=router.db_for_write(CablePath)):
173 for cp in cable_paths:
174 cp.delete()
175 create_cablepaths(cp.origins)
178def rebuild_cable_paths(cable):
179 """
180 Delete and rebuild every CablePath affected by the given Cable, tracing freshly from the Cable's current
181 terminations and from the origins of the affected paths. Used when a Cable's connectivity must be reconciled
182 without its own save() having traced it: the channelization of a terminated interface has changed, or the Cable
183 was written by a process which bypasses save().
184 """
185 from dcim.choices import CableEndChoices
186 from dcim.models import CablePath, CableTermination, PathEndpoint
188 with transaction.atomic(using=router.db_for_write(CablePath)):
189 a_terminations, b_terminations = [], []
190 for ct in CableTermination.objects.filter(cable=cable).prefetch_related('termination'):
191 if ct.cable_end == CableEndChoices.SIDE_A:
192 a_terminations.append(ct.termination)
193 else:
194 b_terminations.append(ct.termination)
196 # Every path traversing the Cable, plus those traversing a termination which is not itself a path endpoint:
197 # the latter may not reach the Cable yet (e.g. an incomplete path through a pass-through port which this
198 # Cable completes).
199 affected = {cp.pk: cp for cp in CablePath.objects.filter(_nodes__contains=cable)}
200 for termination in (*a_terminations, *b_terminations):
201 if not isinstance(termination, PathEndpoint):
202 affected.update({cp.pk: cp for cp in CablePath.objects.filter(_nodes__contains=termination)})
204 # Record each affected path's originating node(s) before deleting it. These are kept as compiled path
205 # nodes; resolving them to objects is deferred to the paths which actually need restoring.
206 origin_keys = {tuple(cp.path[0]) for cp in affected.values()}
208 # Delete existing paths individually so each clears its `_path` back-reference on the originating endpoints.
209 for cp in affected.values():
210 cp.delete()
212 # Trace from the Cable's own terminations first, so that a channelized origin is expanded into its channel
213 # subinterfaces exactly once
214 for nodes in (a_terminations, b_terminations):
215 if nodes and isinstance(nodes[0], PathEndpoint):
216 create_cablepaths(nodes)
217 retraced = {tuple(cp.path[0]) for cp in CablePath.objects.filter(_nodes__contains=cable)}
219 # Restore the affected paths which merely passed through the Cable: those originate elsewhere, so the
220 # tracing above cannot reproduce them.
221 for key in origin_keys - retraced:
222 nodes = [obj for node in key if (obj := path_node_to_object(node))]
223 if not nodes:
224 continue
226 # A path endpoint terminating this Cable belongs to the tracing above: that it produced no path
227 # means the origin no longer has one (e.g. a channel subinterface moved to another parent).
228 if any(isinstance(obj, PathEndpoint) and obj.cable_id == cable.pk for obj in nodes):
229 continue
231 # Nor restore an origin whose path has already been traced through another Cable
232 if key in {tuple(cp.path[0]) for cp in CablePath.objects.filter(_nodes__contains=nodes[0])}:
233 continue
235 create_cablepaths(nodes)
238def update_interface_parents(device, interface_templates, module=None):
239 """
240 Used for device and module instantiation. Iterates all InterfaceTemplates with a parent assigned and applies it to
241 the actual interfaces. Must run after all interfaces have been instantiated (so that every parent interface exists)
242 and before update_interface_bridges() (so that channel subinterfaces validate against a populated parent).
243 """
244 Interface = apps.get_model('dcim', 'Interface')
246 for interface_template in interface_templates.exclude(parent=None): 246 ↛ 247line 246 didn't jump to line 247 because the loop on line 246 never started
247 interface = Interface.objects.get(device=device, name=interface_template.resolve_name(module=module))
248 interface.parent = Interface.objects.get(
249 device=device,
250 name=interface_template.parent.resolve_name(module=module)
251 )
252 interface.full_clean()
253 interface.save()
256def update_interface_bridges(device, interface_templates, module=None):
257 """
258 Used for device and module instantiation. Iterates all InterfaceTemplates with a bridge assigned
259 and applies it to the actual interfaces.
260 """
261 Interface = apps.get_model('dcim', 'Interface')
263 for interface_template in interface_templates.exclude(bridge=None): 263 ↛ 264line 263 didn't jump to line 264 because the loop on line 263 never started
264 interface = Interface.objects.get(
265 device=device,
266 name=interface_template.resolve_name(module=module, device=device)
267 )
269 if interface_template.bridge:
270 interface.bridge = Interface.objects.get(
271 device=device,
272 name=interface_template.bridge.resolve_name(module=module, device=device)
273 )
274 interface.full_clean()
275 interface.save()
278def create_port_mappings(device, device_or_module_type, module=None):
279 """
280 Replicate all front/rear port mappings from a DeviceType or ModuleType to the given device.
281 """
282 from dcim.models import FrontPort, PortMapping, RearPort
284 templates = device_or_module_type.port_mappings.prefetch_related('front_port', 'rear_port')
286 # Cache front & rear ports for efficient lookups by name
287 front_ports = {
288 fp.name: fp for fp in FrontPort.objects.filter(device=device)
289 }
290 rear_ports = {
291 rp.name: rp for rp in RearPort.objects.filter(device=device)
292 }
294 # Replicate PortMappings
295 mappings = []
296 for template in templates: 296 ↛ 297line 296 didn't jump to line 297 because the loop on line 296 never started
297 front_port = front_ports.get(template.front_port.resolve_name(module=module, device=device))
298 rear_port = rear_ports.get(template.rear_port.resolve_name(module=module, device=device))
299 mappings.append(
300 PortMapping(
301 device_id=front_port.device_id,
302 front_port=front_port,
303 front_port_position=template.front_port_position,
304 rear_port=rear_port,
305 rear_port_position=template.rear_port_position,
306 )
307 )
308 # Bulk-created (no per-mapping ObjectChange) to match how every other component is instantiated.
309 PortMapping.objects.bulk_create(mappings)
312def reconcile_port_mappings(mapping_model, parent_field, parent, desired):
313 """
314 Reconcile a parent port's mappings against `desired`, writing only the difference so unchanged
315 mappings keep their PK (and emit no changelog entry). Changed/removed rows are deleted before
316 replacements are created, all in one transaction, so position swaps don't trip the unique
317 constraint. Per-row create()/delete() let the change-logging signals fire naturally.
319 Args:
320 mapping_model: PortMapping or PortTemplateMapping.
321 parent_field: 'front_port' or 'rear_port' — the side being edited; its '<parent_field>_position'
322 is each mapping's stable identity within the set.
323 parent: the parent instance (FrontPort/RearPort or their templates).
324 desired: iterable of dicts of mapping field values EXCLUDING the parent FK, using '<field>_id'
325 for the opposite-port FK, e.g. {'front_port_position': 1, 'rear_port_id': 5,
326 'rear_port_position': 2}. save() derives device/device_type/module_type from the front port.
327 """
328 key_field = f'{parent_field}_position'
329 other_field = 'rear_port' if parent_field == 'front_port' else 'front_port'
330 value_fields = (f'{other_field}_id', f'{other_field}_position')
332 def target(source):
333 # The comparable "value" of a mapping: the opposite port and its position. Two mappings with
334 # the same parent-side position but a different target represent a re-pointing of that slot.
335 get = source.get if isinstance(source, dict) else lambda f: getattr(source, f)
336 return tuple(get(f) for f in value_fields)
338 desired_by_key = {d[key_field]: d for d in desired}
340 with transaction.atomic(using=router.db_for_write(mapping_model)):
341 # Lock the parent's existing mappings for the duration of the reconcile. Two requests editing
342 # the same port would otherwise read the same snapshot and race, the second colliding on a
343 # unique constraint when it recreates rows the first has already committed.
344 existing = {
345 getattr(m, key_field): m
346 for m in mapping_model.objects.filter(**{parent_field: parent}).select_for_update()
347 }
349 # Delete rows that no longer exist or whose target changed (before creating, to free the slots)
350 for key, mapping in existing.items():
351 if key not in desired_by_key or target(mapping) != target(desired_by_key[key]):
352 mapping.delete()
354 # Create rows that are new or whose target changed
355 for key, attrs in desired_by_key.items():
356 if key not in existing or target(existing[key]) != target(attrs):
357 mapping_model.objects.create(**{parent_field: parent, **attrs})