Coverage for netbox_data_import/review_workspace.py: 99%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""Present an Import Plan, resolve explicit review commands, and persist review decisions.""" 

4 

5from __future__ import annotations 

6 

7from dataclasses import dataclass, replace 

8from functools import cached_property 

9from types import MappingProxyType 

10from typing import Any 

11 

12from .cable_target import UNRESOLVED 

13from .import_engine import ImportEngine 

14from .models import ImportProfile, TerminationResolution, locked_profile_policy 

15from .object_permissions import save_permission_scoped_object 

16from .plan import Disposition, ImportPlan, Severity, SynchronizationUnit 

17from .values import ( 

18 effective_device_name, 

19 has_below_rack_position, 

20 identity_text, 

21 normalize_for_compare, 

22 source_position, 

23 source_text, 

24 translation_maps, 

25) 

26 

27 

28def save_termination_resolution_and_replan( 

29 *, 

30 profile, 

31 source_document, 

32 actor, 

33 planning_context, 

34 task_type, 

35 field_key, 

36 selected_object_type, 

37 selected_object_id, 

38 selected_display_name, 

39): 

40 """Persist one manual termination selection, then request a fresh Import Plan.""" 

41 values = { 

42 "selected_object_type": selected_object_type, 

43 "selected_object_id": selected_object_id, 

44 "selected_display_name": selected_display_name, 

45 } 

46 with locked_profile_policy(profile.pk): 

47 locked_profile = ImportProfile.objects.get(pk=profile.pk) 

48 lookup = { 

49 "profile": locked_profile, 

50 "task_type": task_type, 

51 "field_key": field_key, 

52 } 

53 candidate = TerminationResolution(**lookup, **values) 

54 candidate.full_clean(validate_unique=False, validate_constraints=False) 

55 save_permission_scoped_object( 

56 actor, 

57 TerminationResolution, 

58 lookup, 

59 values, 

60 ) 

61 # atomic-exit-safe: decision-saved-and-replanned 

62 return ImportEngine.plan(locked_profile, source_document, actor, planning_context) 

63 

64 

65_DIAGNOSTIC_MESSAGES = { 

66 "cable.ambiguous_mapped_peer": ( 

67 "NetBox maps this port to several peer ports. Choose the peer port this trace continues through." 

68 ), 

69 "cable.attribute_drift": "The existing Cable carries attributes this import would not have written.", 

70 "cable.cableclass_unmapped": "No Cable policy maps this CableClass. Map it on the import profile.", 

71 "cable.multi_termination_conflict": ( 

72 "A Cable with several terminations on one side holds a port this trace needs. Correct that Cable in NetBox." 

73 ), 

74 "cable.pass_through_not_mapped": ( 

75 "No PortMapping joins these two ports, so the stated pass-through cannot be true. " 

76 "Correct the source, or add the PortMapping in NetBox." 

77 ), 

78 "cable.pass_through_verified": "A PortMapping proves the stated pass-through.", 

79 "cable.permission_denied": "Permission denied: you cannot make one of the Cable changes this trace needs.", 

80 "cable.planned_termination_conflict": ( 

81 "Another Source Trace plans a Cable on this termination. Resolve this trace to a different termination." 

82 ), 

83 "cable.resolved_segment_conflict": ( 

84 "Two Source Traces give one shared segment different Cable policies. Make their CableClass values agree." 

85 ), 

86 "cable.same_port_continuation": "A mapped peer port continues the path where the source repeats one port.", 

87 "cable.segment_reused": "An existing Cable already proves this segment, so the import keeps it.", 

88 "cable.segment_self_connection": "Both ends of this segment name one termination. Correct the source path.", 

89 "cable.termination_kind_mismatch": ( 

90 "The saved selection is a different kind of port than the stated PortClass. Choose the termination again." 

91 ), 

92 "cable.termination_occupied": ( 

93 "Another Cable already occupies this termination. " 

94 "Remove that Cable, or resolve this trace to a free termination." 

95 ), 

96 "cable.termination_unresolved": ( 

97 "No single port on the resolved Device matches this name. Choose the termination for it." 

98 ), 

99 "cable.unsupported_termination_kind": "A Cable can end on an Interface, a Front Port, or a Rear Port only.", 

100 "device.add_permission": "Permission denied: dcim.add_device", 

101 "device.already_bound": "Another source row is already linked to this device.", 

102 "device.ambiguous_asset_tag": "Multiple devices have this asset tag.", 

103 "device.ambiguous_field_review": "Saved field reviews identify more than one device.", 

104 "device.ambiguous_name": "Multiple devices have this name.", 

105 "device.ambiguous_serial": "Multiple devices have this serial number.", 

106 "device.ambiguous_stored_source_id": "More than one device stores this source ID.", 

107 "device.contact_invalid": "The primary contact selection is invalid.", 

108 "device.contact_permission": "Permission denied: cannot read or write this row's primary contact.", 

109 "device.contact_resolution_required": "Choose the primary contact values before importing.", 

110 "device.class_ignored": "Ignored source class", 

111 "device.class_unmapped": "No class-to-role mapping exists for this source class.", 

112 "device.cross_site_match": "A strong identity matches a device at another site.", 

113 "device.device_type_missing": ( 

114 "Add the Device Type in NetBox, or map the source make and model to an existing Device Type." 

115 ), 

116 "device.device_type_slug_collision": "A stored device type already uses the slug this model derives.", 

117 "device.duplicate_asset_tag": "The asset tag appears more than once in this import.", 

118 "device.duplicate_name": "The device name appears more than once in this import.", 

119 "device.duplicate_serial": "The serial number appears more than once in this import.", 

120 "device.duplicate_source_id": "The source ID appears more than once in this import.", 

121 "device.face_required": "A rack position needs a device face.", 

122 "device.ignored": "Ignored device", 

123 "device.inaccessible_match": "Permission denied: dcim.view_device", 

124 "device.below_rack": "Skipped because the source position is below rack unit one.", 

125 "device.change_permission": "Permission denied: dcim.change_device", 

126 "device.missing_name": "Missing device name", 

127 "device.name_placement_conflict": "The name matches a device at another placement.", 

128 "device.name_unplaced_match": "The name matches a device with no recorded placement.", 

129 "device.rack_ambiguous": "Multiple racks have this name at the import target.", 

130 "device.rack_missing": "The target rack does not exist.", 

131 "device.rack_position_claimed": "Another row claims this rack position.", 

132 "device.rack_position_occupied": "The rack position is occupied.", 

133 "device.rack_required": "A rack position needs a target rack.", 

134 "device.role_permission": "Permission denied: dcim.add_devicerole", 

135 "device.role_unconfigured": "No device role is configured for this source class.", 

136 "device.zero_u_review_conflict": "A saved review keeps a rack position on a 0U device type.", 

137 "device.unparseable_ip": "The source value is not a valid IP address.", 

138 "device.validation_failed": "The planned device does not pass NetBox validation.", 

139 "profile.dangling_reference": "The import profile names something NetBox no longer offers.", 

140 "rack.add_permission": "Permission denied: dcim.add_rack", 

141 "rack.change_permission": "Permission denied: dcim.change_rack", 

142 "rack.duplicate_name": "The rack name appears more than once in this import.", 

143 "rack.duplicate_source_id": "The source ID appears more than once in this import.", 

144 "rack.ignored": "Ignored rack", 

145 "rack.missing_name": "Missing rack name", 

146 "rack.ambiguous_name": "Multiple racks have this name at the import target.", 

147 "rack.validation_failed": "The planned rack does not pass NetBox validation.", 

148 "trace.device_unresolved": "No single Device matches this name. Correct the source, or add the Device in NetBox.", 

149 "trace.endpoint_evidence_only": ( 

150 "This trace states its two endpoints and no physical path, and no Cable joins them. " 

151 "Add the Segment Evidence rows the path needs." 

152 ), 

153} 

154 

155_IDENTITY_CONFLICTS = { 

156 "device.already_bound": "device_already_bound", 

157 "device.ambiguous_asset_tag": "ambiguous_asset_tag", 

158 "device.ambiguous_field_review": "ambiguous_field_review", 

159 "device.ambiguous_name": "ambiguous_name", 

160 "device.ambiguous_serial": "ambiguous_serial", 

161 "device.ambiguous_stored_source_id": "ambiguous_source_id", 

162 "device.cross_site_match": "cross_site_match", 

163 "device.duplicate_asset_tag": "duplicate_asset_tag", 

164 "device.duplicate_name": "duplicate_name", 

165 "device.duplicate_serial": "duplicate_serial", 

166 "device.duplicate_source_id": "duplicate_source_id", 

167 "device.inaccessible_match": "permission_denied", 

168 "device.name_placement_conflict": "name_placement_conflict", 

169 "device.name_unplaced_match": "name_placement_conflict", 

170 "device.rack_position_claimed": "rack_position_occupied", 

171 "device.rack_position_occupied": "rack_position_occupied", 

172 "rack.ambiguous_name": "ambiguous_rack", 

173 "rack.duplicate_name": "duplicate_rack", 

174 "rack.duplicate_source_id": "duplicate_source_id", 

175 "rack.validation_failed": "rack_validation_failed", 

176} 

177 

178 

179# One table: the action each identity conflict offers, and the extra_data keys that action runs on. 

180_CONFLICT_ACTIONS = MappingProxyType( 

181 { 

182 "duplicate_name": ("use_name", ("suggested_name",)), 

183 "duplicate_serial": ("ignore_serial", ("duplicate_serial",)), 

184 "name_placement_conflict": ("use_name", ("suggested_name",)), 

185 } 

186) 

187 

188 

189def _offered_actions(extra_data, source_id) -> list[str]: 

190 """Return the row actions the preview can run for one unit, in the order the table states them.""" 

191 if not source_id: 

192 return [] 

193 offered: list[str] = [] 

194 for conflict in extra_data.get("identity_conflicts") or (): 

195 entry = _CONFLICT_ACTIONS.get(conflict) 

196 if entry is None: 

197 continue 

198 action, required = entry 

199 if action not in offered and all(extra_data.get(key) for key in required): 

200 offered.append(action) 

201 return offered 

202 

203 

204def _operation(unit: SynchronizationUnit) -> str: 

205 """Return the preview action one unit displays.""" 

206 if unit.disposition == Disposition.ACTIONABLE: 

207 return unit.changes[-1].operation if unit.changes else "update" 

208 if unit.disposition == Disposition.NO_OP: 

209 return "skip" 

210 if unit.disposition == Disposition.EXCLUDED: 

211 return "ignore" 

212 return "error" 

213 

214 

215def _object_type(unit: SynchronizationUnit) -> str: 

216 """Return the target type one unit represents.""" 

217 if unit.changes: 

218 return unit.changes[-1].target_module 

219 return unit.identity.partition(":")[0] 

220 

221 

222def _blocking(unit: SynchronizationUnit) -> list: 

223 """Return the unit's error diagnostics: the first states the row, the rest are what it needs.""" 

