Coverage for rfpy/api/endpoints/scoring.py: 99%
170 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-24 10:52 +0000
1"""
2Operations for viewing & assigning scores for vendors' answers
3"""
5from typing import List, Optional
6from datetime import datetime
8from dataclasses import dataclass
10from sqlalchemy.orm import Session
11from sqlalchemy.orm.exc import NoResultFound
13from rfpy.suxint import http
14from rfpy.model import (
15 ScoreComment,
16 AuditEvent,
17 Score,
18 Project,
19 LiveProject,
20 Issue,
21 User,
22 QuestionInstance,
23)
25from rfpy.api import fetch, validate, update
26from rfpy.auth import AuthorizationFailure, perms
27from rfpy.web import serial
30@http
31def get_project_scores(
32 session, user: User, project_id: int, scoreset_id: str = "__default__"
33) -> serial.ScoringData:
34 """
35 Returns all scores. Used for analysis in the browser.
37 @raise AuthorisationFailue - if the user is not a standard user from the buying organisation
38 @raise NoResultFound - if the specified project cannot be loaded
39 """
40 project = fetch.project(session, project_id)
41 validate.check(
42 user, perms.ISSUE_VIEW_AGREED_SCORES, project=project, deny_restricted=True
43 )
44 scoreset_id = scoreset_id if scoreset_id != "__default__" else ""
45 return serial.ScoringData(
46 scores=[
47 s._asdict() for s in fetch.scoring_data(project, scoreset_id=scoreset_id)
48 ],
49 scoreset_id=scoreset_id,
50 )
53@http
54def get_question_scores(
55 session: Session, user: User, question_id: int, scoreset_id: str = ""
56):
57 question = fetch.question(session, question_id)
58 section = question.section
60 question_filter = QuestionInstance.id == question.id
62 score_permission = Score.get_permission_for_scoreset(user, scoreset_id)
63 validate.check(
64 user, score_permission, project=question.project, section_id=section.id
65 )
67 return fetch.scores(
68 session, section.project, section, scoreset_id, user, question_filter
69 )
72@http
73def post_question_score(
74 session: Session, user: User, question_id: int, score_doc: serial.Score
75) -> serial.Id:
76 """Create or update a single score for a given question/issue combination"""
78 question = fetch.question(session, question_id)
79 validate.check(
80 user,
81 perms.PROJECT_VIEW_QUESTIONNAIRE,
82 project=question.project,
83 section_id=question.section_id,
84 deny_restricted=False,
85 )
86 project = question.project
87 score, created = fetch.or_create_score(session, project, user, question, score_doc)
88 initial_score_value = score.score
89 score_value = score_doc.score_value
90 Score.check_score_value(score_value, project)
91 score.score = score_value
93 score_perm = Score.get_permission_for_scoreset(
94 user, score_doc.scoreset_id, to_save=True
95 )
96 validate.check(
97 user,
98 score_perm,
99 project=question.project,
100 section_id=question.section_id,
101 deny_restricted=False,
102 )
104 # Need the Score record's ID for the audit event record, so flush
105 session.flush()
106 update.log_score_event(session, score, initial_score_value, created, project, user)
107 return serial.Id(id=score.id)
110@http
111def get_project_section_issue_scores(
112 session, user, project_id, section_id, issue_id, scoreset_id=""
113):
114 project = fetch.project(session, project_id)
115 section = fetch.section_of_project(project, section_id)
116 issue = project.get_issue(issue_id)
118 score_permission = Score.get_permission_for_scoreset(user, scoreset_id)
119 validate.check(
120 user, score_permission, project=project, issue=issue, section_id=section.id
121 )
123 issue_filter = Issue.id == issue.id
124 return fetch.scores(session, project, section, scoreset_id, user, issue_filter)
127@http
128def post_question_score_comment(
129 session: Session, user: User, question_id: int, score_doc: serial.Score
130) -> None:
131 """Add a single score comment. A score value can be provided at the same time"""
132 question = fetch.question(session, question_id)
133 validate.check(
134 user,
135 perms.PROJECT_VIEW_QUESTIONNAIRE,
136 project=question.project,
137 section_id=question.section_id,
138 )
139 project = question.project
141 score, created = fetch.or_create_score(session, project, user, question, score_doc)
143 to_save = False if score_doc.score_value is None else True
145 score_permission = Score.get_permission_for_scoreset(
146 user, score_doc.scoreset_id, to_save=to_save
147 )
149 validate.check(
150 user,
151 score_permission,
152 issue=score.issue,
153 section_id=question.section_id,
154 project=question.project,
155 )
157 if score_doc.score_value is not None:
158 initial_score_value = score.score
159 score_value = score_doc.score_value
160 Score.check_score_value(score_value, project)
161 score.score = score_value
162 # Need the Score record's ID for the audit event record, so flush
163 session.flush()
164 update.log_score_event(
165 session, score, initial_score_value, created, project, user
166 )
168 if score_doc.comment is not None:
169 # make the comment and add to database
170 comment = ScoreComment(
171 score=score,
172 comment_time=datetime.now(),
173 user_id=user.id,
174 comment_text=score_doc.comment,
175 )
176 session.add(comment)
177 session.flush()
178 evt = AuditEvent.create(
179 session,
180 "SCORE_COMMENT_ADDED",
181 object_id=comment.id,
182 user=user,
183 project=project,
184 issue_id=score_doc.issue_id,
185 question_id=question.id,
186 )
187 evt.add_change("Comment", "", comment.comment_text)
189 session.add(evt)
192def check_autoscore_permissions(
193 project: Project, initiating_user: User, target_user: User
194):
195 if not project.multiscored:
196 raise ValueError("Project must be using Multiple Score Sets")
197 if target_user.organisation not in project.participants:
198 m = f"User {target_user.id} not a participant in project {project.id}"
199 raise AuthorizationFailure(m)
200 target_user.check_permission(perms.ISSUE_SAVE_SCORES)
201 initiating_user.check_permission(perms.ISSUE_SAVE_AGREED_SCORES)
204@http
205def get_project_calcautoscores(
206 session: Session, user: User, project_id: int, target_user: User
207):
208 project = fetch.project(session, project_id)
210 if not isinstance(project, LiveProject):
211 raise ValueError("Project must be live to generate autoscores")
213 check_autoscore_permissions(project, user, target_user)
214 ascores = project.generate_autoscores(session, target_user)
215 return list(ascores.values())
218@http
219def post_project_calcautoscores(
220 session: Session, user: User, project_id: int, target_user: User
221):
222 """
223 Autoscores are not automatically calculated for User Score Sets. This method
224 allows autoscores to be set for the target_user's score set in the current project.
225 """
226 project = fetch.project(session, project_id)
228 if not isinstance(project, LiveProject):
229 raise ValueError("Project must be live to generate autoscores")
231 check_autoscore_permissions(project, user, target_user)
232 existing_scores = project.scores_dict()
233 unsaved_scores = []
234 for auto_key, score_dict in project.generate_autoscores(
235 session, target_user
236 ).items():
237 if auto_key in existing_scores:
238 score = existing_scores[auto_key]
239 if score.score is not None and int(score.score) == int(
240 score_dict.get("score", -1)
241 ):
242 continue
243 initial_score_value = score.score
244 score.score = score_dict["score"]
245 update.log_score_event(
246 session,
247 score,
248 initial_score_value,
249 False,
250 project,
251 target_user,
252 autoscore=True,
253 )
254 else:
255 score = Score(
256 question_instance_id=score_dict["question_id"],
257 scoreset_id=score_dict["scoreset_id"],
258 issue_id=score_dict["issue_id"],
259 score=score_dict["score"],
260 )
261 unsaved_scores.append(score)
262 session.add(score)
263 # Need ID values for newly created score objects
264 session.flush()
265 for score in unsaved_scores:
266 update.log_score_event(
267 session, score, None, True, project, target_user, autoscore=True
268 )
271@http
272def get_section_scoresummaries(
273 session, user, section_id, scoreset_id
274) -> serial.ScoreSummary:
275 """
276 A report summarising scores for all submitted Issues for a Section
278 For each subsection a summary of the number of questions and the number of scored
279 questions is provided. This data can be used to calculate a percentage completion
280 for each subsection. A score total for that subsection is also provided.
282 For each question within the current section a single score value is provided
284 A user with the permission ISSUE_VIEW_AGREED_SCORES can see other users'
285 scores sets.
286 """
287 section = fetch.section(session, section_id)
288 project = section.project
289 permission = Score.get_permission_for_scoreset(user, scoreset_id)
290 validate.check(user, permission, project=project, section_id=section.id)
291 sub = fetch.subsection_scoressummary(session, user, project, section, scoreset_id)
292 return serial.ScoreSummary(
293 subsections=[
294 serial.SectionScore.model_validate(row)
295 for row in fetch.section_scoresummary(session, user, project, section, sub)
296 ],
297 questions=[
298 serial.QuestionScore.model_validate(q)
299 for q in fetch.question_scoresummary(
300 session, user, project, section, scoreset_id
301 )
302 ],
303 )
306@http
307def get_project_scoresets(
308 session, user, project_id, scoreset_id=""
309) -> List[serial.ScoreSet]:
310 """
311 Returns a list of scoresets that the user can view
312 """
314 project = fetch.project(session, project_id)
316 # VIEW_AGREED_SCORE permission allows a user to view other user's score sets
317 if user.has_permission(perms.ISSUE_VIEW_AGREED_SCORES):
318 sq = (
319 session.query(Score.scoreset_id, User.fullname)
320 .join(Issue)
321 .outerjoin(User, Score.scoreset_id == User.id)
322 .filter(Issue.project == project, Score.scoreset_id != "")
323 .distinct()
324 )
326 sc = [serial.ScoreSet.model_validate(row) for row in sq]
327 sc.append(
328 serial.ScoreSet(scoreset_id="__default__", fullname="Agreed Scoring Set")
329 )
330 return sc
332 else:
333 user.check_permission(perms.ISSUE_VIEW_SCORES)
334 return [serial.ScoreSet(scoreset_id=user.id, fullname=user.fullname)]
337@http
338def get_question_issue_comments(session, user, question_id, issue_id, scoreset_id=""):
339 question = fetch.question(session, question_id)
340 validate.check(
341 user,
342 perms.PROJECT_VIEW_QUESTIONNAIRE,
343 project=question.project,
344 section_id=question.section_id,
345 )
346 issue = question.project.get_issue(issue_id)
348 score_permission = Score.get_permission_for_scoreset(user, scoreset_id)
350 validate.check(
351 user,
352 score_permission,
353 issue=issue,
354 section_id=question.section_id,
355 project=question.project,
356 )
358 try:
359 score = (
360 session.query(Score)
361 .filter(
362 Score.question_instance_id == question.id,
363 Score.issue_id == issue.id,
364 Score.scoreset_id == scoreset_id,
365 )
366 .one()
367 )
368 except NoResultFound:
369 return []
371 return [comment.as_dict() for comment in score.comments]
374@dataclass
375class ScoreData:
376 score: Score
377 initial_score_value: Optional[float]
378 created: bool
381@http
382def post_section_scoreset_scores(
383 session: Session,
384 user: User,
385 section_id: int,
386 scoreset_id: str,
387 section_score_docs: serial.SectionScoreDocs,
388) -> List[serial.Id]:
389 """Bulk create/update scores"""
391 section = fetch.section(session, section_id)
392 project = section.project
393 validate.check(
394 user,
395 perms.PROJECT_VIEW_QUESTIONNAIRE,
396 project=project,
397 section_id=section_id,
398 deny_restricted=False,
399 )
401 score_perm = Score.get_permission_for_scoreset(user, scoreset_id, to_save=True)
402 validate.check(
403 user, score_perm, project=project, section_id=section_id, deny_restricted=False
404 )
406 # Use the dataclass instead of a named tuple
407 data: list[ScoreData] = []
409 for doc in section_score_docs.root:
410 question = fetch.question_of_section(session, section_id, doc.question_id)
411 for score_doc in doc.scores:
412 issue = project.get_issue(score_doc.issue_id)
413 sd = serial.Score(
414 issue_id=issue.id,
415 score_value=score_doc.score_value,
416 scoreset_id=scoreset_id,
417 )
418 score, created = fetch.or_create_score(session, project, user, question, sd)
419 initial_score_value = score.score
420 score_value = sd.score_value
421 Score.check_score_value(score_value, project)
422 score.score = score_value
423 data.append(
424 ScoreData(
425 score=score,
426 initial_score_value=initial_score_value,
427 created=created,
428 )
429 )
431 session.flush()
432 score_ids: list[serial.Id] = []
434 for item in data:
435 session.flush()
436 update.log_score_event(
437 session,
438 item.score,
439 item.initial_score_value,
440 item.created,
441 project,
442 user,
443 )
444 score_ids.append(serial.Id(id=score.id))
446 return score_ids