Coverage for netbox_data_import/preview_coordinator.py: 95.84%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""The Preview Coordinator: the one owner of each browser session's active preview (ADR 0004). 

4 

5`setup_claim` gives the setup page the claim of the session's row, creating the empty row on first 

6use. `read_preview` returns an immutable snapshot for a page or a read. `apply_preview_command` runs 

7one command under the coordinator lock, then the profile lock, then whatever the command locks, and 

8it is the only code that stores a plan or moves the revision. No other module touches preview state, 

9and this one reads nothing from the session beyond its key. 

10""" 

11 

12from __future__ import annotations 

13 

14import hashlib 

15import logging 

16import re 

17import secrets 

18import uuid 

19from collections.abc import Callable, Iterable, Mapping 

20from dataclasses import dataclass, field 

21from functools import cached_property 

22from types import MappingProxyType 

23from typing import Any, ClassVar 

24 

25from core.signals import clear_events 

26from django.conf import settings 

27from django.core.exceptions import ObjectDoesNotExist, PermissionDenied, ValidationError 

28from django.db import IntegrityError, transaction 

29from django.utils import timezone 

30 

31from . import branching 

32from .adapter_config import PREVIEW_PORT_CLASS_MAP 

33from .import_engine import ImportEngine, StaleSourceDocument 

34from .models import ( 

35 ImportProfile, 

36 PreviewCoordinator, 

37 PreviewState, 

38 SourceDocument, 

39 locked_profile_policy, 

40 validate_adapter_target_module, 

41 validate_registered_adapter, 

42) 

43from .plan import ImportPlan, PlanError, canonical_json 

44from .preview_limits import MAX_PLAN_BYTES, preview_plan_byte_limit 

45from .object_permissions import clear_user_permission_caches 

46from .review_workspace import ReviewWorkspace, refuse_moved_policy 

47from .public_refusal import PublicRefusal 

48 

49logger = logging.getLogger(__name__) 

50 

51 

52def _refresh_actor(user) -> None: 

53 """Read the actor's current status and grants after any wait for the profile lock.""" 

54 try: 

55 user.refresh_from_db(fields=["is_active", "is_superuser"]) 

56 except ObjectDoesNotExist as exc: 

57 raise PermissionDenied from exc 

58 if not user.is_active: 

59 raise PermissionDenied 

60 clear_user_permission_caches(user) 

61 

62 

63CLAIM_FIELDS = ("preview_token", "preview_revision", "preview_document", "preview_profile") 

64_TOKEN = re.compile(r"[A-Za-z0-9_-]{43}") 

65PLANNING_KEYS = ("site_id", "location_id", "tenant_id") 

66 

67NO_PREVIEW = "No import preview is in progress. Start a new import." 

68CLAIM_INVALID = "This request does not name a preview. Reload the page and try again." 

69STALE_PREVIEW = "This preview changed in another tab or request. Reload it and try again." 

70NEWER_PREVIEW = "A newer import was started in another tab. Reload the page to continue there." 

71EXPIRED_PREVIEW = "This preview expired. Start a new import." 

72SUBMITTED_PREVIEW = "The import already started, so this preview can no longer take a decision." 

73SYNC_QUEUED = ( 

74 "A trace synchronization waits in the queue and has not started. " 

75 "Wait for it to finish, or cancel it, before you change this workspace." 

76) 

77SYNC_RUNNING = "A trace synchronization is running. Wait for it to finish before you change this workspace." 

78SYNC_FINISHED = "The trace synchronization finished. Re-read the preview before the next decision." 

79SYNC_STARTED = "The trace synchronization already started, so you cannot cancel it." 

80SYNC_CANCELLED = "The operator cancelled this trace synchronization before it started." 

81SYNC_CANCEL_DONE = "The trace synchronization was cancelled. The preview was re-read from NetBox." 

82SYNC_NOT_THIS_PREVIEW = "This preview did not queue that trace synchronization." 

83SYNC_OTHER_ACTIVE = "Another trace synchronization of this source waits or runs. Wait for it to finish." 

84UNREADABLE_PREVIEW = "This preview cannot be read. Re-read the preview." 