224 return [item for item in unit.diagnostics if item.severity == Severity.ERROR] 

225 

226 

227def _states_a_trace(unit: SynchronizationUnit) -> bool: 

228 """Return whether one unit carries a Source Trace, which is what the workspace lists.""" 

229 return unit.display.get("trace") is not None 

230 

231 

232def _diagnostic_message(diagnostic) -> str: 

233 """Return the operator wording for one diagnostic.""" 

234 return str(diagnostic.display.get("message") or "") or _DIAGNOSTIC_MESSAGES.get(diagnostic.code, diagnostic.code) 

235 

236 

237def _detail(unit: SynchronizationUnit, action: str, object_type: str, name: str) -> str: 

238 """Return stable operator wording for one unit.""" 

239 if unit.diagnostics: 

240 blocking = _blocking(unit) 

241 return _diagnostic_message(blocking[0] if blocking else unit.diagnostics[0]) 

242 if detail := unit.display.get("detail"): 

243 return str(detail) 

244 verb = { 

245 "create": "Would create", 

246 "update": "Would update", 

247 "skip": "No changes for", 

248 "ignore": "Ignored", 

249 }.get(action, action.title()) 

250 return f"{verb} {object_type} '{name}'" 

251 

252 

253@dataclass(frozen=True) 

254class WorkspaceUnit: 

