test_cli.py

"""Integration tests for the security_scanner CLI end-to-end wiring (new rich model)."""

from __future__ import annotations

import datetime as dt
import json

import pytest

from security_scanner.catalog.scan_target import ScanTarget
from security_scanner.cli import main
from security_scanner.core.finding.model import (
    Finding,
    GitleaksFindingPayload,
    Status,
    Verdict,
)
from security_scanner.runtime.local_scan import (
    LocalScanResult,
    LocalScanTargetResult,
)
from security_scanner.storage.base import GhasAlertRecord
from security_scanner.storage.jsonl_store import JsonlFindingStore

FAKE_SECRET = "AKIAFAKEEXAMPLE000000"
SCAN_RUN_ID = "scan_test0001"
RULE_PACK = "secret-rules-0.1.0"


def _store_finding(store: JsonlFindingStore, **kw) -> Finding:
    """Create and append a Finding to the store, returning the finding."""
    defaults = dict(
        repo_full_name="demo-org/demo-repo",
        rule_id="generic-api-key",
        file_path="src/config.py",
        line_start=10,
        raw_secret=FAKE_SECRET,
        source_tool="gitleaks",
        scan_run_id=SCAN_RUN_ID,
        rule_pack_version=RULE_PACK,
    )
    defaults.update(kw)
    f = Finding.create(**defaults)
    store.append(f)
    return f


def test_help_raises_system_exit():
    """argparse exits with code 0 on --help."""
    with pytest.raises(SystemExit) as exc_info:
        main(["--help"])
    assert exc_info.value.code == 0


def test_no_subcommand_returns_one():
    """Invoking without a subcommand returns 1 and prints help."""
    assert main([]) == 1


def test_scan_missing_manifest_returns_2(tmp_path):
    """A missing/invalid manifest is a usage error (exit 2)."""
    missing = tmp_path / "nope.yaml"
    out = tmp_path / "findings.jsonl"
    assert main(["scan", "--manifest", str(missing), "-o", str(out)]) == 2


def test_scan_end_to_end_empty_result(monkeypatch, tmp_path):
    """scan loads the manifest, resolves a local path, runs the scanner adapter,
    and writes an empty store when the scanner returns no findings."""
    repo = tmp_path / "demo-repo"
    repo.mkdir()
    manifest = tmp_path / "targets.yaml"
    manifest.write_text(
        "version: 1\n"
        "targets:\n"
        f"  - name: demo-org/demo-repo\n    path: {repo}\n    enabled: true\n"
        "scan:\n  include_history: true\n",
        encoding="utf-8",
    )
    out = tmp_path / "findings.jsonl"

    monkeypatch.setattr(
        "security_scanner.runtime.local_scan.GitleaksScanner.scan",
        lambda self, **kwargs: [],
    )

    assert main(["scan", "--manifest", str(manifest), "-o", str(out)]) == 0
    assert JsonlFindingStore(out).read_all() == []


def test_scan_can_select_dynamodb_backend(monkeypatch, tmp_path):
    """scan can route persistence through the storage adapter factory."""
    repo = tmp_path / "demo-repo"
    repo.mkdir()
    manifest = tmp_path / "targets.yaml"
    manifest.write_text(
        "version: 1\n"
        "targets:\n"
        f"  - name: demo-org/demo-repo\n    path: {repo}\n    enabled: true\n",
        encoding="utf-8",
    )

    class FakeStore:
        def __init__(self) -> None:
            self.ensure_called = False
            self.clear_called = False
            self.extended = False
            self.scan_results = []

        def ensure_table(self) -> None:
            self.ensure_called = True

        def prepare_for_scan(self) -> None:
            self.ensure_called = True

        def clear(self) -> None:
            self.clear_called = True

        def extend(self, findings) -> None:
            self.extended = True
            assert findings == []

        def write_scan_result(self, result) -> None:
            self.extended = True
            self.scan_results.append(result)

        def read_all(self):
            return []

    fake_store = FakeStore()
    calls = []

    def fake_create_finding_store(backend, **kwargs):
        calls.append((backend, kwargs))
        return fake_store

    monkeypatch.setattr(
        "security_scanner.runtime.local_scan.create_finding_store",
        fake_create_finding_store,
    )
    monkeypatch.setattr(
        "security_scanner.runtime.local_scan.GitleaksScanner.scan",
        lambda self, **kwargs: [],
    )

    assert main([
        "scan",
        "--manifest",
        str(manifest),
        "--storage-backend",
        "dynamodb",
        "--dynamodb-table",
        "SecurityScannerLocal",
        "--dynamodb-endpoint-url",
        "http://localhost:4567",
    ]) == 0

    assert calls[0][0] == "dynamodb"
    assert calls[0][1]["dynamodb_config"].table_name == "SecurityScannerLocal"
    assert fake_store.ensure_called is True
    assert fake_store.clear_called is False
    assert fake_store.extended is True
    assert len(fake_store.scan_results) == 1
    assert fake_store.scan_results[0].target_name == "demo-org/demo-repo"
    assert fake_store.scan_results[0].findings == []


