Coverage for webapp/endpoints/models.py: 94%

1# Packages 

2import flask 

3from flask import make_response 

4from canonicalwebteam.exceptions import ( 

5 StoreApiResponseErrorList, 

6 StoreApiResourceNotFound, 

7) 

8from canonicalwebteam.store_api.publishergw import PublisherGW 

9 

10# Local 

11from webapp.decorators import login_required, exchange_required 

12from webapp.helpers import api_publisher_session, get_brand_id 

13 

14publisher_gateway = PublisherGW("snap", api_publisher_session) 

15 

16models = flask.Blueprint( 

17 "models", 

18 __name__, 

19) 

20 

21 

22@models.route("/api/store/<store_id>/models") 

23@login_required 

24@exchange_required 

25def get_models(store_id: str): 

26 """ 

27 Retrieves models associated with a given store ID. 

28 

29 Args: 

30 store_id (int): The ID of the store for which to retrieve models. 

31 

32 Returns: 

33 dict: A dictionary containing the response message, success status, 

34 and data. 

35 """ 

36 

37 res = {} 

38 try: 

39 models = publisher_gateway.get_store_models(flask.session, store_id) 

40 res["success"] = True 

41 res["data"] = models 

42 response = make_response(res, 200) 

43 response.cache_control.max_age = "3600" 

44 except StoreApiResponseErrorList as error_list: 

45 error_messages = [ 

46 f"{error.get('message', 'An error occurred')}" 

47 for error in error_list.errors 

48 ] 

49 if "unauthorized" in error_messages: 

50 res["message"] = "Store not found" 

51 else: 

52 res["message"] = " ".join(error_messages) 

53 res["success"] = False 

54 response = make_response(res, 500) 

55 

56 return response 

57 

58 

59@models.route("/api/store/<store_id>/models/<model_name>/policies") 

60@login_required 

61@exchange_required 

62def get_policies(store_id: str, model_name: str): 

63 """ 

64 Get the policies for a given store model. 

65 

66 Args: 

67 store_id (str): The ID of the store. 

68 model_name (str): The name of the model. 

69 

70 Returns: 

71 dict: A dictionary containing the response message and success 

72 """ 

73 brand_id = get_brand_id(flask.session, store_id) 

74 res = {} 

75 

76 try: 

77 policies = publisher_gateway.get_store_model_policies( 

78 flask.session, brand_id, model_name 

79 ) 

80 res["success"] = True 

81 res["data"] = policies 

82 response = make_response(res, 200) 

83 response.cache_control.max_age = "3600" 

84 return response 

85 except StoreApiResponseErrorList as error_list: 

86 res["success"] = False 

87 res["message"] = " ".join( 

88 [ 

89 f"{error.get('message', 'An error occurred')}" 

90 for error in error_list.errors 

91 ] 

92 ) 

93 except Exception: 

94 res["success"] = False 

95 res["message"] = "An error occurred" 

96 

97 return make_response(res, 500) 

98 

99 

100@models.route( 

101 "/api/store/<store_id>/models/<model_name>/policies", methods=["POST"] 

102) 

103@login_required 

104@exchange_required 

105def create_policy(store_id: str, model_name: str): 

106 """ 

107 Creat policy for a store model. 

108 

109 Args: 

110 store_id (str): The ID of the store. 

111 model_name (str): The name of the model. 

112 

113 Returns: 

114 dict: A dictionary containing the response message and success 

115 """ 

116 signing_key = flask.request.form.get("signing_key") 

117 res = {} 

118 try: 

119 signing_keys_data = publisher_gateway.get_store_signing_keys( 

120 flask.session, store_id 

121 ) 

122 signing_keys = [key["sha3-384"] for key in signing_keys_data] 

123 

124 if not signing_key: 

125 res["message"] = "Signing key required" 

126 res["success"] = False 

127 return make_response(res, 500) 

128 

