Coverage for tests/endpoints/tests_snaps_auditable.py: 100%

1from unittest.mock import patch 

2 

3from tests.endpoints.endpoint_testing import TestEndpoints 

4from webapp.api.exceptions import ApiTimeoutError 

5from webapp.endpoints.snaps import ( 

6 FAILED_PROVENANCE_TTL, 

7 _get_provenance_map, 

8) 

9 

10 

11def _channel(arch, track, risk, revision): 

12 return { 

13 "channel": { 

14 "architecture": arch, 

15 "track": track, 

16 "risk": risk, 

17 "name": f"{track}/{risk}", 

18 "released-at": "2024-03-28T05:16:07.500510+00:00", 

19 }, 

20 "revision": revision, 

21 "version": "1.0", 

22 "confinement": "strict", 

23 "download": {"size": 100}, 

24 } 

25 

26 

27def _details(channel_map, default_track="latest"): 

28 return { 

29 "channel-map": channel_map, 

30 "default-track": default_track, 

31 "snap": {}, 

32 } 

33 

34 

35def _provenance( 

36 revisions, 

37 github_repository="snapcrafters/mumble", 

38 failed=False, 

39 reason=None, 

40): 

41 return { 

42 "github_repository": github_repository, 

43 "git_repository_url": "https://github.com/snapcrafters/mumble", 

44 "revisions": revisions, 

45 "failed": failed, 

46 "reason": reason, 

47 } 

48 

49 

50VERIFIED_BUILD = { 

51 "commit_sha": "10c7c9e1234567890", 

52 "commit_url": ( 

53 "https://github.com/snapcrafters/mumble/commit/10c7c9e1234567890" 

54 ), 

55 "build_id": "216436", 

56 "build_url": ( 

57 "https://launchpad.net/~build.snapcraft.io/+snap/x/+build/216436" 

58 ), 

59} 

60 

61 

62class TestProvenanceMapCaching(TestEndpoints): 

63 """What gets cached decides whether a Launchpad wobble becomes a storm.""" 

64 

65 def setUp(self): 

66 super().setUp() 

67 cache_patcher = patch("webapp.endpoints.snaps.redis_cache") 

68 self.cache_patch = cache_patcher.start() 

69 self.cache_patch.get.return_value = None 

70 self.addCleanup(cache_patcher.stop) 

71 details_patcher = patch( 

72 "webapp.endpoints.snaps.device_gateway.get_item_details", 

73 return_value=_details( 

74 [_channel("amd64", "latest", "stable", 1721)] 

75 ), 

76 ) 

77 details_patcher.start() 

78 self.addCleanup(details_patcher.stop) 

79 

80 def _ttl_for(self, provenance_map): 

81 with patch( 

82 "webapp.endpoints.snaps.launchpad_provenance." 

83 "build_provenance_map", 

84 return_value=provenance_map, 

85 ): 

86 _get_provenance_map("mumble") 

87 return self.cache_patch.set.call_args.kwargs["ttl"] 

88 

89 def test_successful_scan_is_cached_for_an_hour(self): 

90 # Includes scans that stopped at their bounds: partial but valid. 

91 self.assertEqual(self._ttl_for(_provenance({})), 3600) 

92 

93 def test_failed_scan_is_cached_briefly(self): 

94 ttl = self._ttl_for(_provenance({}, failed=True)) 

95 self.assertEqual(ttl, FAILED_PROVENANCE_TTL) 

96 self.assertLess(ttl, 3600) 

97 

98 @patch("webapp.endpoints.snaps.repository_is_public") 

99 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

100 def test_repository_is_checked_once_and_cached( 

101 self, mock_map, mock_public 

102 ): 

103 # One call per map build, not one per page view. 

104 mock_map.return_value = _provenance({}) 

105 mock_public.return_value = False 

106 

107 result = _get_provenance_map("mumble") 

108 

109 mock_public.assert_called_once_with("snapcrafters/mumble") 

110 self.assertFalse(result["source_available"]) 

111 cached = self.cache_patch.set.call_args.args[1] 

112 self.assertFalse(cached["source_available"]) 

113 

114 

115class TestAuditableEndpoint(TestEndpoints): 

116 def setUp(self): 

117 super().setUp() 

118 # Force cache miss so build_provenance_map is always exercised. 

119 cache_patcher = patch("webapp.endpoints.snaps.redis_cache") 

