Coverage for tests / store / tests_details.py: 96%

1import responses 

2from urllib.parse import urlencode 

3from flask_testing import TestCase 

4from webapp.app import create_app 

5from unittest.mock import patch 

6from cache.cache_utility import redis_cache 

7 

8POPULAR_PATH = "webapp.store.views.snap_recommendations.get_popular" 

9RECENT_PATH = "webapp.store.views.snap_recommendations.get_recent" 

10TREND_PATH = "webapp.store.views.snap_recommendations.get_trending" 

11TOP_PATH = "webapp.store.views.snap_recommendations.get_top_rated" 

12CATEGORIES_PATH = "webapp.store.views.device_gateway.get_categories" 

13 

14 

15EMPTY_EXTRA_DETAILS_PAYLOAD = {"aliases": None, "package_name": "vault"} 

16SNAP_PAYLOAD = { 

17 "snap-id": "id", 

18 "name": "toto", 

19 "default-track": None, 

20 "snap": { 

21 "title": "Snap Title", 

22 "summary": "This is a summary", 

23 "description": "this is a description", 

24 "media": [], 

25 "license": "license", 

26 "publisher": { 

27 "display-name": "Toto", 

28 "username": "toto", 

29 "validation": True, 

30 }, 

31 "categories": [{"name": "test"}], 

32 "trending": False, 

33 "unlisted": False, 

34 "links": {}, 

35 }, 

36 "channel-map": [ 

37 { 

38 "channel": { 

39 "architecture": "amd64", 

40 "name": "stable", 

41 "risk": "stable", 

42 "track": "latest", 

43 "released-at": "2018-09-18T14:45:28.064633+00:00", 

44 }, 

45 "created-at": "2018-09-18T14:45:28.064633+00:00", 

46 "version": "1.0", 

47 "confinement": "conf", 

48 "download": {"size": 100000}, 

49 "revision": 1, 

50 } 

51 ], 

52} 

53 

54 

55class GetDetailsPageTest(TestCase): 

56 def setUp(self): 

57 super().setUp() 

58 self.snap_name = "toto" 

59 self.snap_id = "id" 

60 self.revision = 1 

61 self.api_url = "".join( 

62 [ 

63 "https://api.snapcraft.io/v2/", 

64 "snaps/info/", 

65 self.snap_name, 

66 "?", 

67 urlencode( 

68 { 

69 "fields": ",".join( 

70 [ 

71 "title", 

72 "summary", 

73 "description", 

74 "license", 

75 "contact", 

76 "website", 

77 "publisher", 

78 "media", 

79 "download", 

80 "version", 

81 "created-at", 

82 "confinement", 

83 "categories", 

84 "trending", 

85 "unlisted", 

86 "links", 

87 "revision", 

88 ] 

89 ) 

90 } 

91 ), 

92 ] 

93 ) 

94 self.endpoint_url = "/" + self.snap_name 

95 self.api_url_details = "".join( 

96 [ 

97 "https://api.snapcraft.io/api/v1/", 

98 "snaps/details/", 

99 self.snap_name, 

100 "?", 

101 urlencode({"fields": ",".join(["aliases"])}), 

102 ] 

103 ) 

104 self.api_url_sboms = "".join( 

105 [ 

106 "https://api.snapcraft.io/api/v1/", 

107 "sboms/download/", 

108 f"sbom_snap_{self.snap_id}_{self.revision}.spdx2.3.json", 

109 ] 

110 ) 

111 

112 def create_app(self): 

113 app = create_app(testing=True) 

114 app.secret_key = "secret_key" 

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

116 

117 return app 

118 

119 def assert_not_in_context(self, name): 

120 try: 

121 self.get_context_variable(name) 

122 except Exception: 

123 # flask-testing throws exception if context doesn't have "name" 

124 # that's what we expect so we just return and let the test pass 

125 return 

126 # If we reach this point it means the variable IS in context 

127 self.fail(f"Context variable exists: {name}") 

128 

129 @responses.activate 

130 def test_has_sboms_success(self): 

