Coverage for tests/api/tests_launchpad_provenance.py: 100%

1from threading import Lock 

2from unittest import TestCase 

3from unittest.mock import MagicMock 

4 

5from webapp.api.exceptions import ApiTimeoutError 

6from webapp.api.launchpad_provenance import ( 

7 LaunchpadProvenance, 

8 extract_github_repository, 

9 extract_launchpad_repository, 

10) 

11 

12 

13def _response(payload): 

14 resp = MagicMock() 

15 resp.json.return_value = payload 

16 resp.raise_for_status.return_value = None 

17 return resp 

18 

19 

20def _build(arch, revision, commit, status="Uploaded", build_id="216436"): 

21 return { 

22 "arch_tag": arch, 

23 "store_upload_revision": revision, 

24 "revision_id": commit, 

25 "store_upload_status": status, 

26 "self_link": ( 

27 "https://api.launchpad.net/devel/~build.snapcraft.io" 

28 f"/+snap/x/+build/{build_id}" 

29 ), 

30 } 

31 

32 

33class TestExtractGithubRepository(TestCase): 

34 def test_github_url(self): 

35 self.assertEqual( 

36 extract_github_repository( 

37 "https://github.com/snapcrafters/mumble" 

38 ), 

39 "snapcrafters/mumble", 

40 ) 

41 

42 def test_github_url_with_git_suffix(self): 

43 self.assertEqual( 

44 extract_github_repository( 

45 "https://github.com/snapcrafters/mumble.git" 

46 ), 

47 "snapcrafters/mumble", 

48 ) 

49 

50 def test_non_github_url(self): 

51 self.assertIsNone( 

52 extract_github_repository("https://gitlab.com/foo/bar") 

53 ) 

54 

55 def test_trailing_slash(self): 

56 self.assertEqual( 

57 extract_github_repository( 

58 "https://github.com/snapcrafters/mumble/" 

59 ), 

60 "snapcrafters/mumble", 

61 ) 

62 

63 def test_non_github_host_containing_github_com(self): 

64 self.assertIsNone( 

65 extract_github_repository( 

66 "https://git.example.com/mirrors/github.com/torvalds/linux" 

67 ) 

68 ) 

69 

70 def test_github_com_in_query_string(self): 

71 self.assertIsNone( 

72 extract_github_repository("https://evil.com/?x=github.com/foo/bar") 

73 ) 

74 

75 def test_extra_path_segments(self): 

76 self.assertIsNone( 

77 extract_github_repository("https://github.com/a/b/tree/main") 

78 ) 

79 

80 def test_missing_repository_name(self): 

81 self.assertIsNone( 

82 extract_github_repository("https://github.com/snapcrafters") 

83 ) 

84 

85 def test_none(self): 

86 self.assertIsNone(extract_github_repository(None)) 

87 

88 

89class TestExtractLaunchpadRepository(TestCase): 

90 def test_project_repository(self): 

91 self.assertEqual( 

92 extract_launchpad_repository( 

93 "https://api.launchpad.net/devel/~mozilla-snaps" 

94 "/firefox-snap/+git/firefox-snap" 

95 ), 

96 "~mozilla-snaps/firefox-snap/+git/firefox-snap", 

97 ) 

98 

99 def test_distro_source_package_repository(self): 

100 self.assertEqual( 

101 extract_launchpad_repository( 

102 "https://api.launchpad.net/devel/~hellsworth/ubuntu" 

103 "/+source/libreoffice/+git/libreoffice-snap" 

104 ), 

105 "~hellsworth/ubuntu/+source/libreoffice/+git/libreoffice-snap", 

106 ) 

107 

108 def test_personal_repository(self): 

109 self.assertEqual( 

110 extract_launchpad_repository( 

111 "https://api.launchpad.net/devel/~someone/+git/thing" 

112 ), 

113 "~someone/+git/thing", 

114 ) 

115 

116 def test_trailing_slash(self): 

117 self.assertEqual( 

118 extract_launchpad_repository( 

119 "https://api.launchpad.net/devel/~a/b/+git/c/" 

120 ), 

121 "~a/b/+git/c", 

122 ) 

123 

124 def test_redacted_private_repository(self): 

125 self.assertIsNone( 

126 extract_launchpad_repository("tag:launchpad.net:2008:redacted") 

127 ) 