255 """View-supplied presentation data for one Synchronization Unit.""" 

256 

257 identity: str 

258 disposition: str 

259 row_number: int | None 

260 source_id: str 

261 name: str 

262 action: str 

263 object_type: str 

264 detail: str 

265 rack_name: str 

266 netbox_url: str 

267 extra_data: dict[str, Any] 

268 source_row: dict[str, Any] 

269 

270 @classmethod 

271 def from_unit(cls, unit: SynchronizationUnit) -> WorkspaceUnit: 

272 """Build presentation data without changing the accepted plan.""" 

273 unit_data = unit.to_dict() 

274 display = unit_data["display"] 

275 source_row = dict(display.get("source_row") or {}) 

276 object_type = _object_type(unit) 

277 action = _operation(unit) 

278 name = str(display.get("name") or display.get("device_name") or display.get("rack_name") or "") 

279 rack_name = str(display.get("rack_name") or source_row.get("rack_name") or "") 

280 source_id = str(display.get("source_id") or source_row.get("source_id") or "") 

281 row_number = display.get("row_number", source_row.get("_row_number")) 

282 extra_data = dict(display.get("extra_data") or {}) 

283 for key, value in source_row.items(): 

284 if key.startswith("_"): 

285 continue 

286 extra_data.setdefault(f"source_{key}", value) 

