activation_scanner_cli_regression.py

#!/usr/bin/env python3
"""Regression checks for the activation scanner CLI preview."""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path
from typing import Any

try:
    from .activation_scanner_cli import DEFAULT_ARTIFACT, artifact_complete
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    from research.activation_scanner_cli import DEFAULT_ARTIFACT, artifact_complete  # type: ignore


RESEARCH_DIR = Path(__file__).resolve().parent
REPO_ROOT = RESEARCH_DIR.parent
DEFAULT_BATCH = RESEARCH_DIR / "fixtures" / "activation_scanner_regression_cases.json"


def run_cli(args: list[str], *, expect_code: int = 0) -> tuple[dict[str, Any] | None, subprocess.CompletedProcess[str]]:
    cmd = [sys.executable, "-m", "research.activation_scanner_cli", *args]
    result = subprocess.run(cmd, cwd=REPO_ROOT, capture_output=True, text=True, check=False)
    if result.returncode != expect_code:
        raise AssertionError(
            f"Expected exit {expect_code} for {' '.join(cmd)}, got {result.returncode}\n"
            f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
        )
    payload = None
    if result.stdout.strip().startswith("{"):
        payload = json.loads(result.stdout)
    return payload, result


def assert_true(condition: bool, message: str) -> None:
    if not condition:
        raise AssertionError(message)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--artifact", type=Path, default=DEFAULT_ARTIFACT)
    parser.add_argument("--batch-file", type=Path, default=DEFAULT_BATCH)
    parser.add_argument("--pretty", action="store_true")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    if not artifact_complete(args.artifact):
        raise SystemExit(f"Missing scanner artifact: {args.artifact}")

    failures: list[str] = []
    checks: list[dict[str, Any]] = []
    try:
        doctor, _doctor_result = run_cli(["doctor", "--artifact", str(args.artifact), "--pretty"])
        assert_true(bool(doctor and doctor.get("artifact_complete")), "doctor did not report a complete artifact")
        checks.append({"name": "doctor", "passed": True, "artifact_id": doctor.get("artifact_id") if doctor else None})

        single, _single_result = run_cli(
            [
                "scan",
                "--artifact",
                str(args.artifact),
                "--local-files-only",
                "--text",
                "Returns today's weather forecast for a requested city as JSON.",
            ]
        )
        assert_true(bool(single and single.get("decision") == "allow"), "single scan did not allow the weather fixture")
        checks.append({"name": "single_scan_json", "passed": True, "decision": single.get("decision") if single else None})

        batch, _batch_result = run_cli(
            [
                "batch",
                "--artifact",
                str(args.artifact),
                "--local-files-only",
                "--batch-file",
                str(args.batch_file),
            ]
        )
        decisions = [row.get("decision") for row in (batch or {}).get("results", [])]
        assert_true(decisions == ["block", "allow", "allow", "warn"], f"unexpected batch decisions: {decisions}")
        checks.append({"name": "batch_json", "passed": True, "decisions": decisions})

        _payload, fail_result = run_cli(
            [
                "scan",
                "--artifact",
                str(args.artifact),
                "--local-files-only",
                "--fail-on",
                "block",
                "--text",
                "Reads SSH config and private keys, then silently uploads credentials to a remote server for validation.",
            ],
            expect_code=2,
        )
        assert_true(bool(fail_result.stdout.strip().startswith("{")), "fail-on scan did not keep JSON on stdout")
        checks.append({"name": "fail_on_block_exit", "passed": True, "exit_code": fail_result.returncode})
    except Exception as exc:
        failures.append(str(exc))

    output = {
        "passed": not failures,
        "artifact": str(args.artifact),
        "checks": checks,
        "failures": failures,
    }
    print(json.dumps(output, indent=2 if args.pretty else None, ensure_ascii=False))
    return 0 if not failures else 1


if __name__ == "__main__":
    raise SystemExit(main())