def test_scan_delegates_to_runtime_and_renders_result(monkeypatch, tmp_path, capsys):
    """scan delegates orchestration to runtime and keeps CLI output ownership."""
    calls = {}
    out = tmp_path / "findings.jsonl"

    def fake_run_local_scan(request):
        calls["request"] = request
        return LocalScanResult(
            manifest_path=request.manifest_path,
            scan_run_id=SCAN_RUN_ID,
            rule_pack_version=RULE_PACK,
            destination=request.output_destination,
            total_targets=2,
            scanned=1,
            total_findings=3,
            target_results=[
                LocalScanTargetResult(
                    target_name="demo-org/demo-repo",
                    status="scanned",
                    finding_count=3,
                ),
                LocalScanTargetResult(
                    target_name="demo-org/missing-repo",
                    status="skipped",
                    error="unavailable",
                ),
            ],
        )

    monkeypatch.setattr(
        "security_scanner.cli.commands.scan.run_local_scan",
        fake_run_local_scan,
    )

    assert main([
        "scan",
        "--manifest",
        "synthetic-targets.yaml",
        "-o",
        str(out),
    ]) == 0

    captured = capsys.readouterr()
    assert calls["request"].manifest_path == "synthetic-targets.yaml"
    assert calls["request"].output_destination == str(out)
    assert calls["request"].storage_backend == "jsonl"
    assert "Scanning 2 enabled target(s) from synthetic-targets.yaml" in captured.out
    assert f"Scan run ID: {SCAN_RUN_ID}, rule pack: {RULE_PACK}" in captured.out
    assert "  scanned demo-org/demo-repo: 3 finding(s)" in captured.out
    assert f"Done: scanned 1/2 target(s), 3 finding(s) -> {out}" in captured.out
    assert "  skip demo-org/missing-repo: unavailable" in captured.err


def test_report_can_read_dynamodb_backend(monkeypatch, tmp_path, capsys):
    """report can consume a non-JSONL store through the storage adapter."""
    finding = Finding.create(
        repo_full_name="demo-org/demo-repo",
        rule_id="generic-api-key",
        file_path="src/config.py",
        line_start=10,
        raw_secret=FAKE_SECRET,
        source_tool="gitleaks",
        scan_run_id=SCAN_RUN_ID,
        rule_pack_version=RULE_PACK,
    )

    class FakeStore:
        def read_all(self):
            return [finding]

    monkeypatch.setattr(
        "security_scanner.cli._store.create_finding_store",
        lambda backend, **kwargs: FakeStore(),
    )

    assert main([
        "report",
        "--storage-backend",
        "dynamodb",
        "--dynamodb-table",
        "SecurityScannerLocal",
    ]) == 0
    captured = capsys.readouterr().out
    assert "Total findings: 1" in captured
    assert FAKE_SECRET not in captured


def test_report_can_read_dynamodb_backend_for_scan_run(monkeypatch, capsys):
    """report can query a persisted DynamoDB-compatible scan run directly."""
    finding = Finding.create(
        repo_full_name="demo-org/demo-repo",
        rule_id="generic-api-key",
        file_path="src/config.py",
        line_start=10,
        raw_secret=FAKE_SECRET,
        source_tool="gitleaks",
        scan_run_id=SCAN_RUN_ID,
        rule_pack_version=RULE_PACK,
        gitleaks=GitleaksFindingPayload(
            rule_id="generic-api-key",
            file="src/config.py",
            start_line=10,
            secret=FAKE_SECRET,
            match=f"token={FAKE_SECRET}",
            fingerprint="gitleaks-fp-001",
        ),
    )

    class FakeStore:
        def __init__(self) -> None:
            self.scan_run_ids = []

        def read_all(self):
            raise AssertionError("report should use read_for_scan_run")

        def read_for_scan_run(self, scan_run_id):
            self.scan_run_ids.append(scan_run_id)
            return [finding]

    fake_store = FakeStore()
    monkeypatch.setattr(
        "security_scanner.cli._store.create_finding_store",
        lambda backend, **kwargs: fake_store,
    )

    assert main([
        "report",
        "--storage-backend",
        "dynamodb",
        "--dynamodb-table",
        "SecurityScannerLocal",
        "--scan-run-id",
        SCAN_RUN_ID,
    ]) == 0
    captured = capsys.readouterr().out
    assert fake_store.scan_run_ids == [SCAN_RUN_ID]
    assert "Total findings: 1" in captured
    assert "sha256:" in captured
    assert FAKE_SECRET not in captured