287 if unit.changes: 

288 change = unit.changes[-1] 

289 change_data = change.to_dict() 

290 extra_data.update({key: value for key, value in change_data["payload"].items() if key not in extra_data}) 

291 object_id = change.preconditions.get(f"{object_type}_id") 

292 if object_id is not None: 

293 extra_data.setdefault(f"netbox_{object_type}_id", object_id) 

294 if unit.diagnostics: 

295 diagnostic = unit.diagnostics[0] 

296 diagnostic_display = diagnostic.to_dict()["display"] 

297 structural = {"device_name", "extra_data", "message", "name", "object_type", "source_row"} 

298 extra_data.update({key: value for key, value in diagnostic_display.items() if key not in structural}) 

299 if conflict := _IDENTITY_CONFLICTS.get(diagnostic.code): 

300 extra_data.setdefault("identity_conflict", conflict) 

301 rows = diagnostic_display.get("rows") 

302 if rows: 

303 row_key = { 

304 "device.duplicate_asset_tag": "duplicate_asset_tag_rows", 

305 "device.duplicate_serial": "duplicate_serial_rows", 

306 "device.duplicate_source_id": "duplicate_source_id_rows", 

307 "rack.duplicate_source_id": "duplicate_source_id_rows", 

308 }.get(diagnostic.code) 

309 if row_key: 

310 extra_data[row_key] = [number for number in rows if number != row_number] 

311 # The first error states the row; the rest are what it still needs. A warning is neither. 

312 blocking = _blocking(unit) 

313 extra_data["other_issues"] = [ 

314 {"code": item.code, "message": _diagnostic_message(item)} for item in blocking[1:] 

315 ] 

316 # A row action reads these facts, so a second problem is settled like a first one. 

317 extra_data["identity_conflicts"] = [ 

318 conflict for item in blocking if (conflict := _IDENTITY_CONFLICTS.get(item.code)) 

319 ] 

320 for item in blocking[1:]: 

321 secondary = item.to_dict()["display"] 

322 for key, value in secondary.items(): 

323 if key not in structural: 

324 extra_data.setdefault(key, value) 

325 for key, value in (secondary.get("extra_data") or {}).items(): 

326 extra_data.setdefault(key, value) 

327 extra_data["offered_actions"] = _offered_actions(extra_data, source_id) 

328 return cls( 

329 identity=unit.identity, 

330 disposition=unit.disposition, 

331 row_number=row_number, 

332 source_id=source_id, 

333 name=name, 

334 action=action, 

335 object_type=object_type, 

336 detail=_detail(unit, action, object_type, name), 

337 rack_name=rack_name, 

338 netbox_url=str(display.get("netbox_url") or ""), 

339 extra_data=extra_data, 

340 source_row=source_row, 

341 ) 

