Coverage for webapp/api/launchpad_provenance.py: 97%
145 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-25 22:09 +0000
1import os
2import re
3from concurrent.futures import ThreadPoolExecutor
5from webapp.api.exceptions import ApiTimeoutError
6from webapp.api.requests import Session
8MAX_CONCURRENT_SCANS = 10
10LAUNCHPAD_API_URL = os.getenv(
11 "LAUNCHPAD_API_URL", "https://api.launchpad.net/devel/"
12)
15GITHUB_URL_RE = re.compile(
16 r"^(?:https?://)?(?:www\.)?github\.com/"
17 r"(?P<repo>[^/]+/[^/]+?)(?:\.git)?/?$"
18)
20LAUNCHPAD_GIT_RE = re.compile(
21 r"^https?://api\.launchpad\.net/[^/]+/"
22 r"(?P<path>~[^/]+/(?:[^/]+/)*\+git/[^/]+?)/?$"
23)
26def failure_reason(exc):
27 if isinstance(exc, ApiTimeoutError):
28 return "launchpad_timeout"
29 return "launchpad_error"
32def extract_github_repository(git_repository_url):
33 """Extract owner/repo from a GitHub repository URL, or None.
35 Anchored at both ends: the badge attests against this, so a URL merely
36 containing "github.com/" must not match.
37 """
38 if not git_repository_url:
39 return None
41 match = GITHUB_URL_RE.match(git_repository_url)
42 if match:
43 return match.groupdict()["repo"]
44 return None
47def extract_launchpad_repository(git_repository_link):
48 """Extract the git.launchpad.net path from a Launchpad link"""
49 if not git_repository_link:
50 return None
52 match = LAUNCHPAD_GIT_RE.match(git_repository_link)
53 if match:
54 return match.groupdict()["path"]
55 return None
58def build_commit_url(source, commit_sha):
59 """Build a browsable commit URL for a recipe's source"""
60 if source.get("github_repository"):
61 return (
62 f"https://github.com/{source['github_repository']}"
63 f"/commit/{commit_sha}"
64 )
65 if source.get("launchpad_repository"):
66 return (
67 f"https://git.launchpad.net/{source['launchpad_repository']}"
68 f"/commit/?id={commit_sha}"
69 )
70 return None
73def recipe_source(recipe):
74 """Resolve where a recipe's source is hosted."""
75 git_repository_url = recipe.get("git_repository_url")
76 return {
77 "git_repository_url": git_repository_url,
78 "github_repository": extract_github_repository(git_repository_url),
79 "launchpad_repository": extract_launchpad_repository(
80 recipe.get("git_repository_link")
81 ),
82 }
85class LaunchpadProvenance:
86 """Read-only, anonymous client for Launchpad build provenance.
88 Unlike the authenticated ``canonicalwebteam.launchpad`` client, this one
89 sends no OAuth credentials and does not filter recipes by owner, so it can
90 read provenance for *any* public Launchpad recipe. It is used to link a
91 store revision back to the public git commit it was built from.
92 """
94 def __init__(self, session=None, api_url=LAUNCHPAD_API_URL):
95 self.api_url = api_url
96 self._owns_session = session is None
97 self.session = session or Session()
98 self.session.headers["Accept"] = "application/json"
100 def _get(self, url, params=None, session=None):
101 response = (session or self.session).get(url, params=params)
102 response.raise_for_status()
103 return response.json()
105 def _new_session(self):
106 if not self._owns_session:
107 return self.session
108 session = Session()
109 session.headers["Accept"] = "application/json"
110 return session
112 def get_recipes(self, store_name, max_recipes):
113 """Find the public Launchpad recipes for a store name, best first.
115 store_name is free text, so one name matches dozens of recipes
116 (firefox matches 62), mostly personal ones that never upload. Ranked
117 by upload capability, then recency; at most ``max_recipes``.
118 """
119 data = self._get(
120 f"{self.api_url}+snaps",
121 params={
122 "ws.op": "findByStoreName",
123 "store_name": f'"{store_name}"',
124 },
125 )
127 matches = [
128 entry
129 for entry in data.get("entries", [])
130 if entry.get("store_name") == store_name
131 ]
133 # Two stable sorts: the second keeps the date order within each group.
134 matches.sort(
135 key=lambda e: e.get("date_last_modified") or "", reverse=True
136 )
137 matches.sort(key=lambda e: not e.get("can_upload_to_store"))
139 return matches[:max_recipes]
141 def iter_builds(
142 self, collection_link, max_pages, session=None, wanted=None
143 ):
144 """Collect completed builds.
146 `wanted` is the set of store revisions the caller is after. Once a
147 page has covered all of them there is no point requesting the next
148 one: every extra request is time spent and a chance to fail.
149 """
150 entries = []
151 url = collection_link
152 pages = 0
153 missing = {str(r) for r in wanted} if wanted else None
155 while url and pages < max_pages:
156 try:
157 data = self._get(url, session=session)
158 except Exception as exc:
159 return entries, failure_reason(exc)
160 page_entries = data.get("entries", [])
161 entries.extend(page_entries)
162 url = data.get("next_collection_link")
163 pages += 1
165 if missing is not None:
166 missing.difference_update(
167 str(build["store_upload_revision"])
168 for build in page_entries
169 if build.get("store_upload_status") == "Uploaded"
170 and build.get("store_upload_revision")
171 and build.get("revision_id")
172 and build.get("arch_tag")
173 )
174 if not missing:
175 break
177 return entries, None
179 def _scan_recipe(self, recipe, max_pages, wanted=None):
180 session = self._new_session()
181 try:
182 return self.iter_builds(
183 recipe["completed_builds_collection_link"],
184 max_pages,
185 session=session,
186 wanted=wanted,
187 )
188 finally:
189 if session is not self.session:
190 session.close()
192 def _merge_builds(self, builds, source, revisions):
193 """Fold one recipe's builds into the shared revision map.
195 Returns True if anything was added; earlier recipes win on conflict.
196 """
197 added = False
199 for build in builds:
200 if build.get("store_upload_status") != "Uploaded":
201 continue
203 revision = build.get("store_upload_revision")
204 commit_sha = build.get("revision_id")
205 arch = build.get("arch_tag")
207 if not revision or not commit_sha or not arch:
208 continue
210 revision_key = str(revision)
211 # Newest first, so the first build seen for a revision+arch wins.
212 arch_map = revisions.setdefault(revision_key, {})
213 if arch in arch_map:
214 continue
216 commit_url = build_commit_url(source, commit_sha)
218 build_id = None
219 build_url = None
220 self_link = build.get("self_link")
221 if self_link:
222 build_id = self_link.rstrip("/").split("/")[-1]
223 # API self_link -> human-facing web URL.
224 build_url = self_link.replace(
225 "api.launchpad.net/devel/", "launchpad.net/"
226 )
228 arch_map[arch] = {
229 "commit_sha": commit_sha,
230 "commit_url": commit_url,
231 "build_id": build_id,
232 "build_url": build_url,
233 # Per entry: a merged map can span recipes with different
234 # sources.
235 "github_repository": source.get("github_repository"),
236 "launchpad_repository": source.get("launchpad_repository"),
237 }
238 added = True
240 return added
242 def build_provenance_map(
243 self, store_name, max_pages, max_recipes, wanted=None
244 ):
245 """Return a provenance map joining store revisions to git commits.
247 Shape:
248 {
249 "github_repository": "owner/repo" | None,
250 "launchpad_repository": "~owner/proj/+git/name" | None,
251 "git_repository_url": "https://..." | None,
252 "revisions": {
253 "<store_revision>": {
254 "<arch>": {
255 "commit_sha": "...",
256 "commit_url": "https://github.com/.../commit/..."
257 | None,
258 "build_id": "216436",
259 },
260 },
261 },
262 }
264 Builds are merged across candidate recipes, since one store name can
265 span several legitimate recipes each holding part of the history.
266 Only uploaded builds with a ``revision_id`` are included; revision
267 keys are strings so the map survives JSON round-trips.
268 """
269 result = {
270 "github_repository": None,
271 "launchpad_repository": None,
272 "git_repository_url": None,
273 "revisions": {},
274 # Set when an upstream request failed, so callers can avoid
275 # caching a transient failure as a negative answer.
276 "failed": False,
277 "reason": None,
278 }
280 try:
281 recipes = self.get_recipes(store_name, max_recipes)
282 except Exception as exc:
283 result["failed"] = True
284 result["reason"] = failure_reason(exc)
285 return result
287 if not recipes:
288 return result
290 candidates = [
291 recipe
292 for recipe in recipes
293 if recipe.get("completed_builds_collection_link")
294 ]
295 if not candidates:
296 return result
298 # Share the page budget across candidates: revisions being resolved
299 # are recent and builds are newest first, so breadth beats depth.
300 pages_each = max(1, max_pages // len(candidates))
302 # receopes are fetched in parallel
303 with ThreadPoolExecutor(
304 max_workers=min(MAX_CONCURRENT_SCANS, len(candidates))
305 ) as executor:
306 scans = list(
307 executor.map(
308 lambda recipe: self._scan_recipe(
309 recipe, pages_each, wanted
310 ),
311 candidates,
312 )
313 )
315 revisions = result["revisions"]
316 source = None
317 fallback = None
319 for recipe, (builds, reason) in zip(candidates, scans):
320 recipe_src = recipe_source(recipe)
321 if fallback is None:
322 fallback = recipe_src
324 if (
325 self._merge_builds(builds, recipe_src, revisions)
326 and source is None
327 ):
328 # The recipe that produced revisions, not the first sorted.
329 source = recipe_src
331 if reason and not result["failed"]:
332 result["failed"] = True
333 result["reason"] = reason
335 # A failed request is irrelevant if every wanted revision was found
336 # anyway, e.g. another recipe had them or they were on earlier pages.
337 if (
338 result["failed"]
339 and wanted
340 and all(str(revision) in revisions for revision in wanted)
341 ):
342 result["failed"] = False
343 result["reason"] = None
345 if source is None:
346 source = fallback
347 if source:
348 result["git_repository_url"] = source["git_repository_url"]
349 result["github_repository"] = source["github_repository"]
350 result["launchpad_repository"] = source["launchpad_repository"]
352 return result