Coverage for tests/api/tests_launchpad

1from threading import Lock

2from unittest import TestCase

3from unittest.mock import MagicMock

5from webapp.api.exceptions import ApiTimeoutError

6from webapp.api.launchpad_provenance import (

7 LaunchpadProvenance,

8 extract_github_repository,

9 extract_launchpad_repository,

10)

13def _response(payload):

14 resp = MagicMock()

15 resp.json.return_value = payload

16 resp.raise_for_status.return_value = None

17 return resp

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 }

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 )

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 )

50 def test_non_github_url(self):

51 self.assertIsNone(

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

53 )

55 def test_trailing_slash(self):

56 self.assertEqual(

57 extract_github_repository(

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

59 ),

60 "snapcrafters/mumble",

61 )

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 )

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 )

75 def test_extra_path_segments(self):

76 self.assertIsNone(

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

78 )

80 def test_missing_repository_name(self):

81 self.assertIsNone(

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

83 )

85 def test_none(self):

86 self.assertIsNone(extract_github_repository(None))

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 )

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 test_non_github_repo_yields_no_commit_url(self):

307 recipe = {

308 "entries": [

309 {

310 "store_name": "mumble",

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

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

313 }

314 ]

315 }

316 page = {

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

318 "next_collection_link": None,

319 }

320

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

322 result = client.build_provenance_map(

323 "mumble", max_pages=5, max_recipes=5

324 )

325

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

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

328

329 def test_launchpad_hosted_repo_yields_commit_url(self):

330 recipe = {

331 "entries": [

332 {

333 "store_name": "firefox",

334 "git_repository_url": None,

335 "git_repository_link": (

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

337 "/firefox-snap/+git/firefox-snap"

338 ),

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

340 }

341 ]

342 }

343 page = {

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

345 "next_collection_link": None,

346 }

347

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

349 result = client.build_provenance_map(

350 "firefox", max_pages=5, max_recipes=5

351 )

352

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

354 self.assertEqual(

355 result["launchpad_repository"],

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

357 )

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

359 self.assertEqual(

360 amd64["commit_url"],

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

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

363 )

364

365 def test_github_wins_over_launchpad_link(self):

366 recipe = {

367 "entries": [

368 {

369 "store_name": "mumble",

370 "git_repository_url": (

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

372 ),

373 "git_repository_link": (

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

375 ),

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

377 }

378 ]

379 }

380 page = {

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

382 "next_collection_link": None,

383 }

384

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

386 result = client.build_provenance_map(

387 "mumble", max_pages=5, max_recipes=5

388 )

389

390 self.assertEqual(

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

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

393 )

394

395 def test_timeout_is_reported_as_launchpad_timeout(self):

396 recipe = {

397 "entries": [

398 {

399 "store_name": "mumble",

400 "git_repository_url": (

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

402 ),

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

404 }

405 ]

406 }

407 session = MagicMock()

408

409 def get(url, params=None):

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

411 return _response(recipe)

412 raise ApiTimeoutError("took too long")

413

414 session.get.side_effect = get

415 client = LaunchpadProvenance(session=session)

416 result = client.build_provenance_map(

417 "mumble", max_pages=5, max_recipes=5

418 )

419

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

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

422

423 def test_other_errors_are_reported_as_launchpad_error(self):

424 recipe = {

425 "entries": [

426 {

427 "store_name": "mumble",

428 "git_repository_url": (

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

430 ),

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

432 }

433 ]

434 }

435 session = MagicMock()

436

437 def get(url, params=None):

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

439 return _response(recipe)

440 raise Exception("boom")

441

442 session.get.side_effect = get

443 client = LaunchpadProvenance(session=session)

444 result = client.build_provenance_map(

445 "mumble", max_pages=5, max_recipes=5

446 )

447

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

449

450 def test_successful_scan_has_no_reason(self):

451 recipe = {

452 "entries": [

453 {

454 "store_name": "mumble",

455 "git_repository_url": (

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

457 ),

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

459 }

460 ]

461 }

462 page = {

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

464 "next_collection_link": None,

465 }

466

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