342 

343 

344@dataclass(frozen=True) 

345class AutoMatchSummary: 

346 """Counts returned by the device auto-match review command.""" 

347 

348 matched: int = 0 

349 probable: int = 0 

350 ambiguous: int = 0 

351 placement_conflicts: int = 0 

352 already: int = 0 

353 skipped: int = 0 

354 

355 def message(self) -> str: 

356 """Return the existing operator-facing summary.""" 

357 parts = [] 

358 if self.matched: 

359 parts.append(f"{self.matched} auto-matched (serial/asset_tag/name)") 

360 if self.probable: 

361 parts.append(f"{self.probable} probable name match(es): use Link button to confirm") 

362 if self.ambiguous: 

363 parts.append(f"{self.ambiguous} ambiguous (multiple devices)") 

364 if self.placement_conflicts: 

365 parts.append(f"{self.placement_conflicts} placement conflict(s)") 

366 if self.already: 

367 parts.append(f"{self.already} already matched") 

368 if self.skipped: 

369 parts.append(f"{self.skipped} skipped (permission denied or concurrent change)") 

370 return f"Auto-match: {', '.join(parts) or 'nothing found'}." 

371 

372 

373def _resolve_strong_identity(devices, serial: str, asset_tag: str): 

374 """Resolve serial and asset tag to one device, or report ambiguity.""" 

375 serial_matches = list(devices.filter(serial=serial)[:2]) if serial else [] 

376 asset_matches = list(devices.filter(asset_tag__iexact=asset_tag)[:2]) if asset_tag else [] 

377 if len(serial_matches) > 1 or len(asset_matches) > 1: 

378 return None, None, True 

379 serial_device = serial_matches[0] if serial_matches else None 

380 asset_device = asset_matches[0] if asset_matches else None 

381 if serial_device is not None and asset_device is not None and serial_device.pk != asset_device.pk: 

382 return None, None, True 

383 if serial_device is not None: 

384 return serial_device, "serial", False 

385 if asset_device is not None: 

386 return asset_device, "asset tag", False 

387 return None, None, False 

388 

389 

390def _match_existing_device(device_model, visible_devices, name, serial, asset_tag, site, tenant_id): 

391 """Return one exact device match and its method, or report ambiguity.""" 

392 device, method, ambiguous = _resolve_strong_identity(device_model.objects, serial, asset_tag) 

393 if ambiguous: 

394 return None, None, True 

395 if device is None and name: 

396 tenant_filter = {"tenant_id": tenant_id} if tenant_id is not None else {"tenant__isnull": True} 

397 matches = list(device_model.objects.filter(site=site, name__iexact=name, **tenant_filter)[:2]) 

398 if len(matches) > 1: 

399 return None, None, True 

400 if matches: 

401 device, method = matches[0], "name" 

402 if device is not None and (device.site_id != site.pk or not visible_devices.filter(pk=device.pk).exists()): 

403 return None, None, True 

404 return device, method, False 

405 

406 

407def _device_placement_differs(device, source_location_id, rack_name, position, face) -> bool: 

408 """Return whether source placement differs from one matched NetBox device.""" 

409 device_rack_name = device.rack.name if device.rack_id else "" 

410 device_rack_location_id = device.rack.location_id if device.rack_id else None 

411 return ( 

412 device.location_id != source_location_id 

413 or (device.rack_id is not None and device_rack_location_id != source_location_id) 

414 or identity_text(device_rack_name) != identity_text(rack_name) 

415 or normalize_for_compare(device.position) != normalize_for_compare(position) 

416 or (face is not None and (device.face or None) != face) 

417 ) 

418 

419 

420_SUMMARY_KEYS = { 

421 Disposition.ACTIONABLE: "actionable", 

422 Disposition.BLOCKED: "blocked", 

423 Disposition.INVALID: "invalid", 

424 Disposition.NO_OP: "no_change", 

425} 

426 

427 

428_SYNC_URL_NAME = "plugins:netbox_data_import:trace_sync" 

429 

430 

431@dataclass(frozen=True) 

432class TraceAction: 

