Coverage for netbox_data_import/preview_row_actions.py: 96%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright (C) 2026 Marcin Zieba <marcinpsk@gmail.com> 

3"""Manage the materialized preview used by asynchronous row actions.""" 

4 

5import secrets 

6 

7 

8class PreviewActionInvalid(ValueError): 

9 """A row action refused for a reason this plugin wrote, so the response may state it. 

10 

11 Any other exception carries internal detail, so it reaches the operator as a generic message. 

12 """ 

13 

14 

15PREVIEW_DIRTY_SESSION_KEY = "import_preview_dirty" 

16PREVIEW_PLAN_SESSION_KEY = "import_plan" 

17PREVIEW_REVISION_SESSION_KEY = "import_preview_revision" 

18PREVIEW_USE_MATERIALIZED_ONCE_SESSION_KEY = "import_preview_use_materialized_once" 

19 

20 

21def current_preview_revision(session) -> str: 

22 """Return the active preview revision, creating it when needed.""" 

23 revision = session.get(PREVIEW_REVISION_SESSION_KEY) 

24 if not revision: 

25 revision = secrets.token_urlsafe(18) 

26 session[PREVIEW_REVISION_SESSION_KEY] = revision 

27 return revision 

28 

29 

30RETAINED_SYNC_BLOCK_REASON = ( 

31 "A trace synchronization is still running. Wait for it to finish before changing this workspace." 

32) 

33 

34 

35class PreviewLocked(RuntimeError): 

36 """The preview may not move while the trace sync it queued is still running. 

37 

38 A per-trace sync keeps the preview open, so the operator stays on a page whose plan the queued 

39 Job is about to invalidate. Recalculating adopts NetBox state that predates the Job's writes and 

40 clears the guard that stops a second queue, so both are refused until the Job is terminal. 

41 """ 

42 

43 

44def retained_sync_block_reason(session, user) -> str: 

45 """Return why a retained trace sync holds this preview, or ``""``. 

46 

47 The Job rows are the record. A request that loses the race to enqueue still writes the session, 

48 so a guard reading one remembered id can open a preview whose other sync is still writing. 

49 """ 

50 from core.choices import JobStatusChoices 

51 

52 from .jobs import ImportJobRunner 

53 

54 context = session.get("import_context") 

55 if not isinstance(context, dict): 

56 return "" 

57 profile_id, document_id = context.get("profile_id"), context.get("source_document_id") 

58 if not profile_id or not document_id: 

59 return "" 

60 retained = ImportJobRunner.get_jobs().filter( 

61 user=user, 

62 data__job_type=ImportJobRunner.job_type, 

63 data__keeps_preview=True, 

64 data__profile_id=profile_id, 

65 data__source_document_id=document_id, 

66 status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES, 

67 ) 

68 return RETAINED_SYNC_BLOCK_REASON if retained.exists() else "" 

69 

70 

71def assert_preview_may_move(session, user) -> None: 

72 """Raise `PreviewLocked` when the retained trace sync still holds this preview.""" 

73 if reason := retained_sync_block_reason(session, user): 

74 raise PreviewLocked(reason) 

75 

76 

77def _store_preview(session, plan) -> str: 

78 """Write one authoritative preview and return its new revision.""" 

79 revision = secrets.token_urlsafe(18) 

80 session[PREVIEW_PLAN_SESSION_KEY] = plan.to_dict() 

81 session[PREVIEW_DIRTY_SESSION_KEY] = False 

82 session[PREVIEW_REVISION_SESSION_KEY] = revision 

83 return revision 

84 

85 

86def record_recalculated_preview(session, plan, *, user) -> str: 

87 """Replace the current preview with a freshly read one, refusing while a sync holds it.""" 

88 assert_preview_may_move(session, user) 

89 return _store_preview(session, plan) 

90 

91 

92def start_new_preview(session, plan) -> str: 

93 """Store the first preview of a newly uploaded source, replacing whatever came before. 

94 

95 Unguarded on purpose: this is a different import, so it inherits no earlier sync. The upload 

96 stored its own Source Document, which is what stops the previous preview's Job from matching. 

97 """ 

98 return _store_preview(session, plan) 

99 

100 

101def restore_preview_plan(session, plan_data) -> None: 

102 """Adopt the accepted plan a failed Job stored, so its preview can be reviewed again.""" 

103 session[PREVIEW_PLAN_SESSION_KEY] = plan_data 

104 

105 

106def clear_preview_state(session) -> None: 

107 """Drop the stored plan, for a preview that is being discarded.""" 

108 session.pop(PREVIEW_PLAN_SESSION_KEY, None) 

109 

110 

111def retire_preview_revision(session) -> str: 

112 """Invalidate the token any open preview is holding, without storing a new result.""" 

113 revision = secrets.token_urlsafe(18) 

114 session[PREVIEW_REVISION_SESSION_KEY] = revision 

115 return revision 

116 

117 

118def load_cached_preview(request): 

119 """Return the active Import Profile and materialized Review Workspace.""" 

120 from .models import ImportProfile 

121 from .plan import PlanError 

122 from .review_workspace import ReviewWorkspace 

123 

124 context = request.session.get("import_context") 

125 plan_data = request.session.get(PREVIEW_PLAN_SESSION_KEY) 

126 if ( 

127 request.session.get("import_preview_pending") is not True 

128 or not isinstance(context, dict) 

129 or not isinstance(plan_data, dict) 

130 ): 

131 return None 

132 revision = current_preview_revision(request.session) 

133 if "application/json" in request.headers.get("Accept", ""): 

134 # A read carries its revision in the query, because a GET has no posted body to hold it. 

135 posted = request.POST.get("preview_revision", request.GET.get("preview_revision")) 

136 if posted != revision: 

137 return None 

138 profile = ImportProfile.objects.restrict(request.user, "change").filter(pk=context.get("profile_id")).first() 

139 if profile is None: 

140 return None 

141 try: 

142 workspace = ReviewWorkspace.from_dict(plan_data) 

143 except PlanError: 

144 return None 

145 return profile, workspace 

146 

147 

148def mark_preview_dirty(session) -> None: 

149 """Record that saved changes require one authoritative recalculation.""" 

150 session[PREVIEW_DIRTY_SESSION_KEY] = True 

151 

152 

153def pending_preview_payload(row_number: int, message: str, detail: str = "", resolution: dict | None = None) -> dict: 

154 """Return the small response shared by deferred preview-row actions. 

155 

156 `detail` names a write this action already made in NetBox, which the page reports rather than 

157 leaving the operator to discover it. A save that only records a decision carries none. 

158 

159 `resolution` is the decision as it was stored, which the page keeps in place of the one it 

160 posted. Only an action that saves a resolution carries it. 

161 """ 

162 payload = { 

163 "ok": True, 

164 "row_number": row_number, 

165 "preview_state": "recalculation_required", 

166 "message": message, 

167 "detail": detail, 

168 } 

169 if resolution is not None: 

170 payload["resolution"] = resolution 

171 return payload