128 

129 def test_non_launchpad_host(self): 

130 self.assertIsNone( 

131 extract_launchpad_repository( 

132 "https://evil.com/?x=api.launchpad.net/devel/~a/+git/b" 

133 ) 

134 ) 

135 

136 def test_not_a_git_link(self): 

137 self.assertIsNone( 

138 extract_launchpad_repository( 

139 "https://api.launchpad.net/devel/~mozilla-snaps" 

140 ) 

141 ) 

142 

143 def test_none(self): 

144 self.assertIsNone(extract_launchpad_repository(None)) 

145 

146 

147class TestBuildProvenanceMap(TestCase): 

148 def _client(self, recipe, build_pages): 

149 """Build a client whose session returns the recipe for the +snaps 

150 lookup and successive build pages for the collection link.""" 

151 session = MagicMock() 

152 pages = iter(build_pages) 

153 lock = Lock() 

154 

155 def get(url, params=None): 

156 if url.endswith("+snaps"): 

157 return _response(recipe) 

158 with lock: 

159 return _response(next(pages)) 

160 

161 session.get.side_effect = get 

162 return LaunchpadProvenance(session=session) 

163 

164 def test_join_and_filtering(self): 

165 recipe = { 

166 "entries": [ 

167 { 

168 "store_name": "mumble", 

169 "git_repository_url": ( 

170 "https://github.com/snapcrafters/mumble" 

171 ), 

172 "completed_builds_collection_link": ( 

173 "https://api.launchpad.net/devel/x/completed_builds" 

174 ), 

175 } 

176 ] 

177 } 

178 builds_page = { 

179 "entries": [ 

180 _build("amd64", 1721, "aaaaaaa000"), 

181 _build("arm64", 1798, "bbbbbbb111"), 

182 # Skipped: not uploaded. 

183 _build("armhf", 1799, "ccccccc222", status="Failed"), 

184 # Skipped: no commit. 

185 _build("s390x", 1800, None), 

186 ], 

187 "next_collection_link": None, 

188 } 

189 

190 client = self._client(recipe, [builds_page]) 

191 result = client.build_provenance_map( 

192 "mumble", max_pages=5, max_recipes=5 

193 ) 

194 

195 self.assertFalse(result["failed"]) 

196 self.assertEqual(result["github_repository"], "snapcrafters/mumble") 

197 self.assertIn("1721", result["revisions"]) 

198 self.assertIn("1798", result["revisions"]) 

199 self.assertNotIn("1799", result["revisions"]) 

200 self.assertNotIn("1800", result["revisions"]) 

201 

202 amd64 = result["revisions"]["1721"]["amd64"] 

203 self.assertEqual(amd64["commit_sha"], "aaaaaaa000") 

204 self.assertEqual( 

205 amd64["commit_url"], 

206 "https://github.com/snapcrafters/mumble/commit/aaaaaaa000", 

207 ) 

208 self.assertEqual(amd64["build_id"], "216436") 

209 self.assertEqual( 

210 amd64["build_url"], 

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

212 ) 

213 

214 def test_pagination_is_bounded(self): 

215 recipe = { 

216 "entries": [ 

217 { 

218 "store_name": "mumble", 

219 "git_repository_url": ( 

220 "https://github.com/snapcrafters/mumble" 

221 ), 

222 "completed_builds_collection_link": "https://lp/p1", 

223 } 

224 ] 

225 } 

226 page1 = { 

227 "entries": [_build("amd64", 1721, "aaa")], 

228 "next_collection_link": "https://lp/p2", 

229 } 

230 page2 = { 

231 "entries": [_build("arm64", 1798, "bbb")], 

232 "next_collection_link": "https://lp/p3", 

233 } 

234 

235 client = self._client(recipe, [page1, page2]) 

236 result = client.build_provenance_map( 

237 "mumble", max_pages=1, max_recipes=5 

238 ) 

239 

240 # Stopping at the bound is not a failure; the data gathered is valid. 

241 self.assertFalse(result["failed"]) 

242 self.assertIn("1721", result["revisions"]) 

243 self.assertNotIn("1798", result["revisions"]) 

244 

245 def test_not_failed_when_pagination_ends_naturally(self): 