433 """One review command, always visible, carrying its reason when it cannot run. 

434 

435 `url_name` has no default: the page posts every action to it, so a command that named none 

436 would inherit whichever endpoint the template happened to hardcode. 

437 """ 

438 

439 key: str 

440 label: str 

441 enabled: bool 

442 url_name: str 

443 reason: str = "" 

444 

445 

446@dataclass(frozen=True) 

447class TraceWorkspaceUnit: 

448 """One Source Trace as the review workspace shows it.""" 

449 

450 identity: str 

451 trace_identity: str 

452 disposition: str 

453 name: str 

454 row_number: int | None 

455 sheet: str 

456 endpoints: dict[str, str] 

457 segments: list[dict[str, Any]] 

458 logical_cable: dict[str, Any] | None 

459 deletes_logical_cable: bool 

460 topology_known: bool 

461 terminations: list[dict[str, Any]] 

462 findings: list[dict[str, str]] 

463 actions: tuple[TraceAction, ...] 

464 

465 @classmethod 

466 def from_unit(cls, unit: SynchronizationUnit) -> TraceWorkspaceUnit: 

467 """Build the workspace entry one trace unit states, without recomputing its plan.""" 

468 display = unit.to_dict()["display"] 

469 workspace = dict(display.get("trace") or {}) 

470 findings = [ 

471 {"code": item.code, "message": _diagnostic_message(item), "severity": item.severity} 

472 for item in unit.diagnostics 

473 ] 

474 return cls( 

475 identity=unit.identity, 

476 trace_identity=str(workspace.get("identity") or ""), 

477 disposition=unit.disposition, 

478 name=str(display.get("name") or ""), 

479 row_number=display.get("row_number"), 

480 sheet=str(display.get("sheet") or ""), 

481 endpoints=dict(workspace.get("endpoints") or {}), 

482 segments=[dict(segment) for segment in workspace.get("segments") or ()], 

483 logical_cable=workspace.get("logical_cable"), 

484 deletes_logical_cable=bool(workspace.get("deletes_logical_cable")), 

485 topology_known=bool(workspace.get("topology_known")), 

486 terminations=[dict(item) for item in workspace.get("terminations") or ()], 

487 findings=findings, 

488 actions=cls._actions(unit, findings, str(display.get("detail") or "")), 

489 ) 

490 

491 @staticmethod 

492 def _actions(unit: SynchronizationUnit, findings: list[dict[str, str]], detail: str) -> tuple[TraceAction, ...]: 

493 """Return every review command, each stating why it cannot run when it cannot.""" 

494 blocking = next((item["message"] for item in findings if item["severity"] == Severity.ERROR), "") 

495 if unit.disposition == Disposition.ACTIONABLE: 

496 sync = TraceAction(key="sync", label="Sync with dependencies", enabled=True, url_name=_SYNC_URL_NAME) 

497 else: 

498 reason = blocking or detail or f"This trace is {unit.disposition}." 

499 sync = TraceAction( 

500 key="sync", label="Sync with dependencies", enabled=False, url_name=_SYNC_URL_NAME, reason=reason 

501 ) 

502 return (sync,) 

503 

504 

505class ReviewWorkspace: 

506 """Read-only presentation of the accepted Import Plan.""" 

507 

508 def __init__(self, plan: ImportPlan): 

509 self.plan = plan 

510 self.units = tuple(WorkspaceUnit.from_unit(unit) for unit in plan.units) 

511 

512 @classmethod 

513 def from_dict(cls, data: dict) -> ReviewWorkspace: 

514 """Restore a workspace from the session's serialized Import Plan.""" 

515 return cls(ImportPlan.from_dict(data)) 

516 

517 @property 

518 def counts(self) -> MappingProxyType: 

519 """Return preview counts in the existing template vocabulary.""" 

520 counts: dict[str, int] = {} 

521 for unit in self.units: 

522 if unit.action == "error": 

523 key = "errors" 

524 elif unit.action == "skip": 

525 key = "skipped" 

526 elif unit.action == "ignore": 

527 key = "ignored" 

528 elif unit.action in {"create", "update"}: 

529 key = f"{unit.object_type}s_{unit.action}d" 

530 else: 

531 continue 

