Coverage for webapp/api/launchpad_provenance.py: 99%
95 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-20 22:09 +0000
1import os
2import re
4from webapp.api.requests import Session
6LAUNCHPAD_API_URL = os.getenv(
7 "LAUNCHPAD_API_URL", "https://api.launchpad.net/devel/"
8)
11GITHUB_URL_RE = re.compile(
12 r"^(?:https?://)?(?:www\.)?github\.com/"
13 r"(?P<repo>[^/]+/[^/]+?)(?:\.git)?/?$"
14)
17def extract_github_repository(git_repository_url):
18 """Extract owner/repo from a GitHub repository URL, or None.
20 Anchored at both ends: the badge attests against this, so a URL merely
21 containing "github.com/" must not match.
22 """
23 if not git_repository_url:
24 return None
26 match = GITHUB_URL_RE.match(git_repository_url)
27 if match:
28 return match.groupdict()["repo"]
29 return None
32class LaunchpadProvenance:
33 """Read-only, anonymous client for Launchpad build provenance.
35 Unlike the authenticated ``canonicalwebteam.launchpad`` client, this one
36 sends no OAuth credentials and does not filter recipes by owner, so it can
37 read provenance for *any* public Launchpad recipe. It is used to link a
38 store revision back to the public git commit it was built from.
39 """
41 def __init__(self, session=None, api_url=LAUNCHPAD_API_URL):
42 self.api_url = api_url
43 self.session = session or Session()
44 self.session.headers["Accept"] = "application/json"
46 def _get(self, url, params=None):
47 response = self.session.get(url, params=params)
48 response.raise_for_status()
49 return response.json()
51 def get_recipes(self, store_name, max_recipes):
52 """Find the public Launchpad recipes for a store name, best first.
54 store_name is free text, so one name matches dozens of recipes
55 (firefox matches 62), mostly personal ones that never upload. Ranked
56 by upload capability, then recency; at most ``max_recipes``.
57 """
58 data = self._get(
59 f"{self.api_url}+snaps",
60 params={
61 "ws.op": "findByStoreName",
62 "store_name": f'"{store_name}"',
63 },
64 )
66 matches = [
67 entry
68 for entry in data.get("entries", [])
69 if entry.get("store_name") == store_name
70 ]
72 # Two stable sorts: the second keeps the date order within each group.
73 matches.sort(
74 key=lambda e: e.get("date_last_modified") or "", reverse=True
75 )
76 matches.sort(key=lambda e: not e.get("can_upload_to_store"))
78 return matches[:max_recipes]
80 def iter_builds(self, collection_link, max_pages):
81 """Collect completed builds, up to ``max_pages`` pages.
83 Returns ``(entries, failed)``; whatever was gathered is returned
84 even when a page request errored.
85 """
86 entries = []
87 url = collection_link
88 pages = 0
90 while url and pages < max_pages:
91 try:
92 data = self._get(url)
93 except Exception:
94 return entries, True
95 entries.extend(data.get("entries", []))
96 url = data.get("next_collection_link")
97 pages += 1
99 return entries, False
101 def _merge_builds(self, builds, github_repository, revisions):
102 """Fold one recipe's builds into the shared revision map.
104 Returns True if anything was added; earlier recipes win on conflict.
105 """
106 added = False
108 for build in builds:
109 if build.get("store_upload_status") != "Uploaded":
110 continue
112 revision = build.get("store_upload_revision")
113 commit_sha = build.get("revision_id")
114 arch = build.get("arch_tag")
116 if not revision or not commit_sha or not arch:
117 continue
119 revision_key = str(revision)
120 # Newest first, so the first build seen for a revision+arch wins.
121 arch_map = revisions.setdefault(revision_key, {})
122 if arch in arch_map:
123 continue
125 commit_url = None
126 if github_repository:
127 commit_url = (
128 f"https://github.com/{github_repository}"
129 f"/commit/{commit_sha}"
130 )
132 build_id = None
133 build_url = None
134 self_link = build.get("self_link")
135 if self_link:
136 build_id = self_link.rstrip("/").split("/")[-1]
137 # API self_link -> human-facing web URL.
138 build_url = self_link.replace(
139 "api.launchpad.net/devel/", "launchpad.net/"
140 )
142 arch_map[arch] = {
143 "commit_sha": commit_sha,
144 "commit_url": commit_url,
145 "build_id": build_id,
146 "build_url": build_url,
147 # Per entry: a merged map can span recipes with different
148 # sources.
149 "github_repository": github_repository,
150 }
151 added = True
153 return added
155 def build_provenance_map(self, store_name, max_pages, max_recipes):
156 """Return a provenance map joining store revisions to git commits.
158 Shape:
159 {
160 "github_repository": "owner/repo" | None,
161 "git_repository_url": "https://..." | None,
162 "revisions": {
163 "<store_revision>": {
164 "<arch>": {
165 "commit_sha": "...",
166 "commit_url": "https://github.com/.../commit/..."
167 | None,
168 "build_id": "216436",
169 },
170 },
171 },
172 }
174 Builds are merged across candidate recipes, since one store name can
175 span several legitimate recipes each holding part of the history.
176 Only uploaded builds with a ``revision_id`` are included; revision
177 keys are strings so the map survives JSON round-trips.
178 """
179 recipes = self.get_recipes(store_name, max_recipes)
181 result = {
182 "github_repository": None,
183 "git_repository_url": None,
184 "revisions": {},
185 # Set when an upstream request failed, so callers can avoid
186 # caching a transient failure as a negative answer.
187 "failed": False,
188 }
190 if not recipes:
191 return result
193 # Share the page budget across candidates: revisions being resolved
194 # are recent and builds are newest first, so breadth beats depth.
195 pages_each = max(1, max_pages // len(recipes))
196 revisions = result["revisions"]
197 source = None
198 fallback = None
200 for recipe in recipes:
201 collection_link = recipe.get("completed_builds_collection_link")
202 if not collection_link:
203 continue
205 git_repository_url = recipe.get("git_repository_url")
206 github_repository = extract_github_repository(git_repository_url)
207 if fallback is None:
208 fallback = (git_repository_url, github_repository)
210 builds, failed = self.iter_builds(collection_link, pages_each)
212 if (
213 self._merge_builds(builds, github_repository, revisions)
214 and source is None
215 ):
216 # The recipe that produced revisions, not the first sorted.
217 source = (git_repository_url, github_repository)
219 if failed:
220 # Launchpad is struggling, scanning the remaining candidates
221 # would just queue up more 12s timeouts on this request.
222 result["failed"] = True
223 break
225 if source is None:
226 source = fallback
227 if source:
228 result["git_repository_url"] = source[0]
229 result["github_repository"] = source[1]
231 return result