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) 

33 

34 

35class PagerArg(GetArg): 

36 page_number = GetArg( 

37 "page", 

38 default=1, 

39 validator=lambda x: x > 0, 

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

41 ) 

42 page_size = GetArg( 

43 "page_size", 

44 default=20, 

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

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

47 ) 

48 

49 def extract(self, request): 

50 _pagenum = PagerArg.page_number.extract(request) 

51 PagerArg.page_number.validate(_pagenum) 

52 

53 _pagesize = PagerArg.page_size.extract(request) 

54 PagerArg.page_size.validate(_pagesize) 

55 

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

57 # MySQL Limit is a 64 bit integer 

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

59 

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

61 

62 def update_openapi_path_object(self, path_object: dict): 

63 ArgExtractor.update_openapi_path_object(PagerArg.page_size, path_object) 

64 ArgExtractor.update_openapi_path_object(PagerArg.page_number, path_object) 

65 

66 

67def check_node_number(val): 

68 bare = val.strip(". ") 

69 if not bare: 

70 # root is an empty string 

71 return True 

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

73 return False 

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

75 # base 36 maximum for converting into two letters 

76 if int(el) >= 1295: 

77 return False 

78 

79 return True 

80 

81 

82# PATH ARGUMENTS 

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

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

85# preceding component of the url path 

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

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

88# expected by the receiving handler function 

89 

90section_id = PathArg("section") 

91 

92issue_id = PathArg("issue") 

93 

94question_id = PathArg("question") 

95 

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

97 

98score_id = PathArg("score") 

99 

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

101 

102attachment_id = PathArg("attachment") 

103 

104answer_id = PathArg("answer") 

105 

106element_id = PathArg("element") 

107 

108watch_id = PathArg("watch") 

109 

110event_id = PathArg("event") 

111 

112category_id = PathArg("category") 

113 

114reltype_id = PathArg("reltype") 

115 

116tag_id = PathArg("tag") 

117 

118note_id = PathArg("note") 

119 

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

121 

122workflow_id = PathArg("workflow") 

123 

124entity_id = PathArg("entity") 

125 

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

127 

128# GET ARGUMENTS 

129 

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

131 

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

133 

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

135 

136q_debug = GetArg("debug", arg_type="str") 

137 

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

139 

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

141 

142org_type = GetArg( 

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

144) 

145 

146 

147def search_val(val): 

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

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

150 return True 

151 

152 

153search_term = GetArg( 

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

155) 

156 

157search_options = GetArgSet( 

158 "options", 

159 arg_type="str", 

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

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

162) 

163 

164project_statuses = GetArgSet( 

165 "projectStatus", 

166 required=False, 

167 arg_type="str", 

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

169) 

170 

171project_sort = GetArg( 

172 "projectSort", 

173 arg_type="str", 

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

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

176 default="date_created", 

177) 

178 

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

180 

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

182 

183issue_sort = GetArg( 

184 "issueSort", 

185 arg_type="str", 

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

187 doc="Issue list sort order", 

188 default="issue_date", 

189) 

190 

191issue_status = GetArg( 

192 "projectStatus", 

193 arg_type="str", 

194 enum_values=( 

195 "Not Sent", 

196 "Opportunity", 

197 "Accepted", 

198 "Updateable", 

199 "Declined", 

200 "Submitted", 

201 "Retracted", 

202 ), 

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

204) 

205 

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

207 

208 

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

210 

211q_question_id = GetArg("questionId") 

212 

213q_section_id = GetArg("sectionId") 

214 

215scoreset_docs = ( 

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

217 "enabled for this project" 

218) 

219 

220""" 

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

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

223""" 

224 

225 

226def default_scoreset(v): 

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

228 return "" 

229 return v 

230 

231 

232scoreset_id = GetArg( 

233 "scoresetId", 

234 arg_type="str", 

235 default="__default__", 

236 doc=scoreset_docs, 

237 converter=default_scoreset, 

238) 

239 

240 

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

242 

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

244 

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

246 

247q_issue_id = GetArg("issueId") 

248 

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

250 

251node_doc = ( 

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

253 " or section within the questionnaire" 

254) 

255 

256node_number = GetArg( 

257 "nodeNumber", 

258 arg_type="str", 

259 converter=to_b36, 

260 validator=check_node_number, 

261 doc=node_doc, 

262 default="", 

263) 

264 

265q_question_number = GetArg( 

266 "questionNumber", 

267 arg_type="str", 

268 converter=to_b36, 

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

270 required=True, 

271) 

272 

273 

274def val_evt_type(val): 

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

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

277 return True 

278 

279 

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

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

282 

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

284 

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

286 

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

288 

289issue_ids = GetArgSet( 

290 "issueIds", 

291 array_items_type="int", 

292 min_items=1, 

293 max_items=10, 

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

295) 

296 

297project_ids = GetArgSet( 

298 "projectIds", 

299 array_items_type="int", 

300 min_items=0, 

301 max_items=100, 

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

303) 

304 

305scoring_model_docs = ( 

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

307 "Unweighted - no weighting applied, " 

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

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

310) 

311 

312scoring_model = GetArg( 

313 "scoringModel", 

314 arg_type="str", 

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

316 default="Unweighted", 

317 doc=scoring_model_docs, 

318) 

319 

320# POST Args 

321 

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

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

324 

325# POST File Uploads 

326attachment_upload = PostFileArg("attachment_upload") 

327 

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

329 

330# VALIDATED JSON REQUEST BODY ARGUMENTS 

331 

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

333 

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

335 

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

337 

338element_doc = SchemaDocArg(serial.qmodels.QElement) 

339 

340id_doc = SchemaDocArg(serial.Id) 

341 

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

343 

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

345 

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

347 

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

349 

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

351 

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

353 

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

355 

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

357 

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

359 

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

361 

362new_project_doc = SchemaDocArg(serial.NewProject) 

363 

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

365 

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

367 

368participants_list = SchemaDocArg(serial.UpdateParticipantList) 

369 

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

371 

372publish_doc = SchemaDocArg(serial.PublishProject) 

373 

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

375 

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

377 

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

379 

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

381 

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

383 

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

385 

386section_doc = SchemaDocArg(serial.Section) 

387 

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

389 

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

391 

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

393 

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

395 

396update_project_doc = SchemaDocArg(serial.UpdateableProject) 

397 

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

399 

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

401 

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

403 

404watchers_doc = SchemaDocArg(serial.TargetUserList) 

405 

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

407 

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

409 

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

411 

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

413 

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

415 

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

417 

418 

419class UserObject(ArgExtractor): 

420 swagger_in = "query" 

421 swagger_type = "string" 

422 

423 def extract(self, request): 

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

425 if not user_id: 

426 return None 

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

428 

429 

430target_user = UserObject("targetUser") 

431 

432 

433# NON USER DEFINED VALUES ASSOCIATED WITH THE REQUEST 

434 

435 

436def session(request): 

437 return request.session 

438 

439 

440def user(request): 

441 return request.user 

442 

443 

444def effective_user(request): 

445 return request.user