532 counts[key] = counts.get(key, 0) + 1 

533 return MappingProxyType(counts) 

534 

535 @property 

536 def has_errors(self) -> bool: 

537 """Return whether any unit cannot execute.""" 

538 return any(unit.action == "error" for unit in self.units) 

539 

540 @property 

541 def has_traces(self) -> bool: 

542 """Return whether the plan holds a Source Trace, without building one workspace entry.""" 

543 return any(_states_a_trace(unit) for unit in self.plan.units) 

544 

545 @cached_property 

546 def traces(self) -> tuple[TraceWorkspaceUnit, ...]: 

547 """Return one workspace entry per Source Trace, in plan order. 

548 

549 Cached because one page reads it twice, and each build reserializes every change. 

550 """ 

551 return tuple(TraceWorkspaceUnit.from_unit(unit) for unit in self.plan.units if _states_a_trace(unit)) 

552 

553 def sync_selection(self, identity: str) -> tuple[str, ...]: 

554 """Return the unit and every unit owning a change it depends on, transitively. 

555 

556 `merge_changes` refuses a selection whose dependency is absent, so a review command that 

557 synchronizes one trace has to carry the units its changes wait on. 

558 """ 

559 units = {unit.identity: unit for unit in self.plan.units} 

560 selected = units.get(identity) 

561 if selected is None or selected.disposition != Disposition.ACTIONABLE: 

562 return () 

563 owner_of = {change.identity: unit.identity for unit in self.plan.units for change in unit.changes} 

564 chosen: list[str] = [] 

565 queue = [identity] 

566 while queue: 

567 current = queue.pop() 

568 if current in chosen: 

569 continue 

570 chosen.append(current) 

571 for change in units[current].changes: 

572 for dependency in change.dependencies: 

573 owner = owner_of.get(dependency) 

574 if owner is not None and owner not in chosen: 

575 queue.append(owner) 

576 return tuple(chosen) 

577 

578 @property 

579 def trace_summary(self) -> dict[str, int]: 

580 """Return the summary strip: what the reviewer still has to work through.""" 

581 traces = self.traces 

582 summary = { 

583 "traces": len(traces), 

584 "unresolved_terminations": 0, 

585 "resolved_terminations": 0, 

586 } 

587 # A template cannot resolve a key with a hyphen, so the strip names each disposition itself. 

588 for disposition, key in _SUMMARY_KEYS.items(): 

589 summary[key] = sum(1 for trace in traces if trace.disposition == disposition) 

590 for trace in traces: 

591 for termination in trace.terminations: 

592 key = ( 

593 "unresolved_terminations" 

594 if termination.get("state", UNRESOLVED) == UNRESOLVED 

595 else "resolved_terminations" 

596 ) 

597 summary[key] += 1 

598 return summary 

599 

600 @property 

601 def rack_groups(self) -> dict: 

602 """Group Rack and Device units for the existing rack-card view.""" 

603 groups: dict[str, dict[str, Any]] = {} 

604 for unit in self.units: 

605 if unit.object_type == "rack": 

606 if not unit.name: 

607 continue 

608 groups.setdefault(unit.name, {"rack_row": None, "devices": []})["rack_row"] = unit 

609 elif unit.object_type == "device": 

610 groups.setdefault(unit.rack_name or "(No rack)", {"rack_row": None, "devices": []})["devices"].append( 

611 unit 

612 ) 

613 

614 def placement_sort_key(unit): 

615 position = source_position(unit.extra_data.get("u_position")) 

616 return position is None, position or 0 

617 

618 for group in groups.values(): 

619 group["devices"].sort(key=placement_sort_key) 

620 return groups 

621 

622 @property 

623 def source_rows(self) -> list[dict[str, Any]]: 

624 """Return the source rows carried as display data, once per row number.""" 

625 rows: dict[int, dict[str, Any]] = {} 

626 for unit in self.units: 

627 if unit.row_number is not None and unit.source_row: 

628 rows.setdefault(unit.row_number, dict(unit.source_row)) 

629 return [rows[number] for number in sorted(rows)] 

630 

631 @property 

632 def unused_columns(self) -> list[dict[str, Any]]: 