246 recipe = { 

247 "entries": [ 

248 { 

249 "store_name": "mumble", 

250 "git_repository_url": ( 

251 "https://github.com/snapcrafters/mumble" 

252 ), 

253 "completed_builds_collection_link": "https://lp/p1", 

254 } 

255 ] 

256 } 

257 page1 = { 

258 "entries": [_build("amd64", 1721, "aaa")], 

259 "next_collection_link": None, 

260 } 

261 

262 client = self._client(recipe, [page1]) 

263 result = client.build_provenance_map( 

264 "mumble", max_pages=5, max_recipes=5 

265 ) 

266 

267 self.assertFalse(result["failed"]) 

268 

269 def test_partial_result_when_a_page_fails(self): 

270 # A timeout on page 2 must not discard page 1's data. 

271 recipe = { 

272 "entries": [ 

273 { 

274 "store_name": "mumble", 

275 "git_repository_url": ( 

276 "https://github.com/snapcrafters/mumble" 

277 ), 

278 "completed_builds_collection_link": "https://lp/p1", 

279 } 

280 ] 

281 } 

282 page1 = { 

283 "entries": [_build("amd64", 1721, "aaa")], 

284 "next_collection_link": "https://lp/p2", 

285 } 

286 session = MagicMock() 

287 calls = {"builds": 0} 

288 

289 def get(url, params=None): 

290 if url.endswith("+snaps"): 

291 return _response(recipe) 

292 calls["builds"] += 1 

293 if calls["builds"] == 1: 

294 return _response(page1) 

295 raise Exception("read timed out") 

296 

297 session.get.side_effect = get 

298 client = LaunchpadProvenance(session=session) 

299 result = client.build_provenance_map( 

300 "mumble", max_pages=5, max_recipes=5 

301 ) 

302 

303 self.assertTrue(result["failed"]) 

304 self.assertIn("1721", result["revisions"]) 

305 

306 def _github_recipe(self, link="https://lp/p1"): 

307 return { 

308 "entries": [ 

309 { 

310 "store_name": "mumble", 

311 "git_repository_url": ( 

312 "https://github.com/snapcrafters/mumble" 

313 ), 

314 "completed_builds_collection_link": link, 

315 } 

316 ] 

317 } 

318 

319 def test_stops_paginating_once_wanted_revisions_are_found(self): 

320 page1 = { 

321 "entries": [_build("amd64", 1721, "aaa")], 

322 "next_collection_link": "https://lp/p2", 

323 } 

324 client = self._client(self._github_recipe(), [page1]) 

325 result = client.build_provenance_map( 

326 "mumble", max_pages=5, max_recipes=5, wanted={"1721"} 

327 ) 

328 

329 self.assertFalse(result["failed"]) 

330 self.assertIn("1721", result["revisions"]) 

331 self.assertEqual(client.session.get.call_count, 2) 

332 

333 def test_keeps_paginating_while_a_wanted_revision_is_missing(self): 

334 page1 = { 

335 "entries": [_build("amd64", 1721, "aaa")], 

336 "next_collection_link": "https://lp/p2", 

337 } 

338 page2 = { 

339 "entries": [_build("arm64", 1798, "bbb")], 

340 "next_collection_link": None, 

341 } 

342 client = self._client(self._github_recipe(), [page1, page2]) 

343 result = client.build_provenance_map( 

344 "mumble", max_pages=5, max_recipes=5, wanted={"1721", "1798"} 

345 ) 

346 

347 self.assertIn("1721", result["revisions"]) 

348 self.assertIn("1798", result["revisions"]) 

349 self.assertEqual(client.session.get.call_count, 3) 

350 

351 def test_non_uploaded_builds_do_not_satisfy_wanted(self): 

352 page1 = { 

353 "entries": [_build("amd64", 1721, "aaa", status="Failed")], 

354 "next_collection_link": "https://lp/p2", 

355 } 

356 page2 = { 

357 "entries": [_build("amd64", 1721, "aaa")], 

358 "next_collection_link": None, 

359 } 

360 client = self._client(self._github_recipe(), [page1, page2]) 

361 result = client.build_provenance_map( 

362 "mumble", max_pages=5, max_recipes=5, wanted={"1721"} 

363 ) 

364 

365 self.assertEqual(client.session.get.call_count, 3) 

