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 

14 

15publisher_gateway = PublisherGW("snap", api_publisher_session) 

16 

17models = flask.Blueprint( 

18 "models", 

19 __name__, 

20) 

21 

22 

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

24@login_required 

25@exchange_required 

26def get_models(store_id): 

27 """ 

28 Retrieves models associated with a given store ID. 

29 

30 Args: 

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

32 

33 Returns: 

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

35 and data. 

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 res["success"] = True 

207 

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)