Coverage for netbox_interface_name_rules / family / names.py: 98.53%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com> 

3"""Shared live interface-name primitives for family execution and its preview.""" 

4 

5import logging 

6from collections import Counter 

7 

8from dcim.models import Interface 

9from django.core.exceptions import ValidationError 

10from django.db import DatabaseError, IntegrityError 

11 

12from ..transactions import atomic_with_events, on_commit 

13 

14logger = logging.getLogger(__name__) 

15 

16COLLISION_REASON = "target name is already in use" 

17REPEATED_NAME_REASON = "the module's families want it more than once" 

18INTERFACE_NAME_CONSTRAINT = "dcim_interface_unique_device_name" 

19 

20 

21def name_owners(device_id, names) -> dict[str, int]: 

22 """Return the primary key of the interface that owns each of *names* on the device, by name.""" 

23 return dict(Interface.objects.filter(device_id=device_id, name__in=names).values_list("name", "pk")) 

24 

25 

26def first_taken_name(target_names, owners, own_pks): 

27 """Return the first of *target_names* that *owners* gives to an interface outside *own_pks*, or None.""" 

28 return next((name for name in target_names if name in owners and owners[name] not in own_pks), None) 

29 

30 

31def name_holders(device_id, names) -> dict: 

32 """Return the interface that has each of *names* on the device, by name, with its parent.""" 

33 return { 

34 interface.name: interface 

35 for interface in Interface.objects.filter(device_id=device_id, name__in=names).select_related("parent") 

36 } 

37 

38 

39def _owner_refusal(holder, module_id, family_bases): 

40 """Return why *holder* belongs to another module or family, or an empty string.""" 

41 if holder.pk in family_bases: 

42 return "another family is built on it" 

43 if holder.module_id is not None and holder.module_id != module_id: 

44 return "it belongs to another module" 

45 if holder.parent_id is not None or getattr(holder, "channel_id", None) is not None: 

46 return f"it is already a subinterface of {getattr(holder.parent, 'name', 'another interface')}" 

47 if getattr(holder, "channels", None): 

48 return "it is a channelized parent" 

49 return "" 

50 

51 

52def belongs_elsewhere(holder, module_id, family_bases) -> bool: 

53 """Return whether *holder* belongs to another module or family, so that its own outcome reports it.""" 

54 return bool(_owner_refusal(holder, module_id, family_bases)) 

55 

56 

57def channel_holder_refusal(holder, module_id, family_bases) -> str: 

58 """Return why a family built on module *module_id* cannot take over *holder*, or an empty string. 

59 

60 *holder* is the row that has one of the family's channel names, and *family_bases* holds the 

61 primary key of each row another family is built on. Planning, the preview and execution all ask 

62 this one question (ADR 0017). 

63 """ 

64 refusal = _owner_refusal(holder, module_id, family_bases) 

65 if refusal: 

66 return refusal 

67 # NetBox refuses a channel ID on an interface with a cable, and the plugin never edits a cable. 

68 if holder.cable_id is not None: 

69 return "it has a cable" 

70 return "" 

71 

72 

73def channelized_name_refusal(parent_name, channel_names, holders, own_pks, module_id, family_bases): 

74 """Return ``(name, reason)`` for the first name a channelized family cannot take, or None. 

75 

76 *holders* maps each name in use on the device to the row that has it, and a row in *own_pks* is 

77 the family's own. The family takes over a row with a channel name that ``channel_holder_refusal`` 

78 allows, given *family_bases*. 

79 """ 

80 holder = holders.get(parent_name) 

81 if holder is not None and holder.pk not in own_pks: 

82 return parent_name, f"{COLLISION_REASON}: {parent_name}" 

83 for name in channel_names: 

84 holder = holders.get(name) 

85 if holder is None or holder.pk in own_pks: 

86 continue 

87 refusal = channel_holder_refusal(holder, module_id, family_bases) 

88 if refusal: 

89 return name, f"{COLLISION_REASON}: {name} ({refusal})" 

90 return None 

91 

92 

93def repeated_names(families) -> frozenset: 

94 """Return each channel name that the channelized families on one module want more than once. 

95 

96 *families* holds the channel names of each family, so a name repeated inside one family counts as 

97 well as a name two families share. Planning, the preview and execution refuse every family that 

98 wants such a name, and none of them takes over the row that has it (ADR 0017). 

99 """ 

100 counts = Counter(name for channel_names in families for name in channel_names) 

101 return frozenset(name for name, count in counts.items() if count > 1) 

102 

103 

104def repeated_name_refusal(channel_names, repeated): 

105 """Return ``(name, reason)`` for the first of *channel_names* in *repeated*, or None.""" 

