Coverage for netbox_data_import/field_keys.py: 97.06%

1# SPDX-License-Identifier: Apache-2.0 

2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com> 

3"""Share trace field-key vocabulary between source and target modules.""" 

4 

5from __future__ import annotations 

6 

7import json 

8 

9from .identity import identity_text 

10 

11INTERFACE_PORT_CLASSES = frozenset({"NIC", "Switch Port", "Port"}) 

12FRONT_PORT_CLASSES = frozenset({"Position Front", "Fiber Pair Front"}) 

13REAR_PORT_CLASSES = frozenset({"Punch-Down", "Fiber Pair Back"}) 

14POWER_PORT_CLASSES = frozenset({"Power Input Port"}) 

15POWER_OUTLET_CLASSES = frozenset({"Power Output Port"}) 

16POWER_CLASSES = POWER_PORT_CLASSES | POWER_OUTLET_CLASSES 

17PORT_CLASSES = INTERFACE_PORT_CLASSES | FRONT_PORT_CLASSES | REAR_PORT_CLASSES | POWER_CLASSES 

18PORT_CLASS_CHOICES = tuple((port_class, port_class) for port_class in sorted(PORT_CLASSES)) 

19 

20INTERFACE_KIND = "interface" 

21FRONT_PORT_KIND = "front_port" 

22REAR_PORT_KIND = "rear_port" 

23POWER_PORT_KIND = "power_port" 

24POWER_OUTLET_KIND = "power_outlet" 

25 

26PORT_CLASS_CLAIMED_KINDS = { 

27 **dict.fromkeys(INTERFACE_PORT_CLASSES, INTERFACE_KIND), 

28 **dict.fromkeys(FRONT_PORT_CLASSES, FRONT_PORT_KIND), 

29 **dict.fromkeys(REAR_PORT_CLASSES, REAR_PORT_KIND), 

30 **dict.fromkeys(POWER_PORT_CLASSES, POWER_PORT_KIND), 

31 **dict.fromkeys(POWER_OUTLET_CLASSES, POWER_OUTLET_KIND), 

32} 

33 

34FRONT_PORT_MODEL = "dcim.frontport" 

35REAR_PORT_MODEL = "dcim.rearport" 

36 

37#: The NetBox termination models each claimed kind admits, in match order (section 6.1). 

38ADMITTED_TERMINATION_MODELS = { 

39 INTERFACE_KIND: ( 

40 "dcim.interface", 

41 "dcim.consoleport", 

42 "dcim.consoleserverport", 

43 "dcim.powerport", 

44 "dcim.poweroutlet", 

45 ), 

46 FRONT_PORT_KIND: (FRONT_PORT_MODEL,), 

47 REAR_PORT_KIND: (REAR_PORT_MODEL,), 

48 POWER_PORT_KIND: ("dcim.powerport",), 

49 POWER_OUTLET_KIND: ("dcim.poweroutlet",), 

50} 

51#: Every model an explicit selection may name, in the deduplicated union order of the admission table. 

52CABLE_END_MODELS = tuple(dict.fromkeys(model for models in ADMITTED_TERMINATION_MODELS.values() for model in models)) 

53CABLE_END_KINDS = frozenset(CABLE_END_MODELS) 

54 

55TERMINATION_ROLE = "termination" 

56MAPPED_PEER_ROLE = "mapped_peer" 

57TERMINATION_ROLES = frozenset({TERMINATION_ROLE, MAPPED_PEER_ROLE}) 

58 

59SELECT_TERMINATION_TASK = "select_termination" 

60 

61 

62def same_device_and_cards(first, second) -> bool: 

63 """Return whether two Termination References name one device and one cards label. 

64 

65 The Source Adapter reads it to claim a pass-through, and the Cable Target Module reads it to 

66 place one, so the rule has one definition. 

67 """ 

68 return (identity_text(first.device), identity_text(first.cards)) == ( 

69 identity_text(second.device), 

70 identity_text(second.cards), 

71 ) 

72 

73 

74def port_class_mapping_errors(source_label: str, port_class: str) -> dict[str, str]: 

75 """Return why one source PortClass label cannot map to one PortClass, by field, or nothing when it can.""" 

76 errors = {} 

77 if not source_label.strip(): 