120 self.cache_patch = cache_patcher.start() 

121 self.cache_patch.get.return_value = None 

122 self.addCleanup(cache_patcher.stop) 

123 # Real HTTP; keep it off the network. 

124 public_patcher = patch( 

125 "webapp.endpoints.snaps.repository_is_public", return_value=True 

126 ) 

127 self.public_patch = public_patcher.start() 

128 self.addCleanup(public_patcher.stop) 

129 

130 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

131 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

132 def test_scan_is_told_every_released_revision( 

133 self, mock_map, mock_details 

134 ): 

135 mock_details.return_value = _details( 

136 [ 

137 _channel("amd64", "latest", "stable", 1721), 

138 _channel("amd64", "latest", "edge", 1800), 

139 _channel("arm64", "latest", "stable", 1798), 

140 ] 

141 ) 

142 mock_map.return_value = _provenance({}) 

143 

144 self.client.get("/api/mumble/auditable") 

145 

146 mock_details.assert_called_once() 

147 self.assertEqual( 

148 mock_map.call_args.kwargs["wanted"], {"1721", "1798", "1800"} 

149 ) 

150 

151 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

152 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

153 def test_auditable_hit(self, mock_map, mock_details): 

154 mock_details.return_value = _details( 

155 [_channel("amd64", "latest", "stable", 1721)] 

156 ) 

157 mock_map.return_value = _provenance( 

158 {"1721": {"amd64": VERIFIED_BUILD}} 

159 ) 

160 

161 response = self.app.test_client().get("/api/mumble/auditable") 

162 data = response.get_json() 

163 

164 self.assertEqual(response.status_code, 200) 

165 self.assertTrue(data["auditable"]) 

166 self.assertEqual(data["status"], "verified") 

167 self.assertEqual(data["revision"], 1721) 

168 self.assertEqual(data["architecture"], "amd64") 

169 self.assertEqual(data["commit_sha"], "10c7c9e1234567890") 

170 self.assertEqual(data["github_repository"], "snapcrafters/mumble") 

171 self.assertIn("commit/10c7c9e", data["commit_url"]) 

172 self.assertEqual(data["build_id"], "216436") 

173 self.assertEqual(response.cache_control.max_age, 3600) 

174 

175 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

176 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

177 def test_prefers_amd64(self, mock_map, mock_details): 

178 mock_details.return_value = _details( 

179 [ 

180 _channel("arm64", "latest", "stable", 1798), 

181 _channel("amd64", "latest", "stable", 1721), 

182 ] 

183 ) 

184 mock_map.return_value = _provenance( 

185 { 

186 "1721": {"amd64": VERIFIED_BUILD}, 

187 "1798": {"arm64": VERIFIED_BUILD}, 

188 } 

189 ) 

190 

191 data = self.client.get("/api/mumble/auditable").get_json() 

192 

193 self.assertTrue(data["auditable"]) 

194 self.assertEqual(data["architecture"], "amd64") 

195 self.assertEqual(data["revision"], 1721) 

196 

197 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

198 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

199 def test_deterministic_arch_fallback(self, mock_map, mock_details): 

200 # No amd64 -> first architecture sorted alphabetically (arm64). 

201 mock_details.return_value = _details( 

202 [ 

203 _channel("armhf", "latest", "stable", 900), 

204 _channel("arm64", "latest", "stable", 800), 

205 ] 

206 ) 

207 mock_map.return_value = _provenance( 

208 { 

209 "800": {"arm64": VERIFIED_BUILD}, 

210 "900": {"armhf": VERIFIED_BUILD}, 

211 } 

212 ) 

213 

214 data = self.client.get("/api/mumble/auditable").get_json() 

215 

216 self.assertTrue(data["auditable"]) 

217 self.assertEqual(data["architecture"], "arm64") 

218 self.assertEqual(data["revision"], 800) 

219 

220 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

221 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

222 def test_unavailable_when_revision_absent(self, mock_map, mock_details): 

223 # Public recipe exists (github_repository set) but this revision has no 

224 # matching build -> "unavailable", not "not-provided". 

225 mock_details.return_value = _details( 

226 [_channel("amd64", "latest", "stable", 1721)] 

227 ) 

228 mock_map.return_value = _provenance({}) 

229 

230 data = self.client.get("/api/mumble/auditable").get_json() 