106 name = next((name for name in channel_names if name in repeated), None) 

107 return None if name is None else (name, f"{COLLISION_REASON}: {name} ({REPEATED_NAME_REASON})") 

108 

109 

110def validate_or_block(interface, role): 

111 """Run NetBox's own validation on *interface*, restating a rejection as this family's reason.""" 

112 try: 

113 interface.full_clean() 

114 except ValidationError as error: 

115 raise ValidationError(f"{role} {interface.name!r}: {' '.join(error.messages)}") from error 

116 

117 

118def name_is_taken(device_id, target_name, exclude_pk) -> bool: 

119 """Return whether an interface other than *exclude_pk* already owns *target_name* on the device.""" 

120 return Interface.objects.filter(device_id=device_id, name=target_name).exclude(pk=exclude_pk).exists() 

121 

122 

123def is_name_collision(error: IntegrityError) -> bool: 

124 """Return whether PostgreSQL identified NetBox's interface-name constraint.""" 

125 cause = error.__cause__ 

126 diagnostics = getattr(cause, "diag", None) 

127 return getattr(diagnostics, "constraint_name", None) == INTERFACE_NAME_CONSTRAINT 

128 

129 

130class ChannelReconciliationError(RuntimeError): 

131 """The names that NetBox's parent cascade gave the channels this plugin kept could not be restored.""" 

132 

133 

134def restore_deferred_channel_names(reconciliations): 

135 """Restore plugin-owned names that NetBox's parent cascade changed after commit. 

136 

137 A database failure rolls the whole restore back and raises ``ChannelReconciliationError``, which 

138 names each channel and the name it kept, so that an operator can rename it back. 

139 """ 

140 try: 

141 _restore_channel_names(reconciliations) 

142 except DatabaseError as error: 

143 kept = ", ".join(f"`{cascade_name}` to `{final_name}`" for _pk, final_name, cascade_name in reconciliations) 

144 raise ChannelReconciliationError( 

145 f"NetBox's parent cascade renamed channels that kept their names, and restoring those names failed " 

146 f"with {type(error).__name__}. Rename each channel back: {kept}" 

147 ) from error 

148 

149 

150def _restore_channel_names(reconciliations): 

151 """Restore each kept name in one block, and leave a channel that changed since the cascade alone.""" 

152 child_pks = [child_pk for child_pk, _final_name, _cascade_name in reconciliations] 

153 with atomic_with_events(): 

154 children = ( 

155 Interface.objects.select_for_update(of=("self",)).select_related("device").order_by("pk").in_bulk(child_pks) 

156 ) 

157 for child_pk, final_name, cascade_name in reconciliations: 

158 child = children.get(child_pk) 

159 if child is None or child.name == final_name: 

160 continue 

161 if child.name != cascade_name: 

162 logger.warning( 

163 "Channel interface %s changed to unexpected name %r before deferred reconciliation; " 

164 "leaving it unchanged.", 

165 child_pk, 

166 child.name, 

167 ) 

168 continue 

169 previous_name = child.name 

170 child.snapshot() 

171 try: 

172 with atomic_with_events(): 

173 child.name = final_name 

174 child.full_clean() 

175 child.save() 

176 except ValidationError: 

177 child.name = previous_name 

178 logger.exception( 

179 "Failed to restore channel interface %s from NetBox's deferred name %r to %r; skipping.", 

180 child_pk, 

181 cascade_name, 

182 final_name, 

183 ) 

184 except IntegrityError as error: 

185 child.name = previous_name 

186 if not is_name_collision(error): 186 ↛ 188line 186 didn't jump to line 188 because the condition on line 186 was always true

187 raise 

188 logger.warning( 

189 "Channel interface %s could not reclaim name %r after NetBox's deferred rename; skipping.", 

190 child_pk, 

191 final_name, 

192 ) 

193 

194 

195def reconcile_after_parent_cascade(parent_before, parent_after, channels): 

196 """Schedule restoration of the channel names NetBox's deferred parent cascade will overwrite. 

197 

198 *channels* carries ``(child_pk, channel_id, final_name)`` for every channel the caller settled. 

199 The callback runs after the open transactions commit on both connections, so after NetBox's cascade. 

200 """ 

201 if parent_after == parent_before: 

202 return 

203 reconciliations = tuple( 

204 (child_pk, final_name, f"{parent_after}:{channel_id}") 

205 for child_pk, channel_id, final_name in channels 

206 if final_name == f"{parent_before}:{channel_id}" and final_name != f"{parent_after}:{channel_id}" 

207 ) 

208 if not reconciliations: 

209 return 

210 on_commit(lambda: restore_deferred_channel_names(reconciliations))