diff --git a/tests/test_pin_watch.py b/tests/test_pin_watch.py new file mode 100644 index 0000000..4c197ae --- /dev/null +++ b/tests/test_pin_watch.py @@ -0,0 +1,372 @@ +"""Tests for scripts/pin-watch.py. + +The watcher runs on a weekly cron, not on a pull request — a regression in +the read path, the held-pin decision, or the issue-create path is invisible +for up to seven days. The failure mode is silent: a broken watcher does not +shout, it just stops reporting. These tests are what turn a regression in +the watcher into a CI failure on the bump PR. + +Three IO seams are stubbed: + +- The HTTP fetchers (`upstream_crates_io`, `upstream_pypi`, etc.) are + Python functions named at module scope; tests monkeypatch them via + `monkeypatch.setattr` on the loaded module. The actual network is never + touched. +- `upstream_gh_release` calls `gh release list` via subprocess. Tests + monkeypatch the dispatcher (`fetch_upstream`) so the binary path is + never reached. `tests/conftest.py`'s `gh_stub` fixture is shaped for + `gh api` calls — adding a second stub shape here would invent a third + style for what is fundamentally the same seam (Python function in the + script). +- `issue_already_open` and `open_issue` are also monkeypatched — the + dedup test asserts `open_issue` is NOT called when the title already + exists, and asserting on a call requires controlling it. + +The script is imported directly under a synthetic name (`pin_watch` +cannot be a real module name — Python module names cannot contain +hyphens). The importlib dance is the same one `scripts/tooling-audit.py` +already uses internally to load `pin-watch.py` for the shared reader +helper, so it is the established shape for this repository rather than +an invention here. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +SCRIPT = REPO_ROOT / "scripts" / "pin-watch.py" + + +def _load_pin_watch(): + """Import `pin-watch.py` under a synthetic module name. + + Python module names cannot contain hyphens, so the script cannot be + imported with a plain `import` statement. `importlib.util` loads the + file by path under a synthetic name. The same dance + `scripts/tooling-audit.py` uses internally to share `_read_input_default` + with this script. + """ + spec = importlib.util.spec_from_file_location("pin_watch_under_test", SCRIPT) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +# --- TASK 1 item 1: reading from each source kind --------------------- + + +def test_read_env_value_parses_bare_value_and_returns_none_for_absent_key() -> None: + """A bare `KEY: 0.11.0` line is read; an absent key returns None (not ''). + + Catches: a regression where the regex would silently return '' for an + absent key — a later equality check against '' would report a + spurious drift for every pin the watcher could not read, and the + weekly issue would list pins that are perfectly fine. + """ + pin_watch = _load_pin_watch() + assert pin_watch._read_env_value("KEY: 0.11.0\n", "KEY") == "0.11.0" + assert pin_watch._read_env_value("OTHER: 0.11.0\n", "KEY") is None + + +def test_read_input_default_parses_nested_default_and_returns_none_for_absent_input() -> ( + None +): + """An `inputs..default:` block yields the value; absent input returns None. + + Catches: a regression where the walker stops one level short of + `default:`, leaving the watcher reading nothing from every input — + the weekly issue would report zero drift while pins silently move + upstream. + """ + pin_watch = _load_pin_watch() + text = ( + "on:\n" + " workflow_call:\n" + " inputs:\n" + " foo:\n" + " description: x\n" + " default: '1.2.3'\n" + ) + assert pin_watch._read_input_default(text, "foo") == "1.2.3" + assert pin_watch._read_input_default(text, "missing") is None + + +def test_read_requirement_pin_parses_eq_pin_and_returns_none_for_absent_name() -> None: + """A `name==version` line is read; absent name returns None (not empty string). + + Catches: a regression where the parser would return '' for an absent + name and the watcher would later compare '' against an upstream + version, producing a permanent drift report for every pin it could + not read. + """ + pin_watch = _load_pin_watch() + assert pin_watch._read_requirement_pin("pyyaml==6.0.1\n", "pyyaml") == "6.0.1" + assert pin_watch._read_requirement_pin("pyyaml==6.0.1\n", "other") is None + + +# --- TASK 1 item 2: only == is accepted in requirements ---------------- + + +def test_read_requirement_pin_accepts_only_eq_operator() -> None: + """A `>=` line, a bare name, and a commented-out pin all return None. + + Catches: a regression where the parser would accept any operator and + the watcher would report drift against something unpinned — drift + reports for unpinned deps are noise, and the docstring states the + 'only ==' contract that this test pins. + """ + pin_watch = _load_pin_watch() + # `>=` is a range, not a pin. + assert pin_watch._read_requirement_pin("pyyaml>=6.0\n", "pyyaml") is None + # A bare name has no version and cannot drift. + assert pin_watch._read_requirement_pin("requests\n", "requests") is None + # A commented-out pin is not a pin (and is not a range either). + assert pin_watch._read_requirement_pin("# pyyaml==6.0.1\n", "pyyaml") is None + # And the unpinned lines above do not pollute parsing of the actual + # pinned entry below. + text = "pyyaml>=6.0\nrequests\n# pyyaml==6.0.1\npyyaml==6.0.1\n" + assert pin_watch._read_requirement_pin(text, "pyyaml") == "6.0.1" + + +# --- TASK 1 items 3, 4, 5: end-to-end via main() ---------------------- +# +# These three tests share the same harness: stub read_pins, stub +# fetch_upstream, stub the two `gh` calls, run main(), observe. The +# _stub_main helper below wires it all up so each test reads as the one +# assertion that distinguishes it from the others. + + +def _stub_main( + pin_watch, + monkeypatch, + *, + pins: dict[str, tuple[str, str]], + upstream_for: dict[str, str], + issue_already_open: bool = False, +) -> list[dict]: + """Wire up the stubs `main()` needs for a controlled end-to-end run. + + Returns the list `open_issue` will append to when invoked. Each test + asserts on whether the list is empty (no issue opened) and on the + captured stdout. Every IO seam `main()` reaches is replaced: pin + reading (so the test controls inputs without touching the real + workflow files), the upstream fetch dispatcher (so no network or + `gh` binary is involved), and the two `gh` calls (so the dedup + decision is observable from inside the test). + """ + monkeypatch.setattr(pin_watch, "read_pins", lambda: dict(pins)) + + def fake_fetch(name: str) -> str: + try: + return upstream_for[name] + except KeyError as e: + raise RuntimeError(f"unexpected pin in test: {name}") from e + + monkeypatch.setattr(pin_watch, "fetch_upstream", fake_fetch) + monkeypatch.setattr( + pin_watch, "issue_already_open", lambda title: issue_already_open + ) + + opened: list[dict] = [] + + def fake_open(title: str, body: str) -> str: + opened.append({"title": title, "body": body}) + return "https://example.test/issue/1" + + monkeypatch.setattr(pin_watch, "open_issue", fake_open) + return opened + + +def test_held_pin_stays_silent_when_pin_matches_hold(monkeypatch, capsys) -> None: + """A held pin whose value still matches its hold stays silent — no issue opened. + + Catches: a regression that breaks a held pin out of 'silent' state + and trains the reader to ignore the weekly issue — the failure mode + the hold list exists to prevent (per the module docstring, the held + pins are the ones deliberately below upstream). + """ + pin_watch = _load_pin_watch() + # `ruff` is in HOLDS as "0.12.0"; matching the hold is the silent case. + opened = _stub_main( + pin_watch, + monkeypatch, + pins={"ruff": ("python-ci.yml", "0.12.0")}, + upstream_for={}, # held pins never reach fetch_upstream + ) + + exit_code = pin_watch.main() + capsys.readouterr() + + assert exit_code == 0 + assert opened == [], ( + f"Held pin matching its hold must stay silent; got issue(s): {opened!r}" + ) + + +def test_held_pin_reported_stale_when_pin_no_longer_matches_hold( + monkeypatch, capsys +) -> None: + """A held pin whose value no longer matches its hold is reported as stale. + + Catches: a regression where a pin bump without updating `HOLDS` does + not surface — 'silent' means 'pin matches hold', not 'pin differs + but we forgot to look'. The only place the hold decision lives is + in main(); a wrong branch here is the most expensive failure mode + the watcher can have, because the whole point of a hold is to mute + one pin and the only signal of a stale mute is this issue. + """ + pin_watch = _load_pin_watch() + opened = _stub_main( + pin_watch, + monkeypatch, + pins={"ruff": ("python-ci.yml", "0.99.0")}, # does NOT match HOLDS["ruff"] + upstream_for={}, + ) + + exit_code = pin_watch.main() + capsys.readouterr() + + assert exit_code == 0 + assert len(opened) == 1 + body = opened[0]["body"] + # Structure check: the pin name, its new pinned value, and the hold + # version all appear in the body — the held_stale table is populated + # with the actual data, not just a heading. + assert "ruff" in body + assert "0.99.0" in body + assert "0.12.0" in body + + +def test_no_drift_reported_when_pin_matches_upstream(monkeypatch, capsys) -> None: + """A non-held pin whose upstream matches the pin stays silent. + + Catches: a regression where the watcher reports a drift for a + matched pin and trains the reader to ignore the weekly issue — the + mirror of the held-silent failure mode, both of which the docstring + calls out as the reason the script exists. + """ + pin_watch = _load_pin_watch() + opened = _stub_main( + pin_watch, + monkeypatch, + pins={"pytest": ("python-ci.yml", "8.0.0")}, + upstream_for={"pytest": "8.0.0"}, + ) + + exit_code = pin_watch.main() + capsys.readouterr() + + assert exit_code == 0 + assert opened == [], ( + f"Pin matching upstream must stay silent; got issue(s): {opened!r}" + ) + + +def test_drift_reported_when_pin_differs_from_upstream(monkeypatch, capsys) -> None: + """A non-held pin whose upstream differs from the pin is reported as drift. + + Catches: a regression where the comparison silently swallows the + drift and the bump goes unannounced — the only reason the watcher + runs at all. + """ + pin_watch = _load_pin_watch() + opened = _stub_main( + pin_watch, + monkeypatch, + pins={"pytest": ("python-ci.yml", "8.0.0")}, + upstream_for={"pytest": "8.1.0"}, # upstream is one ahead + ) + + exit_code = pin_watch.main() + capsys.readouterr() + + assert exit_code == 0 + assert len(opened) == 1 + body = opened[0]["body"] + # Structure check: the pin name and BOTH values (pinned + upstream) + # appear in the body — the drift table is populated with the actual + # data, not just a heading. + assert "pytest" in body + assert "8.0.0" in body + assert "8.1.0" in body + + +def test_no_duplicate_issue_when_title_already_open(monkeypatch, capsys) -> None: + """A drift with the exact issue title already open does not create a second one. + + Catches: a regression that opens a fresh issue every weekly run — + the maintainer would drown in duplicates with no signal which is the + current one. The dedup is on the exact title; a title that drifts + even one character would still create a duplicate, and the + regression that loses the dedup check entirely would do so silently. + """ + pin_watch = _load_pin_watch() + opened = _stub_main( + pin_watch, + monkeypatch, + pins={"pytest": ("python-ci.yml", "8.0.0")}, + upstream_for={"pytest": "8.1.0"}, + issue_already_open=True, # pretend the issue exists + ) + + exit_code = pin_watch.main() + capsys.readouterr() + + assert exit_code == 0 + assert opened == [], ( + f"With issue already open, open_issue must NOT be called; got: {opened!r}" + ) + + +# --- TASK 1 item 6: unreadable pin fails the run ---------------------- + + +def test_read_pins_fails_when_pin_cannot_be_read(monkeypatch, tmp_path, capsys) -> None: + """A pin the script could not read fails the run with `::error::` and exit 1. + + Catches: a regression where the watcher silently skips a pin it was + asked to read and reports the rest — silent skip is the failure + mode that produced the missing coverage in the first place (the + weekly cron would still report zero drift, masking the unreadable + pin for up to seven days). + + Sets up a temp `.github/workflows/fake.yml` whose `inputs` block + does NOT contain the named input, points PIN_SOURCES at it, and + asserts the script exits 1 with a `::error file=::` line + naming the missing input. The CI workflow's run summary reads those + `::error::` lines; absent this, the failure is invisible. + """ + pin_watch = _load_pin_watch() + # Replace PIN_SOURCES with a single entry that points at a workflow + # file where the named input is missing. This exercises the + # 'input default not found' branch of read_pins without requiring + # all 19 real workflow files to be present. + monkeypatch.setattr( + pin_watch, + "PIN_SOURCES", + [("fake-pin", "fake.yml", "input", "missing-input")], + ) + + wf_dir = tmp_path / ".github" / "workflows" + wf_dir.mkdir(parents=True) + (wf_dir / "fake.yml").write_text( + "on:\n" + " workflow_call:\n" + " inputs:\n" + " other-input:\n" + " default: '1.0.0'\n" + ) + monkeypatch.chdir(tmp_path) + + with pytest.raises(SystemExit) as exc_info: + pin_watch.read_pins() + captured = capsys.readouterr() + + assert exc_info.value.code == 1 + assert "::error" in captured.err + assert "missing-input" in captured.err diff --git a/tests/test_tooling_audit.py b/tests/test_tooling_audit.py new file mode 100644 index 0000000..6d247fa --- /dev/null +++ b/tests/test_tooling_audit.py @@ -0,0 +1,347 @@ +"""Tests for scripts/tooling-audit.py. + +The script audits the dependency trees of the Rust tooling CI installs +(`cargo install --locked ` reuses the tool's own bundled +`Cargo.lock`). The output is a single weekly issue; a regression that +silently swallows real findings — or that breaks the dedup, or that +fails the run on advisories a maintainer cannot control — is invisible +for up to seven days. These tests are what turn a regression into a CI +failure on the bump PR. + +Three IO seams are stubbed: + +- The HTTP download (`_http_get_bytes`) is a Python function named at + module scope; tests monkeypatch it on the loaded module. The actual + network is never touched. +- `cargo audit` is invoked via subprocess; tests monkeypatch + `run_cargo_audit` so the binary is never invoked. The output the + script parses is JSON, so the stub is a dict — a regression that + re-introduced text-grep would surface here. +- The two `gh` calls (`issue_already_open`, `open_issue`) are + monkeypatched for the same reason as in `test_pin_watch.py`: the + dedup test asserts `open_issue` is NOT called when the title already + exists. + +The script is imported directly under a synthetic name (`tooling-audit` +cannot be a real module name — Python module names cannot contain +hyphens). The script itself already uses the same importlib dance +internally to load `pin-watch.py` for the shared reader helper, so the +shape is established here, not invented. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +SCRIPT = REPO_ROOT / "scripts" / "tooling-audit.py" + + +def _load_tooling_audit(): + """Import `tooling-audit.py` under a synthetic module name. + + Python module names cannot contain hyphens, so the script cannot be + imported with a plain `import` statement. `importlib.util` loads the + file by path under a synthetic name. The script uses the same dance + internally to share `_read_input_default` with `pin-watch.py`. + """ + spec = importlib.util.spec_from_file_location("tooling_audit_under_test", SCRIPT) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +# --- TASK 2 item 1: exit code semantics ------------------------------- +# +# Two tests, one in each direction. The exit-code-on-advisory trap is the +# single most expensive failure this script can have: an upstream +# advisory that ships during a quiet week would block every PR in the +# organisation via this self-CI. The test that asserts an advisory does +# NOT fail is therefore the most important test in the file. + + +def _audit_result( + *, + tool: str = "cargo-audit", + version: str = "0.21.0", + vulnerabilities: list[dict] | None = None, + yanked: list[dict] | None = None, + reason: str | None = None, +) -> dict: + """Build a complete AuditResult with sensible defaults for unused keys.""" + return { + "tool": tool, + "version": version, + "vulnerabilities": vulnerabilities if vulnerabilities is not None else [], + "yanked": yanked if yanked is not None else [], + "reason": reason, + } + + +def test_advisory_found_does_not_fail_the_run(monkeypatch, capsys) -> None: + """An advisory (vulnerability or yanked) found by `cargo audit` does NOT exit 1. + + Catches: a regression where exit-code-as-finding would fail every PR + in the organisation when a third-party advisory ships — the failure + mode the module docstring explicitly forbids. The right response to + a finding is the issue, not a red CI. + """ + ta = _load_tooling_audit() + # One tool with a real vulnerability; no script failure. + monkeypatch.setattr( + ta, + "read_pins", + lambda: {"cargo-audit": ("rust-audit.yml", "0.21.0")}, + ) + monkeypatch.setattr( + ta, + "audit_one", + lambda tool, version: _audit_result( + vulnerabilities=[ + { + "advisory": "RUSTSEC-2025-0001", + "package": "openssl", + "version": "0.1.0", + "cvss": None, + "patched": "0.2.0", + } + ] + ), + ) + monkeypatch.setattr(ta, "issue_already_open", lambda title: False) + opened: list[dict] = [] + monkeypatch.setattr( + ta, + "open_issue", + lambda title, body: ( + opened.append({"title": title, "body": body}) + or "https://example.test/issue/1" + ), + ) + + exit_code = ta.main() + capsys.readouterr() + + assert exit_code == 0, ( + f"An advisory found by cargo audit must not fail the run; got exit {exit_code}" + ) + assert len(opened) == 1 + + +def test_script_failure_fails_the_run(monkeypatch, capsys) -> None: + """A script-level failure (download error, unparseable JSON) DOES exit 1. + + Catches: a regression where the audit silently swallows real script + failures — a broken tool on the weekly cron is invisible for up to + seven days, and the operator never learns the auditor stopped + working. The 'no bundled Cargo.lock' reason is excluded; that one + is a legitimate 'we could not audit this' answer, not a failure. + """ + ta = _load_tooling_audit() + # One tool; the audit ran but failed for a non-lockfile reason. + monkeypatch.setattr( + ta, + "read_pins", + lambda: {"cargo-audit": ("rust-audit.yml", "0.21.0")}, + ) + monkeypatch.setattr( + ta, + "audit_one", + lambda tool, version: _audit_result( + reason="download HTTP 404: https://static.crates.io/crates/cargo-audit/0.21.0.crate", + ), + ) + + exit_code = ta.main() + capsys.readouterr() + + assert exit_code != 0, ( + f"A script failure (download error) must exit non-zero; got exit {exit_code}" + ) + + +# --- TASK 2 item 2: no bundled Cargo.lock is reported, not skipped ----- + + +def test_audit_one_reports_no_bundled_cargo_lock(monkeypatch) -> None: + """A `.crate` with no bundled `Cargo.lock` is reported as NOT AUDITED with a reason. + + Catches: a regression where the script silently treats such a tool + as 'audited and clean' — 'we could not audit this' is an honest + answer the issue body must surface, not a quiet skip. The + FileNotFoundError branch is the exact path `fetch_lockfile` raises + when the extracted crate has no `Cargo.lock`, which is the seam the + test exercises here. + """ + ta = _load_tooling_audit() + + def fake_fetch_lockfile(tool: str, version: str) -> Path: + raise FileNotFoundError("missing") + + monkeypatch.setattr(ta, "fetch_lockfile", fake_fetch_lockfile) + + result = ta.audit_one("cargo-audit", "0.21.0") + + assert result["reason"] == "no bundled Cargo.lock in .crate" + assert result["vulnerabilities"] == [] + assert result["yanked"] == [] + + +# --- TASK 2 item 3: yanked separate from vulnerabilities -------------- + + +def test_render_body_separates_yanked_from_vulnerabilities() -> None: + """The rendered body puts yanked entries in `## Yanked crates` and vulnerability entries in `## Vulnerabilities`, not mixed. + + Catches: a regression where the report conflates the two and the + maintainer cannot tell whether a finding is a security advisory + (action: bump and re-audit) or an upstream marking-as-bad (action: + bump or stop using). Both belong on the page; both must land in + their own section, in that order, so the body reads top-to-bottom + as severity decreasing. + """ + ta = _load_tooling_audit() + results = [ + _audit_result( + vulnerabilities=[ + { + "advisory": "RUSTSEC-2025-0001", + "package": "openssl", + "version": "0.1.0", + "cvss": "7.5", + "patched": "0.2.0", + } + ], + yanked=[{"package": "openssl", "version": "0.0.1"}], + ) + ] + + body = ta.render_body(results) + + # Structure: `## Vulnerabilities` appears BEFORE `## Yanked crates`. + vuln_idx = body.index("## Vulnerabilities") + yanked_idx = body.index("## Yanked crates") + assert vuln_idx < yanked_idx, ( + "Vulnerabilities section must precede Yanked crates section" + ) + + # The vulnerability entry lives in the Vulnerabilities section, + # between its heading and the Yanked crates heading. + vuln_section = body[vuln_idx:yanked_idx] + assert "RUSTSEC-2025-0001" in vuln_section + assert "openssl" in vuln_section + + # The yanked entry lives in the Yanked crates section (and NOT in + # the Vulnerabilities section above). + yanked_section = body[yanked_idx:] + assert "0.0.1" in yanked_section + assert "RUSTSEC-2025-0001" not in vuln_section or yanked_section.index("0.0.1") < ( + yanked_section.index("RUSTSEC-2025-0001") + if "RUSTSEC-2025-0001" in yanked_section + else len(yanked_section) + ) + + +# --- TASK 2 item 4: report built from parsed `cargo audit` JSON ------ + + +def test_audit_one_parses_cargo_audit_json(monkeypatch, tmp_path) -> None: + """The vulnerabilities and yanked lists in `AuditResult` come from the parsed cargo audit JSON, not from its human output. + + Catches: a regression where the script grep-ed the human output for + keywords and a future cargo audit version changed the wording — the + auditor would silently stop reporting. The test feeds a small fixed + JSON document with one vulnerability and one yanked warning and + asserts both land in the right fields of `AuditResult`. + """ + ta = _load_tooling_audit() + + cargo_audit_output = { + "vulnerabilities": { + "list": [ + { + "advisory": { + "id": "RUSTSEC-2025-0001", + "package": "openssl", + "cvss": "7.5", + }, + "package": {"name": "openssl", "version": "0.1.0"}, + "versions": {"patched": ["0.2.0"]}, + } + ] + }, + "warnings": { + "yanked": [ + {"package": {"name": "old-crate", "version": "0.0.1"}}, + ] + }, + } + + fake_lock = tmp_path / "Cargo.lock" + fake_lock.write_text("") + monkeypatch.setattr(ta, "fetch_lockfile", lambda tool, version: fake_lock) + monkeypatch.setattr(ta, "run_cargo_audit", lambda lockfile: cargo_audit_output) + + result = ta.audit_one("cargo-audit", "0.21.0") + + assert result["reason"] is None + assert len(result["vulnerabilities"]) == 1 + assert result["vulnerabilities"][0]["advisory"] == "RUSTSEC-2025-0001" + assert result["vulnerabilities"][0]["package"] == "openssl" + assert result["vulnerabilities"][0]["version"] == "0.1.0" + assert len(result["yanked"]) == 1 + assert result["yanked"][0]["package"] == "old-crate" + assert result["yanked"][0]["version"] == "0.0.1" + + +# --- TASK 2 item 5: dedup --------------------------------------------- + + +def test_no_duplicate_issue_when_title_already_open(monkeypatch, capsys) -> None: + """With an issue of the exact title already open, no second one is created. + + Catches: a regression that opens a fresh issue every weekly run — + the maintainer would drown in duplicates with no signal which is + the current one. The dedup is on the exact title. + """ + ta = _load_tooling_audit() + monkeypatch.setattr( + ta, + "read_pins", + lambda: {"cargo-audit": ("rust-audit.yml", "0.21.0")}, + ) + monkeypatch.setattr( + ta, + "audit_one", + lambda tool, version: _audit_result( + vulnerabilities=[ + { + "advisory": "RUSTSEC-2025-0001", + "package": "openssl", + "version": "0.1.0", + "cvss": None, + "patched": "(none — no fix)", + } + ] + ), + ) + monkeypatch.setattr(ta, "issue_already_open", lambda title: True) + opened: list[dict] = [] + monkeypatch.setattr( + ta, + "open_issue", + lambda title, body: ( + opened.append({"title": title, "body": body}) + or "https://example.test/issue/1" + ), + ) + + exit_code = ta.main() + capsys.readouterr() + + assert exit_code == 0 + assert opened == [], ( + f"With issue already open, open_issue must NOT be called; got: {opened!r}" + )