Coverage for postrfp / web / adaptors / core.py: 99%

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""" 

14 

15import re 

16 

17from postrfp.shared import fetch 

18from postrfp.model.questionnaire.b36 import to_b36 

19from postrfp.shared import serial 

20from postrfp.shared.pager import Pager 

21from postrfp.web.suxint.extractors import SchemaDocArg 

22from postrfp.model.audit import evt_types 

23 

24 

25from postrfp.web.suxint import ( 

26 ArgExtractor, 

27 PathArg, 

28 GetArg, 

29 GetArgSet, 

30 PostFileArg, 

31 PostArg, 

32 HeaderArg, 

33) 

34 

35 

36class PagerArg(GetArg): 

37 page_number = GetArg( 

38 "page", 

39 default=1, 

40 validator=lambda x: x > 0, 

41 doc="Page number of records to show, beginning from 1", 

42 ) 

43 page_size = GetArg( 

44 "page_size", 

45 default=20, 

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

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

48 ) 

49 

50 def extract(self, request): 

51 _pagenum = PagerArg.page_number.extract(request) 

52 PagerArg.page_number.validate(_pagenum) 

53 

54 _pagesize = PagerArg.page_size.extract(request) 

55 PagerArg.page_size.validate(_pagesize) 

56 

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

58 # MySQL Limit is a 64 bit integer 

59 raise ValueError("Either page number or page size is too large") 

60 

61 return Pager(page=_pagenum, page_size=_pagesize) 

62 

63 def update_openapi_path_object(self, path_object: dict): 

64 ArgExtractor.update_openapi_path_object(PagerArg.page_size, path_object) 

65 ArgExtractor.update_openapi_path_object(PagerArg.page_number, path_object) 

66 

67 

68def check_node_number(val): 

69 bare = val.strip(". ") 

70 if not bare: 

71 # root is an empty string 

72 return True 

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

74 return False 

75 for el in bare.split("."): 

76 # base 36 maximum for converting into two letters 

77 if int(el) >= 1295: 

78 return False 

79 

80 return True 

81 

82 

83# PATH ARGUMENTS 

84# These are used to extract values from the URL path 

85# the string value passed to the constructor is the name of the 

86# preceding component of the url path 

87# e.g. /api/v1/projects/{project_id}/ is defined by project_id = PathArg("project") 

88# the name of variable set here is the name of the variable 

89# expected by the receiving handler function 

90 

91section_id = PathArg("section") 

92 

93issue_id = PathArg("issue") 

94 

95question_id = PathArg("question") 

96 

97question_number = PathArg("question", converter=to_b36) 

98 

99score_id = PathArg("score") 

100 

101project_id = PathArg("project", arg_type="int") 

102 

103attachment_id = PathArg("attachment") 

104 

105answer_id = PathArg("answer") 

106 

107element_id = PathArg("element") 

108 

109event_id = PathArg("event") 

110 

111category_id = PathArg("category") 

112 

113reltype_id = PathArg("reltype") 

114 

115tag_id = PathArg("tag") 

116 

117note_id = PathArg("note") 

118 

119weightset_path_id = PathArg("weightset", arg_type="int") 

120 

121workflow_id = PathArg("workflow") 

122 

123entity_id = PathArg("entity") 

124 

125approval_id = PathArg("approval", arg_type="int") 

126 

127webhook_id = PathArg("webhook", arg_type="int") 

128 

129feed_id = PathArg("feed", arg_type="str") 

130 

131 

132# HEADER ARGUMENTS 

133 

134if_match = HeaderArg( 

135 "If-Match", doc="ETag of the content to be updated", required=True, arg_type="str" 

136) 

137 

138 

139# GET ARGUMENTS 

140 

141role_id = GetArg("roleId", arg_type="str") 

142 

143q_org_id = GetArg("orgId", arg_type="str", default=None) 

144 

145q_participant_id = GetArg("participantId", arg_type="str") 

146 

147with_ancestors = GetArg("ancestors", arg_type="boolean", default=False) 

148 

149search_doc = "See Path documentation for details of search special characters" 

150 

151org_type = GetArg( 

152 "orgType", arg_type="str", default="vendors", enum_values=("clients", "vendors") 

153) 

154 

155 

156def search_val(val): 

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

158 raise ValueError("Search term must be at least 2 characters long") 

159 return True 

160 

161 

162search_term = GetArg( 

163 "searchTerm", arg_type="str", validator=search_val, doc=search_doc, required=True 

164) 

165 

166search_options = GetArgSet( 

167 "options", 

168 arg_type="str", 

169 enum_values=("answers", "choices", "notes", "questions", "scoreComments"), 

170 doc="Defines which object types should be searched", 

171) 

172 

173project_statuses = GetArgSet( 

174 "projectStatus", 

175 required=False, 

176 arg_type="str", 

177 doc="Filter by provided Project Status values", 

178) 

179 

180project_sort = GetArg( 

181 "projectSort", 

182 arg_type="str", 

183 enum_values=("date_created", "date_published", "deadline", "title"), 

184 doc="Sort projects by dates or title", 

185 default="date_created", 

186) 

187 

188q_title = GetArg("qTitle", arg_type="str", doc="Search for string inside a title") 

189 

190q_name = GetArg("qName", arg_type="str", doc="Search for string inside a name") 

191 

192issue_sort = GetArg( 

193 "issueSort", 

194 arg_type="str", 

195 enum_values=("deadline", "submitted_date", "issue_date"), 

196 doc="Issue list sort order", 

197 default="issue_date", 

198) 

199 

200issue_status = GetArg( 

201 "projectStatus", 

202 arg_type="str", 

203 enum_values=( 

204 "Not Sent", 

205 "Opportunity", 

206 "Accepted", 

207 "Updateable", 

208 "Declined", 

209 "Submitted", 

210 "Retracted", 

211 ), 

212 doc="Filter by provided Project Status values", 

213) 

214 

215sort_order = GetArg("sort", arg_type="str", enum_values=("asc", "desc"), default="desc") 

216 

217 

218offset = GetArg("offset", arg_type="int", doc="For paging results") 

219 

220q_section_id = GetArg("sectionId") 

221 

222scoreset_docs = ( 

223 "Set to the user ID of a scoring user if Multiple Score Sets are " 

224 "enabled for this project" 

225) 

226 

227""" 

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

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