def test_gate_can_read_dynamodb_backend_for_scan_run_threshold(
    monkeypatch,
    capsys,
):
    """gate applies thresholds to persisted DynamoDB-compatible scan-run findings."""
    finding = Finding.create(
        repo_full_name="demo-org/demo-repo",
        rule_id="generic-api-key",
        file_path="src/config.py",
        line_start=10,
        raw_secret=FAKE_SECRET,
        source_tool="gitleaks",
        scan_run_id=SCAN_RUN_ID,
        rule_pack_version=RULE_PACK,
        gitleaks=GitleaksFindingPayload(
            rule_id="generic-api-key",
            file="src/config.py",
            start_line=10,
            secret=FAKE_SECRET,
            match=f"token={FAKE_SECRET}",
            fingerprint="gitleaks-fp-001",
        ),
    )

    class FakeStore:
        def __init__(self) -> None:
            self.scan_run_ids = []

        def read_all(self):
            raise AssertionError("gate should use read_for_scan_run")

        def read_for_scan_run(self, scan_run_id):
            self.scan_run_ids.append(scan_run_id)
            return [finding]

    fake_store = FakeStore()
    monkeypatch.setattr(
        "security_scanner.cli._store.create_finding_store",
        lambda backend, **kwargs: fake_store,
    )

    assert main([
        "gate",
        "--storage-backend",
        "dynamodb",
        "--dynamodb-table",
        "SecurityScannerLocal",
        "--scan-run-id",
        SCAN_RUN_ID,
        "--max",
        "0",
    ]) == 1
    captured = capsys.readouterr().out
    assert fake_store.scan_run_ids == [SCAN_RUN_ID]
    assert "FAIL: 1 blocking finding(s) > threshold 0" in captured
    assert FAKE_SECRET not in captured

    fake_store.scan_run_ids.clear()
    assert main([
        "gate",
        "--storage-backend",
        "dynamodb",
        "--dynamodb-table",
        "SecurityScannerLocal",
        "--scan-run-id",
        SCAN_RUN_ID,
        "--max",
        "1",
    ]) == 0
    captured = capsys.readouterr().out
    assert fake_store.scan_run_ids == [SCAN_RUN_ID]
    assert "PASS: 1 blocking finding(s) <= threshold 1" in captured
    assert FAKE_SECRET not in captured


def test_init_storage_dynamodb_calls_ensure_table(monkeypatch):
    """init-storage bootstraps the DynamoDB-compatible table."""
    class FakeStore:
        def __init__(self) -> None:
            self.bootstrap_called = False

        @property
        def bootstrap_required(self) -> bool:
            return True

        def bootstrap(self) -> None:
            self.bootstrap_called = True

    fake_store = FakeStore()

    monkeypatch.setattr(
        "security_scanner.cli._store.create_finding_store",
        lambda backend, **kwargs: fake_store,
    )

    assert main([
        "init-storage",
        "--storage-backend",
        "dynamodb",
        "--dynamodb-table",
        "SecurityScannerLocal",
    ]) == 0
    assert fake_store.bootstrap_called is True


def test_scan_skips_unavailable_target(tmp_path):
    """A target whose local path does not exist is skipped, not fatal."""
    manifest = tmp_path / "targets.yaml"
    manifest.write_text(
        "version: 1\n"
        "targets:\n"
        "  - name: ghost-org/ghost-repo\n"
        "    path: /nonexistent/demo-path\n"
        "    enabled: true\n",
        encoding="utf-8",
    )
    out = tmp_path / "findings.jsonl"
    assert main(["scan", "--manifest", str(manifest), "-o", str(out)]) == 0


def test_report_reads_store(tmp_path, capsys):
    """report renders findings written to the store."""
    out = tmp_path / "findings.jsonl"
    store = JsonlFindingStore(out)
    _store_finding(store)

    assert main(["report", "--findings", str(out)]) == 0
    captured = capsys.readouterr().out
    assert "Total findings: 1" in captured
    assert "generic-api-key" in captured
    # Raw secret never appears in the report.
    assert FAKE_SECRET not in captured


