Coverage for tests/publisher/endpoint_testing.py: 92%

1from unittest.mock import patch 

2 

3import requests 

4 

5import responses 

6from flask_testing import TestCase 

7from webapp.app import create_app 

8from webapp.authentication import get_authorization_header 

9 

10# Make sure tests fail on stray responses. 

11responses.mock.assert_all_requests_are_fired = True 

12 

13 

14class BaseTestCases: 

15 """ 

16 This class has a set of test classes that should be inherited by endpoint 

17 that have authentication. 

18 

19 It is also used to avoid unittest to run this tests file. 

20 """ 

21 

22 class BaseAppTesting(TestCase): 

23 def setUp(self, snap_name, api_url, endpoint_url): 

24 self.snap_name = snap_name 

25 self.api_url = api_url 

26 self.endpoint_url = endpoint_url 

27 

28 def tearDown(self): 

29 responses.reset() 

30 

31 def create_app(self): 

32 app = create_app(testing=True) 

33 app.secret_key = "secret_key" 

34 app.config["WTF_CSRF_METHODS"] = [] 

35 

36 return app 

37 

38 def _get_location(self): 

39 return "{}".format(self.endpoint_url) 

40 

41 def _log_in(self, client): 

42 """Emulates test client login in the store. 

43 

44 Fill current session with `openid` and `macaroon_exchanged`. 

45 

46 Return the expected `Authorization` header for further verification 

47 in API requests. 

48 """ 

49 exchanged_macaroon = "test-exchanged-macaroon" 

50 

51 with client.session_transaction() as s: 

52 s["publisher"] = { 

53 "image": None, 

54 "nickname": "Toto", 

55 "fullname": "El Toto", 

56 "email": "testing@testing.com", 

57 "stores": [], 

58 } 

59 s["macaroon_exchanged"] = exchanged_macaroon 

60 

61 return get_authorization_header(exchanged_macaroon) 

62 

63 def check_call_by_api_url(self, calls): 

64 found = False 

65 for called in calls: 

66 if self.api_url == called.request.url: 

67 found = True 

68 self.assertEqual( 

69 self.authorization, 

70 called.request.headers.get("Authorization"), 

71 ) 

72 

73 assert found 

74 

75 class EndpointLoggedOut(BaseAppTesting): 

76 def setUp(self, snap_name, endpoint_url, method_endpoint="GET"): 

77 self.method_endpoint = method_endpoint 

78 super().setUp(snap_name, None, endpoint_url) 

79 

80 def test_access_not_logged_in(self): 

81 if self.method_endpoint == "GET": 

82 response = self.client.get(self.endpoint_url) 

83 else: 

84 response = self.client.post(self.endpoint_url, data={}) 

85 

86 self.assertEqual(302, response.status_code) 

87 self.assertEqual( 

88 "/login?next={}".format(self.endpoint_url), 

89 response.location, 

90 ) 

91 

92 class EndpointLoggedIn(BaseAppTesting): 

93 def setUp( 

94 self, 

95 snap_name, 

96 endpoint_url, 

97 api_url, 

98 method_endpoint="GET", 

99 method_api="GET", 

100 data=None, 

101 json=None, 

102 ): 

103 super().setUp( 

104 snap_name=snap_name, api_url=api_url, endpoint_url=endpoint_url 

105 ) 

106 

107 # Stub the blueprint-level "has releases" gate so each test does 

108 # not need to mock the extra dashboard call. Tests that exercise 

109 # the gate directly can override this in their own setUp. 

110 self._release_history_patcher = patch( 

111 "webapp.decorators._dashboard.snap_release_history", 

112 return_value={"revisions": [{"revision": 1}]}, 

113 ) 

114 self._release_history_patcher.start() 

115 self.addCleanup(self._release_history_patcher.stop) 

116 

117 self.method_endpoint = method_endpoint 

118 self.method_api = method_api 

119 self.data = data 

120 self.json = json 

121 self.authorization = self._log_in(self.client) 

122 

123 @responses.activate 

124 def test_timeout(self): 

125 responses.add( 

126 responses.Response( 

127 method=self.method_api, 

128 url=self.api_url, 

129 body=requests.exceptions.Timeout(), 

130 status=504, 

131 ) 

132 ) 

133 

134 if self.method_endpoint == "GET": 

135 response = self.client.get(self.endpoint_url) 

136 else: 

137 if self.data: 

138 response = self.client.post( 

139 self.endpoint_url, data=self.data 

140 ) 

141 else: 

142 response = self.client.post( 

143 self.endpoint_url, json=self.json 

144 ) 

145 

146 self.check_call_by_api_url(responses.calls) 

147 

148 assert response.status_code == 504 

149 

150 @responses.activate 

151 def test_connection_error(self): 

152 responses.add( 

153 responses.Response( 

154 method=self.method_api, 

155 url=self.api_url, 

156 body=requests.exceptions.ConnectionError(), 

157 status=500, 

158 ) 

159 ) 

160 

161 if self.method_endpoint == "GET": 

162 response = self.client.get(self.endpoint_url) 

163 else: 

164 if self.data: 

165 response = self.client.post( 

166 self.endpoint_url, data=self.data 

167 ) 

168 else: 

169 response = self.client.post( 

170 self.endpoint_url, json=self.json 

171 ) 

172 

173 self.check_call_by_api_url(responses.calls) 

174 

175 assert response.status_code == 502 

176 

177 @responses.activate 

178 def test_broken_json(self): 

179 # To test this I return no json from the server, this makes the 

180 # call to the function response.json() raise a ValueError exception 