85PREVIEW_TOO_LARGE = "This preview is too large to store. Split the source file and import each part." 

86SOURCE_GONE = "The stored source is no longer available. Upload it again." 

87 

88 

89class StalePreview(PublicRefusal): 

90 """The command named a preview that is not the active one, or one that cannot take it now.""" 

91 

92 

93class PreviewCommandRefused(PublicRefusal): 

94 """A command refused for a reason this plugin wrote, so the response may state it.""" 

95 

96 def __init__(self, message: str, status: int = 400): 

97 super().__init__(message) 

98 self.status = status 

99 

100 

101@dataclass(frozen=True) 

102class PreviewClaim: 

103 """The preview a page displays: its generation token, revision, Source Document and profile.""" 

104 

105 token: str 

106 revision: int 

107 document_id: int | None 

108 profile_id: int | None 

109 

110 @classmethod 

111 def posted(cls, data) -> PreviewClaim: 

112 """Return the claim a request carries, refusing a missing or malformed one.""" 

113 token = data.get("preview_token", "") 

114 if not isinstance(token, str) or not _TOKEN.fullmatch(token): 

115 raise StalePreview(CLAIM_INVALID) 

116 revision = _claim_number(data.get("preview_revision")) 

117 if revision is None: 

118 raise StalePreview(CLAIM_INVALID) 

119 return cls( 

120 token=token, 

121 revision=revision, 

122 document_id=_optional_claim_number(data.get("preview_document")), 

123 profile_id=_optional_claim_number(data.get("preview_profile")), 

124 ) 

125 

126 def fields(self) -> dict[str, str]: 

127 """Return the claim as the form fields a page renders and posts back.""" 

128 return { 

129 "preview_token": self.token, 

130 "preview_revision": str(self.revision), 

131 "preview_document": "" if self.document_id is None else str(self.document_id), 

132 "preview_profile": "" if self.profile_id is None else str(self.profile_id), 

133 } 

134 

135 

136def _claim_number(value) -> int | None: 

137 """Return one posted positive integer, or None for anything else.""" 

138 if not isinstance(value, str) or not value.isascii() or not value.isdigit() or len(value) > 19: 

139 return None 

140 number = int(value) 

141 return number if number >= 1 else None 

142 

143 

144def _optional_claim_number(value) -> int | None: 

145 """Return one posted id, None for an empty value, and refuse a malformed one.""" 

146 if value in (None, ""): 

147 return None 

148 number = _claim_number(value) 

149 if number is None: 

150 raise StalePreview(CLAIM_INVALID) 

151 return number 

152 

153 

154def _claim_of(row: PreviewCoordinator) -> PreviewClaim: 

155 return PreviewClaim( 

156 token=row.preview_token, 

157 revision=row.revision, 

158 document_id=row.source_document_id, 

159 profile_id=row.profile_id, 

160 ) 

161 

162 

163def _same_generation(row: PreviewCoordinator, expected: PreviewClaim) -> bool: 

164 """Return whether a claim names this row's generation and identity, at any revision up to now.""" 

165 return ( 

166 secrets.compare_digest(row.preview_token, expected.token) 

167 and expected.revision <= row.revision 

168 and expected.document_id == row.source_document_id 

169 and expected.profile_id == row.profile_id 

170 ) 

171 

172 

173def _exact(row: PreviewCoordinator, expected: PreviewClaim) -> bool: 

174 return _same_generation(row, expected) and expected.revision == row.revision 

175 

176 

177def _new_token() -> str: 

178 return secrets.token_urlsafe(32) 

179 

180 

181def _session_binding(request, *, create: bool = False) -> str | None: 

182 """Return the digest of the server-side session key, which selects the session's row.""" 

183 if create and not request.session.session_key: 

184 request.session.save() 

185 key = request.session.session_key 

186 if not key: 

187 return None 

188 return hashlib.sha256(key.encode()).hexdigest() 

189 

190 

191def _session_expiry(request): 

192 return request.session.get_expiry_date() 

193 

194 

195def _payload_expiry(request): 

196 """Return when a new preview expires: with its session, and no later than its Source Document.""" 