131 payload = SNAP_PAYLOAD 

132 

133 responses.add( 

134 responses.Response( 

135 method="GET", url=self.api_url, json=payload, status=200 

136 ) 

137 ) 

138 responses.add( 

139 responses.Response( 

140 method="HEAD", url=self.api_url_sboms, json={}, status=200 

141 ) 

142 ) 

143 

144 metrics_url = "https://api.snapcraft.io/api/v1/snaps/metrics" 

145 responses.add( 

146 responses.Response( 

147 method="POST", url=metrics_url, json={}, status=200 

148 ) 

149 ) 

150 

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

152 

153 assert response.status_code == 200 

154 

155 @responses.activate 

156 def test_has_sboms_error(self): 

157 payload = SNAP_PAYLOAD 

158 

159 responses.add( 

160 responses.Response( 

161 method="GET", url=self.api_url, json=payload, status=200 

162 ) 

163 ) 

164 responses.add( 

165 responses.Response( 

166 method="HEAD", url=self.api_url_sboms, json={}, status=404 

167 ) 

168 ) 

169 

170 metrics_url = "https://api.snapcraft.io/api/v1/snaps/metrics" 

171 responses.add( 

172 responses.Response( 

173 method="POST", url=metrics_url, json={}, status=200 

174 ) 

175 ) 

176 

177 response = self.client.head(self.api_url_sboms) 

178 

179 assert response.status_code == 404 

180 

181 @responses.activate 

182 def test_api_404(self): 

183 payload = {"error-list": [{"code": "resource-not-found"}]} 

184 responses.add( 

185 responses.Response( 

186 method="GET", url=self.api_url, json=payload, status=404 

187 ) 

188 ) 

189 

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

191 

192 called = responses.calls[0] 

193 assert called.request.url == self.api_url 

194 assert len(responses.calls) == 1 

195 

196 assert response.status_code == 404 

197 

198 @responses.activate 

199 def test_extra_details_error(self): 

200 payload = SNAP_PAYLOAD 

201 extra_details_payload = { 

202 "error_list": [ 

203 { 

204 "code": "resource-not-found", 

205 "message": "No snap named 'toto' found in series '16'.", 

206 } 

207 ], 

208 "errors": ["No snap named 'toto' found in series '16'."], 

209 "result": "error", 

210 } 

211 

212 responses.add( 

213 responses.Response( 

214 method="GET", url=self.api_url, json=payload, status=200 

215 ) 

216 ) 

217 responses.add( 

218 responses.Response( 

219 method="GET", 

220 url=self.api_url_details, 

221 json=extra_details_payload, 

222 status=404, 

223 ) 

224 ) 

225 responses.add( 

226 responses.Response( 

227 method="HEAD", url=self.api_url_sboms, json={}, status=200 

228 ) 

229 ) 

230 metrics_url = "https://api.snapcraft.io/api/v1/snaps/metrics" 

231 responses.add( 

232 responses.Response( 

233 method="POST", url=metrics_url, json={}, status=200 

234 ) 

235 ) 

236 

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

238 

239 assert response.status_code == 200 

240 

241 @responses.activate 

242 def test_api_500(self): 

243 payload = {"error-list": []} 

244 responses.add( 

245 responses.Response( 

246 method="GET", url=self.api_url, json=payload, status=500 

247 ) 

248 ) 

249 

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

251 

252 assert len(responses.calls) == 1 

253 called = responses.calls[0] 

254 assert called.request.url == self.api_url 

255 

256 assert response.status_code == 502 

257 

258 @responses.activate 

259 def test_api_500_no_answer(self): 

260 responses.add( 

261 responses.Response(method="GET", url=self.api_url, status=500) 

262 ) 

263 

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

265 

266 assert len(responses.calls) == 1 

267 called = responses.calls[0] 

268 assert called.request.url == self.api_url 

269 

270 assert response.status_code == 502 

271 

272 @responses.activate 

273 def test_no_channel_map(self): 

