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

14 

15import re 

16 

17from rfpy.api import fetch 

18from rfpy.model.questionnaire import NumberString 

19from rfpy.web import serial, Pager 

20from rfpy.web.serial import SchemaDocArg 

21from rfpy.model.audit import evt_types 

22 

23 

24from rfpy.suxint import ArgExtractor, PathArg, GetArg, GetArgSet, PostFileArg, PostArg 

25 

26 

27class PagerArg(GetArg): 

28 page_number = GetArg( 

29 "page", 

30 default=1, 

31 validator=lambda x: x > 0, 

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

33 ) 

34 page_size = GetArg( 

35 "page_size", 

36 default=20, 

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

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

39 ) 

40 

41 def extract(self, request): 

42 _pagenum = PagerArg.page_number.extract(request) 

43 PagerArg.page_number.validate(_pagenum) 

44 

45 _pagesize = PagerArg.page_size.extract(request) 

46 PagerArg.page_size.validate(_pagesize) 

47 

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

49 # MySQL Limit is a 64 bit integer 

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

51 

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

53 

54 def update_openapi_path_object(self, path_object: dict): 

55 ArgExtractor.update_openapi_path_object(PagerArg.page_size, path_object) 

56 ArgExtractor.update_openapi_path_object(PagerArg.page_number, path_object) 

57 

58 

59def check_node_number(val): 

60 bare = val.strip(". ") 

61 if not bare: 

62 # root is an empty string 

63 return True 

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

65 return False 

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

67 # base 36 maximum for converting into two letters 

68 if int(el) >= 1295: 

69 return False 

70 

71 return True 

72 

73 

74# PATH ARGUMENTS 

75 

76 

77section_id = PathArg("section") 

78 

79issue_id = PathArg("issue") 

80 

81question_id = PathArg("question") 

82 

83question_number = PathArg("question", converter=NumberString.from_dotted) 

84 

85score_id = PathArg("score") 

86 

87project_id = PathArg("project") 

88 

89attachment_id = PathArg("attachment") 

90 

91answer_id = PathArg("answer") 

92 

93element_id = PathArg("element") 

94 

95watch_id = PathArg("watch") 

96 

97event_id = PathArg("event") 

98 

99category_id = PathArg("category") 

100 

101reltype_id = PathArg("reltype") 

102 

103tag_id = PathArg("tag") 

104 

105note_id = PathArg("note") 

106 

107weightset_path_id = PathArg("weightset") 

108 

109# GET ARGUMENTS 

110 

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

112 

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

114 

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

116 

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

118 

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

120 

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

122 

123org_type = GetArg( 

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

125) 

126 

127 

128def search_val(val): 

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

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

131 return True 

132 

133 

134search_term = GetArg( 

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

136) 

137 

138search_options = GetArgSet( 

139 "options", 

140 arg_type="str", 

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

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

143) 

144 

145project_statuses = GetArgSet( 

146 "projectStatus", 

147 required=False, 

148 arg_type="str", 

149 enum_values=("Draft", "Live", "Closed"), 

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

151) 

152 

153project_sort = GetArg( 

154 "projectSort", 

155 arg_type="str", 

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

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

158 default="date_created", 

159) 

160 

161q_project_title = GetArg( 

162 "pTitle", arg_type="str", doc="Search for string inside project title" 

163) 

164 

165issue_sort = GetArg( 

166 "issueSort", 

167 arg_type="str", 

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

169 doc="Issue list sort order", 

170 default="issue_date", 

171) 

172 

173issue_status = GetArg( 

174 "projectStatus", 

175 arg_type="str", 

176 enum_values=( 

177 "Not Sent", 

178 "Opportunity", 

179 "Accepted", 

180 "Updateable", 

181 "Declined", 

182 "Submitted", 

183 "Retracted", 

184 ), 

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

186) 

187 

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

189 

190 

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

192 

193q_question_id = GetArg("questionId") 

194 

195q_section_id = GetArg("sectionId") 

196 

197scoreset_docs = ( 

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

199 "enabled for this project" 

200) 

201 

202""" 

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

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

205""" 

206 

207 

208def default_scoreset(v): 

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

210 return "" 

211 return v 

212 

213 

214scoreset_id = GetArg( 

215 "scoresetId", 

216 arg_type="str", 

217 default="__default__", 

218 doc=scoreset_docs, 

219 converter=default_scoreset, 

220) 

221 

222 

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

224 

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

226 

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

228 

229q_issue_id = GetArg("issueId") 

230 

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

232 

233node_doc = ( 

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

235 " or section within the questionnaire" 

236) 

237 

238node_number = GetArg( 

239 "nodeNumber", 

240 arg_type="str", 

241 converter=NumberString.from_dotted, 

242 validator=check_node_number, 

243 doc=node_doc, 

244 default="", 

245) 

246 

247q_question_number = GetArg( 

248 "questionNumber", 

249 arg_type="str", 

250 converter=NumberString.from_dotted, 

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

252 required=True, 

253) 

254 

255 

256def val_evt_type(val): 

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

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

259 return True 

260 

261 

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

263 

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

265 

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

267 

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

269 

270issue_ids = GetArgSet( 

271 "issueIds", 

272 array_items_type="int", 

273 min_items=1, 

274 max_items=10, 

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

276) 

277 

278project_ids = GetArgSet( 

279 "issueId", 

280 array_items_type="int", 

281 min_items=1, 

282 max_items=100, 

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

284) 

285# POST Args 

286 

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

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

289 

290# POST File Uploads 

291attachment_upload = PostFileArg("attachment_upload") 

292 

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

294 

295# VALIDATED JSON REQUEST BODY ARGUMENTS 

296 

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

298 

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

300 

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

302 

303element_doc = SchemaDocArg(serial.qmodels.QElement) 

304 

305id_doc = SchemaDocArg(serial.Id) 

306 

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

308 

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

310 

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

312 

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

314 

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

316 

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

318 

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

320 

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

322 

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

324 

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

326 

327new_project_doc = SchemaDocArg(serial.NewProject) 

328 

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

330 

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

332 

333participants_list = SchemaDocArg(serial.UpdateParticipantList) 

334 

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

336 

337publish_doc = SchemaDocArg(serial.PublishProject) 

338 

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

340 

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

342 

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

344 

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

346 

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

348 

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

350 

351section_doc = SchemaDocArg(serial.Section) 

352 

353section_score_docs = SchemaDocArg(serial.SectionScoreDocs) 

354 

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

356 

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

358 

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

360 

361update_project_doc = SchemaDocArg(serial.UpdateableProject) 

362 

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

364 

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

366 

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

368 

369watchers_doc = SchemaDocArg(serial.TargetUserList) 

370 

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

372 

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

374 

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

376 

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

378 

379 

380class UserObject(ArgExtractor): 

381 swagger_in = "query" 

382 swagger_type = "string" 

383 

384 def extract(self, request): 

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

386 if not user_id: 

387 return None 

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

389 

390 

391target_user = UserObject("targetUser") 

392 

393 

394# NON USER DEFINED VALUES ASSOCIATED WITH THE REQUEST 

395 

396 

397def session(request): 

398 return request.session 

399 

400 

401def user(request): 

402 return request.user 

403 

404 

405def effective_user(request): 

406 return request.user