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

1from unittest import TestCase 

2from unittest.mock import MagicMock 

3 

4from webapp.api.launchpad_provenance import ( 

5 LaunchpadProvenance, 

6 extract_github_repository, 

7) 

8 

9 

10def _response(payload): 

11 resp = MagicMock() 

12 resp.json.return_value = payload 

13 resp.raise_for_status.return_value = None 

14 return resp 

15 

16 

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

18 return { 

19 "arch_tag": arch, 

20 "store_upload_revision": revision, 

21 "revision_id": commit, 

22 "store_upload_status": status, 

23 "self_link": ( 

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

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

26 ), 

27 } 

28 

29 

30class TestExtractGithubRepository(TestCase): 

31 def test_github_url(self): 

32 self.assertEqual( 

33 extract_github_repository( 

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

35 ), 

36 "snapcrafters/mumble", 

37 ) 

38 

39 def test_github_url_with_git_suffix(self): 

40 self.assertEqual( 

41 extract_github_repository( 

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

43 ), 

44 "snapcrafters/mumble", 

45 ) 

46 

47 def test_non_github_url(self): 

48 self.assertIsNone( 

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

50 ) 

51 

52 def test_trailing_slash(self): 

53 self.assertEqual( 

54 extract_github_repository( 

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

56 ), 

57 "snapcrafters/mumble", 

58 ) 

59 

60 def test_non_github_host_containing_github_com(self): 

61 self.assertIsNone( 

62 extract_github_repository( 

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

64 ) 

65 ) 

66 

67 def test_github_com_in_query_string(self): 

68 self.assertIsNone( 

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

70 ) 

71 

72 def test_extra_path_segments(self): 

73 self.assertIsNone( 

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

75 ) 

76 

77 def test_missing_repository_name(self): 

78 self.assertIsNone( 

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

80 ) 

81 

82 def test_none(self): 

83 self.assertIsNone(extract_github_repository(None)) 

84 

85 

86class TestBuildProvenanceMap(TestCase): 

87 def _client(self, recipe, build_pages): 

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

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

90 session = MagicMock() 

91 pages = iter(build_pages) 

92 

93 def get(url, params=None): 

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

95 return _response(recipe) 

96 return _response(next(pages)) 

97 

98 session.get.side_effect = get 

99 return LaunchpadProvenance(session=session) 

100 

101 def test_join_and_filtering(self): 

102 recipe = { 

103 "entries": [ 

104 { 

105 "store_name": "mumble", 

106 "git_repository_url": ( 

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

108 ), 

109 "completed_builds_collection_link": ( 

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

111 ), 

112 } 

113 ] 

114 } 

115 builds_page = { 

116 "entries": [ 

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

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

119 # Skipped: not uploaded. 

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

121 # Skipped: no commit. 

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

123 ], 

124 "next_collection_link": None, 

125 } 

126 

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

128 result = client.build_provenance_map( 

129 "mumble", max_pages=5, max_recipes=5 

130 ) 

131 

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

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

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

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

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

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

138 

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

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

141 self.assertEqual( 

142 amd64["commit_url"], 

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

144 ) 

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

146 self.assertEqual( 

147 amd64["build_url"], 

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

149 ) 

150 

151 def test_pagination_is_bounded(self): 

152 recipe = { 

153 "entries": [ 

154 { 

155 "store_name": "mumble", 

156 "git_repository_url": ( 

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

158 ), 

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

160 } 

161 ] 

162 } 

163 page1 = { 

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

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

166 } 

167 page2 = { 

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

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

170 } 

171 

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

173 result = client.build_provenance_map( 

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

175 ) 

176 

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

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

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

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

181 

182 def test_not_failed_when_pagination_ends_naturally(self): 

183 recipe = { 

184 "entries": [ 

185 { 

186 "store_name": "mumble", 

187 "git_repository_url": ( 

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

189 ), 

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

191 } 

192 ] 

193 } 

194 page1 = { 

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

196 "next_collection_link": None, 

197 } 

198 

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

200 result = client.build_provenance_map( 

201 "mumble", max_pages=5, max_recipes=5 

202 ) 

203 

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

205 

206 def test_partial_result_when_a_page_fails(self): 

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

208 recipe = { 

209 "entries": [ 

210 { 

211 "store_name": "mumble", 

212 "git_repository_url": ( 

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

214 ), 

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

216 } 

217 ] 

218 } 

219 page1 = { 

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

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

222 } 

223 session = MagicMock() 

224 calls = {"builds": 0} 