129 if signing_key in signing_keys: 

130 publisher_gateway.create_store_model_policy( 

131 flask.session, store_id, model_name, signing_key 

132 ) 

133 res["success"] = True 

134 else: 

135 res["message"] = "Invalid signing key" 

136 res["success"] = False 

137 except StoreApiResponseErrorList as error_list: 

138 res["success"] = False 

139 res["message"] = error_list.errors[0]["message"] 

140 

141 if res["success"]: 

142 return make_response(res, 200) 

143 return make_response(res, 500) 

144 

145 

146@models.route( 

147 "/api/store/<store_id>/models/<model_name>/policies/<revision>", 

148 methods=["DELETE"], 

149) 

150@login_required 

151@exchange_required 

152def delete_policy(store_id: str, model_name: str, revision: str): 

153 res = {} 

154 try: 

155 response = publisher_gateway.delete_store_model_policy( 

156 flask.session, store_id, model_name, revision 

157 ) 

158 if response.status_code == 204: 

159 res = {"success": True} 

160 return make_response(res, 200) 

161 elif response.status_code == 404: 

162 res = {"success": False, "message": "Policy not found"} 

163 return make_response(res, 404) 

164 except StoreApiResponseErrorList as error_list: 

165 res["success"] = False 

166 res["message"] = error_list.errors[0]["message"] 

167 return make_response(res, 500) 

168 

169 

170@models.route("/api/store/<store_id>/models", methods=["POST"]) 

171@login_required 

172@exchange_required 

173def create_models(store_id: str): 

174 """ 

175 Create a model for a given store. 

176 

177 Args: 

178 store_id (str): The ID of the store. 

179 

180 Returns: 

181 dict: A dictionary containing the response message and success 

182 status. 

183 """ 

184 

185 # TO DO: Addn validation that name does not exist already 

186 

187 res = {} 

188 

189 try: 

190 name = flask.request.form.get("name") 

191 api_key = flask.request.form.get("api_key", "") 

192 

193 if len(name) > 128: 

194 res["message"] = "Name is too long. Limit 128 characters" 

195 res["success"] = False 

196 return make_response(res, 500) 

197 

198 if api_key and len(api_key) != 50 and not api_key.isalpha(): 

199 res["message"] = "Invalid API key" 

200 res["success"] = False 

201 return make_response(res, 500) 

202 

203 publisher_gateway.create_store_model( 

204 flask.session, store_id, name, api_key 

205 ) 

206 

207 res["success"] = True 

208 return make_response(res, 201) 

209 except StoreApiResponseErrorList as error_list: 

210 res["success"] = False 

211 messages = [ 

212 f"{error.get('message', 'An error occurred')}" 

213 for error in error_list.errors 

214 ] 

215 res["message"] = (" ").join(messages) 

216 

217 except Exception: 

218 res["success"] = False 

219 res["message"] = "An error occurred" 

220 

221 return make_response(res, 500) 

222 

223 

224@models.route("/api/store/<store_id>/models/<model_name>", methods=["PATCH"]) 

225@login_required 

226@exchange_required 

227def update_model(store_id: str, model_name: str): 

228 """ 

229 Update a model for a given store. 

230 

231 Args: 

232 store_id (str): The ID of the store. 

233 model_name (str): The name of the model. 

234 

235 Returns: 

236 dict: A dictionary containing the response message and success 

237 status. 

238 """ 

239 res = {} 

240 

241 try: 

242 api_key = flask.request.form.get("api_key", "") 

243 

244 if len(api_key) != 50 and not api_key.isalpha(): 

245 res["message"] = "Invalid API key" 

246 res["success"] = False 

247 return make_response(res, 500) 

248 

249 publisher_gateway.update_store_model( 

250 flask.session, store_id, model_name, api_key 

251 ) 

252 res["success"] = True 

253 

254 except StoreApiResponseErrorList as error_list: 

255 res["success"] = False 

