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 

72 def _ttl_for(self, provenance_map): 

73 with patch( 

74 "webapp.endpoints.snaps.launchpad_provenance." 

75 "build_provenance_map", 

76 return_value=provenance_map, 

77 ): 

78 _get_provenance_map("mumble") 

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

80 

81 def test_successful_scan_is_cached_for_an_hour(self): 

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

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

84 

85 def test_failed_scan_is_cached_briefly(self): 

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

87 self.assertEqual(ttl, FAILED_PROVENANCE_TTL) 

88 self.assertLess(ttl, 3600) 

89 

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

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

92 def test_repository_is_checked_once_and_cached( 

93 self, mock_map, mock_public 

94 ): 

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

96 mock_map.return_value = _provenance({}) 

97 mock_public.return_value = False 

98 

99 result = _get_provenance_map("mumble") 

100 

101 mock_public.assert_called_once_with("snapcrafters/mumble") 

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

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

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

105 

106 

107class TestAuditableEndpoint(TestEndpoints): 

108 def setUp(self): 

109 super().setUp() 

110 # Force cache miss so build_provenance_map is always exercised. 

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

112 self.cache_patch = cache_patcher.start() 

113 self.cache_patch.get.return_value = None 

114 self.addCleanup(cache_patcher.stop) 

115 # Real HTTP; keep it off the network. 

116 public_patcher = patch( 

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

118 ) 

119 self.public_patch = public_patcher.start() 

120 self.addCleanup(public_patcher.stop) 

121 

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

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

124 def test_auditable_hit(self, mock_map, mock_details): 

125 mock_details.return_value = _details( 

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

127 ) 

128 mock_map.return_value = _provenance( 

129 {"1721": {"amd64": VERIFIED_BUILD}} 

130 ) 

131 

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

133 data = response.get_json() 

134 

135 self.assertEqual(response.status_code, 200) 

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

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

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

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

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

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

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

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

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

145 

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

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

148 def test_prefers_amd64(self, mock_map, mock_details): 

149 mock_details.return_value = _details( 

150 [ 

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

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

153 ] 

154 ) 

155 mock_map.return_value = _provenance( 

156 { 

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

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

159 } 

160 ) 

161 

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

163 

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

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

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

167 

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

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

170 def test_deterministic_arch_fallback(self, mock_map, mock_details): 

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

172 mock_details.return_value = _details( 

173 [ 

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

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

176 ] 

177 ) 

178 mock_map.return_value = _provenance( 

179 { 

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

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

182 } 

183 ) 

184 

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

186 

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

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

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

190 

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

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

193 def test_unavailable_when_revision_absent(self, mock_map, mock_details): 

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

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

196 mock_details.return_value = _details( 

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

198 ) 

199 mock_map.return_value = _provenance({}) 

200 

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

202 

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

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

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

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

207 

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

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

210 def test_not_provided_when_no_public_recipe(self, mock_map, mock_details): 

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

212 mock_details.return_value = _details( 

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

214 ) 

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

216 

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

218 

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

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

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

222 

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

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

225 def test_non_github_repo_is_not_provided(self, mock_map, mock_details): 

226 mock_details.return_value = _details( 

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

228 ) 

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

230 no_link_build = dict(VERIFIED_BUILD, commit_url=None) 

231 mock_map.return_value = _provenance( 

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

233 ) 

234 

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

236 

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

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

239 

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

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

242 def test_upstream_failure_reports_error(self, mock_map, mock_details): 

243 mock_details.return_value = _details( 

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

245 ) 

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

247 

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

249 

250 self.assertEqual(response.status_code, 200) 

251 self.assertEqual( 

252 response.get_json(), 

253 { 

254 "auditable": False, 

255 "status": "error", 

256 "reason": "unexpected_error", 

257 }, 

258 ) 

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

260 

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

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

263 def test_failed_scan_reports_error(self, mock_map, mock_details): 

264 # The one case where retrying might help. 

265 mock_details.return_value = _details( 

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

267 ) 

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

269 

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

271 

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

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

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

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

276 

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

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

279 def test_bounded_scan_is_not_an_error(self, mock_map, mock_details): 

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

281 mock_details.return_value = _details( 

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

283 ) 

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

285 

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

287 

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

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

290 

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

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