78 errors["source_label"] = "Enter the source PortClass label." 

79 elif source_label != source_label.strip(): 

80 # The adapter trims each workbook label, so a padded label never matches one. 

81 errors["source_label"] = "Remove the whitespace at the start and end of the source PortClass label." 

82 elif source_label in PORT_CLASSES: 

83 errors["source_label"] = f"'{source_label}' is a fixed PortClass, so it takes no mapping." 

84 if port_class not in PORT_CLASSES: 

85 errors["port_class"] = f"'{port_class}' is not a fixed PortClass." 

86 return errors 

87 

88 

89def claimed_termination_kind(port_class: str) -> str: 

90 """Return the NetBox termination kind claimed by one fixed PortClass value.""" 

91 try: 

92 return PORT_CLASS_CLAIMED_KINDS[port_class] 

93 except KeyError as exc: 

94 raise ValueError(f"Unknown PortClass '{port_class}'.") from exc 

95 

96 

97def is_outside_claim(kind: str, model_label: str) -> bool: 

98 """Return whether a selected NetBox model is one the claimed kind does not admit to automatic matching.""" 

99 return model_label not in ADMITTED_TERMINATION_MODELS[kind] 

100 

101 

102def termination_field_key(*, device, cards, port, kind: str, role: str = TERMINATION_ROLE) -> str: 

103 """Return the canonical JSON key for one termination-selection role.""" 

104 if kind not in ADMITTED_TERMINATION_MODELS: 

105 raise ValueError(f"Unknown claimed termination kind '{kind}'.") 

106 if role not in TERMINATION_ROLES: 

107 raise ValueError(f"Unknown termination field-key role '{role}'.") 

108 return json.dumps( 

109 { 

110 "cards": identity_text(cards), 

111 "device": identity_text(device), 

112 "kind": kind, 

113 "port": identity_text(port), 

114 "role": role, 

115 }, 

116 ensure_ascii=False, 

117 sort_keys=True, 

118 separators=(",", ":"), 

119 ) 

120 

121 

122def _validate_termination_field_data(data): 

123 """Reject malformed termination field-key data before canonicalization.""" 

124 if not isinstance(data, dict) or set(data) != {"cards", "device", "kind", "port", "role"}: 

125 raise ValueError 

126 if not all(isinstance(data[name], str) for name in data): 

127 raise ValueError 

128 

129 

130def parse_termination_field_key(value: str) -> dict[str, str]: 

131 """Parse an exact canonical termination field key.""" 

132 try: 

133 data = json.loads(value) 

134 _validate_termination_field_data(data) 

135 canonical = termination_field_key(**data) 

136 except (TypeError, ValueError) as exc: 

137 raise ValueError(f"'{value}' is not a canonical termination field key.") from exc 

138 if canonical != value: 

139 raise ValueError(f"'{value}' is not a canonical termination field key.") 

140 return data 

141 

142 

143__all__ = ( 

144 "ADMITTED_TERMINATION_MODELS", 

145 "CABLE_END_KINDS", 

146 "CABLE_END_MODELS", 

147 "FRONT_PORT_CLASSES", 

148 "FRONT_PORT_KIND", 

149 "FRONT_PORT_MODEL", 

150 "INTERFACE_KIND", 

151 "INTERFACE_PORT_CLASSES", 

152 "MAPPED_PEER_ROLE", 

153 "PORT_CLASSES", 

154 "PORT_CLASS_CHOICES", 

155 "PORT_CLASS_CLAIMED_KINDS", 

156 "POWER_CLASSES", 

157 "POWER_OUTLET_CLASSES", 

158 "POWER_OUTLET_KIND", 

159 "POWER_PORT_CLASSES", 

160 "POWER_PORT_KIND", 

161 "REAR_PORT_CLASSES", 

162 "REAR_PORT_KIND", 

163 "REAR_PORT_MODEL", 

164 "SELECT_TERMINATION_TASK", 

165 "TERMINATION_ROLE", 

166 "TERMINATION_ROLES", 

167 "claimed_termination_kind", 

168 "is_outside_claim", 

169 "parse_termination_field_key", 

170 "port_class_mapping_errors", 

171 "same_device_and_cards", 

172 "termination_field_key", 

173)