diff --git a/core/cli.py b/core/cli.py index ab4e73b4..0499df6c 100644 --- a/core/cli.py +++ b/core/cli.py @@ -2014,6 +2014,23 @@ def cmd_teaching_coverage(args: argparse.Namespace) -> int: split = args.split or "train_sample" version = args.version or "v1" + # Validate inputs against a strict whitelist before any path + # construction or subprocess invocation. The runner module name is + # built from these tokens (``f"evals.{lane}.{split}.{version}.runner"``) + # and passed to ``python -m``. Python's module loader would reject + # most malicious payloads, but a strict whitelist is the defense-in- + # depth response to the Sourcery security advisory: reject + # everything except ``[a-z0-9_]+``. + import re as _re + _safe_token_re = _re.compile(r"^[a-z0-9_]+$") + for label, value in (("lane", lane), ("split", split), ("version", version)): + if not _safe_token_re.match(value): + print( + f"ERROR: {label}={value!r} must match ^[a-z0-9_]+$", + file=sys.stderr, + ) + return 1 + repo_root = Path(__file__).resolve().parent.parent lane_dir = repo_root / "evals" / lane / split / version if not lane_dir.is_dir(): diff --git a/teaching/coverage.py b/teaching/coverage.py index c927c2b2..9c48cbd5 100644 --- a/teaching/coverage.py +++ b/teaching/coverage.py @@ -11,6 +11,7 @@ from __future__ import annotations import json import re import subprocess +import tempfile from dataclasses import dataclass, field from pathlib import Path from typing import Mapping @@ -61,24 +62,30 @@ class CoverageReport: def _classify_refusal(reason: str) -> str: """Map a per-case refusal reason string to a stable category bucket. + Matching is case-insensitive throughout — the raw runner output is + consistently lowercase prose today, but normalizing once avoids + drift if upstream casing changes. + Buckets: - ``recognizer_empty_injection()`` — recognizer - matched but injector returned empty - - ``no_admissible_candidate`` — neither parser nor recognizer admitted - - ``no_admissible_question`` — statement(s) admitted; question parser refused + matched but the per-category injector returned empty + - ``no_admissible_question`` — statement(s) admitted; question + parser refused + - ``no_admissible_statement`` — neither parser nor recognizer + admitted any statement - ``unexpected_question_count`` — !=1 question sentence - ``other`` — any unmatched reason text """ if not reason: return "other" lower = reason.lower() - if "recognizer matched but produced no injection" in reason: - m = _REASON_CATEGORY_RE.search(reason) + if "recognizer matched but produced no injection" in lower: + m = _REASON_CATEGORY_RE.search(lower) cat = m.group(1) if m else "unknown" return f"recognizer_empty_injection({cat})" - if "no admissible candidate for question" in reason: + if "no admissible candidate for question" in lower: return "no_admissible_question" - if "no admissible candidate for statement" in reason: + if "no admissible candidate for statement" in lower: return "no_admissible_statement" if "expected exactly one question sentence" in lower: return "unexpected_question_count" @@ -171,9 +178,21 @@ def fetch_committed_baseline( return None if not result.stdout.strip(): return None - tmp = repo_root / ".git" / "coverage_baseline_tmp.json" - tmp.write_text(result.stdout, encoding="utf-8") - return tmp + # Use the system temp dir with a unique filename to avoid: + # (a) failures in non-git checkouts or worktrees where .git is a + # file pointing elsewhere + # (b) concurrent-access collisions if two operators run + # ``core teaching coverage --delta`` simultaneously + fd, tmp_path = tempfile.mkstemp( + prefix="core_coverage_baseline_", suffix=".json" + ) + try: + with open(fd, "w", encoding="utf-8") as fh: + fh.write(result.stdout) + except Exception: + Path(tmp_path).unlink(missing_ok=True) + return None + return Path(tmp_path) __all__ = [ diff --git a/tests/test_teaching_coverage_cli.py b/tests/test_teaching_coverage_cli.py index 8f81ea0f..f2a716a5 100644 --- a/tests/test_teaching_coverage_cli.py +++ b/tests/test_teaching_coverage_cli.py @@ -152,3 +152,58 @@ def test_wrong_zero_invariant_visible_via_as_dict(tmp_path: Path): _write_report(report_path, correct=0, refused=0, wrong=2, per_case=[]) r = build_coverage_report(report_path, lane="t", split="s", version="v1", use_reader=False) assert r.as_dict()["counts"]["wrong"] == 2 + + +# --------------------------------------------------------------------------- +# Sourcery review follow-ups (PR #410 round 1) +# --------------------------------------------------------------------------- + + +def test_classify_refusal_is_case_insensitive(): + """Mixed-case reason text classifies into the same bucket.""" + mixed = ( + "Candidate_Graph: Recognizer Matched But Produced No Injection " + "for statement: 'X.' (category=currency_amount)" + ) + assert _classify_refusal(mixed) == "recognizer_empty_injection(currency_amount)" + + +def test_classify_docstring_matches_implementation_buckets(): + """The docstring's bucket list matches the values _classify_refusal returns.""" + expected_buckets = { + "recognizer_empty_injection", + "no_admissible_question", + "no_admissible_statement", + "unexpected_question_count", + "other", + } + doc = _classify_refusal.__doc__ or "" + for bucket in expected_buckets: + assert bucket in doc, f"docstring missing bucket: {bucket}" + + +def test_fetch_committed_baseline_uses_system_temp(tmp_path, monkeypatch): + """The baseline temp file lives in tempfile.gettempdir(), not .git/.""" + import tempfile as _tempfile + from teaching.coverage import fetch_committed_baseline + + class _FakeCompleted: + def __init__(self, stdout: str) -> None: + self.stdout = stdout + + import subprocess as _sp + + fake_baseline = '{"counts": {"correct": 1, "refused": 1, "wrong": 0}}' + + def _fake_run(*_args, **_kwargs): + return _FakeCompleted(fake_baseline) + + monkeypatch.setattr(_sp, "run", _fake_run) + + out = fetch_committed_baseline("evals/x/report.json", tmp_path) + assert out is not None + sys_temp = Path(_tempfile.gettempdir()).resolve() + assert str(out.resolve()).startswith(str(sys_temp)), ( + f"baseline temp must be under {sys_temp}, got {out!r}" + ) + out.unlink(missing_ok=True)