293 def test_deleted_repository_is_not_verified(self, mock_map, mock_details): 

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

295 mock_details.return_value = _details( 

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

297 ) 

298 mock_map.return_value = _provenance( 

299 {"1721": {"amd64": VERIFIED_BUILD}} 

300 ) 

301 self.public_patch.return_value = False 

302 

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

304 

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

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

307 self.assertNotIn("commit_url", data) 

308 

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

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

311 def test_build_without_commit_url_is_not_an_error( 

312 self, mock_map, mock_details 

313 ): 

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

315 mock_details.return_value = _details( 

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

317 ) 

318 mock_map.return_value = _provenance( 

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

320 github_repository=None, 

321 ) 

322 

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

324 

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

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

327 

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

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

330 def test_hit_in_truncated_scan_is_still_verified( 

331 self, mock_map, mock_details 

332 ): 

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

334 mock_details.return_value = _details( 

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

336 ) 

337 mock_map.return_value = _provenance( 

338 {"1721": {"amd64": VERIFIED_BUILD}} 

339 ) 

340 

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

342 

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

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

345 

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

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

348 def test_default_channel_arch_without_that_risk_is_skipped( 

349 self, mock_map, mock_details 

350 ): 

351 mock_details.return_value = _details( 

352 [ 

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

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

355 ] 

356 ) 

357 mock_map.return_value = _provenance( 

358 {"1721": {"arm64": VERIFIED_BUILD}} 

359 ) 

360 

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

362 

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

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

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

366 

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

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

369 def test_amd64_preferred_when_it_publishes_the_default_channel( 

370 self, mock_map, mock_details 

371 ): 

372 mock_details.return_value = _details( 

373 [ 

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

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

376 ] 

377 ) 

378 mock_map.return_value = _provenance( 

379 {"1721": {"amd64": VERIFIED_BUILD}} 

380 ) 

381 

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

383 

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

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

386 

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

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

389 def test_error_carries_the_failure_reason(self, mock_map, mock_details): 

390 mock_details.return_value = _details( 

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

392 ) 

393 mock_map.return_value = _provenance( 

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

395 ) 

396 

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

398 

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

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

401 

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

403 def test_store_timeout_is_reported(self, mock_details): 

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

405 

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

407 

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

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

410 

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

412 def test_unexpected_errors_are_distinguishable(self, mock_details): 

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

414 

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

416 

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

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

419 

420 

421class TestAuditableRevisionsEndpoint(TestEndpoints): 

422 def setUp(self): 

423 super().setUp() 

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

425 self.cache_patch.get.return_value = None 

426 # Real HTTP; keep it off the network. 

427 self.public_patch = patch( 

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

429 ).start() 

430 self.addCleanup(patch.stopall) 

431 

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

433 def test_returns_flattened_revisions(self, mock_map): 

434 mock_map.return_value = _provenance( 

435 { 

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

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

438 } 

439 ) 

440 

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

442 

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

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

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

446 self.assertEqual( 

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

448 ) 

449 

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

451 def test_revisions_failure_reports_error(self, mock_map): 

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

453 

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

455 "/api/mumble/auditable-revisions" 

456 ) 

457 

458 self.assertEqual(response.status_code, 200) 

459 self.assertEqual( 

460 response.get_json(), 

461 { 

462 "github_repository": None, 

463 "revisions": {}, 

464 "error": True, 

465 "reason": "unexpected_error", 

466 }, 

467 ) 

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

469 

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

471 def test_revisions_omitted_when_repository_is_gone(self, mock_map): 

472 mock_map.return_value = _provenance( 

473 {"1721": {"amd64": VERIFIED_BUILD}} 

474 ) 

475 self.public_patch.return_value = False 

476 

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

478 

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

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

481 

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

483 def test_revisions_failed_scan_sets_error_flag(self, mock_map): 

484 mock_map.return_value = _provenance( 

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

486 ) 

487 

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

489 

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

491 # Partial data is still returned. 

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

493 

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

495 def test_revisions_truncated_scan_is_not_an_error(self, mock_map): 

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

497 mock_map.return_value = _provenance( 

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

499 ) 

500 

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

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

503 "/api/mumble/auditable-revisions" 

504 ) 

505 data = response.get_json() 

506 

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

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

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