Coverage for netbox_data_import/proposal_tasks.py: 100.00%
77 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""The closed proposal task-type registry and the generic Candidate Snapshot (specification 7.9).
5This is the seam that keeps termination specifics out of the lifecycle. A later task type adds a key
6shape and a candidate retrieval, and changes nothing here.
7"""
9from dataclasses import dataclass, replace
11from .proposal_response import candidate_id_for, validate_candidate_ids
12from .public_refusal import PublicRefusal
14#: Why a request cannot be made, or why a snapshot cannot establish freshness.
15NO_CANDIDATES = "no_candidates"
16TOO_MANY_CANDIDATES = "too_many_candidates"
18__all__ = [
19 "NO_CANDIDATES",
20 "TOO_MANY_CANDIDATES",
21 "CandidateSet",
22 "CandidateSnapshot",
23 "CandidateSnapshotEntry",
24 "ProposalInventory",
25 "ProposalStaleness",
26 "UnknownProposalTask",
27 "UnusableCandidateSet",
28 "proposal_inventory_staleness",
29 "proposal_task",
30 "register_proposal_task",
31 "snapshot_from",
32]
35class UnknownProposalTask(Exception):
36 """No task type with that name is registered. The registry is closed on purpose."""
39class UnusableCandidateSet(PublicRefusal):
40 """The eligible set cannot back a proposal. `reason` is one of the two module constants."""
42 def __init__(self, reason, message):
43 super().__init__(message)
44 self.reason = reason
47@dataclass(frozen=True)
48class CandidateSnapshotEntry:
49 """One candidate as it stood at snapshot time (section 7.3)."""
51 candidate_id: str
52 object_type: str
53 object_id: int
54 display_name: str
56 def as_json(self) -> dict:
57 """Return the stored form. Only strings and integers cross into the row."""
58 return {
59 "candidate_id": self.candidate_id,
60 "object_type": self.object_type,
61 "object_id": self.object_id,
62 "display_name": self.display_name,
63 }
66@dataclass(frozen=True)
67class CandidateSnapshot:
68 """The immutable candidate set one request was built from, and the page it offered.
70 The set establishes freshness; the page is only what the prompt carried. Keeping them apart is
71 what lets a dense Device be searched in turns without a truncated read freezing the evidence.
72 """
74 entries: tuple[CandidateSnapshotEntry, ...]
75 total: int
76 page_offset: int = 0
77 page_size: int | None = None
79 @property
80 def candidate_ids(self) -> tuple[str, ...]:
81 """Return the opaque identifiers this snapshot holds, in order."""
82 return tuple(entry.candidate_id for entry in self.entries)
84 @property
85 def page(self) -> tuple[CandidateSnapshotEntry, ...]:
86 """Return the entries this request offered the backend."""
87 size = len(self.entries) if self.page_size is None else self.page_size
88 return self.entries[self.page_offset : self.page_offset + size]
90 @property
91 def page_end(self) -> int:
92 """Return the offset one past this page, which is where the next page starts."""
93 return self.page_offset + len(self.page)
95 @property
96 def has_next_page(self) -> bool:
97 """Return whether the set holds candidates this request never offered."""
98 return self.page_end < self.total
100 def with_page(self, *, offset: int, size: int) -> "CandidateSnapshot":
101 """Return the same set, recording the page one request is about to offer."""
102 return replace(self, page_offset=offset, page_size=size)
104 def as_json(self) -> dict:
105 """Return the stored form of the whole snapshot."""
106 return {
107 "total": self.total,
108 "candidates": [entry.as_json() for entry in self.entries],
109 "page_offset": self.page_offset,
110 "page_size": self.page_size,
111 }
113 @classmethod
114 def from_json(cls, stored) -> "CandidateSnapshot":
115 """Rebuild a stored snapshot so a read-time comparison works on the same shape."""
116 entries = tuple(CandidateSnapshotEntry(**entry) for entry in stored["candidates"])
117 return cls(
118 entries=entries,
119 total=stored["total"],
120 # A row written before paging offered every candidate it held.
121 page_offset=stored.get("page_offset", 0),
122 page_size=stored.get("page_size"),
123 )
125 def matches(self, other) -> bool:
126 """Compare the whole set, display names included: a rename changed the evidence.
128 The page is deliberately excluded: two pages of one set describe the same world.
129 """
130 return self.total == other.total and self.entries == other.entries
133@dataclass(frozen=True)
134class ProposalStaleness:
135 """The two independent reasons a proposal's frozen evidence no longer applies."""
137 resolved_device_changed: bool
138 candidates_changed: bool
140 @property
141 def is_stale(self) -> bool:
142 """Return whether either freshness trigger fired."""
143 return self.resolved_device_changed or self.candidates_changed
146@dataclass(frozen=True)
147class CandidateSet:
148 """What a task type retrieves: the objects themselves and the uncapped matching total."""
150 objects: tuple
151 total: int
154@dataclass(frozen=True)
155class ProposalInventory:
156 """The resolved Device and eligible candidates one freshness read observed together."""
158 resolved_device: object | None
159 candidate_snapshot: CandidateSnapshot | None
160 candidate_error: UnusableCandidateSet | None = None
163def proposal_inventory_staleness(proposal, inventory) -> ProposalStaleness:
164 """Compare one current inventory read with a proposal's frozen evidence."""
165 device = inventory.resolved_device
166 device_changed = (
167 device is None
168 or device.pk != proposal.resolved_device_id
169 or device._meta.label_lower
170 != f"{proposal.resolved_device_type.app_label}.{proposal.resolved_device_type.model}"
171 )
172 current = inventory.candidate_snapshot
173 candidates_changed = current is None or not CandidateSnapshot.from_json(proposal.candidate_snapshot).matches(
174 current
175 )
176 return ProposalStaleness(resolved_device_changed=device_changed, candidates_changed=candidates_changed)
179def snapshot_from(candidate_set, *, name_for, limit) -> CandidateSnapshot:
180 """Turn one retrieved set into a snapshot, refusing a set that cannot back a proposal.
182 A truncated result must never establish freshness, so the retrieved count has to equal the
183 uncapped total. Identifiers are positional and opaque; nothing about the object leaks into them.
184 """
185 total = candidate_set.total
186 if total == 0:
187 raise UnusableCandidateSet(NO_CANDIDATES, "No eligible candidate exists for this field.")
188 if total > limit:
189 raise UnusableCandidateSet(
190 TOO_MANY_CANDIDATES,
191 f"{total} eligible candidates exceed the configured bound of {limit}.",
192 )
193 if len(candidate_set.objects) != total:
194 raise UnusableCandidateSet(
195 TOO_MANY_CANDIDATES,
196 f"The retrieval returned {len(candidate_set.objects)} of {total} candidates.",
197 )
198 entries = tuple(
199 CandidateSnapshotEntry(
200 candidate_id=candidate_id_for(position),
201 object_type=candidate._meta.label_lower,
202 object_id=candidate.pk,
203 display_name=name_for(candidate),
204 )
205 for position, candidate in enumerate(candidate_set.objects)
206 )
207 validate_candidate_ids(entry.candidate_id for entry in entries)
208 return CandidateSnapshot(entries=entries, total=total)
211_REGISTRY: dict = {}
214def register_proposal_task(task_type, task) -> None:
215 """Add one task type to the closed registry, at import time."""
216 _REGISTRY[task_type] = task
219def proposal_task(task_type):
220 """Return the registered task, refusing a name the deployment never declared."""
221 try:
222 return _REGISTRY[task_type]
223 except KeyError as exc:
224 raise UnknownProposalTask(f"'{task_type}' is not a proposal task type.") from exc