274 payload = { 

275 "snap-id": "id", 

276 "name": "toto", 

277 "default-track": None, 

278 "snap": { 

279 "title": "Snap Title", 

280 "summary": "This is a summary", 

281 "description": "this is a description", 

282 "media": [], 

283 "license": "license", 

284 "publisher": { 

285 "display-name": "Toto", 

286 "username": "toto", 

287 "validation": True, 

288 }, 

289 "categories": [{"name": "test"}], 

290 "trending": False, 

291 "unlisted": False, 

292 "links": {}, 

293 }, 

294 } 

295 

296 responses.add( 

297 responses.Response( 

298 method="GET", url=self.api_url, json=payload, status=200 

299 ) 

300 ) 

301 responses.add( 

302 responses.Response( 

303 method="GET", 

304 url=self.api_url_details, 

305 json=EMPTY_EXTRA_DETAILS_PAYLOAD, 

306 status=200, 

307 ) 

308 ) 

309 

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

311 

312 assert response.status_code == 404 

313 

314 @responses.activate 

315 def test_user_connected(self): 

316 payload = SNAP_PAYLOAD 

317 

318 responses.add( 

319 responses.Response( 

320 method="GET", url=self.api_url, json=payload, status=200 

321 ) 

322 ) 

323 responses.add( 

324 responses.Response( 

325 method="GET", 

326 url=self.api_url_details, 

327 json=EMPTY_EXTRA_DETAILS_PAYLOAD, 

328 status=200, 

329 ) 

330 ) 

331 responses.add( 

332 responses.Response( 

333 method="HEAD", url=self.api_url_sboms, json={}, status=200 

334 ) 

335 ) 

336 

337 metrics_url = "https://api.snapcraft.io/api/v1/snaps/metrics" 

338 responses.add( 

339 responses.Response( 

340 method="POST", url=metrics_url, json={}, status=200 

341 ) 

342 ) 

343 

344 with self.client.session_transaction() as s: 

345 # make test session 'authenticated' 

346 s["publisher"] = {"nickname": "toto", "fullname": "Totinio"} 

347 s["macaroon_root"] = "test" 

348 s["macaroon_discharge"] = "test" 

349 # mock test user snaps list 

350 s["user_snaps"] = {"toto": {"snap-id": "test"}} 

351 

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

353 

354 self.assert200(response) 

355 self.assert_context("is_users_snap", True) 

356 

357 @responses.activate 

358 def test_user_not_connected(self): 

359 payload = SNAP_PAYLOAD 

360 

361 responses.add( 

362 responses.Response( 

363 method="GET", url=self.api_url, json=payload, status=200 

364 ) 

365 ) 

366 responses.add( 

367 responses.Response( 

368 method="GET", 

369 url=self.api_url_details, 

370 json=EMPTY_EXTRA_DETAILS_PAYLOAD, 

371 status=200, 

372 ) 

373 ) 

374 responses.add( 

375 responses.Response( 

376 method="HEAD", url=self.api_url_sboms, json={}, status=200 

377 ) 

378 ) 

379 

380 metrics_url = "https://api.snapcraft.io/api/v1/snaps/metrics" 

381 responses.add( 

382 responses.Response( 

383 method="POST", url=metrics_url, json={}, status=200 

384 ) 

385 ) 

386 

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

388 

389 assert response.status_code == 200 

390 self.assert_context("is_users_snap", False) 

391 

392 @responses.activate 

393 def test_user_connected_on_not_own_snap(self): 

394 payload = SNAP_PAYLOAD 

395 

396 responses.add( 

397 responses.Response( 

398 method="GET", url=self.api_url, json=payload, status=200 

399 ) 

400 ) 

401 responses.add( 

402 responses.Response( 

403 method="GET", 

404 url=self.api_url_details, 

405 json=EMPTY_EXTRA_DETAILS_PAYLOAD, 

406 status=200, 

407 ) 

408 ) 

409 responses.add( 

410 responses.Response( 

411 method="HEAD", url=self.api_url_sboms, json={}, status=200 

412 ) 

413 ) 

414 

415 metrics_url = "https://api.snapcraft.io/api/v1/snaps/metrics" 