197 return min(_session_expiry(request), timezone.now() + PreviewCoordinator.PAYLOAD_LIFETIME) 

198 

199 

200def _owned(row: PreviewCoordinator | None, user) -> PreviewCoordinator | None: 

201 if row is not None: 

202 _check_owner(row, user) 

203 return row 

204 

205 

206def _check_owner(row: PreviewCoordinator, user) -> PreviewCoordinator: 

207 if row.owner_id != user.pk: 

208 # Django rotates the key at login, so a binding owned by another user is never this session's. 

209 raise PermissionDenied("This preview belongs to another user.") 

210 return row 

211 

212 

213def _payload_expired(row: PreviewCoordinator) -> bool: 

214 return row.state in PreviewState.ACTIVE and row.expires_at <= timezone.now() 

215 

216 

217def _clear_payload(row: PreviewCoordinator, state: str) -> None: 

218 row.state = state 

219 row.profile_id = None 

220 row.source_document_id = None 

221 row.context = {} 

222 row.plan = None 

223 row.job_id = None 

224 

225 

226def _expire(row: PreviewCoordinator) -> None: 

227 """Clear an expired payload and start a new empty generation, so its claims stay refused.""" 

228 _clear_payload(row, PreviewState.EXPIRED) 

229 row.preview_token = _new_token() 

230 row.revision += 1 

231 row.save() 

232 

233 

234def planning_context_of(context: Mapping) -> dict: 

235 """Return the planning context a preview with this context plans against.""" 

236 projected = {key: context.get(key) for key in PLANNING_KEYS} 

237 # A preview without its own PortClass map keeps the planning context every earlier plan states. 

238 if port_class_map := context.get(PREVIEW_PORT_CLASS_MAP): 

239 projected[PREVIEW_PORT_CLASS_MAP] = dict(port_class_map) 

240 return projected 

241 

242 

243def with_preview_port_class(context: Mapping, label: str, port_class: str | None) -> dict: 

244 """Return the preview context with one PortClass label mapped for this preview alone, or cleared.""" 

245 port_class_map = dict(context.get(PREVIEW_PORT_CLASS_MAP) or {}) 

246 if port_class is None: 

247 port_class_map.pop(label, None) 

248 else: 

249 port_class_map[label] = port_class 

250 updated = {key: value for key, value in context.items() if key != PREVIEW_PORT_CLASS_MAP} 

251 if port_class_map: 

252 updated[PREVIEW_PORT_CLASS_MAP] = dict(sorted(port_class_map.items())) 

253 return updated 

254 

255 

256def sync_hold_reason(statuses: Iterable[str]) -> str: 

257 """Return why trace sync Jobs in these statuses hold their source, or "" when none of them is active.""" 

258 from core.choices import JobStatusChoices 

259 

260 active = set(statuses) & set(JobStatusChoices.ENQUEUED_STATE_CHOICES) 

261 if not active: 

262 return "" 

263 return SYNC_RUNNING if JobStatusChoices.STATUS_RUNNING in active else SYNC_QUEUED 

264 

265 

266def _sync_hold(jobs) -> str: 

267 return sync_hold_reason(jobs.values_list("status", flat=True)) 

268 

269 

270def setup_claim(request) -> PreviewClaim: 

271 """Return the claim the setup page posts, creating the session's empty row on its first visit.""" 

272 branching.refuse_branch() 

273 binding = _session_binding(request, create=True) 

274 row = _owned(PreviewCoordinator.objects.filter(session_binding=binding).defer("plan").first(), request.user) 

275 if row is None: 

276 try: 

277 with transaction.atomic(): 

278 row = PreviewCoordinator.objects.create( 

279 session_binding=binding, 

280 owner=request.user, 

281 preview_token=_new_token(), 

282 expires_at=_session_expiry(request), 

283 ) 

284 except IntegrityError: 

285 # A concurrent first visit created it; the unique binding leaves exactly one row. 

286 row = _check_owner(PreviewCoordinator.objects.defer("plan").get(session_binding=binding), request.user) 

287 return _claim_of(row) 

288 

289 

290@dataclass(frozen=True) 

291class PreviewSnapshot: 

292 """One immutable read of the session's preview, with its profile and document checked live.""" 

