Coverage for netbox_data_import/proposal_decisions.py: 97.67%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""Revalidate proposal evidence and serialize explicit operator decisions.""" 

4 

5from django.db import transaction 

6from utilities.permissions import get_permission_for_model 

7 

8from . import branching 

9from .inference_backend import proposal_eligible_set_limit 

10from .models import ( 

11 ImportProfile, 

12 ProposalDecision, 

13 ProposalOutcome, 

14 ProposalStatus, 

15 ResolutionProposal, 

16 locked_profile_policy, 

17) 

18from .object_permissions import ObjectPermissionDenied 

19from .proposal_tasks import CandidateSnapshot, proposal_inventory_staleness, proposal_task 

20from .resolution_proposals import decide_proposal 

21from .review_workspace import refuse_moved_policy 

22 

23 

24def proposal_staleness(proposal, *, inventory): 

25 """Compare current inventory with frozen evidence without changing the proposal.""" 

26 return proposal_inventory_staleness(proposal, inventory) 

27 

28 

29def accept_proposal(proposal_id, *, source, operator, netbox_reader, reviewed_fingerprint) -> bool: 

30 """Write one fresh candidate decision, with the source spelling the preview states, under the locks. 

31 

32 Acceptance is a workspace command that writes profile policy, so it also refuses a decision made 

33 against a reviewed preview whose profile policy has since moved (section 10.2). 

34 """ 

35 branching.refuse_branch() 

36 if operator is None or netbox_reader.actor != operator: 

37 raise ValueError("Acceptance requires a reader scoped to the deciding operator.") 

38 profile_id = ResolutionProposal.objects.values_list("profile_id", flat=True).get(pk=proposal_id) 

39 with locked_profile_policy(profile_id): 

40 proposal = ResolutionProposal.objects.select_for_update().get(pk=proposal_id, profile_id=profile_id) 

41 task = proposal_task(proposal.task_type) 

42 if ( 

43 proposal.status != ProposalStatus.COMPLETED 

44 or proposal.outcome != ProposalOutcome.CANDIDATE 

45 or proposal.decision 

46 ): 

47 # atomic-exit-safe: proposal-refused-before-write 

48 return False 

49 proposal.profile = ImportProfile.objects.get(pk=profile_id) 

50 refuse_moved_policy(proposal.profile, reviewed_fingerprint) 

51 snapshot = CandidateSnapshot.from_json(proposal.candidate_snapshot) 

52 entry = next( 

53 (entry for entry in snapshot.entries if entry.candidate_id == proposal.selected_candidate_id), None 

54 ) 

55 if entry is None: 

56 # atomic-exit-safe: proposal-refused-before-write 

57 return False 

58 receipt = task.write_resolution_if_fresh( 

59 proposal=proposal, 

60 entry=entry, 

61 source=source, 

62 actor=operator, 

63 netbox_reader=netbox_reader, 

64 limit=proposal_eligible_set_limit(), 

65 ) 

66 if receipt is None: 

67 # atomic-exit-safe: proposal-refused-before-write 

68 return False 

69 decided = decide_proposal( 

70 proposal.pk, 

71 decision=ProposalDecision.ACCEPTED, 

72 operator=operator, 

73 written_resolution=receipt.written_resolution_id, 

74 ) 

75 if not decided: 

76 transaction.set_rollback(True) 

77 return decided 

78 

79 

80def reject_proposal(proposal_id, *, operator) -> bool: 

81 """Record one rejection without changing status or requiring fresh inventory. 

82 

83 Rejection writes no Row Resolution, so it takes the workspace permission and not the permission 

84 to create one (specification 7.6). 

85 """ 

86 branching.refuse_branch() 

87 profile_id = ResolutionProposal.objects.values_list("profile_id", flat=True).get(pk=proposal_id) 

88 if not ImportProfile.objects.restrict(operator, "view").filter(pk=profile_id).exists(): 

89 raise ObjectPermissionDenied(get_permission_for_model(ImportProfile, "view")) 

90 with locked_profile_policy(profile_id): 

91 proposal = ResolutionProposal.objects.select_for_update().get(pk=proposal_id, profile_id=profile_id) 

92 decided = decide_proposal(proposal.pk, decision=ProposalDecision.REJECTED, operator=operator) 

93 return decided