366 self.assertIn("1721", result["revisions"]) 

367 

368 def test_build_without_commit_does_not_satisfy_wanted(self): 

369 incomplete = _build("amd64", 1721, "aaa") 

370 incomplete["revision_id"] = None 

371 page1 = { 

372 "entries": [incomplete], 

373 "next_collection_link": "https://lp/p2", 

374 } 

375 page2 = { 

376 "entries": [_build("amd64", 1721, "aaa")], 

377 "next_collection_link": None, 

378 } 

379 client = self._client(self._github_recipe(), [page1, page2]) 

380 result = client.build_provenance_map( 

381 "mumble", max_pages=5, max_recipes=5, wanted={"1721"} 

382 ) 

383 

384 self.assertEqual(client.session.get.call_count, 3) 

385 self.assertIn("1721", result["revisions"]) 

386 

387 def test_wanted_accepts_ints(self): 

388 page1 = { 

389 "entries": [_build("amd64", 1721, "aaa")], 

390 "next_collection_link": "https://lp/p2", 

391 } 

392 client = self._client(self._github_recipe(), [page1]) 

393 result = client.build_provenance_map( 

394 "mumble", max_pages=5, max_recipes=5, wanted={1721} 

395 ) 

396 

397 self.assertEqual(client.session.get.call_count, 2) 

398 self.assertIn("1721", result["revisions"]) 

399 

400 def test_page_failure_is_not_an_error_when_wanted_is_complete(self): 

401 page1 = { 

402 "entries": [_build("amd64", 1721, "aaa")], 

403 "next_collection_link": "https://lp/p2", 

404 } 

405 session = MagicMock() 

406 calls = {"builds": 0} 

407 

408 def get(url, params=None): 

409 if url.endswith("+snaps"): 

410 return _response(self._github_recipe()) 

411 calls["builds"] += 1 

412 if calls["builds"] == 1: 

413 return _response(page1) 

414 raise Exception("read timed out") 

415 

416 session.get.side_effect = get 

417 client = LaunchpadProvenance(session=session) 

418 result = client.build_provenance_map( 

419 "mumble", max_pages=5, max_recipes=5, wanted={"1721", "1798"} 

420 ) 

421 self.assertTrue(result["failed"]) 

422 self.assertEqual(result["reason"], "launchpad_error") 

423 

424 calls["builds"] = 0 

425 result = client.build_provenance_map( 

426 "mumble", max_pages=5, max_recipes=5, wanted={"1721"} 

427 ) 

428 self.assertFalse(result["failed"]) 

429 self.assertIsNone(result["reason"]) 

430 self.assertIn("1721", result["revisions"]) 

431 

432 def test_page_failure_is_still_an_error_without_wanted(self): 

433 page1 = { 

434 "entries": [_build("amd64", 1721, "aaa")], 

435 "next_collection_link": "https://lp/p2", 

436 } 

437 session = MagicMock() 

438 calls = {"builds": 0} 

439 

440 def get(url, params=None): 

441 if url.endswith("+snaps"): 

442 return _response(self._github_recipe()) 

443 calls["builds"] += 1 

444 if calls["builds"] == 1: 

445 return _response(page1) 

446 raise Exception("read timed out") 

447 

448 session.get.side_effect = get 

449 client = LaunchpadProvenance(session=session) 

450 result = client.build_provenance_map( 

451 "mumble", max_pages=5, max_recipes=5 

452 ) 

453 

454 self.assertTrue(result["failed"]) 

455 

456 def test_non_github_repo_yields_no_commit_url(self): 

457 recipe = { 

458 "entries": [ 

459 { 

460 "store_name": "mumble", 

461 "git_repository_url": "https://gitlab.com/foo/mumble", 

462 "completed_builds_collection_link": "https://lp/p1", 

463 } 

464 ] 

465 } 

466 page = { 

467 "entries": [_build("amd64", 1721, "aaa")], 

468 "next_collection_link": None, 

469 } 

470 

471 client = self._client(recipe, [page]) 

472 result = client.build_provenance_map( 

473 "mumble", max_pages=5, max_recipes=5 

474 ) 

475 

476 self.assertIsNone(result["github_repository"]) 

477 self.assertIsNone(result["revisions"]["1721"]["amd64"]["commit_url"]) 

478 