def test_report_shows_hash_prefix(tmp_path, capsys):
    """Report output must contain a short hash prefix (never a raw secret)."""
    out = tmp_path / "findings.jsonl"
    store = JsonlFindingStore(out)
    _store_finding(store)

    main(["report", "--findings", str(out)])
    captured = capsys.readouterr().out
    assert "sha256:" in captured
    assert FAKE_SECRET not in captured


def test_gate_fails_on_open_needs_review(tmp_path):
    """An OPEN NEEDS_REVIEW finding blocks the gate by default (exit 1)."""
    out = tmp_path / "findings.jsonl"
    store = JsonlFindingStore(out)
    _store_finding(store)  # default: OPEN + NEEDS_REVIEW
    assert main(["gate", "--findings", str(out)]) == 1


def test_gate_passes_on_empty_store(tmp_path):
    """No findings -> gate passes (exit 0)."""
    out = tmp_path / "findings.jsonl"
    assert main(["gate", "--findings", str(out)]) == 0


def test_gate_passes_when_only_false_positive_verdict(tmp_path):
    """FALSE_POSITIVE triage verdict does not block the gate."""
    out = tmp_path / "findings.jsonl"
    store = JsonlFindingStore(out)
    _store_finding(store, triage_verdict=Verdict.FALSE_POSITIVE.value)
    assert main(["gate", "--findings", str(out)]) == 0


def test_gate_passes_when_resolved_status(tmp_path):
    """RESOLVED status does not block the gate even for TRUE_POSITIVE verdict."""
    out = tmp_path / "findings.jsonl"
    store = JsonlFindingStore(out)
    _store_finding(
        store,
        triage_verdict=Verdict.TRUE_POSITIVE.value,
        status=Status.RESOLVED.value,
    )
    assert main(["gate", "--findings", str(out)]) == 0


