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

1from typing import List 

2import logging 

3 

4from rfpy.suxint import http 

5from rfpy.api import fetch, update, attachments 

6from rfpy.web.response import XAccelAttachmentResponse 

7from ..validation import validate 

8from rfpy.auth import perms 

9from rfpy.web import serial 

10from rfpy.model import AAttachment, IssueAttachment 

11 

12log = logging.getLogger(__name__) 

13 

14 

15@http 

16def post_issue_attachment( 

17 session, effective_user, issue_id, attachment_upload, attachment_description 

18) -> serial.Id: 

19 """ 

20 Upload an Issue attachment together with metadata (description etc) 

21 

22 @param Attachment 

23 """ 

24 issue = fetch.issue(session, issue_id) 

25 validate( 

26 effective_user, issue=issue, action=perms.ISSUE_MANAGE_RESPONDENT_ATTACHMENTS 

27 ) 

28 

29 att = attachments.save_issue_attachment( 

30 session, issue_id, effective_user, attachment_upload, attachment_description 

31 ) 

32 return serial.Id(id=att.id) 

33 

34 

35@http 

36def post_issue_answer_attachment( 

37 session, effective_user, issue_id, attachment_upload, el_id 

38) -> serial.AnswerAttachmentIds: 

39 issue = fetch.issue(session, issue_id) 

40 qelement = fetch.qelement(session, el_id) 

41 question_instance = qelement.get_question_instance(issue.project_id) 

42 

43 validate( 

44 effective_user, 

45 issue=issue, 

46 question=question_instance, 

47 action=perms.ISSUE_SAVE_QUESTION_RESPONSE, 

48 ) 

49 

50 answers_lookup = {el_id: attachment_upload.filename} 

51 

52 # First, save and lookup an Answer - this is needed to find the disc 

53 # file path for the AAttachment. Don't use upload.save_answers() 

54 # because we don't want to log an event until we have the correct filename 

55 # with the filesize 

56 question_instance.validate_and_save_answers(answers_lookup, issue) 

57 session.flush() 

58 answer = question_instance._answers.filter_by( 

59 issue_id=issue.id, element_id=el_id 

60 ).one() 

61 

62 answer_attachment = AAttachment() 

63 answer_attachment.answer = answer 

64 answer_attachment.filename = attachment_upload.filename 

65 answer_attachment.guess_set_mimetype(attachment_upload.filename) 

66 session.add(answer_attachment) 

67 session.flush() # Need to flush to get the ID values set 

68 

69 attachments.save_to_disc(answer_attachment, attachment_upload.file) 

70 answer_txt = f"{answer_attachment.filename} ({answer_attachment.size})" 

71 answer_lookup = {el_id: answer_txt} 

72 

73 update.save_answers( 

74 session, effective_user, question_instance, answer_lookup, issue 

75 ) 

76 

77 return serial.AnswerAttachmentIds( 

78 attachment_id=answer_attachment.id, answer_id=answer.id 

79 ) 

80 

81 

82@http 

83def get_issue_answer_attachment(session, effective_user, issue_id, attachment_id): 

84 """Download a single Issue Attachment file""" 

85 issue = fetch.issue(session, issue_id) 

86 supporting_attachment_element = fetch.qelement(session, attachment_id) 

87 answer = supporting_attachment_element.get_answer(issue) 

88 

89 validate( 

90 effective_user, issue=issue, action=perms.ISSUE_VIEW_ANSWERS, answer=answer 

91 ) 

92 

93 return XAccelAttachmentResponse(answer.attachment) 

94 

95 

96@http 

97def get_issue_attachment(session, effective_user, issue_id, attachment_id): 

98 """Download a single Issue Attachment file""" 

99 issue = fetch.issue(session, issue_id) 

100 validate(effective_user, issue=issue, action=perms.ISSUE_VIEW_ANSWERS) 

101 ia = issue.attachments.filter_by(id=attachment_id).one() 

102 return XAccelAttachmentResponse(ia) 

103 

104 

105@http 

106def get_issue_attachments(session, effective_user, issue_id) -> List[serial.Attachment]: 

107 """List Issue Attachments for the given Issue ID""" 

108 issue = fetch.issue(session, issue_id) 

109 validate(effective_user, issue=issue, action=perms.ISSUE_VIEW_ANSWERS) 

110 return [ia.as_dict() for ia in issue.attachments] 

111 

112 

113@http 

114def delete_issue_attachments(session, effective_user, issue_id, ids_doc): 

115 """ 

116 Delete those Issue attachments whose IDs are given in the JSON body 

117 document 

118 """ 

119 

120 issue = fetch.issue(session, issue_id) 

121 validate( 

122 effective_user, issue=issue, action=perms.ISSUE_MANAGE_RESPONDENT_ATTACHMENTS 

123 ) 

124 

125 att_id_list = ids_doc.ids 

126 atts = ( 

127 session.query(IssueAttachment) 

128 .filter(IssueAttachment.id.in_(att_id_list)) 

129 .filter(IssueAttachment.issue_id == issue_id) 

130 ) 

131 

132 for att in atts: 

133 session.delete(att) 

134 attachments.delete_from_disc(att) 

135 

136 

137@http 

138def get_issue_question_attachment( 

139 session, effective_user, issue_id, question_id, attachment_id 

140): 

141 """ 

142 Download a Question Attachment. 

143 @attachment_id is the ID of the QuestionAttachment *Element*, 

144 not QAttachment 

145 """ 

146 issue = fetch.issue(session, issue_id) 

147 qi = fetch.question(session, question_id) 

148 

149 validate(effective_user, issue=issue, question=qi) 

150 

151 q_element = qi.question_def.get_element(attachment_id) 

152 return XAccelAttachmentResponse(q_element.attachment)