479 def test_launchpad_hosted_repo_yields_commit_url(self): 

480 recipe = { 

481 "entries": [ 

482 { 

483 "store_name": "firefox", 

484 "git_repository_url": None, 

485 "git_repository_link": ( 

486 "https://api.launchpad.net/devel/~mozilla-snaps" 

487 "/firefox-snap/+git/firefox-snap" 

488 ), 

489 "completed_builds_collection_link": "https://lp/p1", 

490 } 

491 ] 

492 } 

493 page = { 

494 "entries": [_build("amd64", 8763, "659a47f4")], 

495 "next_collection_link": None, 

496 } 

497 

498 client = self._client(recipe, [page]) 

499 result = client.build_provenance_map( 

500 "firefox", max_pages=5, max_recipes=5 

501 ) 

502 

503 self.assertIsNone(result["github_repository"]) 

504 self.assertEqual( 

505 result["launchpad_repository"], 

506 "~mozilla-snaps/firefox-snap/+git/firefox-snap", 

507 ) 

508 amd64 = result["revisions"]["8763"]["amd64"] 

509 self.assertEqual( 

510 amd64["commit_url"], 

511 "https://git.launchpad.net/~mozilla-snaps/firefox-snap" 

512 "/+git/firefox-snap/commit/?id=659a47f4", 

513 ) 

514 

515 def test_github_wins_over_launchpad_link(self): 

516 recipe = { 

517 "entries": [ 

518 { 

519 "store_name": "mumble", 

520 "git_repository_url": ( 

521 "https://github.com/snapcrafters/mumble" 

522 ), 

523 "git_repository_link": ( 

524 "https://api.launchpad.net/devel/~x/+git/y" 

525 ), 

526 "completed_builds_collection_link": "https://lp/p1", 

527 } 

528 ] 

529 } 

530 page = { 

531 "entries": [_build("amd64", 1721, "aaa")], 

532 "next_collection_link": None, 

533 } 

534 

535 client = self._client(recipe, [page]) 

536 result = client.build_provenance_map( 

537 "mumble", max_pages=5, max_recipes=5 

538 ) 

539 

540 self.assertEqual( 

541 result["revisions"]["1721"]["amd64"]["commit_url"], 

542 "https://github.com/snapcrafters/mumble/commit/aaa", 

543 ) 

544 

545 def test_timeout_is_reported_as_launchpad_timeout(self): 

546 recipe = { 

547 "entries": [ 

548 { 

549 "store_name": "mumble", 

550 "git_repository_url": ( 

551 "https://github.com/snapcrafters/mumble" 

552 ), 

553 "completed_builds_collection_link": "https://lp/p1", 

554 } 

555 ] 

556 } 

557 session = MagicMock() 

558 

559 def get(url, params=None): 

560 if url.endswith("+snaps"): 

561 return _response(recipe) 

562 raise ApiTimeoutError("took too long") 

563 

564 session.get.side_effect = get 

565 client = LaunchpadProvenance(session=session) 

566 result = client.build_provenance_map( 

567 "mumble", max_pages=5, max_recipes=5 

568 ) 

569 

570 self.assertTrue(result["failed"]) 

571 self.assertEqual(result["reason"], "launchpad_timeout") 

572 

573 def test_other_errors_are_reported_as_launchpad_error(self): 

574 recipe = { 

575 "entries": [ 

576 { 

577 "store_name": "mumble", 

578 "git_repository_url": ( 

579 "https://github.com/snapcrafters/mumble" 

580 ), 

581 "completed_builds_collection_link": "https://lp/p1", 

582 } 

583 ] 

584 } 

585 session = MagicMock() 

586 

587 def get(url, params=None): 

588 if url.endswith("+snaps"): 

589 return _response(recipe) 

590 raise Exception("boom") 

591 

592 session.get.side_effect = get 

593 client = LaunchpadProvenance(session=session) 

594 result = client.build_provenance_map( 

595 "mumble", max_pages=5, max_recipes=5 

596 ) 

597 

598 self.assertEqual(result["reason"], "launchpad_error") 

599 

600 def test_successful_scan_has_no_reason(self): 