231 

232 self.assertFalse(data["auditable"]) 

233 self.assertEqual(data["status"], "unavailable") 

234 self.assertEqual(data["revision"], 1721) 

235 self.assertEqual(data["architecture"], "amd64") 

236 

237 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

238 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

239 def test_not_provided_when_no_public_recipe(self, mock_map, mock_details): 

240 # No public GitHub recipe (private / non-GitHub) -> "not-provided". 

241 mock_details.return_value = _details( 

242 [_channel("amd64", "latest", "stable", 1721)] 

243 ) 

244 mock_map.return_value = _provenance({}, github_repository=None) 

245 

246 data = self.client.get("/api/mumble/auditable").get_json() 

247 

248 self.assertFalse(data["auditable"]) 

249 self.assertEqual(data["status"], "not-provided") 

250 self.assertEqual(data["revision"], 1721) 

251 

252 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

253 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

254 def test_non_github_repo_is_not_provided(self, mock_map, mock_details): 

255 mock_details.return_value = _details( 

256 [_channel("amd64", "latest", "stable", 1721)] 

257 ) 

258 # Non-GitHub build: commit_url is None and no github repository. 

259 no_link_build = dict(VERIFIED_BUILD, commit_url=None) 

260 mock_map.return_value = _provenance( 

261 {"1721": {"amd64": no_link_build}}, github_repository=None 

262 ) 

263 

264 data = self.client.get("/api/mumble/auditable").get_json() 

265 

266 self.assertFalse(data["auditable"]) 

267 self.assertEqual(data["status"], "not-provided") 

268 

269 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

270 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

271 def test_upstream_failure_reports_error(self, mock_map, mock_details): 

272 mock_details.return_value = _details( 

273 [_channel("amd64", "latest", "stable", 1721)] 

274 ) 

275 mock_map.side_effect = Exception("launchpad down") 

276 

277 response = self.app.test_client().get("/api/mumble/auditable") 

278 

279 self.assertEqual(response.status_code, 200) 

280 self.assertEqual( 

281 response.get_json(), 

282 { 

283 "auditable": False, 

284 "status": "error", 

285 "reason": "unexpected_error", 

286 }, 

287 ) 

288 self.assertEqual(response.cache_control.max_age, FAILED_PROVENANCE_TTL) 

289 

290 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

291 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

292 def test_failed_scan_reports_error(self, mock_map, mock_details): 

293 # The one case where retrying might help. 

294 mock_details.return_value = _details( 

295 [_channel("amd64", "latest", "stable", 1721)] 

296 ) 

297 mock_map.return_value = _provenance({}, failed=True) 

298 

299 data = self.client.get("/api/mumble/auditable").get_json() 

300 

301 self.assertFalse(data["auditable"]) 

302 self.assertEqual(data["status"], "error") 

303 self.assertEqual(data["revision"], 1721) 

304 self.assertEqual(data["architecture"], "amd64") 

305 

306 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

307 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

308 def test_bounded_scan_is_not_an_error(self, mock_map, mock_details): 

309 # Hitting the scan bounds must not pin an error on a snap forever. 

310 mock_details.return_value = _details( 

311 [_channel("amd64", "latest", "stable", 1721)] 

312 ) 

313 mock_map.return_value = _provenance({}, github_repository=None) 

314 

315 data = self.client.get("/api/mumble/auditable").get_json() 

316 

317 self.assertFalse(data["auditable"]) 

318 self.assertEqual(data["status"], "not-provided") 

319 

320 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

321 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

322 def test_deleted_repository_is_not_verified(self, mock_map, mock_details): 

323 # The commit link would 404, so offering it is worse than nothing. 

324 mock_details.return_value = _details( 

325 [_channel("amd64", "latest", "stable", 1721)] 

326 ) 

327 mock_map.return_value = _provenance( 

328 {"1721": {"amd64": VERIFIED_BUILD}} 

329 ) 

330 self.public_patch.return_value = False 

331 

332 data = self.client.get("/api/mumble/auditable").get_json() 

333 

334 self.assertFalse(data["auditable"]) 

335 self.assertEqual(data["status"], "not-provided") 

336 self.assertNotIn("commit_url", data) 

337 

338 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

339 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

340 def test_build_without_commit_url_is_not_an_error( 

341 self, mock_map, mock_details 

342 ): 