416 responses.add( 

417 responses.Response( 

418 method="POST", url=metrics_url, json={}, status=200 

419 ) 

420 ) 

421 

422 with self.client.session_transaction() as s: 

423 s["publisher"] = {"nickname": "greg"} 

424 

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

426 

427 assert response.status_code == 200 

428 self.assert_context("is_users_snap", False) 

429 

430 @responses.activate 

431 def test_extra_details(self): 

432 payload = SNAP_PAYLOAD 

433 payload_extra_details = { 

434 "aliases": [ 

435 {"name": "nu", "target": "nu"}, 

436 { 

437 "name": "nu_plugin_stress_internals", 

438 "target": "nu-plugin-stress-internals", 

439 }, 

440 {"name": "nu_plugin_gstat", "target": "nu-plugin-gstat"}, 

441 {"name": "nu_plugin_formats", "target": "nu-plugin-formats"}, 

442 {"name": "nu_plugin_polars", "target": "nu-plugin-polars"}, 

443 ], 

444 "package_name": "toto", 

445 } 

446 

447 responses.add( 

448 responses.Response( 

449 method="GET", url=self.api_url, json=payload, status=200 

450 ) 

451 ) 

452 responses.add( 

453 responses.Response( 

454 method="GET", 

455 url=self.api_url_details, 

456 json=payload_extra_details, 

457 status=200, 

458 ) 

459 ) 

460 responses.add( 

461 responses.Response( 

462 method="HEAD", url=self.api_url_sboms, json={}, status=200 

463 ) 

464 ) 

465 metrics_url = "https://api.snapcraft.io/api/v1/snaps/metrics" 

466 responses.add( 

467 responses.Response( 

468 method="POST", url=metrics_url, json={}, status=200 

469 ) 

470 ) 

471 

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

473 assert response.status_code == 200 

474 self.assert_context( 

475 "aliases", 

476 [ 

477 ["toto.nu", "nu"], 

478 [ 

479 "toto.nu-plugin-stress-internals", 

480 "nu_plugin_stress_internals", 

481 ], 

482 ["toto.nu-plugin-gstat", "nu_plugin_gstat"], 

483 ["toto.nu-plugin-formats", "nu_plugin_formats"], 

484 ["toto.nu-plugin-polars", "nu_plugin_polars"], 

485 ], 

486 ) 

487 

488 @responses.activate 

489 def test_explore_uses_redis_cache(self): 

490 """When Redis has cached explore data, the recommendation APIs 

491 and device gateway should not be called and the view should 

492 return successfully using the cached values. 

493 """ 

494 # seed redis 

495 popular = [ 

496 { 

497 "details": { 

498 "name": "/pop1", 

499 "icon": "", 

500 "title": "Pop 1", 

501 "publisher": "Pub 1", 

502 "developer_validation": None, 

503 "summary": "Popular snap", 

504 }, 

505 } 

506 ] 

507 recent = [ 

508 { 

509 "details": { 

510 "name": "/recent1", 

511 "icon": "", 

512 "title": "Recent 1", 

513 "publisher": "Pub 2", 

514 "developer_validation": None, 

515 "summary": "Recent snap", 

516 }, 

517 } 

518 ] 

519 trending = [ 

520 { 

521 "details": { 

522 "name": "/trend1", 

523 "icon": "", 

524 "title": "Trend 1", 

525 "publisher": "Pub 3", 

526 "developer_validation": None, 

527 "summary": "Trending snap", 

528 }, 

529 } 

530 ] 

531 top_rated = [ 

532 { 

533 "details": { 

534 "name": "/top1", 

535 "icon": "", 

536 "title": "Top 1", 

537 "publisher": "Pub 4", 

538 "developer_validation": None, 

539 "summary": "Top rated snap", 

540 }, 

541 } 

542 ] 

543 categories = [{"slug": "cat1", "name": "Cat 1"}] 

544 

545 redis_cache.set("explore:popular-snaps", popular, ttl=3600) 