468 result = client.build_provenance_map(

469 "mumble", max_pages=5, max_recipes=5

470 )

471

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

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

474

475 def test_recipe_lookup_timeout_reports_launchpad_timeout(self):

476 session = MagicMock()

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

478 client = LaunchpadProvenance(session=session)

479

480 result = client.build_provenance_map(

481 "mumble", max_pages=5, max_recipes=5

482 )

483

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

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

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

487

488 def test_recipe_lookup_error_reports_launchpad_error(self):

489 session = MagicMock()

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

491 client = LaunchpadProvenance(session=session)

492

493 result = client.build_provenance_map(

494 "mumble", max_pages=5, max_recipes=5

495 )

496

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

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

499

500 def test_no_recipe_returns_empty(self):

501 session = MagicMock()

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

503 client = LaunchpadProvenance(session=session)

504

505 result = client.build_provenance_map(

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

507 )

508

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

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

511

512

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

514 return {

515 "store_name": "mumble",

516 "git_repository_url": url,

517 "completed_builds_collection_link": link,

518 "can_upload_to_store": uploads,

519 "date_last_modified": modified,

520 "_name": name,

521 }

522

523

524class TestRecipeSelection(TestCase):

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

526

527 def _client(self, entries, pages_by_link):

528 session = MagicMock()

529

530 def get(url, params=None):

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

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

533 return _response(pages_by_link[url])

534

535 session.get.side_effect = get

536 return LaunchpadProvenance(session=session)

537

538 def test_scans_past_a_recipe_with_no_uploads(self):

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

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

541 official = _recipe(

542 "official",

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

544 "https://lp/official",

545 )

546 pages = {

547 "https://lp/personal": {

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

549 "next_collection_link": None,

550 },

551 "https://lp/official": {

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

553 "next_collection_link": None,

554 },

555 }

556

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

558 result = client.build_provenance_map(

559 "mumble", max_pages=5, max_recipes=5

560 )

561

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

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

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

565

566 def test_merges_revisions_across_recipes(self):

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

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

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

570 pages = {

571 "https://lp/old": {

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

573 "next_collection_link": None,

574 },

575 "https://lp/new": {

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

577 "next_collection_link": None,

578 },

579 }

580

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

582 result = client.build_provenance_map(

583 "mumble", max_pages=5, max_recipes=5

584 )

585

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

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

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

589 self.assertEqual(

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

591 )

592 self.assertEqual(

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

594 )

595

596 def test_upload_capable_recipes_are_ranked_first(self):

597 cannot = _recipe(

598 "cannot",

599 None,

600 "https://lp/cannot",

601 uploads=False,

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

603 )

604 can = _recipe(

605 "can",

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

607 "https://lp/can",

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

609 )

610 pages = {

611 "https://lp/can": {

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

613 "next_collection_link": None,

614 },

615 "https://lp/cannot": {

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

617 "next_collection_link": None,

618 },

619 }

620

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

622 result = client.build_provenance_map(

623 "mumble", max_pages=5, max_recipes=5

624 )

625

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

627 self.assertEqual(

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

629 )

630

631 def test_max_recipes_bounds_the_scan(self):

632 entries = [

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

634 ]

635 pages = {

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

637 for i in range(4)

638 }

639 scanned = []

640

641 session = MagicMock()

642

643 def get(url, params=None):

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

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

646 scanned.append(url)

647 return _response(pages[url])

648

649 session.get.side_effect = get

650 client = LaunchpadProvenance(session=session)

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

652

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

654

655 def test_failure_does_not_short_circuit_other_recipes(self):

656 entries = [

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

658 ]

659 attempts = []

660 lock = Lock()

661

662 session = MagicMock()

663

664 def get(url, params=None):

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

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

667 with lock:

668 attempts.append(url)

669 raise Exception("read timed out")

670

671 session.get.side_effect = get

672 client = LaunchpadProvenance(session=session)

673 result = client.build_provenance_map(

674 "mumble", max_pages=5, max_recipes=4

675 )

676

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

678 self.assertEqual(

679 sorted(attempts), [f"https://lp/r{i}" for i in range(4)]

680 )

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

248 statements