Coverage for netbox_data_import/proposal_response.py: 98.57%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""The strict Resolution Proposal response validator (specification 7.8). 

4 

5Pure: it reads the returned text and the immutable request snapshot, and returns one answer or 

6raises. It never repairs a response, never fuzzy-matches an identifier, and never asks again. 

7""" 

8 

9import json 

10 

11from dataclasses import dataclass 

12 

13from .proposal_contract import ( 

14 EXPLANATION_MAX_LENGTH, 

15 OUTCOMES, 

16 OUTCOME_CANDIDATE, 

17 OUTCOME_NO_MATCH, 

18 RESPONSE_MEMBERS, 

19 RESPONSE_SCHEMA_VERSION, 

20) 

21 

22__all__ = [ 

23 "EXPLANATION_MAX_LENGTH", 

24 "OUTCOMES", 

25 "OUTCOME_CANDIDATE", 

26 "OUTCOME_NO_MATCH", 

27 "RESPONSE_SCHEMA_VERSION", 

28 "InvalidProposalResponse", 

29 "ProposalAnswer", 

30 "candidate_id_for", 

31 "validate_candidate_ids", 

32 "validate_response", 

33] 

34 

35 

36class InvalidProposalResponse(Exception): 

37 """The response is not one valid object for this request. Stored reason: `invalid_response`.""" 

38 

39 

40@dataclass(frozen=True) 

41class ProposalAnswer: 

42 """What one valid response concluded. Candidate fields are None exactly for no_match.""" 

43 

44 outcome: str 

45 candidate_id: str | None 

46 candidate_display_name: str | None 

47 explanation: str 

48 

49 

50def candidate_id_for(position: int) -> str: 

51 """Return the opaque id for a candidate at *position*, counted from zero (section 7.8).""" 

52 return f"candidate-{position + 1:04d}" 

53 

54 

55def validate_candidate_ids(candidate_ids) -> tuple[str, ...]: 

56 """Return the request's identifiers, refusing a duplicate before the call is ever made.""" 

57 ids = tuple(candidate_ids) 

58 if len(set(ids)) != len(ids): 

59 raise ValueError("Candidate identifiers must be unique within one request.") 

60 return ids 

61 

62 

63def _no_duplicate_members(pairs): 

64 """Reject a repeated member while decoding, because `json.loads` keeps only the last one.""" 

65 seen = {} 

66 for name, value in pairs: 

67 if name in seen: 

68 raise InvalidProposalResponse(f"The response repeats the member '{name}'.") 

69 seen[name] = value 

70 return seen 

71 

72 

73def _decode(content_text): 

74 """Return the one JSON object the response has to be.""" 

75 if not isinstance(content_text, str) or not content_text.strip(): 

76 raise InvalidProposalResponse("The response carries no content.") 

77 try: 

78 decoded = json.loads(content_text, object_pairs_hook=_no_duplicate_members) 

79 except InvalidProposalResponse: 

80 raise 

81 except (ValueError, RecursionError) as exc: 

82 raise InvalidProposalResponse("The response is not JSON.") from exc 

83 if not isinstance(decoded, dict): 

84 raise InvalidProposalResponse("The response is not one JSON object.") 

85 return decoded 

86 

87 

88def _explanation(value): 

89 """Return the required explanation, which the operator reads to understand the answer.""" 

90 if not isinstance(value, str) or not value.strip(): 

91 raise InvalidProposalResponse("The response carries no explanation.") 

92 if len(value) > EXPLANATION_MAX_LENGTH: 

93 raise InvalidProposalResponse(f"The explanation is longer than {EXPLANATION_MAX_LENGTH} characters.") 

94 return value 

95 

96 

97def validate_response( 

98 content_text, *, candidate_display_names, schema_version=RESPONSE_SCHEMA_VERSION 

99) -> ProposalAnswer: 

100 """Return the answer one response states, or raise if it is not exactly one valid answer. 

101 

102 Identifiers are compared as exact opaque strings: an invented, missing or reshaped one is an 

103 invalid response, never a near match. 

104 """ 

105 decoded = _decode(content_text) 

106 unknown = set(decoded) - RESPONSE_MEMBERS 

107 if unknown: 

108 raise InvalidProposalResponse(f"The response carries unknown members: {', '.join(sorted(unknown))}.") 

109 missing = RESPONSE_MEMBERS - set(decoded) 

110 if missing: 

111 raise InvalidProposalResponse(f"The response is missing: {', '.join(sorted(missing))}.") 

112 

113 # A bool is an int subclass. Keep schema versions restricted to integers supplied as integers. 

114 version = decoded["schema_version"] 

115 if isinstance(version, bool) or not isinstance(version, int) or version != schema_version: 

116 raise InvalidProposalResponse(f"The response declares schema_version {version!r}, not {schema_version}.") 

117 

118 outcome = decoded["outcome"] 

119 if outcome not in OUTCOMES: 

120 raise InvalidProposalResponse(f"The response declares outcome {outcome!r}.") 

121 

122 explanation = _explanation(decoded["explanation"]) 

123 candidate_id = decoded["candidate_id"] 

124 candidate_display_name = decoded["candidate_display_name"] 

125 if outcome == OUTCOME_NO_MATCH: 

126 if candidate_id is not None: 

127 raise InvalidProposalResponse("A no_match response carries a candidate_id.") 

128 if candidate_display_name is not None: 

129 raise InvalidProposalResponse("A no_match response carries a candidate_display_name.") 

130 return ProposalAnswer( 

131 outcome=OUTCOME_NO_MATCH, 

132 candidate_id=None, 

133 candidate_display_name=None, 

134 explanation=explanation, 

135 ) 

136 

137 candidate_ids = validate_candidate_ids(candidate_display_names) 

138 if not isinstance(candidate_id, str) or candidate_id not in candidate_ids: 

139 raise InvalidProposalResponse("The response selects a candidate that this request did not offer.") 

140 if not isinstance(candidate_display_name, str) or not candidate_display_name: 

141 raise InvalidProposalResponse("A candidate response carries no candidate_display_name.") 

142 if candidate_display_name != candidate_display_names[candidate_id]: 

143 raise InvalidProposalResponse("The candidate_id and candidate_display_name select different candidates.") 

144 return ProposalAnswer( 

145 outcome=OUTCOME_CANDIDATE, 

146 candidate_id=candidate_id, 

147 candidate_display_name=candidate_display_name, 

148 explanation=explanation, 

149 )