Coverage for cli / commands / deps_cmd.py: 100.00%
123 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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( # noqa: S603 — 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 envelope: dict[str, object] = {
251 "has_drift": has_drift,
252 "scan_complete": scan_complete,
253 "report_markdown": report_markdown,
254 }
255 if not stream:
256 envelope["log_tail"] = (result.stdout or "").splitlines()[-40:]
257 return envelope
260@click.group()
261def deps() -> None:
262 """Dependency maintenance (update scans, NodePool registry freshness)."""
265@deps.command("scan")
266@click.option(
267 "--nodepools-only",
268 is_flag=True,
269 default=False,
270 help=(
271 "Run only the accelerator-catalog / Karpenter NodePool freshness "
272 "check instead of the full scan."
273 ),
274)
275@click.option(
276 "--report",
277 "report_file",
278 type=click.Path(dir_okay=False, writable=True, path_type=Path),
279 default=None,
280 help="Write the Markdown report to this file instead of stdout.",
281)
282@click.pass_context
283def scan(ctx: click.Context, nodepools_only: bool, report_file: Path | None) -> None:
284 """Generate the dependency update list the monthly deps-scan produces.
286 Runs the same scanner as the ``deps-scan`` GitHub Actions workflow and
287 prints its Markdown report, so the update list in the rolling
288 "[Automated] Dependency updates available" issue can be reproduced on
289 demand. Surfaces that need AWS credentials or missing host tools are
290 skipped and flagged as incomplete rather than failing the run.
292 With ``--nodepools-only``, runs just the accelerator catalog /
293 NodePool policy checks (offline always; live EC2 comparison when AWS
294 credentials resolve).
296 With the global ``-o json``, prints a machine-readable envelope
297 (``has_drift``, ``scan_complete``, ``report_markdown``) instead of
298 the bare report — this is the shape the MCP ``deps_scan`` tool
299 returns.
300 """
301 repo_root = _repo_root()
302 json_output = bool(ctx.obj) and getattr(ctx.obj, "output_format", "table") == "json"
304 if nodepools_only:
305 envelope = _run_nodepools_check(repo_root)
306 else:
307 envelope = _run_full_scan(repo_root, stream=not json_output)
309 if json_output:
310 from ..output import emit_structured_document
312 emit_structured_document(
313 envelope,
314 output_format="json",
315 rendered=json.dumps(envelope, indent=2),
316 )
317 return
319 if nodepools_only:
320 offline = cast("dict[str, object]", envelope["offline"])
321 online = cast("dict[str, object]", envelope["online"])
322 report_markdown = str(offline.get("report_markdown", ""))
323 if online.get("report_markdown"):
324 report_markdown += "\n" + str(online["report_markdown"])
325 click.echo(f"offline policy check: {offline['status']}", err=True)
326 click.echo(f"online EC2 catalog: {online['status']}", err=True)
327 else:
328 report_markdown = str(envelope["report_markdown"])
329 click.echo(f"has_drift: {envelope['has_drift']}", err=True)
330 click.echo(f"scan_complete: {envelope['scan_complete']}", err=True)
332 if report_file is not None:
333 report_file.write_text(report_markdown, encoding="utf-8")
334 click.echo(f"report: {report_file}", err=True)
335 else:
336 click.echo(report_markdown)