293 

294 claim: PreviewClaim 

295 state: str 

296 expired: bool 

297 context: Mapping[str, Any] 

298 job_id: int | None 

299 profile: ImportProfile | None 

300 document: SourceDocument | None 

301 _plan_data: dict | None = field(default=None, repr=False) 

302 

303 @property 

304 def active(self) -> bool: 

305 """Return whether a live, unexpired preview with its profile and document stands behind this read.""" 

306 return ( 

307 self.state in PreviewState.ACTIVE 

308 and not self.expired 

309 and self.profile is not None 

310 and self.document is not None 

311 ) 

312 

313 @property 

314 def planning_context(self) -> dict: 

315 """Return the site, location and tenant the preview plans against.""" 

316 return planning_context_of(self.context) 

317 

318 @cached_property 

319 def plan(self) -> ImportPlan: 

320 """Return the stored plan, which is immutable, raising PlanError when this release cannot read it.""" 

321 if self._plan_data is None: 

322 raise PlanError("The preview stores no Import Plan.") 

323 return ImportPlan.from_dict(self._plan_data) 

324 

325 def workspace(self, viewer) -> ReviewWorkspace: 

326 """Return the stored plan as one viewer's Review Workspace.""" 

327 return ReviewWorkspace(self.plan, viewer) 

328 

329 

330def _frozen(value): 

331 if isinstance(value, Mapping): 

332 return MappingProxyType({key: _frozen(item) for key, item in value.items()}) 

333 if isinstance(value, (list, tuple)): 

334 return tuple(_frozen(item) for item in value) 

335 return value 

336 

337 

338def read_preview( 

339 request, *, expected: PreviewClaim | None = None, profile_action: str = "change", include_plan: bool = True 

340) -> PreviewSnapshot: 

341 """Return the session's preview as it stands now; with `expected`, refuse any other one.""" 

342 branching.refuse_branch() 

343 binding = _session_binding(request) 

344 row = None 

345 if binding is not None: 

346 rows = PreviewCoordinator.objects.filter(session_binding=binding) 

347 row = _owned((rows if include_plan else rows.defer("plan")).first(), request.user) 

348 if row is None: 

349 if expected is not None: 

350 raise StalePreview(NO_PREVIEW) 

351 return PreviewSnapshot( 

352 claim=PreviewClaim("", 1, None, None), 

353 state=PreviewState.EMPTY, 

354 expired=False, 

355 context=MappingProxyType({}), 

356 job_id=None, 

357 profile=None, 

358 document=None, 

359 ) 

360 if expected is not None: 

361 if not _exact(row, expected): 

362 raise StalePreview(STALE_PREVIEW) 

363 if _payload_expired(row): 

364 raise StalePreview(EXPIRED_PREVIEW) 

365 profile = document = None 

366 if row.state in PreviewState.ACTIVE: 

367 profile = ImportProfile.objects.restrict(request.user, profile_action).filter(pk=row.profile_id).first() 

368 if profile is not None: 

369 document = ( 

370 SourceDocument.objects.filter(pk=row.source_document_id, profile=profile).defer("content").first() 

371 ) 

372 return PreviewSnapshot( 

373 claim=_claim_of(row), 

374 state=row.state, 

375 expired=_payload_expired(row), 

376 context=_frozen(row.context or {}), 

377 job_id=row.job_id, 

378 profile=profile, 

379 document=document, 

380 _plan_data=row.plan if include_plan else None, 

381 ) 

382 

383 

384class LockedPreview: 

385 """The preview one command runs against, read under the coordinator and the profile locks.""" 

386 

387 def __init__(self, *, actor, profile, document, context, plan, job_id): 

388 self.actor = actor 

389 self.profile = profile 

390 self.document = document 

391 self.context = MappingProxyType(dict(context)) 

392 self.plan = plan 

393 self.job_id = job_id 

394 

395 @property 

396 def planning_context(self) -> dict: 

397 """Return the site, location and tenant the preview plans against.""" 

398 return planning_context_of(self.context) 

399 

400 @property 

401 def reviewed_fingerprint(self) -> str: 

402 """Return the profile fingerprint the stored plan was made under.""" 