230""" 

231 

232 

233def default_scoreset(v): 

234 if v in (None, "__default__"): 

235 return "" 

236 return v 

237 

238 

239scoreset_id = GetArg( 

240 "scoresetId", 

241 arg_type="str", 

242 default="__default__", 

243 doc=scoreset_docs, 

244 converter=default_scoreset, 

245) 

246 

247 

248weightset_id = GetArg("weightsetId", arg_type="int") 

249 

250q_project_id = GetArg("projectId", arg_type="int", doc="ID of the Project") 

251 

252q_category_id = GetArg("categoryId", arg_type="int", doc="ID of the Project") 

253 

254q_issue_id = GetArg("issueId") 

255 

256q_workflow_id = GetArg("workflowId", arg_type="int") 

257 

258user_id = GetArg("userId", arg_type="str", required=True) 

259 

260node_doc = ( 

261 "Dotted String number e.g. 3.12.5 giving position of the question" 

262 " or section within the questionnaire" 

263) 

264 

265node_number = GetArg( 

266 "nodeNumber", 

267 arg_type="str", 

268 converter=to_b36, 

269 validator=check_node_number, 

270 doc=node_doc, 

271 default="", 

272) 

273 

274q_question_number = GetArg( 

275 "questionNumber", 

276 arg_type="str", 

277 converter=to_b36, 

278 doc='e.g. "5.2.9"', 

279 required=True, 

280) 

281 

282 

283def val_evt_type(val): 

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

285 raise ValueError(f"{val} is not a valid event type") 

286 return True 

287 

288 

289event_type = GetArg("eventType", arg_type="str", validator=val_evt_type) 

290entity = GetArg("entity", arg_type="str") 

291 

292restricted_users = GetArg("restricted", arg_type="boolean", default=False) 

293 

294pager = PagerArg("page", default=1) 

295 

296q_with_questions = GetArg("withQuestions", arg_type="boolean", default=True) 

297 

298issue_ids = GetArgSet( 

299 "issueIds", 

300 array_items_type="int", 

301 min_items=1, 

302 max_items=10, 

303 doc="ID values of Issues to query against", 

304) 

305 

306project_ids = GetArgSet( 

307 "projectIds", 

308 array_items_type="int", 

309 min_items=0, 

310 max_items=100, 

311 doc="ID values of Projects to query against", 

312) 

313 

314scoring_model_docs = ( 

315 "Type of scoring model to return. One of: " 

316 "Unweighted - no weighting applied, " 

317 "Direct - score multiplied by the score of the question or section, " 

318 "Absolute - score multiplied by absolute weight of the question or section" 

319) 

320 

321scoring_model = GetArg( 

322 "scoringModel", 

323 arg_type="str", 

324 enum_values=[v for v in serial.ScoringModel], 

325 default="Unweighted", 

326 doc=scoring_model_docs, 

327) 

328 

329# POST Args 

330 

331attachment_description = PostArg("description", arg_type="str") 

332el_id = PostArg("el_id", arg_type="int") 

333 

334# POST File Uploads 

335attachment_upload = PostFileArg("attachment_upload") 

336 

337data_upload = PostFileArg("data_upload", required=True) 

338 

339# VALIDATED JSON REQUEST BODY ARGUMENTS 

340 

341allocation_doc = SchemaDocArg(serial.AllocatedToList, as_dict=False) 

342 

343answers_doc = SchemaDocArg(serial.ElementAnswerList, as_dict=False) 

344 

345category_doc = SchemaDocArg(serial.NewCategory, as_dict=False) 

346 

347element_doc = SchemaDocArg(serial.qmodels.QElement) 

348 

349feed_doc = SchemaDocArg(serial.Datafeed, as_dict=False) 

350 

351ids_doc = SchemaDocArg(serial.IdList, as_dict=False) 

352 

353name_doc = SchemaDocArg(serial.ShortName, as_dict=False) 

354 

355import_answers_doc = SchemaDocArg(serial.ImportAnswers, as_dict=False) 

356 

357import_section_doc = SchemaDocArg(serial.SectionImportDoc, as_dict=False) 

358 

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

360 

361issue_status_doc = SchemaDocArg(serial.IssueStatus, as_dict=False) 

362 

363issue_workflow_doc = SchemaDocArg(serial.IssueUseWorkflow, as_dict=False) 

364 

365move_section_doc = SchemaDocArg(serial.MoveSection, as_dict=False) 

366 

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

368 

369new_issue_doc = SchemaDocArg(serial.NewIssue, as_dict=False) 

370 

371new_project_doc = SchemaDocArg(serial.NewProject) 

372 

373note_doc = SchemaDocArg(serial.ProjectNote, as_dict=False) 

374 

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

376 

377participants_list = SchemaDocArg(serial.UpdateParticipantList) 

378 

379perm_doc = SchemaDocArg(serial.ProjectPermission, as_dict=False) 

380 

381publish_doc = SchemaDocArg(serial.PublishProject) 

382 

383qdef_doc = SchemaDocArg(serial.QuestionDef, as_dict=False) 

384 

385relationship_doc = SchemaDocArg(serial.Relationship, as_dict=False) 

386 

387reltype_doc = SchemaDocArg(serial.RelationshipType, as_dict=False) 

388 

389replace_doc = SchemaDocArg(serial.TextReplace, as_dict=False) 

390 

391respondent_note_doc = SchemaDocArg(serial.RespondentNote, as_dict=False) 

392 

393score_doc = SchemaDocArg(serial.Score, as_dict=False) 

394 

395section_score_docs = SchemaDocArg(serial.SectionScoreDocs, as_dict=False) 

396 

397edit_sec_doc = SchemaDocArg(serial.EditableSection, as_dict=False) 

398 

399tag_assigns_doc = SchemaDocArg(serial.TagAssigns, as_dict=False) 

400 

401tag_doc = SchemaDocArg(serial.NewTag, as_dict=False) 

402 

403update_project_doc = SchemaDocArg(serial.UpdateableProject) 

404 

405user_doc = SchemaDocArg(serial.EditableUser, as_dict=False) 

406 

407user_id_doc = SchemaDocArg(serial.UserId, as_dict=False) 

408 

409watch_doc = SchemaDocArg(serial.TargetUser, as_dict=False) 

410 

411watchers_doc = SchemaDocArg(serial.TargetUserList) 

412 

413webhook_doc = SchemaDocArg(serial.Webhook, as_dict=False) 

414 

415weights_doc = SchemaDocArg(serial.WeightingsDoc, as_dict=False) 

416 

417weightset_doc = SchemaDocArg(serial.NewWeightSet, as_dict=False) 

418 

419child_nodes_doc = SchemaDocArg(serial.SectionChildNodes, as_dict=False) 

420 

421transition_doc = SchemaDocArg(serial.TransitionRequest, as_dict=False) 

422 

423workflow_doc = SchemaDocArg(serial.WorkflowSchema, as_dict=False) 

424 

425 

426class UserObject(ArgExtractor): 

427 swagger_in = "query" 

428 swagger_type = "string" 

429 

430 def extract(self, request): 

431 user_id = request.GET.get("targetUser", None) 

432 if not user_id: 

433 return None 

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

435 

436 

437target_user = UserObject("targetUser") 

438 

439 

440# NON USER DEFINED VALUES ASSOCIATED WITH THE REQUEST 

441 

442 

443def session(request): 

444 return request.session 

445 

446 

447def user(request): 

448 return request.user 

449 

450 

451def effective_user(request): 

452 return request.user