181 responses.add( 

182 responses.Response( 

183 method=self.method_api, url=self.api_url, status=500 

184 ) 

185 ) 

186 

187 if self.method_endpoint == "GET": 

188 response = self.client.get(self.endpoint_url) 

189 else: 

190 if self.data: 

191 response = self.client.post( 

192 self.endpoint_url, data=self.data 

193 ) 

194 else: 

195 response = self.client.post( 

196 self.endpoint_url, json=self.json 

197 ) 

198 

199 self.check_call_by_api_url(responses.calls) 

200 

201 assert response.status_code == 502 

202 

203 @responses.activate 

204 def test_unknown_error(self): 

205 responses.add( 

206 responses.Response( 

207 method=self.method_api, 

208 url=self.api_url, 

209 json={}, 

210 status=500, 

211 ) 

212 ) 

213 

214 if self.method_endpoint == "GET": 

215 response = self.client.get(self.endpoint_url) 

216 else: 

217 if self.data: 

218 response = self.client.post( 

219 self.endpoint_url, data=self.data 

220 ) 

221 else: 

222 response = self.client.post( 

223 self.endpoint_url, json=self.json 

224 ) 

225 

226 self.check_call_by_api_url(responses.calls) 

227 

228 assert response.status_code == 502 

229 

230 @responses.activate 

231 def test_expired_macaroon(self): 

232 responses.add( 

233 responses.Response( 

234 method=self.method_api, 

235 url=self.api_url, 

236 json={}, 

237 status=401, 

238 headers={"WWW-Authenticate": "Macaroon needs_refresh=1"}, 

239 ) 

240 ) 

241 if self.method_endpoint == "GET": 

242 response = self.client.get(self.endpoint_url) 

243 else: 

244 if self.data: 

245 response = self.client.post( 

246 self.endpoint_url, data=self.data 

247 ) 

248 else: 

249 response = self.client.post( 

250 self.endpoint_url, json=self.json 

251 ) 

252 

253 assert response.status_code == 302 

254 assert response.location == (f"/login?next={self.endpoint_url}") 

255 

256 class EndpointLoggedInErrorHandling(EndpointLoggedIn): 

257 @responses.activate 

258 def test_error_4xx(self): 

259 payload = {"error_list": []} 

260 responses.add( 

261 responses.Response( 

262 method=self.method_api, 

263 url=self.api_url, 

264 json=payload, 

265 status=400, 

266 ) 

267 ) 

268 

269 if self.method_endpoint == "GET": 

270 response = self.client.get(self.endpoint_url) 

271 else: 

272 if self.data: 

273 response = self.client.post( 

274 self.endpoint_url, data=self.data 

275 ) 

276 else: 

277 response = self.client.post( 

278 self.endpoint_url, json=self.json 

279 ) 

280 

281 self.check_call_by_api_url(responses.calls) 

282 

283 assert response.status_code == 502 

284 

285 @responses.activate 

286 def test_custom_error(self): 

287 payload = { 

288 "error_list": [ 

289 {"code": "error-code1"}, 

290 {"code": "error-code2"}, 

291 ] 

292 } 

293 responses.add( 

294 responses.Response( 

295 method=self.method_api, 

296 url=self.api_url, 

297 json=payload, 

298 status=400, 

299 ) 

300 ) 

301 

302 if self.method_endpoint == "GET": 

303 response = self.client.get(self.endpoint_url) 

304 else: 

305 if self.data: 

306 response = self.client.post( 

307 self.endpoint_url, data=self.data 

308 ) 

309 else: 

310 response = self.client.post( 

311 self.endpoint_url, json=self.json 

312 ) 

313 

314 self.check_call_by_api_url(responses.calls) 

315 

316 assert response.status_code == 502 

317 

318 @responses.activate 

319 def test_account_not_signed_agreement_logged_in(self): 

320 payload = { 

321 "error_list": [ 

322 { 

323 "code": "user-not-ready", 

324 "message": "has not signed agreement", 

325 } 

326 ] 

327 } 

328 responses.add( 

329 responses.Response( 

330 method=self.method_api, 

331 url=self.api_url, 

332 json=payload, 

333 status=403, 

334 ) 

335 ) 

336 

337 if self.method_endpoint == "GET": 

338 response = self.client.get(self.endpoint_url) 

339 else: 

340 if self.data: 

341 response = self.client.post( 

342 self.endpoint_url, data=self.data 

343 ) 

344 else: 

345 response = self.client.post( 

346 self.endpoint_url, json=self.json 

347 ) 

348 

349 self.check_call_by_api_url(responses.calls) 

350 

351 self.assertEqual(302, response.status_code) 

352 self.assertEqual("/account/agreement", response.location) 

353 

354 @responses.activate 

355 def test_account_no_username_logged_in(self): 

356 payload = { 

357 "error_list": [ 

358 { 

359 "code": "user-not-ready", 

360 "message": "missing store username", 

361 } 

362 ] 

363 } 

364 responses.add( 

365 responses.Response( 

366 method=self.method_api, 

367 url=self.api_url, 

368 json=payload, 

369 status=403, 

370 ) 

371 ) 

372 

373 if self.method_endpoint == "GET": 

374 response = self.client.get(self.endpoint_url) 

375 else: 

376 if self.data: 

377 response = self.client.post( 

378 self.endpoint_url, data=self.data 

379 ) 

380 else: 

381 response = self.client.post( 

382 self.endpoint_url, json=self.json 

383 ) 

384 

385 self.check_call_by_api_url(responses.calls) 

386 

387 self.assertEqual(302, response.status_code) 

388 self.assertEqual("/account/username", response.location)