546 redis_cache.set("explore:recent-snaps", recent, ttl=3600) 

547 redis_cache.set("explore:trending-snaps", trending, ttl=3600) 

548 redis_cache.set("explore:top-rated-snaps", top_rated, ttl=3600) 

549 redis_cache.set("explore:categories", categories, ttl=3600) 

550 

551 with patch(POPULAR_PATH) as mock_popular: 

552 with patch(RECENT_PATH) as mock_recent: 

553 with patch(TREND_PATH) as mock_trending: 

554 with patch(TOP_PATH) as mock_top_rated: 

555 with patch(CATEGORIES_PATH) as mock_categories: 

556 response = self.client.get("/explore") 

557 

558 self.assert200(response) 

559 

560 mock_popular.assert_not_called() 

561 mock_recent.assert_not_called() 

562 mock_trending.assert_not_called() 

563 mock_top_rated.assert_not_called() 

564 mock_categories.assert_not_called() 

565 

566 @responses.activate 

567 def test_explore_populates_cache_when_empty(self): 

568 """When Redis cache is empty, the recommendation/device methods 

569 should be called and their results stored in Redis for subsequent 

570 requests. 

571 """ 

572 

573 with patch( 

574 POPULAR_PATH, 

575 return_value=[ 

576 { 

577 "details": { 

578 "name": "/popx", 

579 "icon": "", 

580 "title": "Pop X", 

581 "publisher": "Pub X", 

582 "developer_validation": None, 

583 "summary": "Popular x", 

584 } 

585 } 

586 ], 

587 ) as mock_popular: 

588 with patch( 

589 RECENT_PATH, 

590 return_value=[ 

591 { 

592 "details": { 

593 "name": "/recentx", 

594 "icon": "", 

595 "title": "Recent X", 

596 "publisher": "Pub RX", 

597 "developer_validation": None, 

598 "summary": "Recent x", 

599 } 

600 } 

601 ], 

602 ) as mock_recent: 

603 with patch( 

604 TREND_PATH, 

605 return_value=[ 

606 { 

607 "details": { 

608 "name": "/trendx", 

609 "icon": "", 

610 "title": "Trend X", 

611 "publisher": "Pub TX", 

612 "developer_validation": None, 

613 "summary": "Trend x", 

614 } 

615 } 

616 ], 

617 ) as mock_trending: 

618 with patch( 

619 TOP_PATH, 

620 return_value=[ 

621 { 

622 "details": { 

623 "name": "/topx", 

624 "icon": "", 

625 "title": "Top X", 

626 "publisher": "Pub TX", 

627 "developer_validation": None, 

628 "summary": "Top x", 

629 } 

630 } 

631 ], 

632 ) as mock_top_rated: 

633 with patch( 

634 CATEGORIES_PATH, 

635 return_value=[{"slug": "c1", "name": "C1"}], 

636 ) as mock_categories: 

637 response = self.client.get("/explore") 

638 

639 self.assert200(response) 

640 

641 # ensure the methods were called to populate cache 

642 self.assertTrue(mock_popular.called) 

643 self.assertTrue(mock_recent.called) 

644 self.assertTrue(mock_trending.called) 

645 self.assertTrue(mock_top_rated.called) 

646 self.assertTrue(mock_categories.called) 

647 # cached values should now exist 

648 pop_cached = redis_cache.get( 

649 "explore:popular-snaps" 

650 ) 

651 recent_cached = redis_cache.get( 

652 "explore:recent-snaps" 

653 ) 

654 trend_cached = redis_cache.get( 

655 "explore:trending-snaps" 

656 ) 

657 top_cached = redis_cache.get( 

658 "explore:top-rated-snaps" 

659 ) 

660 categories_cached = redis_cache.get( 

661 "explore:categories" 

662 ) 

663 

664 assert pop_cached is not None 

665 assert recent_cached is not None 

666 assert trend_cached is not None 

667 assert top_cached is not None 

668 assert categories_cached is not None 

669 

670 

671if __name__ == "__main__": 

672 import unittest 

673 

674 unittest.main()