Coverage for src/footman/_provision.py: 84%
190 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
1"""Fetch the latest curated tools into a throwaway prefix — the engine behind
2`fm footman tools provision`.
4The stubs are read from the *installed* binaries (`fm footman tools sync`), so
5telling an editor what the newest release accepts means having the newest
6release on `PATH` — across five ecosystems (PyPI, npm, bun, Go, C++), none of
7which should be allowed to touch the machine's own environment.
9One isolated prefix answers all of it. Almost every curated tool ships an
10installable PyPI wheel — including the Rust ones (ruff, uv, prek, git-cliff)
11and the C++ ones (cmake, ninja) — so `uv tool install --upgrade` into a
12private `UV_TOOL_DIR`/`UV_TOOL_BIN_DIR` covers the majority and cleans up with
13one `rm -rf`. What's left is bun (its own release), the node CLIs it installs,
14and the Go CLIs (a prebuilt release asset):
16* **uv** — `uv tool install --upgrade <pkg>`, tools and launchers under the
17 prefix; nothing lands in `~/.local` or the system site-packages.
18* **bun** — bun's GitHub release, unpacked into the prefix. Provisioned
19 *first*, because the node tier runs through it.
20* **node** — `bun add --global` with `BUN_INSTALL` pointed at the prefix.
21* **github / gitlab** — the latest release asset for this platform, matched
22 from the release's own asset list (so `Darwin`/`x86_64` vs `darwin`/`x64`
23 naming needn't be transcribed), unpacked, binary placed in the prefix.
24* **system** — git, docker, the uv running this: already on `PATH`, left be.
25* **deferred** — parked, with a reason (tea, until it stops hanging on
26 `--help`).
28Everything writes under one prefix and `PATH="<prefix>/bin:$PATH"` is all a
29`sync` needs to read the newest binaries; deleting the prefix undoes it. This
30is a maintainer tool: it shells out and downloads, and it is never on the
31completion hot path.
32"""
34from __future__ import annotations
36import json
37import os
38import platform
39import shutil
40import subprocess
41import tarfile
42import urllib.error
43import urllib.parse
44import urllib.request
45import zipfile
46from dataclasses import dataclass
47from pathlib import Path
49from footman._drivers import Driver
52class ProvisionError(Exception):
53 """A tool could not be fetched — reported per tool, never fatal."""
56@dataclass(frozen=True)
57class Outcome:
58 """What became of one tool: the line `provision` prints."""
60 key: str
61 kind: str
62 status: str # "ok" | "fail" | "skip" | "deferred"
63 detail: str = ""
66def bin_dir(prefix: Path) -> Path:
67 """The one directory to put on `PATH`; every tier lands its launchers here."""
68 return prefix / "bin"
71def provision(
72 drivers: tuple[Driver, ...], prefix: Path, *, only: str = ""
73) -> list[Outcome]:
74 """Materialise the latest of each curated tool under *prefix*.
76 Tiers run in the one order that matters: bun before the node CLIs that
77 need it. Each tool's failure is its own line, never the run's — a missing
78 binary should read as one skipped hint, not a broken provision.
79 """
80 prefix = Path(prefix)
81 bin_dir(prefix).mkdir(parents=True, exist_ok=True)
82 chosen = [d for d in drivers if not only or d.key == only]
83 outcomes: list[Outcome] = []
84 by_kind: dict[str, list[Driver]] = {}
85 for driver in chosen:
86 if driver.source == "manual": 86 ↛ 90line 86 didn't jump to line 90 because the condition on line 86 was never true
87 # A hand-written stub (the shells): its stub is curated, not read
88 # from a binary, so there is nothing to fetch — skip it rather than
89 # try `uv tool install bash` and print a spurious failure.
90 outcomes.append(Outcome(driver.key, "manual", "skip", "hand-written"))
91 continue
92 by_kind.setdefault(driver.provision.kind, []).append(driver)
94 for driver in by_kind.get("system", []):
95 outcomes.append(
96 Outcome(driver.key, "system", "skip", f"uses the system {driver.name}")
97 )
98 for driver in by_kind.get("deferred", []):
99 outcomes.append(
100 Outcome(driver.key, "deferred", "deferred", driver.provision.note)
101 )
102 outcomes += _uv_tier(prefix, by_kind.get("uv", []))
103 outcomes += _python_tier(prefix, by_kind.get("python", []))
104 for driver in by_kind.get("bun", []): # before node: node runs through bun 104 ↛ 105line 104 didn't jump to line 105 because the loop on line 104 never started
105 outcomes.append(_release(prefix, driver, host="github"))
106 outcomes += _node_tier(prefix, by_kind.get("node", []))
107 for driver in by_kind.get("github", []) + by_kind.get("gitlab", []):
108 outcomes.append(_release(prefix, driver, host=driver.provision.kind))
109 return outcomes
112# --- uv tier -----------------------------------------------------------------
115def _uv_env(prefix: Path) -> dict[str, str]:
116 """uv's install targets, redirected so nothing escapes the prefix."""
117 return {
118 **os.environ,
119 "UV_TOOL_DIR": str(prefix / "uv-tools"),
120 "UV_TOOL_BIN_DIR": str(bin_dir(prefix)),
121 }
124def _uv_tier(prefix: Path, drivers: list[Driver]) -> list[Outcome]:
125 """`uv tool install --upgrade` each distinct package into the prefix.
127 A driver's `provision.plugins` ride along as `--with` packages in the tool's
128 own isolated environment, so a plugin-extended CLI (pytest + pytest-cov) is
129 installed whole and its plugin flags are there to read.
130 """
131 env = _uv_env(prefix)
132 installed: dict[tuple[str, tuple[str, ...]], bool] = {}
133 outcomes: list[Outcome] = []
134 for driver in drivers:
135 package = driver.provision.target(driver.name)
136 plugins = driver.provision.plugins
137 key = (package, plugins)
138 if key not in installed:
139 withs = [f"--with={p}" for p in plugins]
140 installed[key] = _run(
141 ["uv", "tool", "install", "--upgrade", package, *withs], env=env
142 )
143 ok = installed[key]
144 detail = package if not plugins else f"{package} (+{', '.join(plugins)})"
145 outcomes.append(Outcome(driver.key, "uv", "ok" if ok else "fail", detail))
146 return outcomes
149# --- python tier (an interpreter to read `--help` from) ----------------------
152def _python_tier(prefix: Path, drivers: list[Driver]) -> list[Outcome]:
153 """`uv python install` each requested interpreter, linked into the prefix.
155 python is provisioned like any other tool — an interpreter whose `--help`
156 is read for the stub. The *runtime* `tools.python` always targets
157 `sys.executable`; provisioning only supplies versions to extract from, so
158 the stub reflects real pythons rather than whatever `python`/`python3` a
159 machine happens to have on PATH.
160 """
161 outcomes: list[Outcome] = []
162 for driver in drivers: 162 ↛ 163line 162 didn't jump to line 163 because the loop on line 162 never started
163 version = driver.provision.package or "3"
164 if not _run(["uv", "python", "install", version], env=dict(os.environ)):
165 outcomes.append(
166 Outcome(driver.key, "python", "fail", f"uv python install {version}")
167 )
168 continue
169 try:
170 found = subprocess.run(
171 ["uv", "python", "find", version],
172 capture_output=True,
173 text=True,
174 timeout=60,
175 env=dict(os.environ),
176 )
177 path = Path(found.stdout.strip())
178 except (OSError, subprocess.SubprocessError):
179 path = Path()
180 if not path.name or not path.exists():
181 outcomes.append(
182 Outcome(driver.key, "python", "fail", f"no python {version} found")
183 )
184 continue
185 link = bin_dir(prefix) / "python"
186 link.unlink(missing_ok=True)
187 link.symlink_to(path)
188 outcomes.append(Outcome(driver.key, "python", "ok", f"{version} ({path})"))
189 return outcomes
192# --- node tier (through the provisioned bun) ---------------------------------
195def _node_tier(prefix: Path, drivers: list[Driver]) -> list[Outcome]:
196 """`bun add --global` each package, with bun's install dir the prefix."""
197 if not drivers:
198 return []
199 bun = bin_dir(prefix) / "bun"
200 if not bun.exists():
201 return [
202 Outcome(d.key, "node", "fail", "bun was not provisioned first")
203 for d in drivers
204 ]
205 env = {
206 **os.environ,
207 "BUN_INSTALL": str(prefix), # global bin lands in <prefix>/bin
208 "PATH": f"{bin_dir(prefix)}{os.pathsep}{os.environ.get('PATH', '')}",
209 }
210 packages = sorted({d.provision.target(d.name) for d in drivers})
211 ok = _run([str(bun), "add", "--global", *packages], env=env)
212 return [
213 Outcome(d.key, "node", "ok" if ok else "fail", d.provision.target(d.name))
214 for d in drivers
215 ]
218# --- release tier (github / gitlab, and bun) ---------------------------------
221def _release(prefix: Path, driver: Driver, *, host: str) -> Outcome:
222 """Download the latest release asset for this platform and unpack it."""
223 kind = driver.provision.kind
224 try:
225 assets = _latest_assets(host, driver.provision.repo)
226 name, url = _pick_asset(assets)
227 archive = _download(url, prefix)
228 placed = _extract_binary(archive, driver.name, bin_dir(prefix))
229 except ProvisionError as exc:
230 return Outcome(driver.key, kind, "fail", str(exc))
231 return Outcome(driver.key, kind, "ok", f"{placed.name} ({name})")
234def _latest_assets(host: str, repo: str) -> list[tuple[str, str]]:
235 """`[(asset name, download url)]` for *repo*'s latest release."""
236 if not repo:
237 raise ProvisionError("no repo to fetch from")
238 if host == "github":
239 data = _get_json(f"https://api.github.com/repos/{repo}/releases/latest")
240 assets = data.get("assets", [])
241 return [(a["name"], a["browser_download_url"]) for a in assets]
242 if host == "gitlab":
243 quoted = urllib.parse.quote(repo, safe="")
244 data = _get_json(
245 f"https://gitlab.com/api/v4/projects/{quoted}/releases/permalink/latest"
246 )
247 links = data.get("assets", {}).get("links", [])
248 return [(a["name"], a.get("direct_asset_url") or a["url"]) for a in links]
249 raise ProvisionError(f"unknown release host {host!r}")
252# The alias sets that fold one platform's many spellings into a match: bun
253# says `darwin`/`aarch64`, goreleaser `Darwin`/`x86_64`, gh `macOS`/`amd64`.
254_OS_ALIASES = {
255 "darwin": ("darwin", "macos", "apple", "osx"),
256 "linux": ("linux",),
257 "windows": ("windows", "win"),
258}
259_ARCH_ALIASES = {
260 "arm64": ("arm64", "aarch64"),
261 "aarch64": ("arm64", "aarch64"),
262 "x86_64": ("x86_64", "amd64", "x64", "x86-64"),
263 "amd64": ("x86_64", "amd64", "x64", "x86-64"),
264}
265_ARCHIVES = (".tar.gz", ".tgz", ".tar.xz", ".tar.bz2", ".zip")
266# Sidecar files that ride alongside a real asset — never the binary.
267_SIDECARS = (".sha256", ".sha256sum", ".sig", ".asc", ".txt", ".pem", ".sbom")
268# Build variants that sit beside the canonical asset for the same platform:
269# bun's `-profile`/`-baseline`, a `-debug` build, a `musl` libc. Preferred
270# against, never excluded — the canonical build is what a task wants.
271_VARIANTS = ("profile", "baseline", "debug", "musl", "-static")
274def _platform_tokens() -> tuple[tuple[str, ...], tuple[str, ...]]:
275 """This machine's OS and CPU aliases, for matching an asset name."""
276 os_aliases = _OS_ALIASES.get(
277 platform.system().lower(), (platform.system().lower(),)
278 )
279 machine = platform.machine().lower()
280 arch_aliases = _ARCH_ALIASES.get(machine, (machine,))
281 return os_aliases, arch_aliases
284def _pick_asset(assets: list[tuple[str, str]]) -> tuple[str, str]:
285 """The one asset for this OS and CPU, archives before bare binaries."""
286 os_aliases, arch_aliases = _platform_tokens()
288 def matches(name: str) -> bool:
289 low = name.lower()
290 if low.endswith(_SIDECARS):
291 return False
292 return any(o in low for o in os_aliases) and any(a in low for a in arch_aliases)
294 candidates = [(name, url) for name, url in assets if matches(name)]
295 if not candidates:
296 raise ProvisionError("no release asset for this platform")
298 def rank(asset: tuple[str, str]) -> tuple[bool, bool, int, str]:
299 # Prefer an archive over a bare binary, the canonical build over a
300 # variant (bun ships `-profile`/`-baseline` beside the plain one), and
301 # then the shortest name — a qualifier only ever lengthens it.
302 low = asset[0].lower()
303 variant = any(marker in low for marker in _VARIANTS)
304 return (not low.endswith(_ARCHIVES), variant, len(asset[0]), asset[0])
306 candidates.sort(key=rank)
307 return candidates[0]
310# --- download + unpack -------------------------------------------------------
313def _get_json(url: str) -> dict:
314 """A JSON API response — GitHub/GitLab both want a User-Agent."""
315 request = urllib.request.Request(url, headers={"User-Agent": "footman-provision"})
316 try:
317 with urllib.request.urlopen(request, timeout=30) as response:
318 return json.loads(response.read().decode("utf-8"))
319 except (urllib.error.URLError, ValueError, OSError) as exc:
320 raise ProvisionError(f"{url}: {exc}") from exc
323def _download(url: str, prefix: Path) -> Path:
324 """Fetch *url* into the prefix's cache, reusing a prior download by name."""
325 cache = prefix / ".cache"
326 cache.mkdir(parents=True, exist_ok=True)
327 dest = cache / url.rsplit("/", 1)[-1]
328 if dest.exists() and dest.stat().st_size:
329 return dest
330 request = urllib.request.Request(url, headers={"User-Agent": "footman-provision"})
331 try:
332 with (
333 urllib.request.urlopen(request, timeout=120) as response,
334 open(dest, "wb") as out,
335 ):
336 shutil.copyfileobj(response, out)
337 except (urllib.error.URLError, OSError) as exc:
338 raise ProvisionError(f"{url}: {exc}") from exc
339 return dest
342def _extract_binary(archive: Path, tool: str, into: Path) -> Path:
343 """Unpack *archive* and place its `tool` binary in *into*, executable.
345 Release archives nest the binary under a versioned directory, so the
346 whole tree is searched for a file named `tool` (or `tool.exe`); a bare
347 downloaded binary is taken as-is.
348 """
349 wanted = {tool, f"{tool}.exe"}
350 into.mkdir(parents=True, exist_ok=True)
351 dest = into / tool
352 if archive.name.lower().endswith((".tar.gz", ".tgz", ".tar.xz", ".tar.bz2")):
353 with tarfile.open(archive) as tar:
354 member = next(
355 (m for m in tar.getmembers() if Path(m.name).name in wanted), None
356 )
357 if member is None:
358 raise ProvisionError(f"{tool} not found inside {archive.name}")
359 source = tar.extractfile(member)
360 if source is None: 360 ↛ 361line 360 didn't jump to line 361 because the condition on line 360 was never true
361 raise ProvisionError(f"{tool} is not a file inside {archive.name}")
362 dest.write_bytes(source.read())
363 elif archive.name.lower().endswith(".zip"): 363 ↛ 370line 363 didn't jump to line 370 because the condition on line 363 was always true
364 with zipfile.ZipFile(archive) as zf:
365 name = next((n for n in zf.namelist() if Path(n).name in wanted), None)
366 if name is None: 366 ↛ 367line 366 didn't jump to line 367 because the condition on line 366 was never true
367 raise ProvisionError(f"{tool} not found inside {archive.name}")
368 dest.write_bytes(zf.read(name))
369 else: # a bare binary, downloaded directly
370 dest.write_bytes(archive.read_bytes())
371 dest.chmod(0o755)
372 return dest
375# --- subprocess --------------------------------------------------------------
378def _run(argv: list[str], *, env: dict[str, str]) -> bool:
379 """Run an install command, quietly; its success is all the caller needs."""
380 try:
381 done = subprocess.run(
382 argv, env=env, capture_output=True, text=True, timeout=600
383 )
384 except (OSError, subprocess.SubprocessError):
385 return False
386 return done.returncode == 0