403 return self.plan.profile_fingerprint 

404 

405 @cached_property 

406 def workspace(self) -> ReviewWorkspace: 

407 """Return the stored plan as the acting operator's Review Workspace.""" 

408 return ReviewWorkspace(self.plan, self.actor) 

409 

410 

411@dataclass(frozen=True) 

412class CommandOutcome: 

413 """What one command did. A command never stores a plan: the coordinator stores `plan` for it.""" 

414 

415 message: str = "" 

416 payload: Mapping[str, Any] = field(default_factory=dict) 

417 # A replacement the command already produced, inside its own savepoint. 

418 plan: ImportPlan | None = None 

419 # The preview context the command leaves behind; a replacement plan must be planned against it. 

420 context: Mapping[str, Any] | None = None 

421 # Commit what the command wrote durably, keep the preview unchanged, then raise this. 

422 refusal: Exception | None = None 

423 # The state and Job a queueing command leaves behind, and its after-commit compensation. 

424 state: str | None = None 

425 job_id: int | None = None 

426 compensate: Callable[[], None] | None = None 

427 

428 

429@dataclass(frozen=True) 

430class PreviewResult: 

431 """A committed command: the claim the page now holds, the outcome, and the preview's profile.""" 

432 

433 claim: PreviewClaim 

434 outcome: CommandOutcome 

435 profile_id: int | None 

436 state: str 

437 

438 

439class PreviewCommand: 

440 """One coordinated change of the active preview (ADR 0004). 

441 

442 A subclass implements `apply`, which runs under the coordinator and profile locks against the 

443 stored plan. The coordinator then replans when `replans` is set, and advances the revision when 

444 `advances` is set. `apply` raises to refuse; everything it wrote rolls back. 

445 """ 

446 

447 allowed_states: ClassVar[frozenset[str]] = frozenset({PreviewState.READY}) 

448 profile_action: ClassVar[str] = "change" 

449 reviews_policy: ClassVar[bool] = True 

450 replans: ClassVar[bool] = True 

451 advances: ClassVar[bool] = True 

452 reads_plan: ClassVar[bool] = True 

453 # Only setup replaces an expired preview; every other command finds it gone. 

454 replaces_expired: ClassVar[bool] = False 

455 

456 def accepts(self, row: PreviewCoordinator, expected: PreviewClaim) -> bool: 

457 """Return whether the posted claim names the row exactly.""" 

458 return _exact(row, expected) 

459 

460 def apply(self, preview: LockedPreview) -> CommandOutcome: 

461 """Make this command's own writes against the locked preview.""" 

462 raise NotImplementedError 

463 

464 def run(self, request, row: PreviewCoordinator) -> CommandOutcome: 

465 """Run the command against the locked row; the generic path every writer takes.""" 

466 _refuse_state(row, self.allowed_states) 

467 with locked_profile_policy(row.profile_id): 

468 _refresh_actor(request.user) 

469 profile = ( 

470 ImportProfile.objects.restrict(request.user, self.profile_action).filter(pk=row.profile_id).first() 

471 ) 

472 if profile is None: 

473 raise ImportProfile.DoesNotExist("The import profile is no longer available.") 

474 _refuse_unplannable(profile) 

475 document = SourceDocument.objects.filter(pk=row.source_document_id, profile=profile).first() 

476 if document is None: 

477 raise StaleSourceDocument(SOURCE_GONE) 

478 plan = None 

479 if self.reads_plan: 

480 plan = _stored_plan(row, request.user, document) 

481 if self.reviews_policy: 

482 refuse_moved_policy(profile, plan.profile_fingerprint) 

483 preview = LockedPreview( 

484 actor=request.user, 

485 profile=profile, 

486 document=document, 

487 context=row.context or {}, 

488 plan=plan, 

489 job_id=row.job_id, 

490 ) 

491 outcome = self.apply(preview) 

492 if outcome.refusal is not None: 

493 # atomic-exit-safe: durable-refusal-committed 

494 return outcome 

495 if outcome.context is not None: 

496 row.context = dict(outcome.context) 

497 planning_context = planning_context_of(row.context or {}) 

498 replacement = outcome.plan 

