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

1''' 

2Manage file attachments and uploads 

3''' 

4from typing import List 

5 

6from rfpy.suxint import http 

7from rfpy.auth import perms 

8from rfpy.api import fetch, validate, attachments 

9from rfpy.web.response import XAccelAttachmentResponse 

10from rfpy.web.mime import MimeTypes 

11from rfpy.web import serial 

12from rfpy.model import QuestionInstance 

13from rfpy.model.audit import AuditEvent, evt_types 

14 

15 

16safo = serial.Attachment.from_orm 

17 

18 

19@http 

20def get_project_attachments(session, user, project_id) -> List[serial.Attachment]: 

21 ''' 

22 List details for all Project Attachments (attachments uploaded by the Project author - buyer) 

23 ''' 

24 project = fetch.project(session, project_id) 

25 validate.check(user, perms.PROJECT_ACCESS, project=project, deny_restricted=False) 

26 return [safo(att) for att in project.list_attachments(user)] 

27 

28 

29@http 

30def get_project_issueattachments(session, user, project_id) -> List[serial.IssueAttachment]: 

31 ''' 

32 List all Issue attachments (uploaded by Respondents) for the current Project 

33 ''' 

34 project = fetch.project(session, project_id) 

35 validate.check(user, perms.ISSUE_VIEW_ANSWERS, 

36 project=project, section_id=project.section_id, deny_restricted=False) 

37 return [serial.IssueAttachment.from_orm(att) 

38 for att in project.list_issue_attachments(user)] 

39 

40 

41@http 

42def get_project_answerattachments(session, user, project_id) -> List[serial.AnswerAttachment]: 

43 ''' 

44 List Answer Attachments for the current project - attachments uploaded as answers to 

45 questions with File Upload question elements 

46 ''' 

47 project = fetch.project(session, project_id) 

48 validate.check( 

49 user, 

50 perms.ISSUE_VIEW_ANSWERS, 

51 project=project, 

52 deny_restricted=False, 

53 section_id=project.section_id 

54 ) 

55 return [serial.AnswerAttachment.from_orm(att) 

56 for att in fetch.answer_attachments_q(project, user)] 

57 

58 

59@http 

60def get_project_attachment(session, user, project_id, attachment_id) -> MimeTypes.ANY: 

61 ''' 

62 Download the Project Attachment with the given ID 

63 ''' 

64 project = fetch.project(session, project_id) 

65 validate.check(user, perms.PROJECT_ACCESS, project=project, deny_restricted=False) 

66 attachment = project.get_attachment(user, attachment_id) 

67 

68 return XAccelAttachmentResponse(attachment) 

69 

70 

71@http 

72def post_project_attachment(session, user, project_id, 

73 attachment_upload, attachment_description) -> serial.Id: 

74 ''' 

75 Upload a file as an attachment to the project. 

76 

77 This action can be performed for Projects at status Draft or Live 

78 ''' 

79 

80 project = fetch.project(session, project_id) 

81 validate.check(user, perms.PROJECT_EDIT_COSMETIC, project=project) 

82 attachment = attachments.save_project_attachment(session, project_id, user, attachment_upload, 

83 attachment_description) 

84 session.flush() 

85 changes = [ 

86 ("filename", attachment.filename, "") 

87 ] 

88 evt = AuditEvent.create( 

89 evt_types.PROJECT_ATTACHMENT_ADDED, 

90 project=project, user=user, object_id=attachment.id, change_list=changes 

91 ) 

92 session.add(evt) 

93 return {'id': attachment.id} 

94 

95 

96@http 

97def delete_project_attachment(session, user, project_id, attachment_id): 

98 ''' 

99 Delete the Project Attachment with the given ID 

100 

101 This action can be performed for Projects at status Draft or Live 

102 ''' 

103 project = fetch.project(session, project_id) 

104 validate.check(user, perms.PROJECT_EDIT_COSMETIC, project=project) 

105 attachment = project.get_attachment(user, attachment_id) 

106 filename = attachment.filename 

107 attachments.delete_project_attachment(session, attachment) 

108 changes = [ 

109 ("filename", "", filename) 

110 ] 

111 evt = AuditEvent.create( 

112 evt_types.PROJECT_ATTACHMENT_REMOVED, 

113 project=project, user=user, object_id=attachment_id, change_list=changes 

114 ) 

115 session.add(evt) 

116 

117 

118@http 

119def get_project_issue_attachment(session, 

120 user, 

121 project_id, 

122 issue_id, 

123 attachment_id) -> MimeTypes.ANY: 

124 ''' 

125 Download the Issue Attachment with the given ID 

126 ''' 

127 project = fetch.project(session, project_id) 

128 issue = fetch.issue(session, issue_id) 

129 validate.check(user, perms.ISSUE_VIEW_ANSWERS, issue=issue, 

130 project=project, section_id=project.section_id) 

131 attachment = issue.get_attachment(attachment_id) 

132 

133 return XAccelAttachmentResponse(attachment) 

134 

135 

136@http 

137def get_question_element_attachment(session, user, question_id, element_id) -> MimeTypes.ANY: 

138 ''' 

139 Download the Question Attachment associated with the given Element ID 

140 ''' 

141 q_instance = session.query(QuestionInstance).filter_by(id=question_id).one() 

142 q_element = q_instance.question_def.get_element(element_id) 

143 validate.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE, 

144 project=q_instance.project, 

145 section_id=q_instance.section_id) 

146 return XAccelAttachmentResponse(q_element.attachment) 

147 

148 

149@http 

150def get_answer_attachment(session, user, answer_id) -> MimeTypes.ANY: 

151 ''' 

152 Download the Answer Attachment with the given ID 

153 ''' 

154 answer = fetch.answer(session, answer_id) 

155 issue = answer.issue 

156 validate.check(user, perms.ISSUE_VIEW_ANSWERS, 

157 issue=issue, 

158 project=issue.project, 

159 section_id=answer.question_instance.section_id) 

160 

161 if not answer.attachment: 

162 raise ValueError(f'No attachment found for answer {answer_id}') 

163 

164 return XAccelAttachmentResponse(answer.attachment)