343 # Build found, but no GitHub commit to link: settled, not a failure. 

344 mock_details.return_value = _details( 

345 [_channel("amd64", "latest", "stable", 1721)] 

346 ) 

347 mock_map.return_value = _provenance( 

348 {"1721": {"amd64": {"commit_sha": "aaa", "commit_url": None}}}, 

349 github_repository=None, 

350 ) 

351 

352 data = self.client.get("/api/mumble/auditable").get_json() 

353 

354 self.assertFalse(data["auditable"]) 

355 self.assertEqual(data["status"], "not-provided") 

356 

357 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

358 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

359 def test_hit_in_truncated_scan_is_still_verified( 

360 self, mock_map, mock_details 

361 ): 

362 # A found revision is authoritative however many pages went unread. 

363 mock_details.return_value = _details( 

364 [_channel("amd64", "latest", "stable", 1721)] 

365 ) 

366 mock_map.return_value = _provenance( 

367 {"1721": {"amd64": VERIFIED_BUILD}} 

368 ) 

369 

370 data = self.client.get("/api/mumble/auditable").get_json() 

371 

372 self.assertTrue(data["auditable"]) 

373 self.assertEqual(data["status"], "verified") 

374 

375 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

376 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

377 def test_default_channel_arch_without_that_risk_is_skipped( 

378 self, mock_map, mock_details 

379 ): 

380 mock_details.return_value = _details( 

381 [ 

382 _channel("amd64", "latest", "edge", 900), 

383 _channel("arm64", "latest", "stable", 1721), 

384 ] 

385 ) 

386 mock_map.return_value = _provenance( 

387 {"1721": {"arm64": VERIFIED_BUILD}} 

388 ) 

389 

390 data = self.client.get("/api/mumble/auditable").get_json() 

391 

392 self.assertEqual(data["architecture"], "arm64") 

393 self.assertEqual(data["revision"], 1721) 

394 self.assertTrue(data["auditable"]) 

395 

396 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

397 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

398 def test_amd64_preferred_when_it_publishes_the_default_channel( 

399 self, mock_map, mock_details 

400 ): 

401 mock_details.return_value = _details( 

402 [ 

403 _channel("arm64", "latest", "stable", 1798), 

404 _channel("amd64", "latest", "stable", 1721), 

405 ] 

406 ) 

407 mock_map.return_value = _provenance( 

408 {"1721": {"amd64": VERIFIED_BUILD}} 

409 ) 

410 

411 data = self.client.get("/api/mumble/auditable").get_json() 

412 

413 self.assertEqual(data["architecture"], "amd64") 

414 self.assertEqual(data["revision"], 1721) 

415 

416 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

417 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

418 def test_error_carries_the_failure_reason(self, mock_map, mock_details): 

419 mock_details.return_value = _details( 

420 [_channel("amd64", "latest", "stable", 1721)] 

421 ) 

422 mock_map.return_value = _provenance( 

423 {}, failed=True, reason="launchpad_timeout" 

424 ) 

425 

426 data = self.client.get("/api/mumble/auditable").get_json() 

427 

428 self.assertEqual(data["status"], "error") 

429 self.assertEqual(data["reason"], "launchpad_timeout") 

430 

431 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

432 def test_store_timeout_is_reported(self, mock_details): 

433 mock_details.side_effect = ApiTimeoutError("took too long") 

434 

435 data = self.client.get("/api/mumble/auditable").get_json() 

436 

437 self.assertEqual(data["status"], "error") 

438 self.assertEqual(data["reason"], "store_timeout") 

439 

440 @patch("webapp.endpoints.snaps.device_gateway.get_item_details") 

441 def test_unexpected_errors_are_distinguishable(self, mock_details): 

442 mock_details.side_effect = AttributeError("bug in our own code") 

443 

444 data = self.client.get("/api/mumble/auditable").get_json() 

445 

446 self.assertEqual(data["status"], "error") 

447 self.assertEqual(data["reason"], "unexpected_error") 

448 

449 

450class TestAuditableRevisionsEndpoint(TestEndpoints): 

451 def setUp(self): 

452 super().setUp() 

453 self.cache_patch = patch("webapp.endpoints.snaps.redis_cache").start() 

454 self.cache_patch.get.return_value = None 

455 # Real HTTP; keep it off the network. 

456 self.public_patch = patch( 

457 "webapp.endpoints.snaps.repository_is_public", return_value=True 

458 ).start() 

