Coverage for netbox_data_import/trace_location_resolution.py: 100.00%
199 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"""Map source Location paths and their prefixes to NetBox Locations inside one permission-scoped import target."""
5from __future__ import annotations
7from collections.abc import Iterable, Mapping
8from dataclasses import dataclass, replace
9from typing import Any
11from .identity import identity_text, matching_search
13PATH_SEPARATOR = ">>"
15UNMAPPED = "unmapped"
16MAPPED = "mapped"
17STALE = "stale"
18HIDDEN = "hidden"
20NO_VISIBLE_LOCATION = "No Location in this Site is visible to you."
21SAVE_PERMISSION_REFUSED = "You do not have permission to save this Location mapping."
22CLEAR_PERMISSION_REFUSED = "You do not have permission to clear this Location mapping."
23STALE_REASON = "The mapped Location is no longer available at this import target. Choose it again."
24DIGEST_MISMATCH = "A Trace Location Resolution digest does not match its source Location key."
27def source_location_key(path: str) -> str:
28 """Return the profile-wide identity of one source Location path or prefix."""
29 return identity_text(path)
32def location_prefixes(path: str) -> tuple[tuple[str, str], ...]:
33 """Return each prefix of one source Location path as (key, source text), shortest first."""
34 segments = path.split(PATH_SEPARATOR)
35 prefixes = [
36 (source_location_key(text), text)
37 for end in range(1, len(segments))
38 if identity_text(segments[end - 1])
39 for text in (PATH_SEPARATOR.join(segments[:end]),)
40 ]
41 # The full path is its own last prefix even when its last segment is blank.
42 if key := source_location_key(path):
43 prefixes.append((key, path))
44 return tuple(prefixes)
47def location_prefix_spellings(paths: Mapping[str, str]) -> dict[str, str]:
48 """Return every prefix key of the paths, spelled as the first path in key order that carries it."""
49 spellings: dict[str, str] = {}
50 for path_key in sorted(paths):
51 for key, text in location_prefixes(paths[path_key]):
52 spellings.setdefault(key, text)
53 return spellings
56@dataclass(frozen=True)
57class LocationMapping:
58 """How one stored source Location row maps for one actor at one import target."""
60 key: str
61 state: str
62 # The stored row, only when the actor may view it.
63 row: Any | None
64 # The mapped Location, only while it is visible and inside the selected Site.
65 location: Any | None
68def site_locations(reader):
69 """Return the Locations the actor may view inside the selected Site."""
70 locations = reader.locations()
71 return locations.filter(site=reader.site) if reader.site is not None else locations
74@dataclass(frozen=True)
75class LocationCandidate:
76 """One visible Location the picker offers, with its parent's name when the actor may view it."""
78 location: Any
79 parent: str
82@dataclass(frozen=True)
83class LocationCandidatePage:
84 """One bounded page of Location candidates and its uncapped total."""
86 candidates: tuple[LocationCandidate, ...]
87 total: int
90def eligible_trace_locations(reader, *, search: str = "", limit: int, offset: int = 0) -> LocationCandidatePage:
91 """Return one bounded page of the selected Site's visible Locations, searched by normalized name."""
92 locations = matching_search(site_locations(reader), search).order_by("name", "pk")
93 total = locations.count()
94 page = tuple(locations[offset : offset + limit]) if offset < total else ()
95 # Two Locations can share a name under different parents, so the visible parent tells them apart.
96 parents = dict(
97 site_locations(reader)
98 .filter(pk__in={location.parent_id for location in page if location.parent_id})
99 .values_list("pk", "name")
100 )
101 return LocationCandidatePage(
102 candidates=tuple(LocationCandidate(location, parents.get(location.parent_id, "")) for location in page),
103 total=total,
104 )
107def trace_location_mappings(*, profile, reader, keys: Iterable[str]) -> dict[str, LocationMapping]:
108 """Return the own row state of every source Location key, read in bulk inside the actor's scope."""
109 from .models import TraceLocationResolution, index_digest
111 keys = tuple(dict.fromkeys(key for key in keys if key))
112 if not keys:
113 return {}
114 requested = frozenset(keys)
115 rows = TraceLocationResolution.objects.filter(
116 profile=profile,
117 source_location_key_digest__in=[index_digest(key) for key in keys],
118 )
119 stored = {}
120 for row in rows:
121 # Two requested prefixes can name each other's keys, so the key alone does not prove the digest.
122 if (
123 index_digest(row.source_location_key) != row.source_location_key_digest
124 or row.source_location_key not in requested
125 ):
126 raise ValueError(DIGEST_MISMATCH)
127 stored[row.source_location_key] = row
128 visible_rows = (
129 set(stored)
130 if reader.actor is None
131 else set(
132 TraceLocationResolution.objects.restrict(reader.actor, "view")
133 .filter(pk__in=[row.pk for row in stored.values()])
134 .values_list("source_location_key", flat=True)
135 )
136 )
137 locations = {
138 location.pk: location
139 for location in site_locations(reader).filter(pk__in=[row.selected_location_id for row in stored.values()])
140 }
141 mappings = {}
142 for key in keys:
143 row = stored.get(key)
144 if row is None:
145 mappings[key] = LocationMapping(key=key, state=UNMAPPED, row=None, location=None)
146 elif key not in visible_rows:
147 mappings[key] = LocationMapping(key=key, state=HIDDEN, row=None, location=None)
148 elif row.selected_location_id not in locations:
149 mappings[key] = LocationMapping(key=key, state=STALE, row=row, location=None)
150 else:
151 mappings[key] = LocationMapping(
152 key=key, state=MAPPED, row=row, location=locations[row.selected_location_id]
153 )
154 return mappings
157def stored_location_row(profile, key: str):
158 """Return the row stored for one source Location key, refusing a row whose digest names another key."""
159 from .models import TraceLocationResolution, index_digest
161 row = TraceLocationResolution.objects.filter(profile=profile, source_location_key_digest=index_digest(key)).first()
162 if row is not None and row.source_location_key != key:
163 raise ValueError(DIGEST_MISMATCH)
164 return row
167def decided_location_mappings(*, profile, reader, paths: Iterable[str]) -> dict[str, LocationMapping]:
168 """Return, by path key, the mapping of the longest prefix of each path that has a stored row."""
169 prefixes = {key: location_prefixes(path) for path in paths if (key := source_location_key(path))}
170 own = trace_location_mappings(
171 profile=profile, reader=reader, keys=(key for chain in prefixes.values() for key, _text in chain)
172 )
173 return {path_key: _deciding(own, [key for key, _text in chain]) for path_key, chain in prefixes.items()}
176def _deciding(own: Mapping[str, LocationMapping], chain: list[str]) -> LocationMapping:
177 """Return the own mapping of the longest prefix in *chain* that has a stored row, else the last one's."""
178 # A stale or hidden row decides too, so the search never passes it to a shorter row.
179 return next((own[key] for key in reversed(chain) if own[key].state != UNMAPPED), own[chain[-1]])
182@dataclass(frozen=True)
183class LocationState:
184 """How the workspace shows one mapping state; a hidden row discloses nothing past its existence."""
186 state: str
187 label: str
188 style: str
189 location: str
190 detail: str
193@dataclass(frozen=True)
194class LocationSegment:
195 """One segment of a tree node's text, which opens the picker for the prefix ending at it."""
197 key: str
198 text: str
199 label: str
200 save_reason: str
203@dataclass(frozen=True)
204class LocationNode:
205 """One row of the workspace tree of source Location prefixes, listed flat in page order."""
207 index: int
208 parent: int | None
209 depth: int
210 children: tuple[int, ...]
211 key: str
212 path: str
213 segments: tuple[LocationSegment, ...]
214 own: LocationState
215 # The state of the longest shorter prefix with a stored row, only when this node has no row of its own.
216 inherited: LocationState | None
217 path_count: int
218 expanded: bool
219 shown: bool
220 clearable: bool
221 clear_reason: str
222 reasons: tuple[str, ...]
224 @property
225 def save_reason(self) -> str:
226 """Return why the prefix this node ends at cannot be saved, or an empty string."""
227 return self.segments[-1].save_reason
229 @property
230 def inherited_marker(self) -> str:
231 """Return the muted text that names the inherited state; a hidden row discloses only that it exists."""
232 return _INHERITED_MARKER[self.inherited.state] if self.inherited is not None else ""
235_INHERITED_MARKER = {
236 MAPPED: "inherited",
237 STALE: "inherited stale",
238 HIDDEN: "inherited, a mapping you cannot view",
239}
241_STATE_PRESENTATION = {
242 UNMAPPED: ("unmapped", "unknown"),
243 MAPPED: ("mapped", "manual"),
244 STALE: ("stale", "stale"),
245 HIDDEN: ("a mapping you cannot view", "unknown"),
246}
249def _present_state(mapping: LocationMapping) -> LocationState:
250 label, style = _STATE_PRESENTATION[mapping.state]
251 return LocationState(
252 state=mapping.state,
253 label=label,
254 style=style,
255 location=mapping.location.name if mapping.location is not None else "",
256 detail=STALE_REASON if mapping.state == STALE else "",
257 )
260@dataclass
261class _PrefixTree:
262 """Every prefix the batch paths carry, keyed and spelled once, with its child prefixes and path count."""
264 spellings: dict[str, str]
265 segments: dict[str, str]
266 children: dict[str | None, dict[str, None]]
267 counts: dict[str, int]
268 ends: set[str]
270 @classmethod
271 def of(cls, paths: Mapping[str, str]) -> _PrefixTree:
272 tree = cls(spellings={}, segments={}, children={None: {}}, counts={}, ends=set(paths))
273 for path_key in sorted(paths):
274 parent: str | None = None
275 previous: str | None = None
276 for key, text in location_prefixes(paths[path_key]):
277 if key not in tree.spellings:
278 tree.spellings[key] = text
279 tree.segments[key] = _segment_text(text, previous)
280 tree.children.setdefault(parent, {})[key] = None
281 tree.children[key] = {}
282 tree.counts[key] = tree.counts.get(key, 0) + 1
283 parent, previous = key, text
284 return tree
287def _segment_text(text: str, parent_text: str | None) -> str:
288 """Return the text a prefix adds to its parent prefix, without the separator between them."""
289 if parent_text is None:
290 return text.strip()
291 added = text[len(parent_text) :]
292 return added.removeprefix(PATH_SEPARATOR).strip() or added.strip()
295def present_location_tree(
296 *, profile, viewer, reader, paths: Mapping[str, str], has_locations: bool
297) -> tuple[LocationNode, ...]:
298 """Return the workspace tree of the batch paths' prefixes, each action permission-checked for *viewer*."""
299 from .models import TraceLocationResolution, index_digest
300 from .object_permissions import POLICY_WRITE_REFUSED, assess_loaded_save_option
302 tree = _PrefixTree.of(paths)
303 own = trace_location_mappings(profile=profile, reader=reader, keys=tree.spellings)
304 visible_rows = [mapping.row.pk for mapping in own.values() if mapping.row is not None]
305 deletable = set(
306 TraceLocationResolution.objects.restrict(viewer, "delete")
307 .filter(pk__in=visible_rows)
308 .values_list("pk", flat=True)
309 if visible_rows
310 else ()
311 )
313 def save_reason(key: str) -> str:
314 mapping = own[key]
315 if mapping.state == HIDDEN:
316 return POLICY_WRITE_REFUSED
317 if not has_locations:
318 return NO_VISIBLE_LOCATION
319 assessment = assess_loaded_save_option(
320 viewer,
321 TraceLocationResolution,
322 {"profile": profile, "source_location_key": key, "source_location_key_digest": index_digest(key)},
323 {
324 "source_location_path": tree.spellings[key],
325 "selected_location_id": 1,
326 "selected_display_name": "Pending Location mapping",
327 },
328 current=mapping.row,
329 unknown_fields={"selected_location_id", "selected_display_name"},
330 )
331 return "" if assessment.allowed else SAVE_PERMISSION_REFUSED
333 nodes: list[LocationNode] = []
334 children: dict[int, list[int]] = {}
335 # (first prefix, depth, parent row, row that decides above, shown), popped in page order.
336 pending: list[tuple[str, int, int | None, LocationMapping | None, bool]] = [
337 (root, 0, None, None, True) for root in sorted(tree.children[None], reverse=True)
338 ]
339 while pending:
340 start, depth, parent, inherited, shown = pending.pop()
341 chain = [start]
342 # A node ends at a branch, at a stored row, or at a batch path.
343 while len(tree.children[chain[-1]]) == 1 and own[chain[-1]].state == UNMAPPED and chain[-1] not in tree.ends:
344 chain.append(next(iter(tree.children[chain[-1]])))
345 key = chain[-1]
346 mapping = own[key]
347 stored = mapping.state != UNMAPPED
348 segments = tuple(
349 LocationSegment(
350 key=item, text=tree.segments[item], label=tree.spellings[item], save_reason=save_reason(item)
351 )
352 for item in chain
353 )
354 if mapping.state == HIDDEN:
355 clear_reason = POLICY_WRITE_REFUSED
356 elif mapping.row is not None and mapping.row.pk not in deletable:
357 clear_reason = CLEAR_PERMISSION_REFUSED
358 else:
359 clear_reason = ""
360 index = len(nodes)
361 if parent is not None:
362 children[parent].append(index)
363 children[index] = []
364 expanded = bool(tree.children[key]) and depth == 0 and mapping.state != MAPPED
365 nodes.append(
366 LocationNode(
367 index=index,
368 parent=parent,
369 depth=depth,
370 children=(),
371 key=key,
372 path=tree.spellings[key],
373 segments=segments,
374 own=_present_state(mapping),
375 inherited=None if stored or inherited is None else _present_state(inherited),
376 path_count=tree.counts[key],
377 expanded=expanded,
378 shown=shown,
379 clearable=stored,
380 clear_reason=clear_reason,
381 reasons=tuple(
382 dict.fromkeys(
383 reason for reason in (*(item.save_reason for item in segments), clear_reason) if reason
384 )
385 ),
386 )
387 )
388 below = mapping if stored else inherited
389 pending.extend(
390 (child, depth + 1, index, below, shown and expanded) for child in sorted(tree.children[key], reverse=True)
391 )
392 return tuple(replace(node, children=tuple(children[node.index])) for node in nodes)
395__all__ = (
396 "HIDDEN",
397 "MAPPED",
398 "PATH_SEPARATOR",
399 "STALE",
400 "UNMAPPED",
401 "LocationCandidate",
402 "LocationCandidatePage",
403 "LocationMapping",
404 "LocationNode",
405 "LocationSegment",
406 "LocationState",
407 "decided_location_mappings",
408 "eligible_trace_locations",
409 "location_prefix_spellings",
410 "location_prefixes",
411 "present_location_tree",
412 "site_locations",
413 "source_location_key",
414 "stored_location_row",
415 "trace_location_mappings",
416)