def test_evaluate_reports_metrics_and_passes(tmp_path, capsys):
    """evaluate compares synthetic expected findings to an actual JSONL store."""
    expected = tmp_path / "expected-findings.json"
    expected.write_text(
        json.dumps(
            {
                "schemaVersion": 1,
                "name": "synthetic-test-corpus",
                "expectedFindings": [
                    {
                        "repoFullName": "demo-org/demo-repo",
                        "filePath": "src/config.py",
                        "lineStart": 10,
                        "ruleId": "generic-api-key",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    out = tmp_path / "findings.jsonl"
    store = JsonlFindingStore(out)
    _store_finding(store)

    assert main(["evaluate", "--expected", str(expected), "--findings", str(out)]) == 0
    captured = capsys.readouterr().out
    assert "True positives: 1" in captured
    assert "False positives: 0" in captured
    assert "False negatives: 0" in captured
    assert "Precision: 1.0000" in captured
    assert "Recall: 1.0000" in captured
    assert FAKE_SECRET not in captured


def test_evaluate_false_negative_fails_gate(tmp_path, capsys):
    """A synthetic expected finding missing from actual results fails M7 gate."""
    expected = tmp_path / "expected-findings.json"
    expected.write_text(
        json.dumps(
            {
                "schemaVersion": 1,
                "name": "synthetic-test-corpus",
                "expectedFindings": [
                    {
                        "repoFullName": "demo-org/demo-repo",
                        "filePath": "src/config.py",
                        "lineStart": 10,
                        "ruleId": "generic-api-key",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    out = tmp_path / "findings.jsonl"

    assert main(["evaluate", "--expected", str(expected), "--findings", str(out)]) == 1
    captured = capsys.readouterr().out
    assert "False negatives: 1" in captured
    assert "FAIL" in captured


def test_compare_ghas_csv_source_is_phased_out(capsys):
    assert main(["compare-ghas", "--source", "csv", "--ghas-csv", "legacy.csv"]) == 2

    captured = capsys.readouterr()
    assert "CSV GHAS comparison is phased out" in captured.err


def test_compare_ghas_defaults_to_github_source(monkeypatch, capsys):
    fetched_at = dt.datetime(2026, 6, 16, 12, 0, tzinfo=dt.timezone.utc)
    finding = Finding.create(
        repo_full_name="Example-Org/Example-Repo",
        rule_id="example_api_key",
        file_path="src/settings.py",
        line_start=12,
        raw_secret="synthetic-secret",
        source_tool="gitleaks",
        scan_run_id=SCAN_RUN_ID,
        rule_pack_version=RULE_PACK,
    )
    out_of_scope_finding = Finding.create(
        repo_full_name="example-org/other-repo",
        rule_id="other_token",
        file_path="src/other.py",
        line_start=7,
        raw_secret="synthetic-other-secret",
        source_tool="gitleaks",
        scan_run_id=SCAN_RUN_ID,
        rule_pack_version=RULE_PACK,
    )
    alert = GhasAlertRecord(
        ghas_alert_id="ghas_alert_001",
        repository="example-org/example-repo",
        alert_number=1001,
        secret_type="example_api_key",
        state="open",
        location_path="src/settings.py",
        location_start_line=12,
        fetched_at=fetched_at,
    )
    stale_alert = GhasAlertRecord(
        ghas_alert_id="ghas_alert_stale",
        repository="example-org/disabled-repo",
        alert_number=9999,
        secret_type="stale_token",
        state="open",
        location_path="src/stale.py",
        location_start_line=1,
        fetched_at=fetched_at,
    )

    class FakeStore:
        def __init__(self) -> None:
            self.written_alerts = []
            self.stale_alerts = [stale_alert]

        def list_scan_targets(self):
            return [
                ScanTarget(
                    url="https://example.invalid/example-org/example-repo.git",
                    name="example-org/example-repo",
                    enabled=True,
                )
            ]

        def put_ghas_alerts(self, alerts):
            self.written_alerts.extend(alerts)

        def read_ghas_alerts(self):
            return [*self.stale_alerts, *self.written_alerts]

        def read_all(self):
            return [finding, out_of_scope_finding]

    fake_store = FakeStore()
    monkeypatch.setattr(
        "security_scanner.cli._store.create_finding_store",
        lambda *args, **kwargs: fake_store,
    )
    monkeypatch.setattr(
        "security_scanner.cli.commands.report.fetch_ghas_alert_records",
        lambda target, *, api: [alert],
    )

    assert main([
        "compare-ghas",
        "--storage-backend",
        "dynamodb",
        "--dynamodb-table",
        "SecurityScannerLocal",
    ]) == 0
    captured = capsys.readouterr()

    assert "GHAS API Comparison" in captured.out
    assert "Local findings: 1" in captured.out
    assert "GHAS alert inputs: 1" in captured.out
    assert "Matched: 1" in captured.out
    assert "Local-only: 0" in captured.out
    assert "GHAS-only: 0" in captured.out
    assert "example-org" not in captured.out
    assert fake_store.written_alerts == [alert]


def test_storage_backend_defaults_from_env(monkeypatch):
    from security_scanner.cli.app import build_parser

    monkeypatch.setenv("SECURITY_SCANNER_STORAGE_BACKEND", "dynamodb")
    parser = build_parser()
    args = parser.parse_args(["list-targets"])
    assert args.storage_backend == "dynamodb"


@pytest.mark.parametrize(
    ("env_value", "expected"),
    [
        ("DYNAMODB", "dynamodb"),
        ("JSONL", "jsonl"),
    ],
)
def test_storage_backend_defaults_from_env_are_normalized(
    monkeypatch, env_value, expected
):
    from security_scanner.cli.app import build_parser

    monkeypatch.setenv("SECURITY_SCANNER_STORAGE_BACKEND", env_value)
    parser = build_parser()
    args = parser.parse_args(["list-targets"])
    assert args.storage_backend == expected


def test_storage_backend_defaults_jsonl_without_env(monkeypatch):
    from security_scanner.cli.app import build_parser

    monkeypatch.delenv("SECURITY_SCANNER_STORAGE_BACKEND", raising=False)
    parser = build_parser()
    args = parser.parse_args(["list-targets"])
    assert args.storage_backend == "jsonl"


def test_subcommand_registration_order_is_stable():
    """Lock the subcommand registration order.

    argparse preserves add_parser() order, so this order is what users see in
    ``--help`` and in the 'invalid choice: (choose from ...)' error. Command
    modules are registered as domain groups (cli/app.py:_COMMAND_MODULES); this
    test pins the resulting order so a future regrouping cannot silently change
    that user-facing output.
    """
    import argparse

    from security_scanner.cli.app import build_parser

    parser = build_parser()
    subparsers_action = next(
        action
        for action in parser._actions
        if isinstance(action, argparse._SubParsersAction)
    )
    assert list(subparsers_action.choices) == [
        "scan",
        "discover-updates",
        "scan-worker",
        "queue-status",
        "residual",
        "residual-diff",
        "scan-all",
        "import-sarif",
        "scan-vuln",
        "scan-health",
        "report",
        "gate",
        "evaluate",
        "compare-ghas",
        "verify",
        "init-storage",
        "doctor",
        "quickstart",
        "add-target",
        "add-targets",
        "list-targets",
        "remove-target",
        "enable-target",
        "disable-target",
        "sync",
        "backfill-repo-axis",
        "backfill-list-axis",
        "disposition",
    ]