Coverage for netbox_data_import/termination_proposal.py: 100.00%
78 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 `select_termination` proposal task (specification 7.1).
5The only module that knows termination kinds and the `TerminationResolution` row. The lifecycle sees
6a task type, a field key and a generic Candidate Snapshot, and nothing else.
7"""
9from dataclasses import dataclass
11from .cable_target import eligible_terminations, resolved_device_for
12from .field_keys import SELECT_TERMINATION_TASK, TERMINATION_ROLE, parse_termination_field_key
13from .proposal_tasks import (
14 CandidateSet,
15 ProposalInventory,
16 UnusableCandidateSet,
17 proposal_inventory_staleness,
18 register_proposal_task,
19 snapshot_from,
20)
22__all__ = [
23 "DecisionReceipt",
24 "InvalidProposalCandidate",
25 "SelectTerminationTask",
26 "UnsupportedProposalRole",
27]
30class UnsupportedProposalRole(Exception):
31 """Proposals are requested for the termination role in this delivery (section 7.1)."""
34class InvalidProposalCandidate(Exception):
35 """A stored proposal candidate cannot identify its target model."""
38@dataclass(frozen=True)
39class DecisionReceipt:
40 """What an accepted decision wrote. The coordinator stores the id and never reads the row."""
42 written_resolution_id: int
45def _name_for(candidate) -> str:
46 """Return the display name at snapshot time, which a rename changes."""
47 return str(candidate)
50class SelectTerminationTask:
51 """Retrieve, snapshot and resolve one termination field key."""
53 task_type = SELECT_TERMINATION_TASK
55 def _require_termination_role(self, field_key) -> dict:
56 parsed = parse_termination_field_key(field_key)
57 if parsed["role"] != TERMINATION_ROLE:
58 raise UnsupportedProposalRole(f"A proposal is not requested for the '{parsed['role']}' role.")
59 return parsed
61 def current(self, *, profile, field_key, netbox_reader, limit):
62 """Return the Candidate Snapshot as the world stands now, for a request or a freshness read."""
63 self._require_termination_role(field_key)
64 device = resolved_device_for(field_key, netbox_reader, profile=profile)
65 return self._current_for_device(
66 profile=profile,
67 field_key=field_key,
68 netbox_reader=netbox_reader,
69 limit=limit,
70 device=device,
71 )
73 def _current_for_device(self, *, profile, field_key, netbox_reader, limit, device, lock_rows=False):
74 """Build the snapshot without resolving a Device the caller already read."""
75 # The picker and a proposal request share this query, so both see one eligibility rule.
76 eligible = eligible_terminations(
77 field_key,
78 netbox_reader,
79 profile=profile,
80 limit=limit,
81 _resolved_device=device,
82 _lock_rows=lock_rows,
83 )
84 return snapshot_from(
85 CandidateSet(objects=eligible.candidates, total=eligible.total),
86 name_for=_name_for,
87 limit=limit,
88 )
90 def inventory(self, *, profile, field_key, netbox_reader, limit) -> ProposalInventory:
91 """Read the resolved Device and its candidate snapshot once for display and freshness."""
92 return self._inventory(
93 profile=profile,
94 field_key=field_key,
95 netbox_reader=netbox_reader,
96 limit=limit,
97 lock_rows=False,
98 )
100 def _inventory(self, *, profile, field_key, netbox_reader, limit, lock_rows) -> ProposalInventory:
101 """Read one inventory, locking its target rows when a resolution write follows."""
102 self._require_termination_role(field_key)
103 device = resolved_device_for(field_key, netbox_reader, profile=profile, _lock_rows=lock_rows)
104 candidate_error: UnusableCandidateSet | None
105 try:
106 candidate_snapshot = self._current_for_device(
107 profile=profile,
108 field_key=field_key,
109 netbox_reader=netbox_reader,
110 limit=limit,
111 device=device,
112 lock_rows=lock_rows,
113 )
114 except UnusableCandidateSet as exc:
115 candidate_snapshot = None
116 candidate_error = exc
117 else:
118 candidate_error = None
119 return ProposalInventory(
120 resolved_device=device,
121 candidate_snapshot=candidate_snapshot,
122 candidate_error=candidate_error,
123 )
125 def resolved_device(self, *, profile, field_key, netbox_reader):
126 """Return the one Device this key resolves to now, or None when it does not resolve to one."""
127 self._require_termination_role(field_key)
128 return resolved_device_for(field_key, netbox_reader, profile=profile)
130 def _resolution_write(self, *, profile, field_key, entry, source):
131 """Return the validated lookup and values shared by assessment and execution."""
132 from core.models import ObjectType
134 from .models import TerminationResolution
136 try:
137 app_label, model = entry.object_type.split(".", 1)
138 object_type = ObjectType.objects.get(app_label=app_label, model=model)
139 except (AttributeError, TypeError, ValueError, ObjectType.DoesNotExist) as exc:
140 raise InvalidProposalCandidate("The stored proposal candidate has an invalid object type.") from exc
141 spelling = {f"source_{part}": source[part] for part in ("device", "cards", "port")}
142 candidate = TerminationResolution(
143 profile=profile,
144 task_type=self.task_type,
145 field_key=field_key,
146 **spelling,
147 selected_object_type=object_type,
148 selected_object_id=entry.object_id,
149 selected_display_name=entry.display_name,
150 )
151 candidate.full_clean(validate_unique=False, validate_constraints=False)
152 lookup = {
153 "profile": profile,
154 "task_type": self.task_type,
155 "field_key": field_key,
156 "field_key_digest": candidate.field_key_digest,
157 }
158 values = {
159 **spelling,
160 "selected_object_type": object_type,
161 "selected_object_id": entry.object_id,
162 "selected_display_name": entry.display_name,
163 }
164 return lookup, values
166 def assess_resolution_write(self, *, profile, field_key, entry, source, actor):
167 """Return whether the exact resolution write stays inside the actor's current scope."""
168 from .models import TerminationResolution
169 from .object_permissions import assess_permission_scoped_save
171 lookup, values = self._resolution_write(profile=profile, field_key=field_key, entry=entry, source=source)
172 return assess_permission_scoped_save(actor, TerminationResolution, lookup, values)
174 def write_resolution(self, *, profile, field_key, entry, source, actor) -> DecisionReceipt:
175 """Upsert the Row Resolution the accepted candidate names, and return only its id."""
176 from .models import TerminationResolution
177 from .object_permissions import save_permission_scoped_object
179 lookup, values = self._resolution_write(profile=profile, field_key=field_key, entry=entry, source=source)
180 saved = save_permission_scoped_object(actor, TerminationResolution, lookup, values)
181 return DecisionReceipt(written_resolution_id=saved.instance.pk)
183 def write_resolution_if_fresh(self, *, proposal, entry, source, actor, netbox_reader, limit):
184 """Lock and recheck the target inventory, then write one still-fresh resolution."""
185 assessment = self.assess_resolution_write(
186 profile=proposal.profile,
187 field_key=proposal.field_key,
188 entry=entry,
189 source=source,
190 actor=actor,
191 )
192 if not assessment.allowed:
193 from .object_permissions import ObjectPermissionDenied
195 raise ObjectPermissionDenied(assessment.permission)
196 inventory = self._inventory(
197 profile=proposal.profile,
198 field_key=proposal.field_key,
199 netbox_reader=netbox_reader,
200 limit=limit,
201 lock_rows=True,
202 )
203 if proposal_inventory_staleness(proposal, inventory).is_stale:
204 return None
205 receipt = self.write_resolution(
206 profile=proposal.profile,
207 field_key=proposal.field_key,
208 entry=entry,
209 source=source,
210 actor=actor,
211 )
212 from .review_workspace import pin_trace_device
214 pin_trace_device(
215 profile=proposal.profile, actor=actor, source_device=source["device"], device=inventory.resolved_device
216 )
217 return receipt
220register_proposal_task(SELECT_TERMINATION_TASK, SelectTerminationTask())