459 self.details_patch = patch( 

460 "webapp.endpoints.snaps.device_gateway.get_item_details", 

461 return_value=_details( 

462 [ 

463 _channel("amd64", "latest", "stable", 1721), 

464 _channel("arm64", "latest", "stable", 1798), 

465 ] 

466 ), 

467 ).start() 

468 self.addCleanup(patch.stopall) 

469 

470 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

471 def test_scan_is_told_which_revisions_are_released(self, mock_map): 

472 mock_map.return_value = _provenance({}) 

473 

474 self.client.get("/api/mumble/auditable-revisions") 

475 

476 self.assertEqual(mock_map.call_args.kwargs["wanted"], {"1721", "1798"}) 

477 

478 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

479 def test_details_are_not_fetched_on_cache_hit(self, mock_map): 

480 self.cache_patch.get.return_value = _provenance( 

481 {"1721": {"amd64": VERIFIED_BUILD}} 

482 ) 

483 

484 data = self.client.get("/api/mumble/auditable-revisions").get_json() 

485 

486 self.assertIn("1721", data["revisions"]) 

487 self.details_patch.assert_not_called() 

488 mock_map.assert_not_called() 

489 

490 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

491 def test_returns_flattened_revisions(self, mock_map): 

492 mock_map.return_value = _provenance( 

493 { 

494 "1721": {"amd64": VERIFIED_BUILD}, 

495 "1798": {"arm64": VERIFIED_BUILD}, 

496 } 

497 ) 

498 

499 data = self.client.get("/api/mumble/auditable-revisions").get_json() 

500 

501 self.assertEqual(data["github_repository"], "snapcrafters/mumble") 

502 self.assertIn("1721", data["revisions"]) 

503 self.assertIn("1798", data["revisions"]) 

504 self.assertEqual( 

505 data["revisions"]["1721"]["commit_sha"], "10c7c9e1234567890" 

506 ) 

507 

508 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

509 def test_revisions_failure_reports_error(self, mock_map): 

510 mock_map.side_effect = Exception("launchpad down") 

511 

512 response = self.app.test_client().get( 

513 "/api/mumble/auditable-revisions" 

514 ) 

515 

516 self.assertEqual(response.status_code, 200) 

517 self.assertEqual( 

518 response.get_json(), 

519 { 

520 "github_repository": None, 

521 "revisions": {}, 

522 "error": True, 

523 "reason": "unexpected_error", 

524 }, 

525 ) 

526 self.assertEqual(response.cache_control.max_age, FAILED_PROVENANCE_TTL) 

527 

528 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

529 def test_revisions_omitted_when_repository_is_gone(self, mock_map): 

530 mock_map.return_value = _provenance( 

531 {"1721": {"amd64": VERIFIED_BUILD}} 

532 ) 

533 self.public_patch.return_value = False 

534 

535 data = self.client.get("/api/mumble/auditable-revisions").get_json() 

536 

537 self.assertEqual(data["revisions"], {}) 

538 self.assertFalse(data["error"]) 

539 

540 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

541 def test_revisions_failed_scan_sets_error_flag(self, mock_map): 

542 mock_map.return_value = _provenance( 

543 {"1721": {"amd64": VERIFIED_BUILD}}, failed=True 

544 ) 

545 

546 data = self.client.get("/api/mumble/auditable-revisions").get_json() 

547 

548 self.assertTrue(data["error"]) 

549 # Partial data is still returned. 

550 self.assertIn("1721", data["revisions"]) 

551 

552 @patch("webapp.endpoints.snaps.launchpad_provenance.build_provenance_map") 

553 def test_revisions_truncated_scan_is_not_an_error(self, mock_map): 

554 # Bounded by LP_MAX_BUILD_PAGES by design, so not a failure. 

555 mock_map.return_value = _provenance( 

556 {"1721": {"amd64": VERIFIED_BUILD}}, failed=False 

557 ) 

558 

559 # Logged-in responses are rewritten to `private`, dropping max-age. 

560 response = self.app.test_client().get( 

561 "/api/mumble/auditable-revisions" 

562 ) 

563 data = response.get_json() 

564 

565 self.assertFalse(data["error"]) 

566 self.assertIn("1721", data["revisions"]) 

567 self.assertEqual(response.cache_control.max_age, 3600)