499 if replacement is not None and dict(replacement.planning_context) != planning_context: 

500 raise RuntimeError("A preview command returned a plan made for another planning context.") 

501 if replacement is None and self.replans: 

502 replacement = ImportEngine.plan(profile, document, request.user, planning_context) 

503 if replacement is not None: 

504 _store_plan(row, replacement) 

505 row.state = PreviewState.READY 

506 row.job_id = None 

507 if outcome.state is not None: 

508 row.state = outcome.state 

509 row.job_id = outcome.job_id 

510 if self.advances: 

511 row.revision += 1 

512 row.save() 

513 # atomic-exit-safe: command-published 

514 return outcome 

515 

516 

517def _refuse_unplannable(profile) -> None: 

518 """Refuse a profile whose adapter this release cannot plan, because no command can replan it.""" 

519 try: 

520 validate_registered_adapter(profile) 

521 validate_adapter_target_module(profile.source_adapter) 

522 except ValidationError as exc: 

523 raise PreviewCommandRefused("; ".join(exc.messages), 409) from exc 

524 

525 

526def _refuse_state(row: PreviewCoordinator, allowed) -> None: 

527 """Refuse a command the row's state does not take, naming why.""" 

528 if row.state in allowed: 

529 return 

530 if row.state == PreviewState.SYNC_PENDING: 

531 from core.models import Job 

532 

533 raise StalePreview(_sync_hold(Job.objects.filter(pk=row.job_id)) or SYNC_FINISHED) 

534 if row.state == PreviewState.SUBMITTED: 

535 raise StalePreview(SUBMITTED_PREVIEW) 

536 if row.state == PreviewState.EXPIRED: 

537 raise StalePreview(EXPIRED_PREVIEW) 

538 raise StalePreview(NO_PREVIEW) 

539 

540 

541def _stored_plan(row: PreviewCoordinator, actor, document) -> ImportPlan: 

542 """Return the row's plan, refusing one this release cannot read or one made for another preview.""" 

543 try: 

544 plan = ImportPlan.from_dict(row.plan) 

545 except PlanError as exc: 

546 raise StalePreview(UNREADABLE_PREVIEW) from exc 

547 if ( 

548 plan.actor != str(actor.pk) 

549 or plan.source_fingerprint != document.content_fingerprint 

550 or dict(plan.planning_context) != planning_context_of(row.context or {}) 

551 ): 

552 logger.warning("Preview coordinator %s holds a plan made for another preview.", row.pk) 

553 raise StalePreview(UNREADABLE_PREVIEW) 

554 return plan 

555 

556 

557def validate_preview_plan(plan: ImportPlan) -> dict: 

558 """Return a storable preview, or refuse it before the command's writes commit.""" 

559 data = plan.to_dict() 

560 limit = preview_plan_byte_limit( 

561 settings.PLUGINS_CONFIG["netbox_data_import"].get("preview_max_plan_bytes", MAX_PLAN_BYTES) 

562 ) 

563 if len(canonical_json(data).encode()) > limit: 

564 raise PreviewCommandRefused(PREVIEW_TOO_LARGE, status=413) 

565 return data 

566 

567 

568def _store_plan(row: PreviewCoordinator, plan: ImportPlan) -> None: 

569 row.plan = validate_preview_plan(plan) 

570 

571 

572def apply_preview_command(request, expected: PreviewClaim, command: PreviewCommand) -> PreviewResult: 

573 """Run one command under the coordinator lock and publish what it changed; refuse a stale claim.""" 

574 branching.refuse_branch() 

575 binding = _session_binding(request) 

576 if binding is None: 

577 raise StalePreview(NO_PREVIEW) 

578 committed: list[bool] = [] 

579 outcome: CommandOutcome | None = None 

580 refusal: Exception | None = None 

581 try: 

582 with transaction.atomic(): 

583 # Registered first, so it runs before a queue push that may raise after the commit. 

584 transaction.on_commit(lambda: committed.append(True)) 

585 row = _owned( 

586 PreviewCoordinator.objects.select_for_update().filter(session_binding=binding).first(), 

587 request.user, 

588 ) 

