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

1""" 

2Team Workflow methods. Allocate and track responsibility for answering questions. 

3""" 

4 

5from sqlalchemy.orm import Bundle, Session 

6 

7from rfpy.model import Section, QuestionInstance, QuestionResponseState, User 

8from rfpy.suxint import http 

9from rfpy.api import fetch 

10from rfpy.web import serial 

11from rfpy.auth import perms 

12from rfpy.vendor.validation import validate 

13 

14 

15QRB = QuestionResponseState 

16QResBundle: Bundle = Bundle( 

17 "qresponse_bundle", 

18 QRB.status, 

19 QRB.allocated_by, 

20 QRB.allocated_to, 

21 QRB.approved_by, 

22 QRB.date_updated, 

23 QRB.updated_by, 

24 QuestionInstance.number, 

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

26 single_entity=True, 

27) 

28 

29 

30@http 

31def get_issue_section_workflow( 

32 session, effective_user, issue_id, section_id, pager 

33) -> serial.WorkflowSection: 

34 issue = fetch.issue(session, issue_id) 

35 section = session.get(Section, section_id) 

36 pager.pagesize = 20 

37 

38 # ownership of section should be checked in validate below 

39 validate(effective_user, issue, section=section) 

40 

41 q = ( 

42 session.query(QResBundle) 

43 .join(QuestionInstance) 

44 .filter( 

45 QuestionInstance.number.startswith(section.number), 

46 QuestionResponseState.issue_id == issue_id, 

47 ) 

48 ) 

49 

50 qlist = [] 

51 

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

53 r_dict = row._asdict() 

54 r_dict["number"] = row.number.dotted 

55 qlist.append(r_dict) 

56 

57 return serial.WorkflowSection( 

58 section=serial.Section( 

59 title=section.title, 

60 number=section.number.dotted, 

61 id=section.id, 

62 description=section.description, 

63 parent_id=section.parent_id, 

64 ), 

65 questions=qlist, 

66 count=q.count(), 

67 ) 

68 

69 

70@http 

71def post_issue_allocate( 

72 session: Session, 

73 effective_user: User, 

74 issue_id: int, 

75 allocation_doc: serial.AllocatedToList, 

76) -> None: 

77 issue = fetch.issue(session, issue_id) 

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

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

80 q = ( 

81 session.query(QuestionResponseState) 

82 .filter(QuestionResponseState.issue_id == issue_id) 

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

84 ) 

85 

86 for qrs in q: 

87 qrs.allocated_to = qlookup[qrs.question_instance_id] 

88 qrs.allocated_by = effective_user.id