601 recipe = { 

602 "entries": [ 

603 { 

604 "store_name": "mumble", 

605 "git_repository_url": ( 

606 "https://github.com/snapcrafters/mumble" 

607 ), 

608 "completed_builds_collection_link": "https://lp/p1", 

609 } 

610 ] 

611 } 

612 page = { 

613 "entries": [_build("amd64", 1721, "aaa")], 

614 "next_collection_link": None, 

615 } 

616 

617 client = self._client(recipe, [page]) 

618 result = client.build_provenance_map( 

619 "mumble", max_pages=5, max_recipes=5 

620 ) 

621 

622 self.assertFalse(result["failed"]) 

623 self.assertIsNone(result["reason"]) 

624 

625 def test_recipe_lookup_timeout_reports_launchpad_timeout(self): 

626 session = MagicMock() 

627 session.get.side_effect = ApiTimeoutError("took too long") 

628 client = LaunchpadProvenance(session=session) 

629 

630 result = client.build_provenance_map( 

631 "mumble", max_pages=5, max_recipes=5 

632 ) 

633 

634 self.assertTrue(result["failed"]) 

635 self.assertEqual(result["reason"], "launchpad_timeout") 

636 self.assertEqual(result["revisions"], {}) 

637 

638 def test_recipe_lookup_error_reports_launchpad_error(self): 

639 session = MagicMock() 

640 session.get.side_effect = Exception("boom") 

641 client = LaunchpadProvenance(session=session) 

642 

643 result = client.build_provenance_map( 

644 "mumble", max_pages=5, max_recipes=5 

645 ) 

646 

647 self.assertTrue(result["failed"]) 

648 self.assertEqual(result["reason"], "launchpad_error") 

649 

650 def test_no_recipe_returns_empty(self): 

651 session = MagicMock() 

652 session.get.return_value = _response({"entries": []}) 

653 client = LaunchpadProvenance(session=session) 

654 

655 result = client.build_provenance_map( 

656 "ghost-snap", max_pages=5, max_recipes=5 

657 ) 

658 

659 self.assertEqual(result["github_repository"], None) 

660 self.assertEqual(result["revisions"], {}) 

661 

662 

663def _recipe(name, url, link, uploads=True, modified="2026-01-01T00:00:00Z"): 

664 return { 

665 "store_name": "mumble", 

666 "git_repository_url": url, 

667 "completed_builds_collection_link": link, 

668 "can_upload_to_store": uploads, 

669 "date_last_modified": modified, 

670 "_name": name, 

671 } 

672 

673 

674class TestRecipeSelection(TestCase): 

675 """One store name matches many recipes, so the first is a lottery.""" 

676 

677 def _client(self, entries, pages_by_link): 

678 session = MagicMock() 

679 

680 def get(url, params=None): 

681 if url.endswith("+snaps"): 

682 return _response({"entries": entries}) 

683 return _response(pages_by_link[url]) 

684 

685 session.get.side_effect = get 

686 return LaunchpadProvenance(session=session) 

687 

688 def test_scans_past_a_recipe_with_no_uploads(self): 

689 # The firefox shape: a personal recipe sorts above the real one. 

690 personal = _recipe("personal", None, "https://lp/personal") 

691 official = _recipe( 

692 "official", 

693 "https://github.com/snapcrafters/mumble", 

694 "https://lp/official", 

695 ) 

696 pages = { 

697 "https://lp/personal": { 

698 "entries": [_build("amd64", 999, "zzz", status="Failed")], 

699 "next_collection_link": None, 

700 }, 

701 "https://lp/official": { 

702 "entries": [_build("amd64", 1721, "aaa")], 

703 "next_collection_link": None, 

704 }, 

705 } 

706 

707 client = self._client([personal, official], pages) 

708 result = client.build_provenance_map( 

709 "mumble", max_pages=5, max_recipes=5 

710 ) 

711 

712 self.assertIn("1721", result["revisions"]) 

713 # The reported source is the recipe that produced revisions. 

714 self.assertEqual(result["github_repository"], "snapcrafters/mumble") 

715 

716 def test_merges_revisions_across_recipes(self): 

717 # Real recipes are split by series, each holding part of the history. 

718 old = _recipe("old", "https://github.com/x/old", "https://lp/old") 

719 new = _recipe("new", "https://github.com/x/new", "https://lp/new") 

