Coverage for netbox_data_import/source_trace.py: 100.00%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""Adapter-neutral source values for traced physical paths.""" 

4 

5from __future__ import annotations 

6 

7from dataclasses import dataclass 

8from typing import TYPE_CHECKING 

9 

10from .field_keys import PORT_CLASS_CLAIMED_KINDS, REAR_PORT_CLASSES 

11from .identity import identity_text 

12 

13if TYPE_CHECKING: 

14 from .adapters import SourceDiagnostic 

15 

16 

17IdentityKey = tuple[str, str, str, str] 

18 

19 

20@dataclass(frozen=True) 

21class TerminationReference: 

22 """Name one source termination and retain its corroboration values.""" 

23 

24 device: str 

25 cards: str 

26 port: str 

27 port_class: str 

28 u_position: str = "" 

29 rack: str = "" 

30 location: str = "" 

31 

32 @property 

33 def identity_key(self) -> IdentityKey: 

34 """Return the normalized device, cards, port, and claimed-kind identity.""" 

35 return ( 

36 identity_text(self.device), 

37 identity_text(self.cards), 

38 identity_text(self.port), 

39 PORT_CLASS_CLAIMED_KINDS.get(self.port_class, identity_text(self.port_class)), 

40 ) 

41 

42 

43@dataclass(frozen=True) 

44class EndpointSummary: 

45 """Retain the source From and To statements in their original direction.""" 

46 

47 from_termination: TerminationReference 

48 to_termination: TerminationReference 

49 from_text: str 

50 to_text: str 

51 

52 

53@dataclass(frozen=True) 

54class SegmentEvidence: 

55 """Describe one source-claimed cable between two Termination References.""" 

56 

57 left: TerminationReference 

58 cable_class: str 

59 right: TerminationReference 

60 

61 

62@dataclass(frozen=True) 

63class PassThroughClaim: 

64 """Describe the source-claimed continuation through one device.""" 

65 

66 device: str 

67 cards: str 

68 entry_port: str 

69 exit_port: str 

70 

71 

72@dataclass(frozen=True, order=True) 

73class PortClassLabel: 

74 """Name one source PortClass label outside the fixed vocabulary, and the PortClass a map gave it.""" 

75 

76 label: str 

77 port_class: str = "" 

78 

79 

80@dataclass(frozen=True) 

81class TraceProvenance: 

82 """Locate one Source Trace occurrence in its source document.""" 

83 

84 workbook_fingerprint: str 

85 sheet: str 

86 block_ordinal: int 

87 row_start: int 

88 row_end: int 

89 export_timestamp: str 

90 from_text: str 

91 to_text: str 

92 direction: str 

93 

94 

95@dataclass(frozen=True) 

96class SourceTrace: 

97 """Carry canonical Segment Evidence for one path or an Endpoint Summary fallback.""" 

98 

99 endpoint_summary: EndpointSummary 

100 segments: tuple[SegmentEvidence, ...] 

101 pass_through_claims: tuple[PassThroughClaim, ...] 

102 corroboration: tuple[TerminationReference, ...] 

103 identity: str 

104 content_fingerprint: str 

105 provenance: tuple[TraceProvenance, ...] 

106 errors: tuple[SourceDiagnostic, ...] = () 

107 port_class_labels: tuple[PortClassLabel, ...] = () 

108 

109 def __post_init__(self) -> None: 

110 """Reject a trace that states no source position, which every consumer reads.""" 

111 if not self.provenance: 

112 raise ValueError("A Source Trace must carry at least one TraceProvenance record.") 

113 

114 @property 

115 def valid(self) -> bool: 

116 """Return whether source validation found no errors on this trace.""" 

117 return not self.errors 

118 

119 @property 

120 def ends_at_rear_port(self) -> bool: 

121 """Return whether the original To termination is a rear port.""" 

122 return self.endpoint_summary.to_termination.port_class in REAR_PORT_CLASSES 

123 

124 

125__all__ = ( 

126 "EndpointSummary", 

127 "IdentityKey", 

128 "PassThroughClaim", 

129 "PortClassLabel", 

130 "SegmentEvidence", 

131 "SourceTrace", 

132 "TerminationReference", 

133 "TraceProvenance", 

134)