589 row = _refuse_claim(row, expected, command) 

590 if _payload_expired(row) and not command.replaces_expired: 

591 _expire(row) 

592 refusal = StalePreview(EXPIRED_PREVIEW) 

593 else: 

594 outcome = command.run(request, row) 

595 refusal = outcome.refusal 

596 claim, profile_id, state = _claim_of(row), row.profile_id, row.state 

597 except BaseException: 

598 if not committed: 

599 # Everything the command wrote rolled back, so NetBox must not send the events it queued. 

600 clear_events.send(sender=PreviewCommand) 

601 elif outcome is not None and outcome.compensate is not None: 

602 outcome.compensate() 

603 raise 

604 if refusal is not None: 

605 raise refusal 

606 if outcome is None: 

607 raise RuntimeError("A preview command finished without an outcome.") 

608 return PreviewResult(claim=claim, outcome=outcome, profile_id=profile_id, state=state) 

609 

610 

611def _refuse_claim( 

612 row: PreviewCoordinator | None, expected: PreviewClaim, command: PreviewCommand 

613) -> PreviewCoordinator: 

614 """Refuse a command whose claim does not name this session's row as the command requires.""" 

615 if row is None: 

616 raise StalePreview(NO_PREVIEW) 

617 if not command.accepts(row, expected): 

618 raise StalePreview( 

619 STALE_PREVIEW if secrets.compare_digest(row.preview_token, expected.token) else NEWER_PREVIEW 

620 ) 

621 return row 

622 

623 

624class StartPreview(PreviewCommand): 

625 """Store a new upload, plan it, and make it the session's preview (the setup command). 

626 

627 It may replace any revision of the generation the setup page showed, never a newer one. 

628 """ 

629 

630 allowed_states = frozenset({value for value, _label in PreviewState.CHOICES}) 

631 replaces_expired = True 

632 

633 def __init__(self, *, profile, content: bytes, filename: str, site, location=None, tenant=None): 

634 self.profile = profile 

635 self.content = content 

636 self.filename = filename 

637 self.context = { 

638 "site_id": site.pk, 

639 "location_id": location.pk if location else None, 

640 "tenant_id": tenant.pk if tenant else None, 

641 "filename": filename, 

642 } 

643 

644 def accepts(self, row, expected): 

645 """Return whether the claim names the row's generation, at any revision it reached.""" 

646 return _same_generation(row, expected) 

647 

648 def run(self, request, row): 

649 """Store the upload, plan it, and start a new generation in the locked row.""" 

650 with locked_profile_policy(self.profile.pk): 

651 _refresh_actor(request.user) 

652 profile = ImportProfile.objects.restrict(request.user, "change").filter(pk=self.profile.pk).first() 

653 if profile is None: 

654 raise ImportProfile.DoesNotExist("The import profile is no longer available.") 

655 document = SourceDocument.store( 

656 profile=profile, content=self.content, filename=self.filename, uploaded_by=request.user 

657 ) 

658 plan = ImportEngine.plan(profile, document, request.user, planning_context_of(self.context)) 

659 # A new generation: every claim on the old one is refused from here on. 

660 row.preview_token = _new_token() 

661 row.revision += 1 

662 row.state = PreviewState.READY 

663 row.profile_id = profile.pk 

664 row.source_document_id = document.pk 

665 row.context = self.context 

666 row.job_id = None 

667 _store_plan(row, plan) 

668 row.expires_at = _payload_expiry(request) 

669 row.save() 

670 # atomic-exit-safe: new-generation-published 

671 return CommandOutcome(payload={"plan": plan}) 

672 

673 

674class DiscardPreview(PreviewCommand): 

675 """End the preview; the next setup starts from an empty generation.""" 

676 

677 allowed_states = frozenset(PreviewState.ACTIVE) 

678 # An expired preview is still discarded, rather than refused for having expired. 

679 replaces_expired = True 

680 

681 def run(self, request, row): 

682 """Clear the locked row and start an empty generation.""" 

683 _refuse_state(row, self.allowed_states) 

684 _clear_payload(row, PreviewState.EMPTY) 

685 row.preview_token = _new_token() 

686 row.revision += 1 

