Coverage for postrfp/vendor/api/workflow.py: 100%

1""" 

2Team Workflow endpoints. 

3 

4Workflow here = assigning a responsible user to draft an answer (allocated_to) and 

5tracking review / approval (status transitions handled elsewhere). These endpoints: 

6 * List per‑question response state (allocation + review status) for a section (paged). 

7 * Bulk assign (or reassign) responsible users for selected questions in an Issue. 

8""" 

9 

10from sqlalchemy.orm import Bundle, Session 

11 

12from postrfp.model import Section, QuestionInstance, User 

13from postrfp.model.questionnaire.answering import QuestionResponseState 

14from postrfp.model.questionnaire.b36 import from_b36 

15from postrfp.shared.decorators import http 

16from postrfp.shared import fetch, serial 

17from postrfp.shared.pager import Pager 

18from postrfp.authorisation import perms 

19from postrfp.vendor.validation import validate 

20from postrfp.fsm.service import execute_transition # noqa 

21 

22 

23QRB = QuestionResponseState 

24QResBundle: Bundle = Bundle( 

25 "qresponse_bundle", 

26 QRB.status, 

27 QRB.allocated_by, 

28 QRB.allocated_to, 

29 QRB.approved_by, 

30 QRB.date_updated, 

31 QRB.updated_by, 

32 QuestionInstance.b36_number, 

33 QuestionInstance.id.label("question_instance_id"), 

34 single_entity=True, 

35) 

36 

37 

38@http 

39def get_issue_section_workflow( 

40 session: Session, effective_user: User, issue_id: int, section_id: int, pager: Pager 

41) -> serial.WorkflowSection: 

42 """ 

43 Section workflow snapshot for an Issue: returns section metadata plus a paged list 

44 (page size 20) of question response state rows (status, allocated/review fields, 

45 question number & IDs). Supports pagination via standard pager params. 

46 Permission: access to Issue & Section (implicit respondent visibility rules). 

47 """ 

48 issue = fetch.issue(session, issue_id) 

49 section = session.get_one(Section, section_id) 

50 pager.page_size = 20 

51 

52 # ownership of section should be checked in validate below 

53 validate(effective_user, issue, section=section) 

54 

55 q = ( 

56 session.query(QResBundle) 

57 .join(QuestionInstance) 

58 .filter( 

59 QuestionInstance.b36_number.startswith(section.b36_number), 

60 QuestionResponseState.issue_id == issue_id, 

61 ) 

62 ) 

63 

64 qlist = [] 

65 

66 for row in q.slice(pager.startfrom, pager.goto): 

67 r_dict = row._asdict() 

68 r_dict["number"] = from_b36(row.b36_number) 

69 qlist.append(r_dict) 

70 

71 return serial.WorkflowSection( 

72 section=serial.Section( 

73 title=section.title, 

74 number=section.number, 

75 id=section.id, 

76 description=section.description, 

77 parent_id=section.parent_id, 

78 ), 

79 questions=qlist, 

80 count=q.count(), 

81 ) 

82 

83 

84@http 

85def post_issue_allocate( 

86 session: Session, 

87 effective_user: User, 

88 issue_id: int, 

89 allocation_doc: serial.AllocatedToList, 

90) -> None: 

91 """ 

92 Bulk (re)assign allocated_to for specified questions in an Issue. Overwrites any 

93 existing allocation and stamps allocated_by with the caller. No response body. 

94 Permission: ALLOCATE_QUESTIONS. 

95 """ 

96 issue = fetch.issue(session, issue_id) 

97 validate(effective_user, issue, action=perms.ALLOCATE_QUESTIONS) 

98 qlookup = {a.question_instance_id: a.allocated_to for a in allocation_doc} 

99 q = ( 

100 session.query(QuestionResponseState) 

101 .filter(QuestionResponseState.issue_id == issue_id) 

102 .filter(QuestionResponseState.question_instance_id.in_(qlookup.keys())) 

103 ) 

104 

105 for qrs in q: 

106 qrs.allocated_to = qlookup[qrs.question_instance_id] 

107 qrs.allocated_by = effective_user.id