Coverage for cli / commands / deps_cmd.py: 100.00%
128 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Dependency maintenance commands.
3``gco deps scan`` wraps the repository's dependency scanner
4(``.github/scripts/dependency-scan.sh``) — the same script the monthly
5``deps-scan`` workflow runs to build the rolling
6"[Automated] Dependency updates available" issue — so an operator or agent
7can generate the exact same update list on demand instead of waiting for
8the schedule or hand-assembling the invocation.
10The scanner communicates through the GitHub Actions file-output protocol
11(``$GITHUB_OUTPUT``); this wrapper points that at a private temp file and
12reads back ``has_drift`` / ``scan_complete`` / ``report_path``, so the
13script itself runs bit-for-bit the way CI runs it.
15Two operating modes:
17* Full scan (default) — every surface the workflow checks: Python/npm
18 pins, Docker images, Helm charts, EKS add-ons, Dockerfile.dev ARGs,
19 autopilot pins, pre-commit hooks, CI tooling, version consistency,
20 suppression expiries, lockfile freshness, and the accelerator-catalog /
21 Karpenter NodePool policy. Surfaces that need AWS credentials or tools
22 the host is missing are skipped and reported as incomplete, exactly as
23 in CI.
24* ``--nodepools-only`` — just the accelerator-catalog / NodePool freshness
25 check (``scripts/accelerator_catalog.py``): the deterministic offline
26 validation always runs; the live EC2 catalog comparison runs when AWS
27 credentials resolve and is reported as skipped otherwise.
29Honest side-effect warning: the full scan's Python surface runs
30``pip install -e ".[<every extra>]"`` into the *active environment* (that
31is how it asks pip for outdated direct pins), mirroring the throwaway CI
32environment. Run it from the dev container or a dedicated venv if that
33matters to you.
34"""
36from __future__ import annotations
38import contextlib
39import json
40import os
41import re
42import shutil
43import subprocess
44import sys
45import tempfile
46from pathlib import Path
47from typing import NoReturn, cast
49import click
51#: Tools the full scan shells out to, and the surfaces that go incomplete
52#: without them. Missing entries are warnings, not errors — the scanner
53#: records the gap and keeps going, exactly as it does in CI.
54_OPTIONAL_TOOLS: tuple[tuple[str, str], ...] = (
55 ("jq", "Python-package report rendering"),
56 ("curl", "npm / GitHub / endoflife.date lookups"),
57 ("skopeo", "Docker image tag and digest checks"),
58 ("helm", "Helm chart version checks"),
59 ("aws", "EKS / Aurora / EMR / Bedrock / online accelerator checks"),
60)
62_SCAN_SCRIPT = Path(".github") / "scripts" / "dependency-scan.sh"
63_CATALOG_SCRIPT = Path("scripts") / "accelerator_catalog.py"
66def _fail(message: str) -> NoReturn:
67 raise click.ClickException(message)
70def _repo_root() -> Path:
71 """Resolve the checkout root; the scanner only exists in a git checkout."""
72 result = subprocess.run(
73 ["git", "rev-parse", "--show-toplevel"],
74 capture_output=True,
75 text=True,
76 check=False,
77 )
78 if result.returncode != 0:
79 _fail(
80 "gco deps scan must run from inside a GCO checkout "
81 "(the dependency scanner lives under .github/scripts/)"
82 )
83 root = Path(result.stdout.strip())
84 if not (root / _SCAN_SCRIPT).is_file():
85 _fail(f"{root} has no {_SCAN_SCRIPT} — not a GCO checkout?")
86 return root
89def _parse_github_output(path: Path) -> dict[str, str]:
90 """Parse the ``key=value`` lines the scanner writes to $GITHUB_OUTPUT."""
91 outputs: dict[str, str] = {}
92 try:
93 text = path.read_text(encoding="utf-8")
94 except OSError:
95 return outputs
96 for line in text.splitlines():
97 key, sep, value = line.partition("=")
98 if sep:
99 outputs[key.strip()] = value.strip()
100 return outputs
103def _sts_identity_available() -> bool:
104 """Mirror the scanner's credential preflight for the nodepools fast path."""
105 if shutil.which("aws") is None:
106 return False
107 probe = subprocess.run(
108 ["aws", "sts", "get-caller-identity"],
109 capture_output=True,
110 text=True,
111 check=False,
112 )
113 return probe.returncode == 0
116def _run_nodepools_check(repo_root: Path) -> dict[str, object]:
117 """Run the accelerator-catalog / NodePool freshness checks.
119 Returns a JSON-friendly envelope with an ``offline`` section (always
120 runs; deterministic) and an ``online`` section (runs when AWS
121 credentials resolve, mirrors the scanner's STS preflight).
122 """
123 offline_report = subprocess.run(
124 [sys.executable, str(_CATALOG_SCRIPT), "validate", "--format", "markdown"],
125 cwd=repo_root,
126 capture_output=True,
127 text=True,
128 check=False,
129 )
130 if offline_report.returncode not in (0, 1):
131 _fail(
132 "accelerator catalog validation failed operationally: "
133 + (offline_report.stderr.strip() or f"exit {offline_report.returncode}")
134 )
135 finding_count = len(re.findall(r"^### ", offline_report.stdout, re.M))
136 offline: dict[str, object] = {
137 "status": "pass" if offline_report.returncode == 0 else "findings",
138 "finding_count": finding_count,
139 "report_markdown": offline_report.stdout,
140 }
142 online: dict[str, object]
143 if not _sts_identity_available():
144 online = {
145 "status": "skipped",
146 "skip_reason": (
147 "No AWS credentials available for the online EC2 catalog check "
148 "(needs ec2:DescribeRegions and ec2:DescribeInstanceTypes); "
149 "offline policy validation still ran."
150 ),
151 }
152 else:
153 with tempfile.TemporaryDirectory(prefix="gco-deps-") as tmp:
154 online_report_path = Path(tmp) / "online.md"
155 online_run = subprocess.run(
156 [
157 sys.executable,
158 str(_CATALOG_SCRIPT),
159 "check-online",
160 "--report",
161 str(online_report_path),
162 "--json-summary",
163 ],
164 cwd=repo_root,
165 capture_output=True,
166 text=True,
167 check=False,
168 )
169 if online_run.returncode not in (0, 1):
170 _fail(
171 "online accelerator catalog check failed operationally: "
172 + (online_run.stderr.strip() or f"exit {online_run.returncode}")
173 )
174 try:
175 summary = json.loads(online_run.stdout)
176 except json.JSONDecodeError:
177 _fail("online accelerator catalog check emitted a malformed JSON summary")
178 online = {
179 "status": summary.get("status", "error"),
180 "drift_count": summary.get("drift_count"),
181 "regions_checked": summary.get("regions_checked"),
182 }
183 with contextlib.suppress(OSError):
184 online["report_markdown"] = online_report_path.read_text(encoding="utf-8")
186 has_drift = offline["status"] != "pass" or online.get("status") == "drift"
187 return {
188 "nodepools_only": True,
189 "has_drift": has_drift,
190 "scan_complete": online.get("status") != "skipped",
191 "offline": offline,
192 "online": online,
193 }
196def _run_full_scan(repo_root: Path, *, stream: bool) -> dict[str, object]:
197 """Run the full dependency scanner and return its parsed envelope."""
198 missing = [
199 f"{tool} ({surfaces})" for tool, surfaces in _OPTIONAL_TOOLS if shutil.which(tool) is None
200 ]
201 if missing:
202 click.echo(
203 "warning: missing tools — these surfaces will be reported as "
204 "incomplete: " + "; ".join(missing),
205 err=True,
206 )
208 with tempfile.TemporaryDirectory(prefix="gco-deps-") as tmp:
209 github_output = Path(tmp) / "github-output"
210 github_output.touch()
211 env = dict(os.environ)
212 env["GITHUB_OUTPUT"] = str(github_output)
213 # Never leak into a real Actions job summary if the caller's
214 # environment happens to carry one.
215 env.pop("GITHUB_STEP_SUMMARY", None)
216 env.setdefault("WORKFLOWS_DIR", ".github/workflows")
218 result = subprocess.run( # fixed argv, repo-owned script
219 ["bash", str(_SCAN_SCRIPT)],
220 cwd=repo_root,
221 env=env,
222 check=False,
223 capture_output=not stream,
224 text=True,
225 )
226 if result.returncode != 0:
227 detail = "" if stream else f"\n{(result.stderr or '')[-2000:]}"
228 _fail(f"dependency scanner exited with status {result.returncode}{detail}")
230 outputs = _parse_github_output(github_output)
231 has_drift = outputs.get("has_drift") == "true"
232 scan_complete = outputs.get("scan_complete") == "true"
234 report_markdown: str
235 if has_drift:
236 report_path = outputs.get("report_path", "")
237 try:
238 report_markdown = Path(report_path).read_text(encoding="utf-8")
239 except OSError:
240 _fail("the scanner reported drift but its report file is unreadable")
241 else:
242 report_markdown = "# Dependency Update Report\n\n" + (
243 "All dependencies are up to date.\n"
244 if scan_complete
245 else "No drift was found in completed checks, but the scan is "
246 "incomplete — zero-count surfaces are provisional. See the "
247 "scan log for skipped checks.\n"
248 )
250 # The scanner writes the same findings as a JSON document next to the
251 # report, on every run (``findings_path``). It rides along in the
252 # envelope so ``-o json`` callers and the MCP tool get the rows
253 # without parsing the Markdown. Absent or unreadable is ``None``
254 # rather than an error: the report is still the primary output.
255 findings: object = None
256 findings_path = outputs.get("findings_path", "")
257 if findings_path:
258 with contextlib.suppress(OSError, json.JSONDecodeError):
259 findings = json.loads(Path(findings_path).read_text(encoding="utf-8"))
261 envelope: dict[str, object] = {
262 "has_drift": has_drift,
263 "scan_complete": scan_complete,
264 "report_markdown": report_markdown,
265 "findings": findings,
266 }
267 if not stream:
268 envelope["log_tail"] = (result.stdout or "").splitlines()[-40:]
269 return envelope
272@click.group()
273def deps() -> None:
274 """Dependency maintenance (update scans, NodePool registry freshness)."""
277@deps.command("scan")
278@click.option(
279 "--nodepools-only",
280 is_flag=True,
281 default=False,
282 help=(
283 "Run only the accelerator-catalog / Karpenter NodePool freshness "
284 "check instead of the full scan."
285 ),
286)
287@click.option(
288 "--report",
289 "report_file",
290 type=click.Path(dir_okay=False, writable=True, path_type=Path),
291 default=None,
292 help="Write the Markdown report to this file instead of stdout.",
293)
294@click.pass_context
295def scan(ctx: click.Context, nodepools_only: bool, report_file: Path | None) -> None:
296 """Generate the dependency update list the monthly deps-scan produces.
298 Runs the same scanner as the ``deps-scan`` GitHub Actions workflow and
299 prints its Markdown report, so the update list in the rolling
300 "[Automated] Dependency updates available" issue can be reproduced on
301 demand. Surfaces that need AWS credentials or missing host tools are
302 skipped and flagged as incomplete rather than failing the run.
304 With ``--nodepools-only``, runs just the accelerator catalog /
305 NodePool policy checks (offline always; live EC2 comparison when AWS
306 credentials resolve).
308 With the global ``-o json``, prints a machine-readable envelope
309 (``has_drift``, ``scan_complete``, ``report_markdown``) instead of
310 the bare report — this is the shape the MCP ``deps_scan`` tool
311 returns.
312 """
313 repo_root = _repo_root()
314 json_output = bool(ctx.obj) and getattr(ctx.obj, "output_format", "table") == "json"
316 if nodepools_only:
317 envelope = _run_nodepools_check(repo_root)
318 else:
319 envelope = _run_full_scan(repo_root, stream=not json_output)
321 if json_output:
322 from ..output import emit_structured_document
324 emit_structured_document(
325 envelope,
326 output_format="json",
327 rendered=json.dumps(envelope, indent=2),
328 )
329 return
331 if nodepools_only:
332 offline = cast("dict[str, object]", envelope["offline"])
333 online = cast("dict[str, object]", envelope["online"])
334 report_markdown = str(offline.get("report_markdown", ""))
335 if online.get("report_markdown"):
336 report_markdown += "\n" + str(online["report_markdown"])
337 click.echo(f"offline policy check: {offline['status']}", err=True)
338 click.echo(f"online EC2 catalog: {online['status']}", err=True)
339 else:
340 report_markdown = str(envelope["report_markdown"])
341 click.echo(f"has_drift: {envelope['has_drift']}", err=True)
342 click.echo(f"scan_complete: {envelope['scan_complete']}", err=True)
344 if report_file is not None:
345 report_file.write_text(report_markdown, encoding="utf-8")
346 click.echo(f"report: {report_file}", err=True)
347 else:
348 click.echo(report_markdown)