225 

226 def get(url, params=None): 

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

228 return _response(recipe) 

229 calls["builds"] += 1 

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

231 return _response(page1) 

232 raise Exception("read timed out") 

233 

234 session.get.side_effect = get 

235 client = LaunchpadProvenance(session=session) 

236 result = client.build_provenance_map( 

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

238 ) 

239 

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

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

242 

243 def test_non_github_repo_yields_no_commit_url(self): 

244 recipe = { 

245 "entries": [ 

246 { 

247 "store_name": "mumble", 

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

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

250 } 

251 ] 

252 } 

253 page = { 

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

255 "next_collection_link": None, 

256 } 

257 

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

259 result = client.build_provenance_map( 

260 "mumble", max_pages=5, max_recipes=5 

261 ) 

262 

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

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

265 

266 def test_no_recipe_returns_empty(self): 

267 session = MagicMock() 

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

269 client = LaunchpadProvenance(session=session) 

270 

271 result = client.build_provenance_map( 

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

273 ) 

274 

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

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

277 

278 

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

280 return { 

281 "store_name": "mumble", 

282 "git_repository_url": url, 

283 "completed_builds_collection_link": link, 

284 "can_upload_to_store": uploads, 

285 "date_last_modified": modified, 

286 "_name": name, 

287 } 

288 

289 

290class TestRecipeSelection(TestCase): 

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

292 

293 def _client(self, entries, pages_by_link): 

294 session = MagicMock() 

295 

296 def get(url, params=None): 

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

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

299 return _response(pages_by_link[url]) 

300 

301 session.get.side_effect = get 

302 return LaunchpadProvenance(session=session) 

303 

304 def test_scans_past_a_recipe_with_no_uploads(self): 

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

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

307 official = _recipe( 

308 "official", 

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

310 "https://lp/official", 

311 ) 

312 pages = { 

313 "https://lp/personal": { 

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

315 "next_collection_link": None, 

316 }, 

317 "https://lp/official": { 

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

319 "next_collection_link": None, 

320 }, 

321 } 

322 

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

324 result = client.build_provenance_map( 

325 "mumble", max_pages=5, max_recipes=5 

326 ) 

327 

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

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

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

331 

332 def test_merges_revisions_across_recipes(self): 

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

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

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

336 pages = { 

337 "https://lp/old": { 

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

339 "next_collection_link": None, 

340 }, 

341 "https://lp/new": { 

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

343 "next_collection_link": None, 

344 }, 

345 } 

346 

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

348 result = client.build_provenance_map( 

349 "mumble", max_pages=5, max_recipes=5 

350 ) 

351 

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

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

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

355 self.assertEqual( 

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

357 ) 

358 self.assertEqual( 

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

360 ) 

361 

362 def test_upload_capable_recipes_are_ranked_first(self): 

363 cannot = _recipe( 

364 "cannot", 

365 None, 

366 "https://lp/cannot", 

367 uploads=False, 

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

369 ) 

370 can = _recipe( 

371 "can", 

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

373 "https://lp/can", 

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

375 ) 

376 pages = { 

377 "https://lp/can": { 

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

379 "next_collection_link": None, 

380 }, 

381 "https://lp/cannot": { 

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

383 "next_collection_link": None, 

384 }, 

385 } 

386 

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

388 result = client.build_provenance_map( 

389 "mumble", max_pages=5, max_recipes=5 

390 ) 

391 

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

393 self.assertEqual( 

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

395 ) 

396 

397 def test_max_recipes_bounds_the_scan(self): 

398 entries = [ 

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

400 ] 

401 pages = { 

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

403 for i in range(4) 

404 } 

405 scanned = [] 

406 

407 session = MagicMock() 

408 

409 def get(url, params=None): 

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

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

412 scanned.append(url) 

413 return _response(pages[url]) 

414 

415 session.get.side_effect = get 

416 client = LaunchpadProvenance(session=session) 

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

418 

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

420 

421 def test_a_failing_recipe_stops_the_scan(self): 

422 # Each request costs up to 12s, so failures must not stack up. 

423 entries = [ 

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

425 ] 

426 attempts = [] 

427 

428 session = MagicMock() 

429 

430 def get(url, params=None): 

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

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

433 attempts.append(url) 

434 raise Exception("read timed out") 

435 

436 session.get.side_effect = get 

437 client = LaunchpadProvenance(session=session) 

438 result = client.build_provenance_map( 

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

440 ) 

441 

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

443 self.assertEqual(len(attempts), 1)