Coverage for postrfp/vendor/api/issue.py: 98%
58 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-22 21:34 +0000
1"""
2Respondent Issue endpoints.
4Exposes a limited Issue view for respondents: list/retrieve Issues, perform
5allowed status transitions, toggle workflow usage, and create/list respondent
6notes (with audit logging where applicable).
7"""
9from sqlalchemy import not_, and_
10from sqlalchemy.orm import Session
12from postrfp.model import (
13 Issue,
14 Project,
15 IssueWatchList,
16 Organisation,
17 ProjectNote,
18 AuditEvent,
19 User,
20)
21from postrfp.authorisation import perms
22from postrfp.shared import serial, fetch
23from postrfp.shared.decorators import http
24from postrfp.shared.issue_transition import change_issue_status
25from postrfp.vendor.validation import validate
27# columns exposed to respondent users
28respondent_columns = (
29 Issue.id,
30 Issue.status,
31 Issue.issue_date,
32 Issue.accepted_date,
33 Issue.deadline,
34 Issue.label,
35 Issue.winloss_exposed,
36 Issue.winloss_expiry,
37 Issue.submitted_date,
38 Issue.respondent_id,
39 Issue.use_workflow,
40 Project.title,
41 Project.org_id,
42 IssueWatchList.date_created.isnot(None).label("is_watched"),
43)
46def _issue_q(session, user):
47 return (
48 session.query(Issue)
49 .join(Project)
50 .join(Organisation)
51 .outerjoin(
52 IssueWatchList,
53 and_(
54 IssueWatchList.user_id == user.id, IssueWatchList.issue_id == Issue.id
55 ),
56 )
57 .with_entities(*respondent_columns)
58 )
61@http
62def get_issue(
63 session: Session, effective_user: User, issue_id: int
64) -> serial.VendorIssue:
65 """
66 Retrieve a single Issue visible to the respondent (restricted field set including
67 project title and watch status). Visibility validated against respondent org.
68 """
69 # Using subscript notation here because we want to
70 # reuse the list of columns but this won't work for returning
71 # a single value
72 issue = _issue_q(session, effective_user).filter(Issue.id == issue_id)[0]
74 validate(effective_user, issue)
75 return issue._asdict()
78@http
79def get_issues(session: Session, effective_user: User) -> list[serial.VendorIssue]:
80 """
81 List respondent Issues (excludes Not Sent / Retracted). Ordered by most recent
82 issue_date descending.
83 """
84 issues = (
85 _issue_q(session, effective_user)
86 .filter(
87 Issue.respondent_id == effective_user.organisation.id,
88 not_(Issue.status.in_(("Not Sent", "Retracted"))),
89 )
90 .order_by(Issue.issue_date.desc())
91 )
92 return [i._asdict() for i in issues]
95""" Status Changes """
98@http
99def post_issue_status(
100 session: "Session",
101 effective_user: "User",
102 issue_id: int,
103 issue_status_doc: serial.IssueStatus,
104) -> None:
105 """
106 Perform an allowed respondent status transition: Accepted, Submitted, Declined.
107 Enforces mandatory answer / deadline rules on submission. Emits audit events.
108 """
109 issue = fetch.issue(session, issue_id)
110 status = issue_status_doc.new_status
112 # This check is already effectively performed in Issue.changed_status
113 # but double checking here
114 permitted_statuses = ("Accepted", "Submitted", "Declined")
115 if status not in permitted_statuses:
116 raise ValueError("new status must be one of %s" % str(permitted_statuses))
117 status_perms = {
118 "Accepted": perms.ISSUE_ACCEPT,
119 "Submitted": perms.ISSUE_SUBMIT,
120 "Declined": perms.ISSUE_DECLINE,
121 }
122 action = status_perms[status]
123 validate(effective_user, issue=issue, action=action)
124 change_issue_status(issue, effective_user, status)
127@http
128def post_issue_workflow(
129 session: Session,
130 effective_user: User,
131 issue_id: int,
132 issue_workflow_doc: serial.IssueUseWorkflow,
133) -> None:
134 """
135 Toggle workflow usage flag for this Issue (no audit event). Permission:
136 ISSUE_UPDATE_WORKFLOW.
137 """
138 issue = fetch.issue(session, issue_id)
139 validate(effective_user, issue=issue, action=perms.ISSUE_UPDATE_WORKFLOW)
140 issue.use_workflow = issue_workflow_doc.use_workflow
143""" NOTES """
146@http
147def post_issue_note(
148 session: Session,
149 effective_user: User,
150 issue_id: int,
151 respondent_note_doc: serial.RespondentNote,
152) -> serial.Id:
153 """
154 Create a respondent note (kind=RespondentNote). Private flag restricts visibility
155 to respondent org. Emits PROJECT_NOTE_ADDED audit event with field changes.
156 """
157 issue = fetch.issue(session, issue_id)
158 validate(effective_user, issue, action=perms.PROJECT_ADD_NOTE)
159 project = issue.project
161 note = ProjectNote(
162 kind="RespondentNote",
163 project=project,
164 note_text=respondent_note_doc.note_text,
165 private=respondent_note_doc.private,
166 user_id=effective_user.id,
167 org_id=effective_user.org_id,
168 target_org_id=None, # RespondentNotes don't set target_org_id
169 )
171 session.add(note)
172 session.flush()
174 evt = AuditEvent.create(
175 session,
176 "PROJECT_NOTE_ADDED",
177 project=project,
178 object_id=note.id,
179 user_id=note.user_id,
180 org_id=note.org_id,
181 )
182 evt.add_change("note_text", "", note.note_text)
183 evt.add_change("private", "", note.private)
184 session.add(evt)
185 return serial.Id(id=note.id)
188@http
189def get_issue_notes(
190 session: Session, effective_user: User, issue_id: int
191) -> list[serial.ReadNote]:
192 """
193 List notes visible to the respondent for this Issue (privacy / distribution
194 filtering applied).
195 """
196 issue = fetch.issue(session, issue_id)
197 validate(effective_user, issue, action=perms.ISSUE_VIEW_ANSWERS)
198 nq = fetch.vendor_notes(issue, effective_user)
199 rn = serial.ReadNote
200 return [rn.model_validate(note) for note in nq]