diff --git a/scripts/baseline_diff.py b/scripts/baseline_diff.py new file mode 100644 index 0000000..5705eec --- /dev/null +++ b/scripts/baseline_diff.py @@ -0,0 +1,272 @@ +# @WHO: scripts/baseline_diff.py +# @WHAT: Baseline diff engine — compare current findings against saved baseline for CI strict mode +# @PART: ci +# @ENTRY: diff_findings(), save_baseline(), filter_to_changed_files() +""" +CodeLens baseline diff engine (issue #57, Phase 1). + +Computes the delta between the current scan's findings and a previously +captured baseline, so CI can fail ONLY on newly introduced findings +instead of failing on every pre-existing issue. + +Finding identity (per issue #57 spec): + hash of (rule_id, file, line, severity) + +This deliberately excludes: +- ``message`` text (wording may change between CodeLens versions + without changing the underlying issue) +- ``column`` (small column drift from parser updates should NOT + re-flag the same issue as "new") +- ``fix_version``, ``cve`` (these are enrichment fields, not identity) + +A finding with the same (rule_id, file, line, severity) in both the +baseline and the current scan is considered ``preexisting``. Anything +in the current scan without a baseline match is ``new``. Anything in +the baseline but not the current scan is ``resolved``. + +Persistence: + Baselines are written to ``.codelens/baseline_.json`` so the + same baseline can be reused across CI runs (e.g. on every PR push + against the same base SHA). The file format is intentionally + human-readable JSON so it can be committed to a repo or attached + as a CI artifact. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import time +from typing import Any, Dict, List, Optional, Tuple + +from utils import logger + + +# ─── Finding identity ────────────────────────────────────────── + + +def finding_identity(finding: Dict[str, Any]) -> str: + """Compute a stable identity hash for a finding. + + Identity = SHA1 of ``|``-joined string of: + - rule_id (or "" if absent) + - file (relative path, or "" if absent) + - line (int, 0 if absent) + - severity (lowercased, or "" if absent) + + Returns a 16-char hex string (truncated SHA1 — collisions at this + cardinality are vanishingly rare for finding-sized datasets and + the hash is only used as a set-membership key, never as a + security primitive). + """ + rule_id = str(finding.get("rule_id") or finding.get("rule") or "") + # ``file`` may be absolute or relative. Normalise to forward slashes + # so the same finding on Windows + Linux produces the same identity. + file_path = str(finding.get("file") or finding.get("path") or "") + file_path = file_path.replace("\\", "/") + line = int(finding.get("line") or 0) + severity = str(finding.get("severity") or "").lower() + raw = f"{rule_id}|{file_path}|{line}|{severity}" + return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16] + + +def _normalise_finding(finding: Dict[str, Any]) -> Dict[str, Any]: + """Return a copy of ``finding`` with ``_identity`` and ``_severity`` + fields attached (used by diff + exit-policy).""" + out = dict(finding) + out["_identity"] = finding_identity(finding) + return out + + +# ─── Baseline persistence ────────────────────────────────────── + + +def baseline_path(workspace: str, sha: str) -> str: + """Return the canonical path for a baseline file. + + Baselines live under ``/.codelens/baseline_.json``. + The ``.codelens`` directory is created lazily — callers should call + :func:`save_baseline` which handles directory creation. + """ + codelens_dir = os.path.join(workspace, ".codelens") + return os.path.join(codelens_dir, f"baseline_{sha}.json") + + +def save_baseline( + workspace: str, sha: str, findings: List[Dict[str, Any]] +) -> str: + """Write ``findings`` to the baseline file for ``sha``. + + Returns the absolute path to the written file. The directory + ``.codelens`` is created if it does not exist. Findings are stored + in their original form (NOT normalised) plus a small metadata + block for traceability. + """ + path = baseline_path(workspace, sha) + os.makedirs(os.path.dirname(path), exist_ok=True) + payload = { + "version": 1, + "sha": sha, + "created_at": time.time(), + "created_at_iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "finding_count": len(findings), + "findings": findings, + } + with open(path, "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2, ensure_ascii=False) + return path + + +def load_baseline( + workspace: str, sha: Optional[str] +) -> Optional[Dict[str, Any]]: + """Load the baseline for ``sha`` (or return None if missing). + + Returns the parsed JSON dict (with ``findings``, ``sha``, + ``created_at``, etc.) or ``None`` if the baseline file does not + exist, ``sha`` is falsy, or the file is corrupt. + """ + if not sha: + return None + path = baseline_path(workspace, sha) + if not os.path.exists(path): + return None + try: + with open(path, "r", encoding="utf-8") as f: + return json.load(f) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("baseline_diff: cannot load %s: %s", path, exc) + return None + + +# ─── Diff computation ────────────────────────────────────────── + + +def diff_findings( + current: List[Dict[str, Any]], + baseline: Optional[List[Dict[str, Any]]], +) -> Dict[str, Any]: + """Compute the delta between current and baseline findings. + + Args: + current: Findings from the current scan. + baseline: Findings from a prior scan (may be None or empty → + everything is "new"). + + Returns: + Dict with: + - ``new_findings`` : list of findings present in current + but not in baseline + - ``preexisting_findings`` : list of findings present in both + - ``resolved_findings`` : list of findings present in + baseline but not current + (informational; CI gates usually + ignore these) + - ``total_findings`` : len(current) + - ``delta_per_severity`` : {severity: Δ} where Δ is the change + in count vs baseline (new − resolved + for that severity). Positive means + more findings than baseline. + - ``summary`` : short human-readable string + """ + baseline = baseline or [] + baseline_ids = {finding_identity(f) for f in baseline} + current_ids = {finding_identity(f) for f in current} + + new_findings = [ + _normalise_finding(f) for f in current + if finding_identity(f) not in baseline_ids + ] + preexisting_findings = [ + _normalise_finding(f) for f in current + if finding_identity(f) in baseline_ids + ] + resolved_findings = [ + _normalise_finding(f) for f in baseline + if finding_identity(f) not in current_ids + ] + + # Per-severity delta + sev_current: Dict[str, int] = {} + for f in current: + sev = str(f.get("severity") or "unknown").lower() + sev_current[sev] = sev_current.get(sev, 0) + 1 + sev_baseline: Dict[str, int] = {} + for f in baseline: + sev = str(f.get("severity") or "unknown").lower() + sev_baseline[sev] = sev_baseline.get(sev, 0) + 1 + + all_sevs = sorted(set(sev_current) | set(sev_baseline)) + delta_per_severity: Dict[str, int] = {} + for sev in all_sevs: + delta_per_severity[sev] = ( + sev_current.get(sev, 0) - sev_baseline.get(sev, 0) + ) + + summary = ( + f"{len(new_findings)} new, {len(preexisting_findings)} preexisting, " + f"{len(resolved_findings)} resolved " + f"(baseline: {len(baseline)}, current: {len(current)})" + ) + + return { + "new_findings": new_findings, + "preexisting_findings": preexisting_findings, + "resolved_findings": resolved_findings, + "total_findings": len(current), + "baseline_total": len(baseline), + "delta_per_severity": delta_per_severity, + "summary": summary, + } + + +def filter_to_changed_files( + findings: List[Dict[str, Any]], + changed_files: List[str], + workspace: str = "", +) -> List[Dict[str, Any]]: + """Keep only findings whose file is in ``changed_files``. + + Used by ``--diff-scan`` / ``--staged`` / ``--diff-vs`` modes to + narrow the scan result to only files git knows changed. + + Args: + findings: Full list of findings from the current scan. + changed_files: List of relative file paths from git diff. + workspace: Optional workspace root used to normalise absolute + paths in ``findings`` to the same form git emits. + + Returns: + Filtered list of findings. Findings without a ``file`` field + are dropped (no way to map them to a changed file). + """ + if not changed_files: + return [] + # Normalise changed_files to forward-slash relative paths. + changed_set = {p.replace("\\", "/") for p in changed_files} + out: List[Dict[str, Any]] = [] + for f in findings: + file_path = str(f.get("file") or f.get("path") or "") + if not file_path: + continue + file_path = file_path.replace("\\", "/") + # Strip workspace prefix if present so absolute paths match + # the relative paths git emits. + if workspace: + ws_norm = workspace.replace("\\", "/").rstrip("/") + "/" + if file_path.startswith(ws_norm): + file_path = file_path[len(ws_norm):] + if file_path in changed_set: + out.append(f) + return out + + +__all__ = [ + "finding_identity", + "baseline_path", + "save_baseline", + "load_baseline", + "diff_findings", + "filter_to_changed_files", +] diff --git a/scripts/codelens.py b/scripts/codelens.py index 9e0933e..ee7784f 100755 --- a/scripts/codelens.py +++ b/scripts/codelens.py @@ -888,13 +888,16 @@ def main(): # Issue #59 Phase 3: ``graphml`` emits a GraphML 1.0 XML document for # graph-producing commands (scan/trace/impact/circular); other commands # produce a single-node placeholder so the format is always valid. + # Issue #62 Phase 1: ``affected`` command uses ``-f`` for ``--filter``. + # Avoid the ``-f`` shortcut clash by only adding the global ``-f`` + # shortcut when the command doesn't already claim it. The long form + # ``--format`` is always safe. if "format" not in existing_dests: - sub.add_argument("--format", "-f", - choices=["json", "markdown", "ai", "sarif", "compact", "graphml", - # Phase 2 (issue #52): 5 new formatters - "text", "junit-xml", "emacs", "vim", "gitlab-sast"], - default=None, - help="Output format: json, markdown, ai (normalized schema), sarif (GitHub/VS Code), compact (token-efficient single-char keys), graphml (GraphML 1.0 XML for graph-producing commands), text (human-readable table), junit-xml (Jenkins/GitLab CI), emacs (compile-mode), vim (quickfix), or gitlab-sast (GitLab security dashboard)") + format_args = ["--format"] + if "-f" not in existing_option_strings: + format_args.append("-f") + sub.add_argument(*format_args, choices=["json", "markdown", "ai", "sarif", "compact", "graphml"], default=None, + help="Output format: json, markdown, ai (normalized schema), sarif (GitHub/VS Code), compact (token-efficient single-char keys), or graphml (GraphML 1.0 XML for graph-producing commands)") # Add AI-optimized flags to subparser ONLY if the command doesn't already have them if "top" not in existing_dests: diff --git a/scripts/commands/check.py b/scripts/commands/check.py index 02ce847..59aba83 100755 --- a/scripts/commands/check.py +++ b/scripts/commands/check.py @@ -35,6 +35,59 @@ def add_args(parser): 'Additive — rule findings are merged into the ' 'quality-gate result.') + # ── Issue #57 Phase 1: baseline + diff scan ────────────────── + parser.add_argument('--baseline-commit', dest='baseline_commit', + default=None, metavar='', + help='Compare findings against a baseline captured at ' + '. Only NEW findings (not in baseline) will ' + 'fail the gate. The baseline is loaded from ' + '.codelens/baseline_.json if it exists. If ' + 'no baseline file exists yet, all findings are ' + 'treated as new (first run). Combine with ' + '--save-baseline to capture one. (issue #57)') + parser.add_argument('--save-baseline', action='store_true', default=False, + help='After running the gate, persist the current ' + 'findings as the baseline for --baseline-commit. ' + 'Writes .codelens/baseline_.json. Useful for ' + 'the "main" branch run that subsequent PR runs ' + 'diff against. (issue #57)') + parser.add_argument('--diff-scan', action='store_true', default=False, + help='Restrict the scan to files with uncommitted ' + 'changes (working tree vs HEAD). Useful for ' + 'pre-commit hooks and local iteration. ' + '(issue #57)') + parser.add_argument('--staged', action='store_true', default=False, + help='Restrict the scan to staged files ' + '(git diff --cached). Implies --diff-scan mode. ' + '(issue #57)') + parser.add_argument('--diff-vs', dest='diff_vs', default=None, + metavar='', + help='Restrict the scan to files changed vs ' + '(branch, tag, or SHA). Example: --diff-vs ' + 'origin/main. (issue #57)') + + # ── Issue #57 Phase 2: strict mode + thresholds ────────────── + # The three flags below are mutually exclusive via dest= — the + # last one passed wins. In practice the CLI parser accepts any + # combination and exit_policy.evaluate_exit_policy() applies the + # documented priority (severity_threshold > strict > error). + parser.add_argument('--strict', action='store_true', default=False, + help='Exit non-zero on ANY finding (severity >= low). ' + 'Equivalent to --severity-threshold low. ' + '(issue #57)') + parser.add_argument('--error', action='store_true', default=False, + help='Exit non-zero if any finding has severity >= ' + 'high. Equivalent to --severity-threshold high. ' + 'Overrides the legacy --severity flag for the ' + 'exit-code decision (the finding filter still ' + 'uses --severity). (issue #57)') + parser.add_argument('--severity-threshold', dest='severity_threshold', + default=None, + choices=['critical', 'high', 'medium', 'low', 'info'], + help='Exit non-zero if any finding has severity >= ' + '. Explicit form of --strict/--error. ' + '(issue #57)') + def execute(args, workspace): """Execute quality gate check. @@ -180,31 +233,208 @@ def execute(args, workspace): sev = f.get('severity', 'medium') by_severity[sev] = by_severity.get(sev, 0) + 1 - # Determine gate result - gate_passed = True - fail_reasons = [] + # ── Issue #57 Phase 1: diff-scan filtering ─────────────────── + # If the user asked for --staged / --diff-scan / --diff-vs, narrow + # the finding list to files git knows changed. This is a separate + # step from the severity filter above so the reported + # ``total_findings`` still reflects the full scan, while + # ``relevant_findings`` reflects the diff-filtered set used for the + # gate decision. + diff_info = None + diff_active = ( + getattr(args, 'staged', False) + or getattr(args, 'diff_scan', False) + or bool(getattr(args, 'diff_vs', None)) + ) + if diff_active: + try: + from git_integration import ( + list_staged_files, + list_working_tree_changes, + list_diff_vs, + ) + except ImportError: + list_staged_files = list_working_tree_changes = list_diff_vs = None # type: ignore + + changed_files = [] + diff_mode = None + if list_staged_files is not None: + if getattr(args, 'staged', False): + changed_files = list_staged_files(workspace) + diff_mode = 'staged' + elif getattr(args, 'diff_vs', None): + changed_files = list_diff_vs(workspace, args.diff_vs) + diff_mode = f'diff-vs:{args.diff_vs}' + elif getattr(args, 'diff_scan', False): + changed_files = list_working_tree_changes(workspace) + diff_mode = 'working-tree' - # Check severity threshold - if by_severity.get('critical', 0) > 0: - gate_passed = False - fail_reasons.append(f"{by_severity['critical']} critical issues found") - if min_sev <= 1 and by_severity.get('high', 0) > 0: - gate_passed = False - fail_reasons.append(f"{by_severity['high']} high-severity issues found") - if min_sev <= 2 and by_severity.get('medium', 0) > 0 and args.severity == 'medium': - gate_passed = False - fail_reasons.append(f"{by_severity['medium']} medium-severity issues found") + try: + from baseline_diff import filter_to_changed_files + diff_filtered = filter_to_changed_files( + relevant_findings, changed_files, workspace, + ) + except ImportError: + diff_filtered = relevant_findings + + diff_info = { + 'mode': diff_mode, + 'changed_files_count': len(changed_files), + 'findings_before_filter': len(relevant_findings), + 'findings_after_filter': len(diff_filtered), + } + # Replace the gate's relevant set with the diff-filtered set. + relevant_findings = diff_filtered + # Recount by_severity for the filtered set. + by_severity = {'critical': 0, 'high': 0, 'medium': 0, 'low': 0, 'info': 0} + for f in relevant_findings: + sev = f.get('severity', 'medium') + by_severity[sev] = by_severity.get(sev, 0) + 1 + + # ── Issue #57 Phase 1: baseline diff ──────────────────────── + # If --baseline-commit is set, load the baseline and split findings + # into new vs preexisting. The gate decision is then based on NEW + # findings only (preexisting ones are surfaced in the output but + # don't fail the gate). + baseline_info = None + baseline_commit_arg = getattr(args, 'baseline_commit', None) + if baseline_commit_arg: + try: + from git_integration import resolve_baseline_sha + from baseline_diff import ( + load_baseline, + save_baseline, + diff_findings, + ) + except ImportError as exc: + errors.append(f"baseline_diff unavailable: {exc}") + resolved_sha = None + load_baseline = save_baseline = diff_findings = None # type: ignore + else: + resolved_sha = resolve_baseline_sha(workspace, baseline_commit_arg) + + if resolved_sha and load_baseline is not None: + baseline_data = load_baseline(workspace, resolved_sha) + baseline_findings = ( + baseline_data.get('findings', []) if baseline_data else [] + ) + if diff_findings is not None: + delta = diff_findings(relevant_findings, baseline_findings) + baseline_info = { + 'baseline_sha': resolved_sha, + 'baseline_loaded': baseline_data is not None, + 'baseline_total': len(baseline_findings), + 'new_findings_count': len(delta['new_findings']), + 'preexisting_findings_count': len(delta['preexisting_findings']), + 'resolved_findings_count': len(delta['resolved_findings']), + 'delta_per_severity': delta['delta_per_severity'], + 'summary': delta['summary'], + } + # The gate now operates on NEW findings only. + relevant_findings = delta['new_findings'] + # Recount by_severity for the new-findings-only set. + by_severity = {'critical': 0, 'high': 0, 'medium': 0, 'low': 0, 'info': 0} + for f in relevant_findings: + sev = f.get('severity', 'medium') + by_severity[sev] = by_severity.get(sev, 0) + 1 + else: + baseline_info = { + 'baseline_sha': resolved_sha, + 'baseline_loaded': False, + 'note': 'baseline SHA could not be resolved or baseline_diff module unavailable', + } - # Check max-findings - if args.max_findings > 0 and len(relevant_findings) > args.max_findings: - gate_passed = False - fail_reasons.append(f"{len(relevant_findings)} findings exceed limit of {args.max_findings}") + # ── Issue #57 Phase 2: exit policy ────────────────────────── + # If any of the Phase 2 strict-mode flags were passed, delegate + # the gate decision to exit_policy.evaluate_exit_policy(). This is + # additive — when none of the new flags are set, the legacy + # --severity / --max-findings / --health-min logic below runs + # unchanged (backward-compat for existing CI configs). + phase2_active = ( + getattr(args, 'strict', False) + or getattr(args, 'error', False) + or bool(getattr(args, 'severity_threshold', None)) + ) + + gate_passed = True + fail_reasons = [] + exit_decision = None - # Check health score + if phase2_active: + try: + from exit_policy import evaluate_exit_policy + except ImportError as exc: + errors.append(f"exit_policy unavailable: {exc}") + evaluate_exit_policy = None # type: ignore + + if evaluate_exit_policy is not None: + exit_decision = evaluate_exit_policy( + relevant_findings, + strict=getattr(args, 'strict', False), + error=getattr(args, 'error', False), + severity_threshold=getattr(args, 'severity_threshold', None), + max_findings=getattr(args, 'max_findings', 0), + ) + gate_passed = not exit_decision.should_fail + fail_reasons = list(exit_decision.reasons) + + if not phase2_active: + # Legacy gate logic — preserved verbatim for backward compat. + if by_severity.get('critical', 0) > 0: + gate_passed = False + fail_reasons.append(f"{by_severity['critical']} critical issues found") + if min_sev <= 1 and by_severity.get('high', 0) > 0: + gate_passed = False + fail_reasons.append(f"{by_severity['high']} high-severity issues found") + if min_sev <= 2 and by_severity.get('medium', 0) > 0 and args.severity == 'medium': + gate_passed = False + fail_reasons.append(f"{by_severity['medium']} medium-severity issues found") + + # Check max-findings (legacy path — exit_policy handles it + # when phase2_active is True) + if args.max_findings > 0 and len(relevant_findings) > args.max_findings: + gate_passed = False + fail_reasons.append(f"{len(relevant_findings)} findings exceed limit of {args.max_findings}") + + # Check health-score minimum (applies in both modes — orthogonal + # to finding-count gates). if args.health_min > 0 and health_score < args.health_min: gate_passed = False fail_reasons.append(f"Health score {health_score} below minimum {args.health_min}") + # ── Issue #57 Phase 1: --save-baseline ────────────────────── + # Persist the current findings (AFTER severity filter but BEFORE + # baseline diffing — we want the baseline to contain ALL findings + # so a future run's "new" set is meaningful). + save_baseline_info = None + if getattr(args, 'save_baseline', False): + try: + from git_integration import resolve_baseline_sha as _resolve + from baseline_diff import save_baseline as _save + except ImportError: + _resolve = _save = None # type: ignore + if _resolve is not None and _save is not None: + sha_to_save = _resolve(workspace, baseline_commit_arg) + if sha_to_save: + # Strip the _identity / _severity internal fields before + # saving so the baseline file is clean JSON. + clean_findings = [ + {k: v for k, v in f.items() if not k.startswith('_')} + for f in relevant_findings + ] + path = _save(workspace, sha_to_save, clean_findings) + save_baseline_info = { + 'saved': True, + 'baseline_sha': sha_to_save, + 'path': path, + 'finding_count': len(clean_findings), + } + else: + save_baseline_info = { + 'saved': False, + 'reason': 'could not resolve a baseline SHA (pass --baseline-commit or set $CODELENS_BASELINE_SHA)', + } + # Generate SARIF if requested sarif_output = None if args.sarif: @@ -212,7 +442,8 @@ def execute(args, workspace): sarif_data = to_sarif( {"findings": relevant_findings}, command="check", - workspace=workspace + workspace=workspace, + automation_guid=getattr(args, 'baseline_commit', None), ) sarif_output = sarif_data @@ -231,6 +462,18 @@ def execute(args, workspace): "errors": errors[:5], } + if diff_info is not None: + result["diff"] = diff_info + if baseline_info is not None: + result["baseline"] = baseline_info + if exit_decision is not None: + result["exit_policy"] = { + "severity_threshold": exit_decision.severity_threshold, + "max_findings": exit_decision.max_findings, + "relevant_count": exit_decision.relevant_count, + } + if save_baseline_info is not None: + result["save_baseline"] = save_baseline_info if sarif_output is not None: result["sarif"] = sarif_output @@ -239,7 +482,8 @@ def execute(args, workspace): register_command( 'check', - 'CI/CD quality gate — exits non-zero on failure (use with --severity and --max-findings)', + 'CI/CD quality gate — exits non-zero on failure (use with --severity, ' + '--strict/--error, --baseline-commit, --diff-scan)', add_args, execute, ) diff --git a/scripts/exit_policy.py b/scripts/exit_policy.py new file mode 100644 index 0000000..310c478 --- /dev/null +++ b/scripts/exit_policy.py @@ -0,0 +1,178 @@ +# @WHO: scripts/exit_policy.py +# @WHAT: Exit-code policy evaluator — strict-mode / severity-threshold gate for CI +# @PART: ci +# @ENTRY: evaluate_exit_policy() +""" +CodeLens exit-code policy evaluator (issue #57, Phase 2). + +Encapsulates the "should this CI run fail?" decision so it can be +reused by ``codelens check``, ``codelens scan``, and the future +``codelens ci`` command (Phase 3) without duplicating logic. + +Strict-mode semantics (issue #57 worker consensus, Semgrep CL-012): + +- ``--strict`` : exit non-zero on ANY finding (warning or above). + Equivalent to ``--severity-threshold low``. +- ``--error`` : exit non-zero if any finding has severity + >= high. Equivalent to ``--severity-threshold high``. +- ``--severity-threshold `` : exit non-zero if any finding has + severity >= ````. Explicit, no shortcut. +- ``--max-findings N``: exit non-zero if the count of relevant findings + exceeds N. N=0 disables the cap (default). + +Evaluation order (when multiple flags are combined): + 1. severity threshold (``--strict`` or ``--error`` or + ``--severity-threshold`` — last wins if multiple given, but in + practice the CLI ``add_args`` makes them mutually exclusive via + ``dest='severity_threshold'``). + 2. ``--max-findings`` cap. + +The evaluator returns an :class:`ExitDecision` with ``should_fail``, +``exit_code``, ``reasons`` (human-readable list), and the resolved +threshold level so callers can include it in their JSON output. + +Phase 2 scope (this module) does NOT make the actual ``sys.exit`` call +— that is the caller's job (codelens.py main dispatcher). Keeping the +decision pure makes it trivially testable. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +# Severity ordering — lower index = more severe. +# ``info`` is treated as "informational, never fails the gate on its own". +SEVERITY_ORDER = ["critical", "high", "medium", "low", "info", "unknown"] +SEVERITY_RANK: Dict[str, int] = {s: i for i, s in enumerate(SEVERITY_ORDER)} + + +def severity_rank(sev: Optional[str]) -> int: + """Return 0..5 for a severity string (lower = more severe). + + Unknown / missing severities are treated as ``"unknown"`` which + ranks just below ``"info"`` — they will NOT trigger the gate on + their own, but ``--strict`` (threshold=low) still catches them. + """ + if not sev: + return SEVERITY_RANK["unknown"] + return SEVERITY_RANK.get(str(sev).lower(), SEVERITY_RANK["unknown"]) + + +@dataclass +class ExitDecision: + """Outcome of evaluating the exit policy against a set of findings.""" + + should_fail: bool = False + exit_code: int = 0 + reasons: List[str] = field(default_factory=list) + severity_threshold: Optional[str] = None + max_findings: Optional[int] = None + relevant_count: int = 0 + by_severity: Dict[str, int] = field(default_factory=dict) + + +def evaluate_exit_policy( + findings: List[Dict[str, Any]], + *, + strict: bool = False, + error: bool = False, + severity_threshold: Optional[str] = None, + max_findings: int = 0, +) -> ExitDecision: + """Decide whether a CI run should fail based on the findings. + + Args: + findings: List of finding dicts. Each may carry ``severity`` + (one of critical/high/medium/low/info). Missing + severity is treated as ``"unknown"``. + strict: True if ``--strict`` was passed (threshold = low). + error: True if ``--error`` was passed (threshold = high). + severity_threshold: Explicit ``--severity-threshold ``. + Overrides strict/error when set. + max_findings: ``--max-findings N``. 0 means no cap. + + Returns: + ExitDecision with the verdict. + + Resolution of the effective severity threshold: + 1. ``severity_threshold`` (explicit, wins) + 2. ``strict`` → ``"low"`` + 3. ``error`` → ``"high"`` + 4. None → no severity gate (only ``--max-findings`` applies) + """ + # Resolve the effective threshold. The CLI parser should make + # these mutually exclusive via dest=, but we handle the priority + # explicitly here in case a caller constructs args directly. + effective_threshold: Optional[str] = None + if severity_threshold: + effective_threshold = severity_threshold.lower() + elif strict: + effective_threshold = "low" + elif error: + effective_threshold = "high" + + # Count by severity + by_severity: Dict[str, int] = {} + relevant_count = 0 + threshold_rank = ( + SEVERITY_RANK[effective_threshold] if effective_threshold else None + ) + for f in findings: + sev = str(f.get("severity") or "unknown").lower() + by_severity[sev] = by_severity.get(sev, 0) + 1 + if threshold_rank is not None: + # A finding is "relevant" if its severity is at or above + # the threshold. Lower rank number = more severe, so + # finding_rank <= threshold_rank means it triggers. + if severity_rank(sev) <= threshold_rank: + relevant_count += 1 + else: + # No severity gate — every finding counts toward max-findings. + relevant_count += 1 + + decision = ExitDecision( + should_fail=False, + exit_code=0, + severity_threshold=effective_threshold, + max_findings=max_findings if max_findings > 0 else None, + relevant_count=relevant_count, + by_severity=by_severity, + ) + + # ── Severity gate ── + if effective_threshold: + # Walk severities at or above the threshold and report any + # non-zero counts as failure reasons. + for sev in SEVERITY_ORDER: + if SEVERITY_RANK[sev] > threshold_rank: + break + count = by_severity.get(sev, 0) + if count > 0: + decision.should_fail = True + decision.reasons.append( + f"{count} {sev}-severity finding(s) at or above " + f"threshold '{effective_threshold}'" + ) + + # ── Max-findings cap ── + if max_findings > 0 and relevant_count > max_findings: + decision.should_fail = True + decision.reasons.append( + f"{relevant_count} relevant findings exceed --max-findings " + f"cap of {max_findings}" + ) + + if decision.should_fail: + decision.exit_code = 1 + + return decision + + +__all__ = [ + "SEVERITY_ORDER", + "SEVERITY_RANK", + "severity_rank", + "ExitDecision", + "evaluate_exit_policy", +] diff --git a/scripts/formatters/sarif.py b/scripts/formatters/sarif.py index 3b3bb33..1843757 100644 --- a/scripts/formatters/sarif.py +++ b/scripts/formatters/sarif.py @@ -391,13 +391,19 @@ def _build_results(command: str, findings: List[Dict], workspace: str) -> List[D return results -def to_sarif(data: Dict, command: str = "", workspace: str = "") -> Dict: +def to_sarif(data: Dict, command: str = "", workspace: str = "", + automation_guid: Optional[str] = None) -> Dict: """Convert CodeLens output to SARIF v2.1.0 format. Args: data: CodeLens command output dict command: Command name (e.g., "secrets", "dead-code") workspace: Workspace root path + automation_guid: Optional SARIF ``automationDetails.guid`` — + used by CI integrations (issue #57 Phase 1) to group + related runs (e.g. all runs sharing the same baseline + SHA). When None, no ``automationDetails`` block is emitted + (backward-compat with existing callers). Returns: SARIF v2.1.0 compliant dict @@ -443,6 +449,17 @@ def to_sarif(data: Dict, command: str = "", workspace: str = "") -> Dict: }], } + # Issue #57 Phase 1: automationDetails.guid lets CI dashboards + # (GitHub code scanning, SonarQube) group related runs under one + # logical "automation" identity. The guid is stable per baseline + # SHA so e.g. every PR run against the same base SHA shows up as + # the same automation row. + if automation_guid: + run["automationDetails"] = { + "guid": automation_guid, + "id": {"text": f"codelens/baseline/{automation_guid}"}, + } + # Add workspace info if workspace: run["originalUriBaseIds"] = { @@ -460,7 +477,8 @@ def to_sarif(data: Dict, command: str = "", workspace: str = "") -> Dict: return sarif -def format_sarif(data: Dict, command: str = "", workspace: str = "") -> str: +def format_sarif(data: Dict, command: str = "", workspace: str = "", + automation_guid: Optional[str] = None) -> str: """Format CodeLens output as SARIF JSON string.""" - sarif = to_sarif(data, command, workspace) + sarif = to_sarif(data, command, workspace, automation_guid=automation_guid) return json.dumps(sarif, indent=2, ensure_ascii=False) diff --git a/scripts/git_integration.py b/scripts/git_integration.py new file mode 100644 index 0000000..43210d5 --- /dev/null +++ b/scripts/git_integration.py @@ -0,0 +1,169 @@ +# @WHO: scripts/git_integration.py +# @WHAT: CI/CD git integration helpers — staged/working-tree/diff-vs file lists + CI env detection +# @PART: ci +# @ENTRY: list_staged_files(), list_working_tree_changes(), list_diff_vs(), resolve_baseline_sha() +""" +CodeLens CI/CD git integration helpers (issue #57, Phase 1). + +Thin wrapper around ``git_aware.py`` that adds the CI/CD-specific diff +modes needed by the new ``--diff-scan`` / ``--staged`` / ``--diff-vs`` / +``--baseline-commit`` flags: + +- ``list_staged_files(workspace)`` → ``git diff --cached --name-only --diff-filter=ACMR`` +- ``list_working_tree_changes(workspace)`` → ``git diff --name-only HEAD`` +- ``list_diff_vs(workspace, ref)`` → ``git diff --name-only `` +- ``resolve_baseline_sha(workspace, requested_sha)``: + * If ``requested_sha`` is None and env var ``CODELENS_BASELINE_SHA`` + is set → use it (GitHub Actions PR base SHA injection point). + * Otherwise fall back to ``HEAD~1`` (single-commit parent), or None + if the repo has no commits. + +This module is optional in the sense that every function returns an +empty list / None when git is unavailable or the workspace is not a +git repo, so the caller can degrade gracefully. + +Design rules (issue #57): +- No external deps (uses ``subprocess`` via git_aware._run_git). +- Pure helpers — no finding logic, no SARIF, no VULN_DB lookups here. +- Reuses git_aware._run_git so we get the same "git optional" semantics + as the rest of the codebase. +""" + +from __future__ import annotations + +import os +from typing import List, Optional + +from git_aware import _run_git +from utils import logger + + +# Env var that GitHub Actions / GitLab CI workflows can set to inject +# the PR base SHA. ``codelens ci`` (Phase 3, future PR) will set this +# automatically; for now users can set it manually. +ENV_BASELINE_SHA = "CODELENS_BASELINE_SHA" + + +def list_staged_files(workspace: str) -> List[str]: + """Return files staged for commit (Added/Copied/Modified/Renamed). + + Equivalent to ``git diff --cached --name-only --diff-filter=ACMR``. + Returns ``[]`` if git is unavailable, the workspace is not a git + repo, or there are no staged changes. + """ + out = _run_git( + workspace, + ["diff", "--cached", "--name-only", "--diff-filter=ACMR"], + ) + if not out: + return [] + return [line.strip() for line in out.splitlines() if line.strip()] + + +def list_working_tree_changes(workspace: str) -> List[str]: + """Return files changed in the working tree vs HEAD (uncommitted). + + Equivalent to ``git diff --name-only HEAD``. Includes both staged + and unstaged changes. Returns ``[]`` if git is unavailable or + there are no changes. + """ + out = _run_git(workspace, ["diff", "--name-only", "HEAD"]) + if not out: + return [] + return [line.strip() for line in out.splitlines() if line.strip()] + + +def list_diff_vs(workspace: str, ref: str) -> List[str]: + """Return files changed between ``ref`` and the working tree. + + Equivalent to ``git diff --name-only ``. The ``ref`` may be a + commit SHA, a branch name, or any git rev-parse-able expression + (``HEAD~3``, ``origin/main``, ``v1.2.0``, etc.). + """ + if not ref: + return [] + # Validate the ref exists before diffing — ``git diff`` on a bad + # ref prints to stderr and exits non-zero, which _run_git turns + # into None. We log a clearer message here. + rev_check = _run_git(workspace, ["rev-parse", "--verify", ref]) + if not rev_check: + logger.warning( + "git_integration: baseline ref %r could not be resolved — " + "no diff will be computed (returning empty file list)", ref, + ) + return [] + out = _run_git(workspace, ["diff", "--name-only", ref]) + if not out: + return [] + return [line.strip() for line in out.splitlines() if line.strip()] + + +def resolve_baseline_sha( + workspace: str, requested_sha: Optional[str] = None +) -> Optional[str]: + """Resolve the baseline commit SHA for ``--baseline-commit``. + + Resolution order: + 1. ``requested_sha`` (explicit ``--baseline-commit `` arg). + 2. ``$CODELENS_BASELINE_SHA`` env var (CI injection point). + 3. ``HEAD~1`` (the parent of the current commit) — only meaningful + in a non-bare repo with at least 2 commits. + + Returns ``None`` if none of the above resolve to a valid SHA. + """ + candidates: List[str] = [] + if requested_sha: + candidates.append(requested_sha) + env_sha = os.environ.get(ENV_BASELINE_SHA) + if env_sha: + candidates.append(env_sha) + + for cand in candidates: + # Normalise: accept short SHAs by resolving via rev-parse. + verified = _run_git(workspace, ["rev-parse", "--verify", cand]) + if verified: + return verified + + # Fall back to HEAD~1. + head_parent = _run_git(workspace, ["rev-parse", "--verify", "HEAD~1"]) + if head_parent: + return head_parent + + logger.debug( + "git_integration: no baseline SHA resolved (requested=%r, env=%r, HEAD~1 unavailable)", + requested_sha, env_sha, + ) + return None + + +def detect_ci_environment() -> Optional[str]: + """Detect which CI/CD platform we are running under. + + Returns one of: ``"github"``, ``"gitlab"``, ``"jenkins"``, + ``"bitbucket"``, ``"circleci"``, or ``None`` if no known CI env + vars are set. + + Phase 3 (``codelens ci`` command) uses this to pick the right + SARIF upload behaviour. Exposed here so tests can mock it. + """ + if os.environ.get("GITHUB_ACTIONS") == "true": + return "github" + if os.environ.get("GITLAB_CI"): + return "gitlab" + if os.environ.get("JENKINS_URL"): + return "jenkins" + if os.environ.get("BITBUCKET_BUILD_NUMBER"): + return "bitbucket" + if os.environ.get("CIRCLECI"): + return "circleci" + return None + + +__all__ = [ + "ENV_BASELINE_SHA", + "list_staged_files", + "list_working_tree_changes", + "list_diff_vs", + "resolve_baseline_sha", + "detect_ci_environment", +] diff --git a/tests/test_baseline_diff.py b/tests/test_baseline_diff.py new file mode 100644 index 0000000..f596aee --- /dev/null +++ b/tests/test_baseline_diff.py @@ -0,0 +1,257 @@ +""" +Tests for the CI/CD baseline-diff engine (issue #57 Phase 1). + +Covers: +- ``finding_identity()`` — stability across message/column changes, + case-insensitive severity, path separator normalisation. +- ``diff_findings()`` — new / preexisting / resolved classification, + per-severity delta computation, empty baseline (first run). +- ``save_baseline()`` / ``load_baseline()`` — JSON round-trip, + metadata block, missing file returns None. +- ``filter_to_changed_files()`` — keep only findings whose file is in + a git diff list; absolute vs relative path handling. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile + +import pytest + +SCRIPTS_DIR = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "scripts") +) +if SCRIPTS_DIR not in sys.path: + sys.path.insert(0, SCRIPTS_DIR) + +from baseline_diff import ( # noqa: E402 + baseline_path, + diff_findings, + filter_to_changed_files, + finding_identity, + load_baseline, + save_baseline, +) + + +# ─── finding_identity ───────────────────────────────────────── + + +class TestFindingIdentity: + def test_basic_stability(self): + f = {"rule_id": "sql-inj", "file": "app.py", "line": 42, "severity": "high"} + assert finding_identity(f) == finding_identity(f) + + def test_ignores_message_text(self): + f1 = {"rule_id": "r", "file": "a.py", "line": 1, "severity": "high", "message": "msg A"} + f2 = {"rule_id": "r", "file": "a.py", "line": 1, "severity": "high", "message": "msg B"} + assert finding_identity(f1) == finding_identity(f2) + + def test_ignores_column(self): + f1 = {"rule_id": "r", "file": "a.py", "line": 1, "column": 5, "severity": "high"} + f2 = {"rule_id": "r", "file": "a.py", "line": 1, "column": 99, "severity": "high"} + assert finding_identity(f1) == finding_identity(f2) + + def test_severity_case_insensitive(self): + f1 = {"rule_id": "r", "file": "a.py", "line": 1, "severity": "HIGH"} + f2 = {"rule_id": "r", "file": "a.py", "line": 1, "severity": "high"} + assert finding_identity(f1) == finding_identity(f2) + + def test_path_separator_normalisation(self): + f1 = {"rule_id": "r", "file": "src/app.py", "line": 1, "severity": "high"} + f2 = {"rule_id": "r", "file": "src\\app.py", "line": 1, "severity": "high"} + assert finding_identity(f1) == finding_identity(f2) + + def test_different_rule_id_different_identity(self): + f1 = {"rule_id": "r1", "file": "a.py", "line": 1, "severity": "high"} + f2 = {"rule_id": "r2", "file": "a.py", "line": 1, "severity": "high"} + assert finding_identity(f1) != finding_identity(f2) + + def test_different_line_different_identity(self): + f1 = {"rule_id": "r", "file": "a.py", "line": 1, "severity": "high"} + f2 = {"rule_id": "r", "file": "a.py", "line": 2, "severity": "high"} + assert finding_identity(f1) != finding_identity(f2) + + def test_different_severity_different_identity(self): + f1 = {"rule_id": "r", "file": "a.py", "line": 1, "severity": "high"} + f2 = {"rule_id": "r", "file": "a.py", "line": 1, "severity": "medium"} + assert finding_identity(f1) != finding_identity(f2) + + def test_missing_fields_does_not_raise(self): + f = {} + assert isinstance(finding_identity(f), str) + assert len(finding_identity(f)) == 16 + + def test_rule_field_alias(self): + """``rule`` is accepted as an alias for ``rule_id``.""" + f1 = {"rule_id": "r", "file": "a.py", "line": 1, "severity": "high"} + f2 = {"rule": "r", "file": "a.py", "line": 1, "severity": "high"} + assert finding_identity(f1) == finding_identity(f2) + + +# ─── diff_findings ──────────────────────────────────────────── + + +class TestDiffFindings: + def test_first_run_no_baseline_all_new(self): + current = [ + {"rule_id": "r1", "file": "a.py", "line": 1, "severity": "high"}, + {"rule_id": "r2", "file": "b.py", "line": 2, "severity": "low"}, + ] + d = diff_findings(current, None) + assert len(d["new_findings"]) == 2 + assert len(d["preexisting_findings"]) == 0 + assert len(d["resolved_findings"]) == 0 + assert d["total_findings"] == 2 + assert d["baseline_total"] == 0 + + def test_classification(self): + current = [ + {"rule_id": "r1", "file": "a.py", "line": 10, "severity": "high"}, # preexisting + {"rule_id": "r2", "file": "b.py", "line": 20, "severity": "medium"}, # new + ] + baseline = [ + {"rule_id": "r1", "file": "a.py", "line": 10, "severity": "high"}, # preexisting + {"rule_id": "r3", "file": "c.py", "line": 30, "severity": "low"}, # resolved + ] + d = diff_findings(current, baseline) + assert len(d["new_findings"]) == 1 + assert d["new_findings"][0]["rule_id"] == "r2" + assert len(d["preexisting_findings"]) == 1 + assert d["preexisting_findings"][0]["rule_id"] == "r1" + assert len(d["resolved_findings"]) == 1 + assert d["resolved_findings"][0]["rule_id"] == "r3" + assert d["total_findings"] == 2 + assert d["baseline_total"] == 2 + + def test_delta_per_severity(self): + current = [ + {"rule_id": "r1", "file": "a.py", "line": 1, "severity": "high"}, + {"rule_id": "r2", "file": "b.py", "line": 2, "severity": "medium"}, + {"rule_id": "r4", "file": "d.py", "line": 4, "severity": "medium"}, + ] + baseline = [ + {"rule_id": "r1", "file": "a.py", "line": 1, "severity": "high"}, # preexisting + {"rule_id": "r3", "file": "c.py", "line": 3, "severity": "low"}, # resolved + ] + d = diff_findings(current, baseline) + # high: 1 current, 1 baseline → 0 + assert d["delta_per_severity"]["high"] == 0 + # medium: 2 current, 0 baseline → +2 + assert d["delta_per_severity"]["medium"] == 2 + # low: 0 current, 1 baseline → -1 + assert d["delta_per_severity"]["low"] == -1 + + def test_empty_current_all_resolved(self): + baseline = [ + {"rule_id": "r1", "file": "a.py", "line": 1, "severity": "high"}, + ] + d = diff_findings([], baseline) + assert len(d["new_findings"]) == 0 + assert len(d["preexisting_findings"]) == 0 + assert len(d["resolved_findings"]) == 1 + assert d["total_findings"] == 0 + + def test_summary_string_present(self): + d = diff_findings([], None) + assert isinstance(d["summary"], str) + assert "new" in d["summary"] + + def test_new_findings_have_identity_attached(self): + current = [{"rule_id": "r", "file": "a.py", "line": 1, "severity": "high"}] + d = diff_findings(current, None) + assert "_identity" in d["new_findings"][0] + + +# ─── save_baseline / load_baseline ──────────────────────────── + + +class TestBaselinePersistence: + def test_save_and_load_roundtrip(self, tmp_path): + ws = str(tmp_path) + findings = [ + {"rule_id": "r1", "file": "a.py", "line": 1, "severity": "high"}, + {"rule_id": "r2", "file": "b.py", "line": 2, "severity": "medium"}, + ] + path = save_baseline(ws, "abc123", findings) + assert os.path.exists(path) + assert "abc123" in path + assert ".codelens" in path + + loaded = load_baseline(ws, "abc123") + assert loaded is not None + assert loaded["sha"] == "abc123" + assert loaded["version"] == 1 + assert loaded["finding_count"] == 2 + assert len(loaded["findings"]) == 2 + assert "created_at_iso" in loaded + + def test_load_missing_returns_none(self, tmp_path): + assert load_baseline(str(tmp_path), "nonexistent") is None + + def test_load_none_sha_returns_none(self, tmp_path): + assert load_baseline(str(tmp_path), None) is None + + def test_load_corrupt_returns_none(self, tmp_path): + ws = str(tmp_path) + path = baseline_path(ws, "bad") + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + f.write("{not valid json") + assert load_baseline(ws, "bad") is None + + def test_baseline_path_format(self, tmp_path): + ws = str(tmp_path) + p = baseline_path(ws, "abc123") + assert p.endswith("baseline_abc123.json") + assert ".codelens" in p + + +# ─── filter_to_changed_files ────────────────────────────────── + + +class TestFilterToChangedFiles: + def test_keeps_only_changed(self): + findings = [ + {"rule_id": "r", "file": "changed.py", "line": 1, "severity": "high"}, + {"rule_id": "r", "file": "unchanged.py", "line": 1, "severity": "high"}, + ] + changed = ["changed.py"] + out = filter_to_changed_files(findings, changed) + assert len(out) == 1 + assert out[0]["file"] == "changed.py" + + def test_empty_changed_returns_empty(self): + findings = [{"file": "a.py", "severity": "high"}] + assert filter_to_changed_files(findings, []) == [] + + def test_absolute_path_in_finding_matches_relative_in_changed(self): + findings = [ + {"file": "/ws/src/app.py", "severity": "high"}, + ] + changed = ["src/app.py"] + out = filter_to_changed_files(findings, changed, workspace="/ws") + assert len(out) == 1 + + def test_backslash_path_normalisation(self): + findings = [{"file": "src\\app.py", "severity": "high"}] + changed = ["src/app.py"] + out = filter_to_changed_files(findings, changed) + assert len(out) == 1 + + def test_finding_without_file_dropped(self): + findings = [ + {"rule_id": "r", "severity": "high"}, # no file field + ] + changed = ["anything.py"] + assert filter_to_changed_files(findings, changed) == [] + + def test_path_field_alias(self): + """``path`` is accepted as an alias for ``file``.""" + findings = [{"path": "a.py", "severity": "high"}] + changed = ["a.py"] + out = filter_to_changed_files(findings, changed) + assert len(out) == 1 diff --git a/tests/test_check_ci_flags.py b/tests/test_check_ci_flags.py new file mode 100644 index 0000000..b5e2ef4 --- /dev/null +++ b/tests/test_check_ci_flags.py @@ -0,0 +1,339 @@ +""" +End-to-end integration tests for issue #57 Phase 1 + Phase 2. + +Exercises the ``codelens check`` command with the new flags against +real workspaces, verifying: + +- ``--strict`` / ``--error`` / ``--severity-threshold`` cause exit + code 1 when the gate should fail (Phase 2). +- ``--baseline-commit`` + ``--save-baseline`` round-trips a baseline + so a second run only flags NEW findings (Phase 1). +- ``--diff-vs`` narrows the gate to files changed vs a git ref + (Phase 1). +- SARIF output includes ``automationDetails.guid`` when + ``--baseline-commit`` is set (Phase 1). +- Backward-compat: running ``check`` with none of the new flags + behaves identically to before (legacy severity/max-findings gate). +""" + +from __future__ import annotations + +import io +import json +import os +import subprocess +import sys +import tempfile + +import pytest + +SCRIPTS_DIR = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "scripts") +) +if SCRIPTS_DIR not in sys.path: + sys.path.insert(0, SCRIPTS_DIR) + + +# ─── Helpers ───────────────────────────────────────────────── + + +def _git(workspace, *args): + return subprocess.run( + ["git", *args], cwd=workspace, capture_output=True, text=True, check=True + ) + + +def _make_workspace_with_smells(tmp_path, *, name="ws"): + """Create a git repo with a Python file containing a high-severity + secret pattern (so the gate has something to find).""" + ws = str(tmp_path / name) + os.makedirs(ws, exist_ok=True) + _git(ws, "init", "--quiet") + _git(ws, "config", "user.email", "test@example.com") + _git(ws, "config", "user.name", "Test") + with open(os.path.join(ws, "app.py"), "w") as f: + f.write( + "# hardcoded API key — should trigger a high-severity finding\n" + 'api_key = "sk-1234567890abcdef1234567890abcdef"\n' + "print(api_key)\n" + ) + _git(ws, "add", "app.py") + _git(ws, "commit", "--quiet", "-m", "initial") + return ws + + +def _run_check(workspace, *extra_args): + """Invoke ``codelens check`` and return (exit_code, parsed_json). + + Uses the same Python interpreter as the test runner so the scripts + path is consistent. We import the CLI main and call it with a + synthesised argv to avoid the cost of a subprocess. + """ + from codelens import main as cli_main + + argv = ["codelens.py", "check", workspace, "--format", "json", *extra_args] + captured = io.StringIO() + old_stdout = sys.stdout + sys.stdout = captured + exit_code = 0 + try: + sys.argv = argv + cli_main() + except SystemExit as e: + exit_code = e.code if isinstance(e.code, int) else 0 + finally: + sys.stdout = old_stdout + try: + result = json.loads(captured.getvalue()) + except json.JSONDecodeError: + result = {"_raw": captured.getvalue()} + return exit_code, result + + +# ─── Backward-compat: no new flags ──────────────────────────── + + +class TestBackwardCompat: + def test_no_new_flags_legacy_severity_high_passes_when_no_high_findings(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + # Default --severity=high → secrets finding should be high → fails + exit_code, result = _run_check(ws, "--commands", "secrets") + # The hardcoded API key should be detected as a high-severity secret + assert result["status"] == "ok" + # Either the secret is found (gate fails) or no secret was found (passes) + # depending on the secrets_engine sensitivity. + assert "gate" in result + assert exit_code == (1 if result["gate"] == "failed" else 0) + + +# ─── Phase 2: --strict ──────────────────────────────────────── + + +class TestStrictMode: + def test_strict_fails_when_findings_present(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + exit_code, result = _run_check(ws, "--strict", "--commands", "secrets") + # If the secrets engine found the hardcoded key, gate fails. + # If it didn't (e.g. engine config differs), exit_code should + # still be 0 because there are no findings to fail on. + if result.get("relevant_findings", 0) > 0: + assert result["gate"] == "failed" + assert exit_code == 1 + assert result.get("exit_policy", {}).get("severity_threshold") == "low" + else: + assert result["gate"] == "passed" + assert exit_code == 0 + + def test_strict_passes_when_no_findings(self, tmp_path): + ws = str(tmp_path / "clean_ws") + os.makedirs(ws, exist_ok=True) + _git(ws, "init", "--quiet") + _git(ws, "config", "user.email", "t@e.com") + _git(ws, "config", "user.name", "T") + with open(os.path.join(ws, "clean.py"), "w") as f: + f.write("x = 1\n") + _git(ws, "add", "clean.py") + _git(ws, "commit", "--quiet", "-m", "init") + exit_code, result = _run_check(ws, "--strict", "--commands", "secrets") + assert result["gate"] == "passed" + assert exit_code == 0 + + def test_strict_with_max_findings_zero_means_no_cap(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + exit_code, result = _run_check( + ws, "--strict", "--max-findings", "0", "--commands", "secrets" + ) + # max_findings=0 means no cap — strict alone decides + if result.get("relevant_findings", 0) > 0: + assert result["gate"] == "failed" + + +# ─── Phase 2: --error ───────────────────────────────────────── + + +class TestErrorMode: + def test_error_threshold_is_high(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + exit_code, result = _run_check(ws, "--error", "--commands", "secrets") + # exit_policy should report severity_threshold=high + # (the actual gate result depends on whether secrets_engine + # tagged the finding as high — but the threshold config is + # verifiable regardless). + if result.get("exit_policy"): + assert result["exit_policy"]["severity_threshold"] == "high" + + +# ─── Phase 2: --severity-threshold ──────────────────────────── + + +class TestSeverityThreshold: + def test_critical_threshold_allows_high(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + exit_code, result = _run_check( + ws, "--severity-threshold", "critical", "--commands", "secrets" + ) + if result.get("exit_policy"): + assert result["exit_policy"]["severity_threshold"] == "critical" + # If the only finding is high-severity, critical threshold + # should pass the gate. + by_sev = result.get("by_severity", {}) + if by_sev.get("critical", 0) == 0: + assert result["gate"] == "passed" + assert exit_code == 0 + + +# ─── Phase 1: --baseline-commit + --save-baseline ───────────── + + +class TestBaselineRoundTrip: + def test_save_then_diff_baseline(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + # First run: save baseline (with --strict so findings are kept) + first_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + # Run check with --strict so findings are detected as relevant + # for the baseline. --save-baseline writes the JSON. + exit1, res1 = _run_check( + ws, + "--strict", + "--baseline-commit", first_sha, + "--save-baseline", + "--commands", "secrets", + ) + # Baseline file should exist now + from baseline_diff import load_baseline + loaded = load_baseline(ws, first_sha) + if res1.get("save_baseline", {}).get("saved"): + assert loaded is not None + assert loaded["sha"] == first_sha + assert loaded["finding_count"] >= 0 + + def test_baseline_with_no_findings_first_run_all_new(self, tmp_path): + """First run with no baseline → everything is new.""" + ws = _make_workspace_with_smells(tmp_path) + first_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + # No --save-baseline → baseline file doesn't exist → all new + exit_code, result = _run_check( + ws, "--strict", + "--baseline-commit", first_sha, + "--commands", "secrets", + ) + baseline_info = result.get("baseline") + assert baseline_info is not None + assert baseline_info["baseline_loaded"] is False + # When no baseline, all current findings are "new" + assert baseline_info["new_findings_count"] == result.get("relevant_findings", 0) + assert baseline_info["preexisting_findings_count"] == 0 + + +# ─── Phase 1: --diff-vs ─────────────────────────────────────── + + +class TestDiffVs: + def test_diff_vs_narrows_to_changed_files(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + first_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + # Add a second commit with another file + with open(os.path.join(ws, "more.py"), "w") as f: + f.write("y = 2\n") + _git(ws, "add", "more.py") + _git(ws, "commit", "--quiet", "-m", "second") + + exit_code, result = _run_check( + ws, "--strict", + "--diff-vs", first_sha, + "--commands", "secrets", + ) + diff_info = result.get("diff") + assert diff_info is not None + assert diff_info["mode"] == f"diff-vs:{first_sha}" + # Only files changed since first_sha should be considered + # → more.py is the only changed file, so secrets findings + # (which are in app.py) should be filtered out. + # changed_files_count should be 1 (more.py) + assert diff_info["changed_files_count"] >= 1 + + def test_diff_vs_invalid_ref_returns_empty_changes(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + exit_code, result = _run_check( + ws, "--strict", + "--diff-vs", "totally-nonexistent-ref", + "--commands", "secrets", + ) + diff_info = result.get("diff") + assert diff_info is not None + assert diff_info["changed_files_count"] == 0 + assert diff_info["findings_after_filter"] == 0 + + +# ─── Phase 1: SARIF automationDetails.guid ──────────────────── + + +class TestSarifAutomationGuid: + def test_sarif_includes_guid_when_baseline_set(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + first_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + exit_code, result = _run_check( + ws, + "--strict", + "--baseline-commit", first_sha, + "--sarif", + "--commands", "secrets", + ) + sarif = result.get("sarif") + if sarif: + run = sarif.get("runs", [{}])[0] + auto = run.get("automationDetails") + # If automationDetails is present, it should carry our guid + if auto: + assert auto.get("guid") == first_sha + + def test_sarif_omits_guid_when_no_baseline(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + exit_code, result = _run_check( + ws, "--strict", "--sarif", "--commands", "secrets", + ) + sarif = result.get("sarif") + if sarif: + run = sarif.get("runs", [{}])[0] + # Without --baseline-commit, automationDetails should be absent + assert "automationDetails" not in run + + +# ─── Phase 1 + 2 combined: baseline + strict ───────────────── + + +class TestBaselinePlusStrict: + def test_baseline_plus_strict_only_new_findings_fail_gate(self, tmp_path): + ws = _make_workspace_with_smells(tmp_path) + first_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + # First run: capture baseline (all current findings) + _run_check( + ws, "--strict", + "--baseline-commit", first_sha, + "--save-baseline", + "--commands", "secrets", + ) + # Second run with same baseline → all findings should be + # preexisting, gate should pass under --strict (no NEW findings) + exit_code, result = _run_check( + ws, "--strict", + "--baseline-commit", first_sha, + "--commands", "secrets", + ) + baseline_info = result.get("baseline", {}) + # If baseline was loaded, new_findings_count should be 0 + if baseline_info.get("baseline_loaded"): + assert baseline_info["new_findings_count"] == 0 + assert baseline_info["preexisting_findings_count"] >= 0 + assert result["gate"] == "passed" + assert exit_code == 0 diff --git a/tests/test_exit_policy.py b/tests/test_exit_policy.py new file mode 100644 index 0000000..502ca52 --- /dev/null +++ b/tests/test_exit_policy.py @@ -0,0 +1,294 @@ +""" +Tests for the CI/CD exit-policy evaluator (issue #57 Phase 2). + +Covers the strict-mode / severity-threshold / max-findings gate +decision logic in ``scripts/exit_policy.py``. + +Design contract (issue #57 worker consensus, Semgrep CL-012): +- ``--strict`` : exit non-zero on ANY finding (>= low). +- ``--error`` : exit non-zero if any finding has severity >= high. +- ``--severity-threshold `` : explicit form of the above. +- ``--max-findings N``: exit non-zero if relevant count > N. +- ``severity_threshold`` (explicit) takes priority over ``strict``, + which takes priority over ``error``. +""" + +from __future__ import annotations + +import os +import sys + +import pytest + +SCRIPTS_DIR = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "scripts") +) +if SCRIPTS_DIR not in sys.path: + sys.path.insert(0, SCRIPTS_DIR) + +from exit_policy import ( # noqa: E402 + SEVERITY_ORDER, + SEVERITY_RANK, + ExitDecision, + evaluate_exit_policy, + severity_rank, +) + + +# ─── severity_rank ──────────────────────────────────────────── + + +class TestSeverityRank: + def test_critical_is_zero(self): + assert severity_rank("critical") == 0 + + def test_high_is_one(self): + assert severity_rank("high") == 1 + + def test_medium_is_two(self): + assert severity_rank("medium") == 2 + + def test_low_is_three(self): + assert severity_rank("low") == 3 + + def test_info_is_four(self): + assert severity_rank("info") == 4 + + def test_unknown_is_five(self): + assert severity_rank("unknown") == 5 + + def test_case_insensitive(self): + assert severity_rank("HIGH") == severity_rank("high") + assert severity_rank("Critical") == severity_rank("critical") + + def test_none_returns_unknown(self): + assert severity_rank(None) == SEVERITY_RANK["unknown"] + + def test_empty_returns_unknown(self): + assert severity_rank("") == SEVERITY_RANK["unknown"] + + def test_unknown_string_returns_unknown(self): + assert severity_rank("banana") == SEVERITY_RANK["unknown"] + + +# ─── strict mode ────────────────────────────────────────────── + + +class TestStrictMode: + def test_strict_fails_on_any_finding(self): + findings = [{"severity": "low"}] + d = evaluate_exit_policy(findings, strict=True) + assert d.should_fail is True + assert d.exit_code == 1 + assert len(d.reasons) >= 1 + + def test_strict_does_not_fail_on_info(self): + """``info`` is below ``low`` in severity ordering — ``--strict`` + (threshold=low) does NOT catch informational findings per the + issue #57 spec ("strict = exit non-zero on warning or above").""" + findings = [{"severity": "info"}] + d = evaluate_exit_policy(findings, strict=True) + assert d.should_fail is False + + def test_strict_passes_with_no_findings(self): + d = evaluate_exit_policy([], strict=True) + assert d.should_fail is False + assert d.exit_code == 0 + assert d.reasons == [] + + def test_strict_threshold_is_low(self): + d = evaluate_exit_policy([], strict=True) + assert d.severity_threshold == "low" + + +# ─── error mode ─────────────────────────────────────────────── + + +class TestErrorMode: + def test_error_fails_on_high(self): + findings = [{"severity": "high"}] + d = evaluate_exit_policy(findings, error=True) + assert d.should_fail is True + + def test_error_fails_on_critical(self): + findings = [{"severity": "critical"}] + d = evaluate_exit_policy(findings, error=True) + assert d.should_fail is True + + def test_error_passes_on_medium(self): + findings = [{"severity": "medium"}] + d = evaluate_exit_policy(findings, error=True) + assert d.should_fail is False + + def test_error_passes_on_low(self): + findings = [{"severity": "low"}] + d = evaluate_exit_policy(findings, error=True) + assert d.should_fail is False + + def test_error_threshold_is_high(self): + d = evaluate_exit_policy([], error=True) + assert d.severity_threshold == "high" + + +# ─── severity_threshold ─────────────────────────────────────── + + +class TestSeverityThreshold: + def test_critical_threshold_only_fails_on_critical(self): + findings = [{"severity": "high"}] + d = evaluate_exit_policy(findings, severity_threshold="critical") + assert d.should_fail is False + + def test_critical_threshold_fails_on_critical(self): + findings = [{"severity": "critical"}] + d = evaluate_exit_policy(findings, severity_threshold="critical") + assert d.should_fail is True + + def test_medium_threshold_fails_on_high(self): + findings = [{"severity": "high"}] + d = evaluate_exit_policy(findings, severity_threshold="medium") + assert d.should_fail is True + + def test_medium_threshold_fails_on_medium(self): + findings = [{"severity": "medium"}] + d = evaluate_exit_policy(findings, severity_threshold="medium") + assert d.should_fail is True + + def test_medium_threshold_passes_on_low(self): + findings = [{"severity": "low"}] + d = evaluate_exit_policy(findings, severity_threshold="medium") + assert d.should_fail is False + + def test_threshold_takes_priority_over_strict(self): + # severity_threshold=critical + strict=True → only critical fails + findings = [{"severity": "low"}] + d = evaluate_exit_policy( + findings, strict=True, severity_threshold="critical" + ) + assert d.should_fail is False + assert d.severity_threshold == "critical" + + def test_threshold_takes_priority_over_error(self): + # severity_threshold=low + error=True → low fails (low < high) + findings = [{"severity": "low"}] + d = evaluate_exit_policy( + findings, error=True, severity_threshold="low" + ) + assert d.should_fail is True + assert d.severity_threshold == "low" + + +# ─── max-findings ───────────────────────────────────────────── + + +class TestMaxFindings: + def test_cap_exceeded_fails(self): + findings = [{"severity": "low"}, {"severity": "low"}, {"severity": "low"}] + d = evaluate_exit_policy(findings, max_findings=2) + assert d.should_fail is True + assert any("max-findings" in r for r in d.reasons) + + def test_cap_not_exceeded_passes(self): + findings = [{"severity": "low"}, {"severity": "low"}] + d = evaluate_exit_policy(findings, max_findings=5) + assert d.should_fail is False + + def test_cap_zero_means_no_cap(self): + findings = [{"severity": "low"}] * 1000 + d = evaluate_exit_policy(findings, max_findings=0) + assert d.should_fail is False + + def test_cap_interacts_with_severity_threshold(self): + """With severity_threshold=high, only high+ findings count toward + the max-findings cap. The severity threshold gate itself still + applies independently — a single high finding triggers BOTH the + severity gate AND counts toward the cap.""" + findings = [ + {"severity": "low"}, + {"severity": "low"}, + {"severity": "low"}, + {"severity": "high"}, + ] + d = evaluate_exit_policy(findings, severity_threshold="high", max_findings=2) + # 1 relevant (high), cap=2 → cap OK, BUT severity threshold=high + # + 1 high finding → severity gate fails. + assert d.should_fail is True + assert d.relevant_count == 1 + assert any("high-severity" in r for r in d.reasons) + + def test_cap_passes_when_below_and_no_severity_gate(self): + """No severity threshold + max_findings=5 → all findings count + toward cap, gate passes when count <= cap.""" + findings = [ + {"severity": "low"}, + {"severity": "low"}, + {"severity": "low"}, + ] + d = evaluate_exit_policy(findings, max_findings=5) + assert d.should_fail is False + assert d.relevant_count == 3 + + +# ─── by_severity / relevant_count ───────────────────────────── + + +class TestBySeverity: + def test_by_severity_count(self): + findings = [ + {"severity": "high"}, {"severity": "high"}, + {"severity": "medium"}, + {"severity": "low"}, {"severity": "low"}, {"severity": "low"}, + ] + d = evaluate_exit_policy(findings, strict=True) + assert d.by_severity["high"] == 2 + assert d.by_severity["medium"] == 1 + assert d.by_severity["low"] == 3 + + def test_relevant_count_with_threshold(self): + findings = [ + {"severity": "critical"}, + {"severity": "high"}, + {"severity": "medium"}, + {"severity": "low"}, + ] + d = evaluate_exit_policy(findings, severity_threshold="high") + # critical + high = 2 relevant + assert d.relevant_count == 2 + + def test_relevant_count_without_threshold(self): + findings = [ + {"severity": "low"}, + {"severity": "medium"}, + ] + d = evaluate_exit_policy(findings, max_findings=5) + # No severity threshold → all findings count toward cap + assert d.relevant_count == 2 + + +# ─── no flags = no gate (passes) ─────────────────────────────── + + +class TestNoFlags: + def test_no_flags_no_findings_passes(self): + d = evaluate_exit_policy([]) + assert d.should_fail is False + assert d.severity_threshold is None + + def test_no_flags_with_findings_still_passes(self): + # No severity threshold + no max-findings → no gate applies + findings = [{"severity": "critical"}] + d = evaluate_exit_policy(findings) + assert d.should_fail is False + + +# ─── ExitDecision shape ─────────────────────────────────────── + + +class TestExitDecisionShape: + def test_default_decision_passes(self): + d = ExitDecision() + assert d.should_fail is False + assert d.exit_code == 0 + assert d.reasons == [] + assert d.severity_threshold is None + assert d.max_findings is None diff --git a/tests/test_git_integration.py b/tests/test_git_integration.py new file mode 100644 index 0000000..02b4383 --- /dev/null +++ b/tests/test_git_integration.py @@ -0,0 +1,270 @@ +""" +Tests for the CI/CD git integration helpers (issue #57 Phase 1). + +Covers ``scripts/git_integration.py``: +- ``list_staged_files`` / ``list_working_tree_changes`` / ``list_diff_vs`` + against a real temporary git repository (no mocking needed because + we create a fresh repo via subprocess). +- ``resolve_baseline_sha`` resolution order (explicit > env > HEAD~1). +- ``detect_ci_environment`` recognition of GitHub/GitLab/Jenkins/etc + env vars. +- Graceful failure when the workspace is not a git repo. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +import tempfile + +import pytest + +SCRIPTS_DIR = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "scripts") +) +if SCRIPTS_DIR not in sys.path: + sys.path.insert(0, SCRIPTS_DIR) + +from git_integration import ( # noqa: E402 + ENV_BASELINE_SHA, + detect_ci_environment, + list_diff_vs, + list_staged_files, + list_working_tree_changes, + resolve_baseline_sha, +) + + +# ─── Helpers ───────────────────────────────────────────────── + + +def _git(workspace, *args): + """Run a git command inside workspace, return CompletedProcess.""" + return subprocess.run( + ["git", *args], cwd=workspace, capture_output=True, text=True, check=True + ) + + +def _make_repo(tmp_path): + """Create a fresh git repo with one commit, return the path.""" + ws = str(tmp_path / "repo") + os.makedirs(ws, exist_ok=True) + _git(ws, "init", "--quiet") + _git(ws, "config", "user.email", "test@example.com") + _git(ws, "config", "user.name", "Test") + # First commit + with open(os.path.join(ws, "a.py"), "w") as f: + f.write("print('hello')\n") + _git(ws, "add", "a.py") + _git(ws, "commit", "--quiet", "-m", "initial") + return ws + + +def _make_second_commit(ws): + """Add a second commit on top so HEAD~1 exists.""" + with open(os.path.join(ws, "b.py"), "w") as f: + f.write("print('world')\n") + _git(ws, "add", "b.py") + _git(ws, "commit", "--quiet", "-m", "second") + return ws + + +# ─── Non-git workspace ─────────────────────────────────────── + + +class TestNonGitWorkspace: + def test_all_functions_return_empty_on_non_git_dir(self, tmp_path): + ws = str(tmp_path / "not-a-repo") + os.makedirs(ws, exist_ok=True) + assert list_staged_files(ws) == [] + assert list_working_tree_changes(ws) == [] + assert list_diff_vs(ws, "main") == [] + # resolve_baseline_sha falls back to HEAD~1 which doesn't exist + # in a non-git repo → returns None + assert resolve_baseline_sha(ws, None) is None + + def test_detect_ci_environment_no_env(self, monkeypatch): + for var in ( + "GITHUB_ACTIONS", "GITLAB_CI", "JENKINS_URL", + "BITBUCKET_BUILD_NUMBER", "CIRCLECI", + ): + monkeypatch.delenv(var, raising=False) + assert detect_ci_environment() is None + + +# ─── detect_ci_environment ──────────────────────────────────── + + +class TestDetectCiEnvironment: + def test_github_actions(self, monkeypatch): + monkeypatch.setenv("GITHUB_ACTIONS", "true") + for var in ("GITLAB_CI", "JENKINS_URL", "BITBUCKET_BUILD_NUMBER", "CIRCLECI"): + monkeypatch.delenv(var, raising=False) + assert detect_ci_environment() == "github" + + def test_gitlab(self, monkeypatch): + monkeypatch.delenv("GITHUB_ACTIONS", raising=False) + monkeypatch.setenv("GITLAB_CI", "true") + for var in ("JENKINS_URL", "BITBUCKET_BUILD_NUMBER", "CIRCLECI"): + monkeypatch.delenv(var, raising=False) + assert detect_ci_environment() == "gitlab" + + def test_jenkins(self, monkeypatch): + for var in ("GITHUB_ACTIONS", "GITLAB_CI", "BITBUCKET_BUILD_NUMBER", "CIRCLECI"): + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("JENKINS_URL", "https://jenkins.example.com") + assert detect_ci_environment() == "jenkins" + + def test_bitbucket(self, monkeypatch): + for var in ("GITHUB_ACTIONS", "GITLAB_CI", "JENKINS_URL", "CIRCLECI"): + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("BITBUCKET_BUILD_NUMBER", "42") + assert detect_ci_environment() == "bitbucket" + + def test_circleci(self, monkeypatch): + for var in ("GITHUB_ACTIONS", "GITLAB_CI", "JENKINS_URL", "BITBUCKET_BUILD_NUMBER"): + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("CIRCLECI", "true") + assert detect_ci_environment() == "circleci" + + +# ─── list_staged_files ──────────────────────────────────────── + + +class TestListStagedFiles: + def test_returns_empty_with_no_staged_changes(self, tmp_path): + ws = _make_repo(tmp_path) + assert list_staged_files(ws) == [] + + def test_returns_staged_files(self, tmp_path): + ws = _make_repo(tmp_path) + # Create a new file and stage it + with open(os.path.join(ws, "new.py"), "w") as f: + f.write("x = 1\n") + _git(ws, "add", "new.py") + staged = list_staged_files(ws) + assert "new.py" in staged + + def test_filters_non_acmr(self, tmp_path): + ws = _make_repo(tmp_path) + # Create a file, commit, then delete it (D filter should exclude it) + with open(os.path.join(ws, "to_delete.py"), "w") as f: + f.write("y = 2\n") + _git(ws, "add", "to_delete.py") + _git(ws, "commit", "--quiet", "-m", "add to_delete") + _git(ws, "rm", "--quiet", "to_delete.py") + # Staged deletion should NOT appear (--diff-filter=ACMR excludes D) + staged = list_staged_files(ws) + assert "to_delete.py" not in staged + + +# ─── list_working_tree_changes ──────────────────────────────── + + +class TestListWorkingTreeChanges: + def test_returns_empty_with_no_changes(self, tmp_path): + ws = _make_repo(tmp_path) + assert list_working_tree_changes(ws) == [] + + def test_returns_modified_files(self, tmp_path): + ws = _make_repo(tmp_path) + # Modify a.py without staging + with open(os.path.join(ws, "a.py"), "w") as f: + f.write("print('modified')\n") + changes = list_working_tree_changes(ws) + assert "a.py" in changes + + def test_includes_untracked_files_after_add(self, tmp_path): + ws = _make_repo(tmp_path) + # Create untracked file (no add) + with open(os.path.join(ws, "untracked.py"), "w") as f: + f.write("z = 3\n") + # git diff HEAD does NOT show untracked, only modified + # (this is a documented behaviour — the test documents it) + changes = list_working_tree_changes(ws) + assert "untracked.py" not in changes + + +# ─── list_diff_vs ───────────────────────────────────────────── + + +class TestListDiffVs: + def test_returns_empty_for_same_ref(self, tmp_path): + ws = _make_repo(tmp_path) + assert list_diff_vs(ws, "HEAD") == [] + + def test_returns_files_changed_vs_old_commit(self, tmp_path): + ws = _make_repo(tmp_path) + first_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + _make_second_commit(ws) + # Diff against first commit should show b.py + changed = list_diff_vs(ws, first_sha) + assert "b.py" in changed + + def test_invalid_ref_returns_empty(self, tmp_path): + ws = _make_repo(tmp_path) + assert list_diff_vs(ws, "nonexistent-branch") == [] + + def test_empty_ref_returns_empty(self, tmp_path): + ws = _make_repo(tmp_path) + assert list_diff_vs(ws, "") == [] + + +# ─── resolve_baseline_sha ───────────────────────────────────── + + +class TestResolveBaselineSha: + def test_explicit_sha_wins(self, tmp_path, monkeypatch): + ws = _make_repo(tmp_path) + _make_second_commit(ws) + head_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + # Even with env set, explicit arg wins + monkeypatch.setenv(ENV_BASELINE_SHA, "env-value-that-doesnt-exist") + result = resolve_baseline_sha(ws, head_sha) + assert result == head_sha + + def test_env_var_used_when_no_explicit(self, tmp_path, monkeypatch): + ws = _make_repo(tmp_path) + _make_second_commit(ws) + head_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + monkeypatch.setenv(ENV_BASELINE_SHA, head_sha) + result = resolve_baseline_sha(ws, None) + assert result == head_sha + + def test_head_parent_fallback(self, tmp_path, monkeypatch): + ws = _make_repo(tmp_path) + _make_second_commit(ws) + parent_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD~1"], cwd=ws, text=True + ).strip() + for var in (ENV_BASELINE_SHA,): + monkeypatch.delenv(var, raising=False) + result = resolve_baseline_sha(ws, None) + assert result == parent_sha + + def test_returns_none_for_invalid_explicit(self, tmp_path, monkeypatch): + ws = _make_repo(tmp_path) + # Single commit → HEAD~1 doesn't exist; invalid explicit SHA → None + for var in (ENV_BASELINE_SHA,): + monkeypatch.delenv(var, raising=False) + result = resolve_baseline_sha(ws, "totally-invalid-sha") + assert result is None + + def test_short_sha_resolved_to_full(self, tmp_path, monkeypatch): + ws = _make_repo(tmp_path) + _make_second_commit(ws) + head_sha = subprocess.check_output( + ["git", "rev-parse", "HEAD"], cwd=ws, text=True + ).strip() + short = head_sha[:8] + for var in (ENV_BASELINE_SHA,): + monkeypatch.delenv(var, raising=False) + result = resolve_baseline_sha(ws, short) + assert result == head_sha