633 """Return unmapped source columns carried by plan diagnostics.""" 

634 columns = [ 

635 dict(diagnostic.display) 

636 for diagnostic in self.plan.diagnostics 

637 if diagnostic.code.endswith(".unused_column") 

638 ] 

639 columns.sort(key=lambda column: -int(column.get("count") or 0)) 

640 return columns 

641 

642 def auto_match_devices(self, profile, actor, target) -> AutoMatchSummary: # noqa: C901 

643 """Save safe exact device matches for every eligible plan source row.""" 

644 from django.core.exceptions import ValidationError 

645 from django.db import IntegrityError 

646 from dcim.models import Device 

647 

648 from .models import DeviceExistingMatch 

649 from .object_permissions import ObjectPermissionDenied, save_permission_scoped_object 

650 

651 site = target["site"] 

652 tenant_id = target["tenant"].pk if target["tenant"] else None 

653 visible_devices = Device.objects.restrict(actor, "view") 

654 ignored_source_ids = set(profile.ignored_devices.values_list("source_id", flat=True)) 

655 class_mappings = {mapping.source_class: mapping for mapping in profile.class_role_mappings.all()} 

656 eligible_rows = [] 

657 for row in self.source_rows: 

658 source_id = source_text(row.get("source_id")) 

659 mapping = class_mappings.get(source_text(row.get("device_class"))) 

660 if ( 

661 mapping is None 

662 or mapping.creates_rack 

663 or mapping.ignore 

664 or source_id in ignored_source_ids 

665 or has_below_rack_position(row) 

666 ): 

667 continue 

668 eligible_rows.append(row) 

669 

670 bound_device_by_source = dict(profile.device_matches.values_list("source_id", "netbox_device_id")) 

671 bound_source_by_device = {device_id: source for source, device_id in bound_device_by_source.items()} 

672 source_counts: dict[str, int] = {} 

673 name_counts: dict[str, int] = {} 

674 serial_counts: dict[str, int] = {} 

675 asset_tag_counts: dict[str, int] = {} 

676 for row in eligible_rows: 

677 values = ( 

678 (source_text(row.get("source_id")), source_counts), 

679 (identity_text(effective_device_name(row)), name_counts), 

680 (source_text(row.get("serial")), serial_counts), 

681 (identity_text(source_text(row.get("asset_tag"))[:50]), asset_tag_counts), 

682 ) 

683 for value, counts in values: 

684 if value: 

685 counts[value] = counts.get(value, 0) + 1 

686 

687 counts = { 

688 "matched": 0, 

689 "probable": 0, 

690 "ambiguous": 0, 

691 "placement_conflicts": 0, 

692 "already": 0, 

693 "skipped": 0, 

694 } 

695 side_map, _, _ = translation_maps() 

696 for row in eligible_rows: 

697 source_id = source_text(row.get("source_id")) 

698 name = effective_device_name(row) 

699 serial = source_text(row.get("serial")) 

700 asset_tag = source_text(row.get("asset_tag"))[:50] 

701 if not source_id: 

702 continue 

703 if source_counts.get(source_id, 0) > 1: 

704 counts["ambiguous"] += 1 

705 continue 

706 if source_id in bound_device_by_source: 

707 counts["already"] += 1 

708 continue 

709 

710 device, method, ambiguous = _match_existing_device( 

711 Device, 

712 visible_devices, 

713 name if name_counts.get(identity_text(name), 0) == 1 else "", 

714 serial if serial_counts.get(serial, 0) == 1 else "", 

715 asset_tag if asset_tag_counts.get(identity_text(asset_tag), 0) == 1 else "", 

716 site, 

717 tenant_id, 

718 ) 

719 if ambiguous: 

720 counts["ambiguous"] += 1 

721 continue 

722 if device is not None and method == "name": 

723 face = side_map.get(source_text(row.get("face")).lower()) 

724 if _device_placement_differs( 

725 device, 

726 target["location"].pk if target["location"] else None, 

727 source_text(row.get("rack_name")), 

728 source_position(row.get("u_position")), 

729 face, 

730 ): 

731 counts["placement_conflicts"] += 1 

732 continue 

733 if device is not None: