Coverage for rfpy/adaptors.py: 100%

1''' 

2Functions that accept http request objects (webob) 

3and return values to be consumed by api functions 

4 

5Purpose is to abstract api functions from http and 

6to perform common validation and marshalling jobs 

7 

8Request is expected to be HttpRequest, webob subclass defined in rfy.web.request 

9 

10ORM model instances returned by functions here should be assumed to be validated as 

11being visible to the calling user. Primitives (e.g. ID values like section_id) are not 

12validated. 

13''' 

14import re 

15 

16from rfpy.api import fetch 

17from rfpy.model.questionnaire import NumberString 

18from rfpy.web import serial, Pager 

19from rfpy.web.serial import SchemaDocArg 

20from rfpy.model.audit import evt_types 

21 

22 

23from rfpy.suxint import ( 

24 ArgExtractor, PathArg, GetArg, GetArgSet, PostFileArg, PostArg 

25) 

26 

27 

28class PagerArg(GetArg): 

29 

30 page_number = GetArg( 

31 'page', 

32 default=1, 

33 validator=lambda x: x > 0, 

34 doc='Page number of records to show, beginning from 1' 

35 ) 

36 page_size = GetArg( 

37 'page_size', 

38 default=20, 

39 validator=lambda x: 0 < x < 201, 

40 doc="The number of records to return in a single page, maximum 200, minimum 1" 

41 ) 

42 

43 def extract(self, request): 

44 _pagenum = PagerArg.page_number.extract(request) 

45 PagerArg.page_number.validate(_pagenum) 

46 

47 _pagesize = PagerArg.page_size.extract(request) 

48 PagerArg.page_size.validate(_pagesize) 

49 

50 if (_pagenum * _pagesize) > 2**64: 

51 # MySQL Limit is a 64 bit integer 

52 raise ValueError('Either page number or page size is too large') 

53 

54 return Pager(page=_pagenum, pagesize=_pagesize) 

55 

56 def update_openapi_path_object(self, path_object: dict): 

57 ArgExtractor.update_openapi_path_object(PagerArg.page_size, path_object) 

58 ArgExtractor.update_openapi_path_object(PagerArg.page_number, path_object) 

59 

60 

61def check_node_number(val): 

62 bare = val.strip('. ') 

63 if not bare: 

64 # root is an empty string 

65 return True 

66 if re.match(r'^(\d+\.?)+$', bare) is None: 

67 return False 

68 for el in bare.split('.'): 

69 # base 36 maximum for converting into two letters 

70 if int(el) >= 1295: 

71 return False 

72 

73 return True 

74 

75# PATH ARGUMENTS 

76 

77 

78section_id = PathArg('section') 

79 

80issue_id = PathArg('issue') 

81 

82question_id = PathArg('question') 

83 

84question_number = PathArg('question', converter=NumberString.from_dotted) 

85 

86score_id = PathArg('score') 

87 

88project_id = PathArg('project') 

89 

90attachment_id = PathArg('attachment') 

91 

92answer_id = PathArg('answer') 

93 

94element_id = PathArg('element') 

95 

96watch_id = PathArg('watch') 

97 

98event_id = PathArg('event') 

99 

100category_id = PathArg('category') 

101 

102reltype_id = PathArg('reltype') 

103 

104tag_id = PathArg('tag') 

105 

106note_id = PathArg('note') 

107 

108weightset_path_id = PathArg('weightset') 

109 

110# GET ARGUMENTS 

111 

112role_id = GetArg('roleId', arg_type='str') 

113 

114q_org_id = GetArg('orgId', arg_type='str', default=None) 

115 

116q_participant_id = GetArg('participantId', arg_type='str') 

117 

118q_debug = GetArg('debug', arg_type='str') 

119 

120with_qdefs = GetArg('question_bodies', arg_type='boolean', default=False) 

121 

122with_ancestors = GetArg('ancestors', arg_type='boolean', default=False) 

123 

124search_doc = 'See Path documentation for details of search special characters' 

125 

126org_type = GetArg( 

127 'orgType', 

128 arg_type='str', 

129 default='vendors', 

130 enum_values=('clients', 'vendors') 

131) 

132 

133 

134def search_val(val): 

135 if val is None or len(val) < 2: 

136 raise ValueError('Search term must be at least 2 characters long') 

137 return True 

138 

139 

140search_term = GetArg( 

141 'searchTerm', 

142 arg_type='str', 

143 validator=search_val, 

144 doc=search_doc, 

145 required=True) 

146 

147search_options = GetArgSet( 

148 'options', 

149 arg_type='str', 

150 enum_values=('answers', 'choices', 'notes', 'questions', 'scoreComments'), 

151 doc='Defines which object types should be searched' 

152) 

153 

154project_statuses = GetArgSet( 

155 'projectStatus', 

156 required=False, 

157 arg_type='str', 

158 enum_values=('Draft', 'Live', 'Closed'), 

159 doc='Filter by provided Project Status values' 

160) 

161 

162project_sort = GetArg( 

163 'projectSort', 

164 arg_type='str', 

165 enum_values=('date_created', 'date_published', 'deadline', 'title'), 

166 doc='Sort projects by dates or title', 

167 default='date_created' 

168) 

169 

170q_project_title = GetArg( 

171 'pTitle', 

172 arg_type='str', 

173 doc='Search for string inside project title' 

174) 

175 

176issue_sort = GetArg( 

177 'issueSort', 

178 arg_type='str', 

179 enum_values=('deadline', 'submitted_date', 'issue_date'), 

180 doc='Issue list sort order', 

181 default='issue_date' 

182) 

183 

184issue_status = GetArg( 

185 'projectStatus', 

186 arg_type='str', 

187 enum_values=( 

188 'Not Sent', 'Opportunity', 'Accepted', 'Updateable', 'Declined', 'Submitted', 'Retracted' 

189 ), 

190 doc='Filter by provided Project Status values' 

191) 

192 

193sort_order = GetArg('sort', arg_type='str', enum_values=('asc', 'desc'), default='desc') 

194 

195 

196offset = GetArg('offset', arg_type='int', doc='For paging results') 

197 

198q_question_id = GetArg('questionId') 

199 

200q_section_id = GetArg('sectionId') 

201 

202scoreset_docs = ('Set to the user ID of a scoring user if Multiple Score Sets are ' 

203 'enabled for this project') 

204 

205''' 

206scoreset_id is the ID of the user that created a score set. __default__ is used when returning 

207data to indicate the default user. For db queries an empty string is used. 

208''' 

209 

210 

211def default_scoreset(v): 

212 if v in (None, '__default__'): 

213 return '' 

214 return v 

215 

216 

217scoreset_id = GetArg('scoresetId', arg_type='str', default='__default__', 

218 doc=scoreset_docs, converter=default_scoreset) 

219 

220 

221weightset_id = GetArg('weightsetId', arg_type='int') 

222 

223q_project_id = GetArg('projectId', arg_type='int', doc='ID of the Project') 

224 

225q_category_id = GetArg('categoryId', arg_type='int', doc='ID of the Project') 

226 

227q_issue_id = GetArg('issueId') 

228 

229user_id = GetArg('userId', arg_type="str") 

230 

231node_doc = ('Dotted String number e.g. 3.12.5 giving position of the question' 

232 ' or section within the questionnaire') 

233 

234node_number = GetArg('nodeNumber', 

235 arg_type='str', 

236 converter=NumberString.from_dotted, 

237 validator=check_node_number, 

238 doc=node_doc, 

239 default='') 

240 

241q_question_number = GetArg('questionNumber', arg_type='str', 

242 converter=NumberString.from_dotted, 

243 doc='e.g. "5.2.9"', required=True) 

244 

245 

246def val_evt_type(val): 

247 if val is not None and not hasattr(evt_types, val): 

248 raise ValueError(f'{val} is not a valid event type') 

249 return True 

250 

251 

252event_type = GetArg('eventType', arg_type='str', validator=val_evt_type) 

253 

254restricted_users = GetArg('restricted', arg_type='boolean', default=False) 

255 

256pager = PagerArg('page', default=1) 

257 

258q_with_questions = GetArg('withQuestions', arg_type='boolean', default=True) 

259 

260issue_ids = GetArgSet('issueIds', array_items_type='int', min_items=1, max_items=10, 

261 doc='ID values of Issues to query against') 

262 

263project_ids = GetArgSet('issueId', array_items_type='int', min_items=1, max_items=100, 

264 doc='ID values of Projects to query against') 

265# POST Args 

266 

267attachment_description = PostArg('description', arg_type='str') 

268el_id = PostArg('el_id', arg_type='int') 

269 

270# POST File Uploads 

271attachment_upload = PostFileArg('attachment_upload') 

272 

273data_upload = PostFileArg('data_upload', required=True) 

274 

275# VALIDATED JSON REQUEST BODY ARGUMENTS 

276 

277allocation_doc = SchemaDocArg(serial.AllocatedToList) 

278 

279answers_doc = SchemaDocArg(serial.ElementAnswerList) 

280 

281category_doc = SchemaDocArg(serial.NewCategory) 

282 

283element_doc = SchemaDocArg(serial.qmodels.QElement) 

284 

285id_doc = SchemaDocArg(serial.Id) 

286 

287ids_doc = SchemaDocArg(serial.IdList) 

288 

289name_doc = SchemaDocArg(serial.ShortName) 

290 

291import_answers_doc = SchemaDocArg(serial.ImportAnswers) 

292 

293import_section_doc = SchemaDocArg(serial.SectionImportDoc) 

294 

295issue_doc = SchemaDocArg(serial.UpdateableIssue, exclude_unset=True) 

296 

297issue_status_doc = SchemaDocArg(serial.IssueStatus) 

298 

299issue_workflow_doc = SchemaDocArg(serial.IssueUseWorkflow) 

300 

301move_section_doc = SchemaDocArg(serial.MoveSection) 

302 

303new_client_doc = SchemaDocArg(serial.NewClient, as_dict=False) 

304 

305new_issue_doc = SchemaDocArg(serial.NewIssue) 

306 

307new_project_doc = SchemaDocArg(serial.NewProject) 

308 

309note_doc = SchemaDocArg(serial.ProjectNote) 

310 

311org_doc = SchemaDocArg(serial.Organisation, as_dict=False) 

312 

313participants_list = SchemaDocArg(serial.UpdateParticipantList) 

314 

315perm_doc = SchemaDocArg(serial.ProjectPermission) 

316 

317publish_doc = SchemaDocArg(serial.PublishProject) 

318 

319qdef_doc = SchemaDocArg(serial.QuestionDef) 

320 

321relationship_doc = SchemaDocArg(serial.Relationship) 

322 

323reltype_doc = SchemaDocArg(serial.RelationshipType) 

324 

325replace_doc = SchemaDocArg(serial.TextReplace) 

326 

327respondent_note_doc = SchemaDocArg(serial.RespondentNote) 

328 

329score_doc = SchemaDocArg(serial.Score) 

330 

331section_doc = SchemaDocArg(serial.Section) 

332 

333section_score_docs = SchemaDocArg(serial.SectionScoreDocs) 

334 

335edit_sec_doc = SchemaDocArg(serial.EditableSection) 

336 

337tag_assigns_doc = SchemaDocArg(serial.TagAssigns) 

338 

339tag_doc = SchemaDocArg(serial.NewTag) 

340 

341update_project_doc = SchemaDocArg(serial.UpdateableProject) 

342 

343user_doc = SchemaDocArg(serial.EditableUser) 

344 

345user_id_doc = SchemaDocArg(serial.UserId) 

346 

347watch_doc = SchemaDocArg(serial.TargetUser) 

348 

349watchers_doc = SchemaDocArg(serial.TargetUserList) 

350 

351webhook_doc = SchemaDocArg(serial.NewWebhook, as_dict=False) 

352 

353weights_doc = SchemaDocArg(serial.WeightingsDoc) 

354 

355weightset_doc = SchemaDocArg(serial.NewWeightSet) 

356 

357child_nodes_doc = SchemaDocArg(serial.SectionChildNodes) 

358 

359 

360class UserObject(ArgExtractor): 

361 swagger_in = 'query' 

362 swagger_type = 'string' 

363 

364 def extract(self, request): 

365 user_id = request.GET.get('targetUser', None) 

366 if not user_id: 

367 return None 

368 return fetch.user(request.session, user_id) 

369 

370 

371target_user = UserObject('targetUser') 

372 

373 

374# NON USER DEFINED VALUES ASSOCIATED WITH THE REQUEST 

375 

376 

377def session(request): 

378 return request.session 

379 

380 

381def user(request): 

382 return request.user 

383 

384 

385def effective_user(request): 

386 return request.user