687 row.expires_at = _session_expiry(request) 

688 row.save() 

689 return CommandOutcome(message="The import preview was discarded.") 

690 

691 

692class RereadPreview(PreviewCommand): 

693 """Replan the preview from its stored source against live NetBox. 

694 

695 It is also the schema recovery, so it never reads the stored plan, and it ends a finished trace 

696 sync's hold. It refuses while that sync's Job still runs, because NetBox is mid-write then. 

697 """ 

698 

699 allowed_states = frozenset({PreviewState.READY, PreviewState.SYNC_PENDING, PreviewState.SUBMITTED}) 

700 reviews_policy = False 

701 reads_plan = False 

702 

703 def run(self, request, row): 

704 """Re-read a submitted preview only if its native Job has disappeared.""" 

705 from core.models import Job 

706 

707 if row.state == PreviewState.SUBMITTED and Job.objects.filter(pk=row.job_id).exists(): 

708 raise StalePreview(SUBMITTED_PREVIEW) 

709 return super().run(request, row) 

710 

711 def apply(self, preview): 

712 """Refuse while any sync of this source still writes; the coordinator replans.""" 

713 from .jobs import recover_abandoned_import_job, retained_sync_jobs 

714 

715 for job in retained_sync_jobs(preview.actor, preview.profile.pk, preview.document.pk): 

716 recover_abandoned_import_job(job) 

717 

718 if reason := _sync_hold(retained_sync_jobs(preview.actor, preview.profile.pk, preview.document.pk)): 

719 raise StalePreview(reason) 

720 return CommandOutcome(message="The preview was re-read from NetBox.") 

721 

722 

723class CancelTraceSync(PreviewCommand): 

724 """Cancel the preview's own trace sync before a worker starts it, then replan the preview. 

725 

726 A worker that holds the Job lock owns the delivery, so the command refuses instead of waiting. 

727 """ 

728 

729 allowed_states = frozenset({PreviewState.SYNC_PENDING}) 

730 reviews_policy = False 

731 reads_plan = False 

732 

733 def __init__(self, job_id: int): 

734 self.job_id = job_id 

735 

736 def apply(self, preview): 

737 """Refuse another preview's Job, a Job a worker took, or a source another sync holds.""" 

738 from core.choices import JobStatusChoices 

739 

740 from .jobs import ImportJobRunner, cancel_queued_import_job, retained_sync_jobs 

741 

742 job = ( 

743 ImportJobRunner.get_jobs() 

744 .filter( 

745 pk=self.job_id, 

746 user=preview.actor, 

747 data__job_type=ImportJobRunner.job_type, 

748 data__keeps_preview=True, 

749 data__profile_id=preview.profile.pk, 

750 data__source_document_id=preview.document.pk, 

751 ) 

752 .first() 

753 ) 

754 if job is None or job.pk != preview.job_id: 

755 raise StalePreview(SYNC_NOT_THIS_PREVIEW) 

756 # The replan below reads NetBox, which another sync of this source may be writing. 

757 if retained_sync_jobs(preview.actor, preview.profile.pk, preview.document.pk).exclude(pk=job.pk).exists(): 

758 raise StalePreview(SYNC_OTHER_ACTIVE) 

759 if not cancel_queued_import_job(job, SYNC_CANCELLED): 

760 raise StalePreview(SYNC_FINISHED if job.status in JobStatusChoices.TERMINAL_STATE_CHOICES else SYNC_STARTED) 

761 return CommandOutcome(message=SYNC_CANCEL_DONE) 

762 

763 

764class RestorePreview(PreviewCommand): 

765 """Return to the preview whose final import failed, replanned against live NetBox.""" 

766 

767 allowed_states = frozenset({PreviewState.SUBMITTED}) 

768 reviews_policy = False 

769 reads_plan = False 

770 

771 def __init__(self, job_id: int): 

772 self.job_id = job_id 

773 

774 def apply(self, preview): 

775 """Refuse a Job this preview did not submit, or one that did not fail.""" 

776 from core.choices import JobStatusChoices 

777 from core.models import Job 

778 from .jobs import recover_abandoned_import_job 

779 

780 if preview.job_id != self.job_id: