Coverage for netbox_interface_name_rules / family / template_names.py: 100.00%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""Resolve current and historical NetBox interface-template names. 

4 

5The refetch and template queries here use the default manager, so NetBox's router sends them to the 

6active netbox-branching branch, or to main. The block cache is keyed by primary key alone. That holds 

7because every read of one block goes to one alias: netbox-branching activates one branch for a whole 

8request or job, and the plugin does not change the active branch inside a block. 

9""" 

10 

11import contextlib 

12import copy 

13import re 

14import threading 

15from dataclasses import dataclass, replace 

16from re import Pattern 

17 

18from dcim.models import InterfaceTemplate, Module, VirtualChassis 

19 

20from ..rule_selection import compile_stored_pattern 

21 

22BAY_CHAIN_RELATIONS = ( 

23 "device", 

24 "device__virtual_chassis", 

25 "module_bay", 

26 "module_bay__parent", 

27 "module_bay__module", 

28 "module_bay__module__module_bay", 

29 "module_bay__module__module_bay__parent", 

30 "module_bay__module__module_bay__module", 

31) 

32 

33# PostgreSQL text cannot hold NUL, so no stored template name or bay position can spell this marker. 

34_VC_SENTINEL = "\x00{}\x00" 

35_VC_SENTINEL_RE = re.compile(r"\x00(\d+)\x00") 

36# NetBox stores vc_position in a PositiveIntegerField, so ten digits cover every valid value. 

37VC_POSITION_DIGITS = r"\d{1,10}" 

38 

39 

40@dataclass(frozen=True, slots=True) 

41class ResolvedTemplateName: 

42 """One template's current name, its optional historical-name matcher, and what NetBox resolved it from. 

43 

44 ``source`` holds the template and the module that NetBox resolved the name against. It is None 

45 on a NetBox without the ``{vc_position}`` token, because the name is then the same at every position. 

46 """ 

47 

48 pk: int 

49 template_name: str 

50 resolved: str 

51 historical_pattern: Pattern[str] | None 

52 parent_id: int | None 

53 channel_id: int | None 

54 channels: int | None 

55 source: tuple | None = None 

56 

57 def at_chassis_position(self, vc_position, *, move=False): 

58 """Return this template as NetBox names it at *vc_position*, which None puts off a chassis. 

59 

60 NetBox names an installed component with ``resolve_name``, and a moved one with the resolver of 

61 its move planner; *move* selects that one. 

62 """ 

63 if self.source is None: 

64 return self 

65 template, module = self.source 

66 stand_in = _module_at_chassis_position(module, vc_position) 

67 return replace(self, resolved=_move_name(template, stand_in) if move else template.resolve_name(stand_in)) 

68 

69 

70def _move_name(template, module): # pragma: no cover - requires a NetBox that renames moved components 

71 """Return the name that NetBox's move planner gives *template* on *module* in its bay. 

72 

73 Unlike ``resolve_name``, the planner resolves ``{vc_position}`` also when only a bay position brings 

74 the token in. Positions that do not fit the template raise here as they do in ``resolve_name``. 

75 """ 

76 from dcim.models.module_moves import ModuleMovePlan 

77 from dcim.utils import get_module_bay_positions 

78 

79 positions = get_module_bay_positions(module.module_bay) 

80 return ModuleMovePlan._resolve(template, template.name, positions, module.device) 

81 

82 

83def _module_at_chassis_position(module, vc_position): # pragma: no cover - requires virtual-chassis token support 

84 """Return a copy of *module* whose device is a copy at *vc_position*; None puts that device off a chassis. 

85 

86 NetBox reads only whether the device has a virtual chassis, and its position. The copies share the 

87 bay chain that *module* already loaded, change nothing on *module* or its device, and are not saved. 

88 """ 

89 device = copy.copy(module.device) 

90 device.virtual_chassis = None if vc_position is None else VirtualChassis() 

91 device.vc_position = vc_position 

92 stand_in = copy.copy(module) 

93 stand_in.device = device 

94 return stand_in 

95 

96 

97def vc_position_re(): 

98 """Return NetBox's virtual-chassis template-token pattern when available.""" 

99 try: 

100 from dcim.constants import VC_POSITION_RE 

101 except ImportError: 

102 return None 

103 return VC_POSITION_RE # pragma: no cover - only available on NetBox releases with the token 

104 

105 

106def _vc_position_alternatives(fallback): # pragma: no cover - requires virtual-chassis token support 

107 """Return every value represented by one virtual-chassis position token.""" 

108 if fallback is None: 

109 return VC_POSITION_DIGITS 

110 return f"(?:{VC_POSITION_DIGITS}|{re.escape(fallback)})" 

111 

112 

113def _historical_pattern(template, module, token_re): # pragma: no cover - requires virtual-chassis token support 

114 """Return a matcher for every historical resolution of *template*.""" 

115 fallbacks = [] 

116 

117 def mark(match): 

118 fallbacks.append(match.group(1)) 

119 return _VC_SENTINEL.format(len(fallbacks) - 1) 

120 

121 marked = token_re.sub(mark, template.name) 

122 if not fallbacks: 

123 return None 

124 stub = copy.copy(template) 

125 stub.name = marked 

126 literals = _VC_SENTINEL_RE.split(stub.resolve_name(module))[0::2] 

127 # Adjacent tokens cannot be told apart, and their alternatives would backtrack without bound. 

128 if any(not literal for literal in literals[1:-1]): 

129 return None 

130 pattern = re.escape(literals[0]) 

131 for fallback, literal in zip(fallbacks, literals[1:], strict=True): 

132 pattern += _vc_position_alternatives(fallback) + re.escape(literal) 

133 return compile_stored_pattern(pattern) 

134 

135 

136# One batch of modules shares its module chains, template rows and resolved names, thread-locally. 

137_pin = threading.local() 

138 

139 

140@contextlib.contextmanager 

141def pinned_template_cache(modules=()): 

142 """Share resolved interface templates across every module in one batch. 

143 

144 Each module in *modules* must already carry ``BAY_CHAIN_RELATIONS``, so its templates resolve 

145 without a refetch. Modules the block meets later are chained and cached on first use, and one 

146 module type's template rows are read once. Nested blocks share the outermost cache. 

147 """ 

148 depth = getattr(_pin, "depth", 0) 

149 _pin.depth = depth + 1 

150 try: 

151 # Setting up the cache must sit inside the try, or a raise here strands the depth forever. 

152 if depth == 0: 

153 _pin.chained = {} 

154 _pin.templates = {} 

155 _pin.resolved = {} 

156 _pin.chained.update({module.pk: module for module in modules}) 

157 yield 

158 finally: 

159 _pin.depth -= 1 

160 if _pin.depth == 0: 

161 for attr in ("chained", "templates", "resolved"): 

162 _pin.__dict__.pop(attr, None) 

163 

164 

165def module_with_bay_chain(module): 

166 """Re-fetch *module* with every relation template name resolution uses.""" 

167 chained = getattr(_pin, "chained", None) 

168 if chained is None: 

169 return Module.objects.select_related(*BAY_CHAIN_RELATIONS).get(pk=module.pk) 

170 if module.pk not in chained: 

171 chained[module.pk] = Module.objects.select_related(*BAY_CHAIN_RELATIONS).get(pk=module.pk) 

172 return chained[module.pk] 

173 

174 

175def _interface_templates(module_type_id): 

176 """Return one module type's interface templates in primary-key order.""" 

177 templates = getattr(_pin, "templates", None) 

178 if templates is None: 

179 return list(InterfaceTemplate.objects.filter(module_type_id=module_type_id).order_by("pk")) 

180 if module_type_id not in templates: 

181 templates[module_type_id] = list(InterfaceTemplate.objects.filter(module_type_id=module_type_id).order_by("pk")) 

182 return templates[module_type_id] 

183 

184 

185def _resolve_template(template, module, token_re) -> ResolvedTemplateName: 

186 """Resolve one interface template against *module*.""" 

187 return ResolvedTemplateName( 

188 pk=template.pk, 

189 template_name=template.name, 

190 resolved=template.resolve_name(module), 

191 historical_pattern=None if token_re is None else _historical_pattern(template, module, token_re), 

192 parent_id=getattr(template, "parent_id", None), 

193 channel_id=getattr(template, "channel_id", None), 

194 channels=getattr(template, "channels", None), 

195 source=None if token_re is None else (template, module), 

196 ) 

197 

198 

199def resolve_templates(templates, module) -> tuple[ResolvedTemplateName, ...]: 

200 """Resolve already-loaded interface templates against *module*.""" 

201 token_re = vc_position_re() 

202 return tuple(_resolve_template(template, module, token_re) for template in templates) 

203 

204 

205def resolved_template_names(module) -> tuple[ResolvedTemplateName, ...]: 

206 """Load and resolve every interface template for *module* once.""" 

207 resolved = getattr(_pin, "resolved", None) 

208 if resolved is not None and module.pk in resolved: 

209 return resolved[module.pk] 

210 chained = module_with_bay_chain(module) 

211 names = resolve_templates(_interface_templates(chained.module_type_id), chained) 

212 if resolved is not None: 

213 resolved[module.pk] = names 

214 return names 

215 

216 

217class TemplateNames: 

218 """The module type's resolved template names, read only where a plan needs them.""" 

219 

220 def __init__(self, module): 

221 self._module = module 

222 self._templates = None 

223 

224 def get(self): 

225 """Return every resolved template name for the module, loading them once.""" 

226 if self._templates is None: 

227 self._templates = resolved_template_names(self._module) 

228 return self._templates