Coverage for netbox_data_import/field_keys.py: 93%

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 .values 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"}) 

14PORT_CLASSES = INTERFACE_PORT_CLASSES | FRONT_PORT_CLASSES | REAR_PORT_CLASSES 

15 

16INTERFACE_KIND = "interface" 

17FRONT_PORT_KIND = "front_port" 

18REAR_PORT_KIND = "rear_port" 

19 

20PORT_CLASS_CLAIMED_KINDS = { 

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

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

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

24} 

25 

26TERMINATION_ROLE = "termination" 

27MAPPED_PEER_ROLE = "mapped_peer" 

28TERMINATION_ROLES = frozenset({TERMINATION_ROLE, MAPPED_PEER_ROLE}) 

29 

30SELECT_TERMINATION_TASK = "select_termination" 

31 

32 

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

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

35 

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

37 place one, so the rule has one definition. 

38 """ 

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

40 identity_text(second.device), 

41 identity_text(second.cards), 

42 ) 

43 

44 

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

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

47 try: 

48 return PORT_CLASS_CLAIMED_KINDS[port_class] 

49 except KeyError as exc: 

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

51 

52 

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

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

55 if kind not in {INTERFACE_KIND, FRONT_PORT_KIND, REAR_PORT_KIND}: 

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

57 if role not in TERMINATION_ROLES: 

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

59 return json.dumps( 

60 { 

61 "cards": identity_text(cards), 

62 "device": identity_text(device), 

63 "kind": kind, 

64 "port": identity_text(port), 

65 "role": role, 

66 }, 

67 ensure_ascii=False, 

68 sort_keys=True, 

69 separators=(",", ":"), 

70 ) 

71 

72 

73def _validate_termination_field_data(data): 

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

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

76 raise ValueError 

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

78 raise ValueError 

79 

80 

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

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

83 try: 

84 data = json.loads(value) 

85 _validate_termination_field_data(data) 

86 canonical = termination_field_key(**data) 

87 except (TypeError, ValueError) as exc: 

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

89 if canonical != value: 

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

91 return data 

92 

93 

94__all__ = ( 

95 "FRONT_PORT_CLASSES", 

96 "FRONT_PORT_KIND", 

97 "INTERFACE_KIND", 

98 "INTERFACE_PORT_CLASSES", 

99 "MAPPED_PEER_ROLE", 

100 "PORT_CLASSES", 

101 "PORT_CLASS_CLAIMED_KINDS", 

102 "REAR_PORT_CLASSES", 

103 "REAR_PORT_KIND", 

104 "SELECT_TERMINATION_TASK", 

105 "TERMINATION_ROLE", 

106 "TERMINATION_ROLES", 

107 "claimed_termination_kind", 

108 "parse_termination_field_key", 

109 "same_device_and_cards", 

110 "termination_field_key", 

111)