256 res["message"] = error_list.errors[0]["message"] 

257 

258 except StoreApiResourceNotFound: 

259 res["success"] = False 

260 res["message"] = "Model not found" 

261 if res["success"]: 

262 return make_response(res, 200) 

263 return make_response(res, 500) 

264 

265 

266@models.route("/api/store/<store_id>/models/remodel-allowlist") 

267@login_required 

268@exchange_required 

269def get_remodel_allowlist(store_id: str): 

270 """ 

271 Retrieves remodels associated with a given store ID. 

272 

273 Args: 

274 store_id (int): The ID of the store for which to retrieve remodels. 

275 

276 Returns: 

277 dict: A dictionary containing the response message, success status, 

278 and data. 

279 """ 

280 

281 params = { 

282 # rename `cursor` to `page` on our side for clarity 

283 "cursor": flask.request.args.get("page"), 

284 "from-model": flask.request.args.get("from-model"), 

285 "page-size": flask.request.args.get("page-size"), 

286 } 

287 

288 res = {} 

289 

290 try: 

291 allowlist = publisher_gateway.get_remodel_allowlist( 

292 flask.session, store_id, params 

293 ) 

294 res["success"] = True 

295 res["data"] = allowlist 

296 response = make_response(res, 200) 

297 response.cache_control.max_age = "3600" 

298 except StoreApiResponseErrorList as error_list: 

299 error_messages = [ 

300 f"{error.get('message', 'An error occurred')}" 

301 for error in error_list.errors 

302 ] 

303 if "unauthorized" in error_messages: 

304 res["message"] = "Store not found" 

305 else: 

306 res["message"] = " ".join(error_messages) 

307 res["success"] = False 

308 response = make_response(res, 500) 

309 

310 return response 

311 

312 

313@models.route( 

314 "/api/store/<store_id>/models/remodel-allowlist", methods=["POST"] 

315) 

316@login_required 

317@exchange_required 

318def create_remodel_allowlist(store_id: str): 

319 """ 

320 Create a remodel allowlist for a given store. 

321 

322 Args: 

323 store_id (str): The ID of the store. 

324 

325 Returns: 

326 dict: A dictionary containing the response message and success 

327 status. 

328 """ 

329 res = {} 

330 

331 try: 

332 allowlist = flask.request.json 

333 publisher_gateway.create_remodel_allowlist( 

334 flask.session, store_id, allowlist 

335 ) 

336 

337 res["success"] = True 

338 return make_response(res, 201) 

339 except StoreApiResponseErrorList as error_list: 

340 res["success"] = False 

341 messages = [ 

342 f"{error.get('message', 'An error occurred')}" 

343 for error in error_list.errors 

344 ] 

345 res["message"] = " ".join(messages) 

346 return make_response(res, error_list.status_code) 

347 

348 except StoreApiResourceNotFound: 

349 res["success"] = False 

350 res["message"] = "Models not found" 

351 return make_response(res, 404) 

352 

353 except Exception: 

354 res["success"] = False 

355 res["message"] = "An error occurred" 

356 

357 return make_response(res, 500) 

358 

359 

360@models.route( 

361 "/api/store/<store_id>/models/remodel-allowlist", methods=["PATCH"] 

362) 

363@login_required 

364@exchange_required 

365def update_remodel_allowlist(store_id: str): 

366 """ 

367 Update a remodel allowlist for a given store. 

368 

369 Args: 

370 store_id (str): The ID of the store. 

371 

372 Returns: 

373 dict: A dictionary containing the response message and success 

374 status. 

375 """ 

376 res = {} 

377 

378 try: 

379 allowlist = flask.request.get_json(silent=True) 

380 if allowlist is None: 

381 res["success"] = False 

382 res["message"] = "Missing or invalid JSON payload" 

383 return make_response(res, 400) 

384 

385 # Only wrap in list if payload is a dict, not already a list 