720 pages = { 

721 "https://lp/old": { 

722 "entries": [_build("amd64", 1000, "aaa")], 

723 "next_collection_link": None, 

724 }, 

725 "https://lp/new": { 

726 "entries": [_build("amd64", 2000, "bbb")], 

727 "next_collection_link": None, 

728 }, 

729 } 

730 

731 client = self._client([old, new], pages) 

732 result = client.build_provenance_map( 

733 "mumble", max_pages=5, max_recipes=5 

734 ) 

735 

736 self.assertIn("1000", result["revisions"]) 

737 self.assertIn("2000", result["revisions"]) 

738 # Each row carries its own repository, since they differ. 

739 self.assertEqual( 

740 result["revisions"]["1000"]["amd64"]["github_repository"], "x/old" 

741 ) 

742 self.assertEqual( 

743 result["revisions"]["2000"]["amd64"]["github_repository"], "x/new" 

744 ) 

745 

746 def test_upload_capable_recipes_are_ranked_first(self): 

747 cannot = _recipe( 

748 "cannot", 

749 None, 

750 "https://lp/cannot", 

751 uploads=False, 

752 modified="2026-06-01T00:00:00Z", 

753 ) 

754 can = _recipe( 

755 "can", 

756 "https://github.com/x/can", 

757 "https://lp/can", 

758 modified="2020-01-01T00:00:00Z", 

759 ) 

760 pages = { 

761 "https://lp/can": { 

762 "entries": [_build("amd64", 1721, "aaa")], 

763 "next_collection_link": None, 

764 }, 

765 "https://lp/cannot": { 

766 "entries": [_build("amd64", 1721, "zzz")], 

767 "next_collection_link": None, 

768 }, 

769 } 

770 

771 client = self._client([cannot, can], pages) 

772 result = client.build_provenance_map( 

773 "mumble", max_pages=5, max_recipes=5 

774 ) 

775 

776 # Ranked first despite being far older, and first writer wins. 

777 self.assertEqual( 

778 result["revisions"]["1721"]["amd64"]["commit_sha"], "aaa" 

779 ) 

780 

781 def test_max_recipes_bounds_the_scan(self): 

782 entries = [ 

783 _recipe(f"r{i}", None, f"https://lp/r{i}") for i in range(4) 

784 ] 

785 pages = { 

786 f"https://lp/r{i}": {"entries": [], "next_collection_link": None} 

787 for i in range(4) 

788 } 

789 scanned = [] 

790 

791 session = MagicMock() 

792 

793 def get(url, params=None): 

794 if url.endswith("+snaps"): 

795 return _response({"entries": entries}) 

796 scanned.append(url) 

797 return _response(pages[url]) 

798 

799 session.get.side_effect = get 

800 client = LaunchpadProvenance(session=session) 

801 client.build_provenance_map("mumble", max_pages=5, max_recipes=2) 

802 

803 self.assertEqual(len(scanned), 2) 

804 

805 def test_failed_recipe_is_not_an_error_if_others_cover_wanted(self): 

806 good = _recipe("good", "https://github.com/a/b", "https://lp/good") 

807 bad = _recipe("bad", None, "https://lp/bad") 

808 session = MagicMock() 

809 

810 def get(url, params=None): 

811 if url.endswith("+snaps"): 

812 return _response({"entries": [good, bad]}) 

813 if url == "https://lp/good": 

814 return _response( 

815 { 

816 "entries": [_build("amd64", 1721, "aaa")], 

817 "next_collection_link": None, 

818 } 

819 ) 

820 raise Exception("read timed out") 

821 

822 session.get.side_effect = get 

823 client = LaunchpadProvenance(session=session) 

824 

825 result = client.build_provenance_map( 

826 "mumble", max_pages=5, max_recipes=5, wanted={"1721"} 

827 ) 

828 self.assertFalse(result["failed"]) 

829 self.assertIn("1721", result["revisions"]) 

830 

831 result = client.build_provenance_map( 

832 "mumble", max_pages=5, max_recipes=5, wanted={"1721", "1798"} 

833 ) 

834 self.assertTrue(result["failed"]) 

835 

836 def test_failure_does_not_short_circuit_other_recipes(self): 

837 entries = [ 

838 _recipe(f"r{i}", None, f"https://lp/r{i}") for i in range(4) 

839 ] 

840 attempts = []