386 allowlist_param = ( 

387 [allowlist] if isinstance(allowlist, dict) else allowlist 

388 ) 

389 publisher_gateway.update_remodel_allowlist( 

390 flask.session, store_id, allowlist_param 

391 ) 

392 

393 res["success"] = True 

394 return make_response(res, 200) 

395 except StoreApiResponseErrorList as error_list: 

396 res["success"] = False 

397 messages = [ 

398 f"{error.get('message', 'An error occurred')}" 

399 for error in error_list.errors 

400 ] 

401 res["message"] = " ".join(messages) 

402 return make_response(res, error_list.status_code) 

403 

404 except StoreApiResourceNotFound: 

405 res["success"] = False 

406 res["message"] = "Remodel allowlist not found" 

407 return make_response(res, 404) 

408 

409 except Exception: 

410 res["success"] = False 

411 res["message"] = "An error occurred" 

412 

413 return make_response(res, 500) 

414 

415 

416@models.route( 

417 "/api/store/<store_id>/models/remodel-allowlist", methods=["DELETE"] 

418) 

419@login_required 

420@exchange_required 

421def delete_remodel_allowlist(store_id: str): 

422 """ 

423 Delete a remodel allowlist for a given store. 

424 

425 Args: 

426 store_id (str): The ID of the store. 

427 

428 Returns: 

429 dict: A dictionary containing the response message and success 

430 status. 

431 """ 

432 res = {} 

433 

434 try: 

435 allowlist = flask.request.get_json(silent=True) 

436 if allowlist is None: 

437 res["success"] = False 

438 res["message"] = "Missing or invalid JSON payload" 

439 return make_response(res, 400) 

440 

441 # Only wrap in list if payload is a dict, not already a list 

442 allowlist_param = ( 

443 [allowlist] if isinstance(allowlist, dict) else allowlist 

444 ) 

445 publisher_gateway.delete_remodel_allowlist( 

446 flask.session, store_id, allowlist_param 

447 ) 

448 

449 res["success"] = True 

450 return make_response(res, 200) 

451 except StoreApiResponseErrorList as error_list: 

452 res["success"] = False 

453 messages = [ 

454 f"{error.get('message', 'An error occurred')}" 

455 for error in error_list.errors 

456 ] 

457 res["message"] = " ".join(messages) 

458 return make_response(res, error_list.status_code) 

459 

460 except StoreApiResourceNotFound: 

461 res["success"] = False 

462 res["message"] = "Remodel allowlist not found" 

463 return make_response(res, 404) 

464 

465 except Exception: 

466 res["success"] = False 

467 res["message"] = "An error occurred" 

468 

469 return make_response(res, 500) 

470 

471 

472@models.route("/api/store/<store_id>/models/<model_name>/serial-log") 

473@login_required 

474@exchange_required 

475def get_serial_log(store_id: str, model_name: str): 

476 params = { 

477 # rename `cursor` to `page` on our side for clarity 

478 "cursor": flask.request.args.get("page"), 

479 "start_time": flask.request.args.get("start-time"), 

480 "end_time": flask.request.args.get("end-time"), 

481 "page_size": flask.request.args.get("page-size"), 

482 } 

483 

484 res = {} 

485 

486 try: 

487 logs = publisher_gateway.get_store_model_serial_logs( 

488 flask.session, store_id, model_name, params 

489 ) 

490 res["data"] = logs 

491 res["success"] = True 

492 response = make_response(res, 200) 

493 except StoreApiResponseErrorList as error_list: 

494 error_messages = [ 

495 f"{error.get('message', 'An error occurred')}" 

496 for error in error_list.errors 

497 ] 

498 if "unauthorized" in error_messages: 

499 res["message"] = "Store not found" 

500 else: 

501 res["message"] = " ".join(error_messages) 

502 res["success"] = False 

503 response = make_response(res, 500) 

504 

505 return response