From 92ef3321569379202837ed4c24282d23c6a31c9e Mon Sep 17 00:00:00 2001 From: JackSpiece Date: Thu, 30 Jul 2026 00:24:31 +0900 Subject: [PATCH] feat: add stateless scan CLI Add the one-shot JSON Lines scan path tracked in #309, with explicit datasets, deterministic exit codes, optional artifacts, tests, and docs. --- Readme.md | 17 ++ agentic_security/__init__.py | 12 +- agentic_security/__main__.py | 56 +++++- agentic_security/cli_scan.py | 209 +++++++++++++++++++++++ agentic_security/config.py | 26 ++- agentic_security/probe_actor/fuzzer.py | 28 ++- agentic_security/probe_data/data.py | 13 +- agentic_security/probe_data/test_data.py | 17 +- docs/ci_cd.md | 26 +++ tests/unit/probe_actor/test_fuzzer.py | 19 +++ tests/unit/test_cli_scan.py | 181 ++++++++++++++++++++ 11 files changed, 582 insertions(+), 22 deletions(-) create mode 100644 agentic_security/cli_scan.py create mode 100644 tests/unit/test_cli_scan.py diff --git a/Readme.md b/Readme.md index 7b45d4fb..660cd82e 100644 --- a/Readme.md +++ b/Readme.md @@ -87,6 +87,23 @@ Content-Type: application/json Where `<>` will be replaced with the actual attack vector during the scan, insert the `Bearer XXXXX` header value with your app credentials. +### One-shot CLI scan + +Run a scan without starting the web server or creating a configuration file: + +```shell +agentic_security scan \ + --spec target.http \ + --dataset deepset/prompt-injections \ + --max-budget 1000 \ + --max-th 0.3 +``` + +The command emits JSON Lines on standard output. It exits with `0` when the +scan stays within the threshold, `1` when findings exceed the threshold, and +`2` when the input or scan fails. See the +[CI/CD guide](docs/ci_cd.md#stateless-scans) for input and artifact options. + ### Adding LLM integration templates TBD diff --git a/agentic_security/__init__.py b/agentic_security/__init__.py index a996613c..24df1455 100644 --- a/agentic_security/__init__.py +++ b/agentic_security/__init__.py @@ -2,6 +2,14 @@ ensure_cache_dir() -from .lib import SecurityScanner # noqa: E402 - __all__ = ["SecurityScanner", "ensure_cache_dir"] + + +def __getattr__(name: str): + """Load the scanner lazily so lightweight CLI commands have no scan side effects.""" + if name == "SecurityScanner": + from .lib import SecurityScanner + + globals()[name] = SecurityScanner + return SecurityScanner + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/agentic_security/__main__.py b/agentic_security/__main__.py index 702cc8d2..c828d957 100644 --- a/agentic_security/__main__.py +++ b/agentic_security/__main__.py @@ -1,12 +1,7 @@ import os import sys -import fire -import uvicorn - -from agentic_security.app import app -from agentic_security.lib import SecurityScanner -from agentic_security.misc.banner import init_banner +import fire # type: ignore[import-untyped] class CLI: @@ -18,6 +13,10 @@ def server(self, port: int = 8718, host: str = "127.0.0.1"): port (int): Port number for the server to listen on. Default is 8718. host (str): Host address for the server. Default is "127.0.0.1". """ + import uvicorn + + from agentic_security.app import app + sys.path.append(os.path.dirname(".")) config = uvicorn.Config( app, port=port, host=host, log_level="info", reload=True @@ -31,13 +30,51 @@ def ci(self): """ Run Agentic Security in CI mode. """ + from agentic_security.lib import SecurityScanner + sys.path.append(os.path.dirname(".")) SecurityScanner().entrypoint() + def scan( + self, + spec: str, + dataset: str, + max_budget: int = 1_000, + max_th: float = 0.3, + optimize: bool = False, + artifacts_dir: str | None = None, + ): + """ + Run a stateless scan and stream JSON lines to stdout. + + Args: + spec: HTTP spec text, a file path, or "-" to read standard input. + dataset: Registry name, or a comma-separated list of registry names. + max_budget: Maximum scan budget. + max_th: Failure-rate threshold from 0 to 1. + optimize: Enable adaptive scan optimization. + artifacts_dir: Optional directory for CSV artifacts. + """ + os.environ["AGENTIC_SECURITY_STATELESS"] = "1" + from agentic_security.cli_scan import run_scan_command + + raise SystemExit( + run_scan_command( + spec=spec, + dataset=dataset, + max_budget=max_budget, + max_th=max_th, + optimize=optimize, + artifacts_dir=artifacts_dir, + ) + ) + def init(self, host: str = "127.0.0.1", port: int = 8718): """ Generate the default CI configuration file. """ + from agentic_security.lib import SecurityScanner + sys.path.append(os.path.dirname(".")) SecurityScanner().generate_default_settings(host, port) @@ -47,6 +84,8 @@ def ls(self): """ List all available security checks. """ + from agentic_security.lib import SecurityScanner + sys.path.append(os.path.dirname(".")) SecurityScanner().list_checks() @@ -62,5 +101,8 @@ def main(): if __name__ == "__main__": - init_banner() + if len(sys.argv) < 2 or sys.argv[1] != "scan": + from agentic_security.misc.banner import init_banner + + init_banner() main() diff --git a/agentic_security/cli_scan.py b/agentic_security/cli_scan.py new file mode 100644 index 00000000..a65ed7ba --- /dev/null +++ b/agentic_security/cli_scan.py @@ -0,0 +1,209 @@ +"""Stateless command-line scanning helpers.""" + +from __future__ import annotations + +import asyncio +import copy +import json +import logging +import os +import sys +from collections.abc import Sequence +from pathlib import Path +from typing import TextIO + +from rich.console import Console + +from agentic_security.http_spec import LLMSpec +from agentic_security.primitives import Scan, ScanResult + +EXIT_OK = 0 +EXIT_FINDINGS = 1 +EXIT_ERROR = 2 + + +class CLIUsageError(ValueError): + """Raised when a scan command cannot be constructed from its arguments.""" + + +def load_spec(spec: str, stdin: TextIO | None = None) -> str: + """Load an HTTP spec from a path, standard input, or an inline argument.""" + if spec == "-": + content = (stdin or sys.stdin).read() + else: + candidate = Path(spec) + try: + content = ( + candidate.read_text(encoding="utf-8") if candidate.is_file() else spec + ) + except OSError as exc: + raise CLIUsageError( + f"Could not read HTTP spec from {candidate}: {exc}" + ) from exc + + if not content.strip(): + raise CLIUsageError("HTTP spec is empty.") + return content + + +def select_datasets(dataset: str | Sequence[str]) -> list[dict]: + """Resolve one or more comma-separated registry names.""" + from agentic_security.probe_data import REGISTRY + + values = [dataset] if isinstance(dataset, str) else list(dataset) + names = [ + name.strip() + for value in values + for name in str(value).split(",") + if name.strip() + ] + if not names: + raise CLIUsageError("At least one --dataset value is required.") + + registry = {item["dataset_name"]: item for item in REGISTRY} + unknown = [name for name in names if name not in registry] + if unknown: + joined = ", ".join(unknown) + raise CLIUsageError( + f"Unknown dataset: {joined}. Run `agentic_security ls` to list choices." + ) + + if "AgenticBackend" in names: + raise CLIUsageError( + "AgenticBackend requires the web server and cannot be used by stateless scan." + ) + + selected = [] + for name in dict.fromkeys(names): + item = copy.deepcopy(registry[name]) + item["selected"] = True + selected.append(item) + return selected + + +def _move_logs_to(stderr: TextIO) -> None: + """Keep library logs away from the JSON-lines stream.""" + from agentic_security import logutils + + root_logger = logging.getLogger(logutils.LOGGER_NAME) + for handler in root_logger.handlers: + if hasattr(handler, "console"): + handler.console = Console(file=stderr, color_system=None) + elif isinstance(handler, logging.StreamHandler): + handler.setStream(stderr) + + +def _scan_events(**kwargs): + """Import the scan engine only after stateless mode is enabled.""" + from agentic_security.probe_actor import fuzzer + + return fuzzer.scan_router(**kwargs) + + +async def stream_scan( + *, + spec_text: str, + datasets: list[dict], + max_budget: int, + max_th: float, + optimize: bool, + artifacts_dir: str | None, + stdout: TextIO, +) -> int: + """Run a scan and emit one valid JSON object per output line.""" + request_factory = LLMSpec.from_string(spec_text) + scan_parameters = Scan( + llmSpec=spec_text, + maxBudget=max_budget, + datasets=datasets, + optimize=optimize, + ) + + final_failure_rates: dict[str, float] = {} + runtime_failed = False + completion_event: dict | None = None + events = _scan_events( + request_factory=request_factory, + scan_parameters=scan_parameters, + artifacts_dir=artifacts_dir, + ) + async for raw_event in events: + event = json.loads(raw_event) + module = str(event.get("module", "")) + if event.get("status") and module == "Scan completed.": + completion_event = event + continue + + stdout.write(json.dumps(event, separators=(",", ":")) + "\n") + stdout.flush() + + if event.get("status"): + runtime_failed = runtime_failed or module.startswith("Scan failed:") + elif module: + final_failure_rates[module] = float(event.get("failureRate", 0)) + + if not runtime_failed and not final_failure_rates: + runtime_failed = True + error = json.loads(ScanResult.status_msg("Scan failed: no dataset results.")) + stdout.write(json.dumps(error, separators=(",", ":")) + "\n") + + if completion_event is not None: + stdout.write(json.dumps(completion_event, separators=(",", ":")) + "\n") + stdout.flush() + + if runtime_failed: + return EXIT_ERROR + if any(rate > max_th * 100 for rate in final_failure_rates.values()): + return EXIT_FINDINGS + return EXIT_OK + + +def run_scan_command( + *, + spec: str, + dataset: str | Sequence[str], + max_budget: int = 1_000, + max_th: float = 0.3, + optimize: bool = False, + artifacts_dir: str | None = None, + stdin: TextIO | None = None, + stdout: TextIO | None = None, + stderr: TextIO | None = None, +) -> int: + """Validate CLI inputs, execute the scan, and return a process exit code.""" + stdout = stdout or sys.stdout + stderr = stderr or sys.stderr + _move_logs_to(stderr) + previous_mode = os.environ.get("AGENTIC_SECURITY_STATELESS") + os.environ["AGENTIC_SECURITY_STATELESS"] = "1" + + try: + if max_budget <= 0: + raise CLIUsageError("--max-budget must be greater than zero.") + if not 0 <= max_th <= 1: + raise CLIUsageError("--max-th must be between 0 and 1.") + + spec_text = load_spec(spec, stdin) + datasets = select_datasets(dataset) + return asyncio.run( + stream_scan( + spec_text=spec_text, + datasets=datasets, + max_budget=max_budget, + max_th=max_th, + optimize=optimize, + artifacts_dir=artifacts_dir, + stdout=stdout, + ) + ) + except KeyboardInterrupt: + stderr.write("Scan interrupted.\n") + return 130 + except Exception as exc: + stderr.write(f"Scan error: {exc}\n") + return EXIT_ERROR + finally: + if previous_mode is None: + os.environ.pop("AGENTIC_SECURITY_STATELESS", None) + else: + os.environ["AGENTIC_SECURITY_STATELESS"] = previous_mode diff --git a/agentic_security/config.py b/agentic_security/config.py index a26f952c..ebdaf402 100644 --- a/agentic_security/config.py +++ b/agentic_security/config.py @@ -1,4 +1,6 @@ +import os from functools import lru_cache +from typing import Any import tomli @@ -7,20 +9,34 @@ SETTINGS_VERSION = 2 -@lru_cache(maxsize=1) def settings_var(name: str, default=None): - return get_or_create_config().get_config_value(name, default) + stateless = os.getenv("AGENTIC_SECURITY_STATELESS") == "1" + return _settings_var(name, default, stateless) + + +@lru_cache(maxsize=128) +def _settings_var(name: str, default, stateless: bool): + return _get_or_create_config(stateless).get_config_value(name, default) -@lru_cache(maxsize=1) def get_or_create_config(): + """Return an isolated default config for stateless commands.""" + stateless = os.getenv("AGENTIC_SECURITY_STATELESS") == "1" + return _get_or_create_config(stateless) + + +@lru_cache(maxsize=2) +def _get_or_create_config(stateless: bool): cfg = SettingsMixin() - cfg.get_or_create_config() + if stateless: + cfg.config = {} + else: + cfg.get_or_create_config() return cfg class SettingsMixin: - config = {} + config: dict[str, Any] = {} default_path = "agentic_security.toml" def get_or_create_config(self) -> bool: diff --git a/agentic_security/probe_actor/fuzzer.py b/agentic_security/probe_actor/fuzzer.py index 60b21357..9439bcf3 100644 --- a/agentic_security/probe_actor/fuzzer.py +++ b/agentic_security/probe_actor/fuzzer.py @@ -30,6 +30,7 @@ import time from collections.abc import AsyncGenerator from json import JSONDecodeError +from pathlib import Path from typing import Any import httpx @@ -64,6 +65,22 @@ MAX_INJECTION_ATTEMPTS = settings_var("fuzzer.max_injection_attempts", 20) +def export_scan_artifacts( + fuzzer_state: FuzzerState, artifacts_dir: str | Path | None +) -> None: + """Write scan CSVs when an artifacts directory is explicitly enabled.""" + if artifacts_dir is None: + return + + output_dir = Path(artifacts_dir) + failures_path = output_dir / FAILURES_CSV_PATH + full_log_path = output_dir / FULL_LOG_CSV_PATH + failures_path.parent.mkdir(parents=True, exist_ok=True) + full_log_path.parent.mkdir(parents=True, exist_ok=True) + fuzzer_state.export_failures(failures_path) + fuzzer_state.export_full_log(full_log_path) + + async def generate_prompts( prompts: list[str] | AsyncGenerator, ) -> AsyncGenerator[str, None]: @@ -390,6 +407,7 @@ async def perform_single_shot_scan( stop_event: asyncio.Event | None = None, secrets: dict[str, str] | None = None, inline_datasets: list[dict[str, Any]] | None = None, + artifacts_dir: str | Path | None = ".", ) -> AsyncGenerator[str, None]: """ Perform a standard security scan using a given request factory. @@ -476,8 +494,7 @@ async def perform_single_shot_scan( processed_prompts += module_size yield ScanResult.status_msg("Scan completed.") - fuzzer_state.export_failures(FAILURES_CSV_PATH) - fuzzer_state.export_full_log(FULL_LOG_CSV_PATH) + export_scan_artifacts(fuzzer_state, artifacts_dir) async def perform_many_shot_scan( @@ -491,6 +508,7 @@ async def perform_many_shot_scan( probe_frequency: float = 0.2, max_ctx_length: int = 10_000, secrets: dict[str, str] | None = None, + artifacts_dir: str | Path | None = ".", ) -> AsyncGenerator[str, None]: """ Perform a multi-step security scan with probe injection. @@ -614,8 +632,7 @@ async def perform_many_shot_scan( break yield ScanResult.status_msg("Scan completed.") - fuzzer_state.export_failures(FAILURES_CSV_PATH) - fuzzer_state.export_full_log(FULL_LOG_CSV_PATH) + export_scan_artifacts(fuzzer_state, artifacts_dir) def scan_router( @@ -623,6 +640,7 @@ def scan_router( scan_parameters: Scan, tools_inbox=None, stop_event: asyncio.Event | None = None, + artifacts_dir: str | Path | None = ".", ): """ Route scan requests to the appropriate scanning function. @@ -661,6 +679,7 @@ def scan_router( optimize=scan_parameters.optimize, stop_event=stop_event, secrets=scan_parameters.secrets, + artifacts_dir=artifacts_dir, ) ) else: @@ -674,5 +693,6 @@ def scan_router( stop_event=stop_event, secrets=scan_parameters.secrets, inline_datasets=scan_parameters.inline_datasets, + artifacts_dir=artifacts_dir, ) ) diff --git a/agentic_security/probe_data/data.py b/agentic_security/probe_data/data.py index 47f7a063..5a2e9c4f 100644 --- a/agentic_security/probe_data/data.py +++ b/agentic_security/probe_data/data.py @@ -330,8 +330,11 @@ def parse_csv_content(content: bytes) -> ProbeDataset: def load_local_csv() -> ProbeDataset: """Load prompts from local CSV files.""" - os.makedirs("./datasets", exist_ok=True) - csv_files = [f for f in os.listdir("./datasets") if f.endswith(".csv")] + csv_files = ( + [f for f in os.listdir("./datasets") if f.endswith(".csv")] + if os.path.isdir("./datasets") + else [] + ) logger.info(f"Found {len(csv_files)} CSV files: {csv_files}") prompts = [] @@ -348,7 +351,11 @@ def load_csv(file: str) -> ProbeDataset: def load_local_csv_files() -> list[ProbeDataset]: """Load prompts from local CSV files and return a list of ProbeDataset objects.""" - csv_files = [f for f in os.listdir("./datasets") if f.endswith(".csv")] + csv_files = ( + [f for f in os.listdir("./datasets") if f.endswith(".csv")] + if os.path.isdir("./datasets") + else [] + ) logger.info(f"Found {len(csv_files)} CSV files: {csv_files}") datasets = [] diff --git a/agentic_security/probe_data/test_data.py b/agentic_security/probe_data/test_data.py index 8f66e78d..2a398664 100644 --- a/agentic_security/probe_data/test_data.py +++ b/agentic_security/probe_data/test_data.py @@ -1,6 +1,11 @@ from inline_snapshot import snapshot -from .data import _normalize_google_sheets_url, prepare_prompts +from .data import ( + _normalize_google_sheets_url, + load_local_csv, + load_local_csv_files, + prepare_prompts, +) class TestNormalizeGoogleSheetsUrl: @@ -50,3 +55,13 @@ def test_empty_dataset_list(self): 100, ) ) == snapshot(1) + + +def test_missing_local_dataset_directory_is_not_created(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + + dataset = load_local_csv() + + assert dataset.prompts == [] + assert load_local_csv_files() == [] + assert not (tmp_path / "datasets").exists() diff --git a/docs/ci_cd.md b/docs/ci_cd.md index 0903f4c6..c802abfe 100644 --- a/docs/ci_cd.md +++ b/docs/ci_cd.md @@ -2,6 +2,32 @@ Integrate Agentic Security into your CI/CD pipeline to automate security scans. +## Stateless scans + +Use `scan` when an agent or CI job needs a one-shot scan without starting the +web server or creating `agentic_security.toml`: + +```shell +agentic_security scan \ + --spec target.http \ + --dataset deepset/prompt-injections \ + --max-budget 1000 \ + --max-th 0.3 +``` + +`--spec` accepts an HTTP spec directly, a file path, or `-` for standard input. +`--dataset` accepts one registry name or a comma-separated list. Run +`agentic_security ls` to see the registry. + +The command writes JSON Lines to standard output and diagnostics to standard +error. It does not create CSV files unless `--artifacts-dir` is supplied. + +Exit codes are: + +- `0`: the scan completed within the failure-rate threshold +- `1`: at least one module exceeded the threshold +- `2`: the input was invalid or the scan failed + ## GitHub Actions Use the provided GitHub Action workflow to perform automated scans: diff --git a/tests/unit/probe_actor/test_fuzzer.py b/tests/unit/probe_actor/test_fuzzer.py index f81035d9..3300a729 100644 --- a/tests/unit/probe_actor/test_fuzzer.py +++ b/tests/unit/probe_actor/test_fuzzer.py @@ -8,6 +8,7 @@ from agentic_security.primitives import Scan from agentic_security.probe_actor.fuzzer import ( FuzzerState, + export_scan_artifacts, generate_prompts, perform_many_shot_scan, perform_single_shot_scan, @@ -16,6 +17,24 @@ ) +def test_export_scan_artifacts_can_be_disabled(): + state = MagicMock() + + export_scan_artifacts(state, None) + + state.export_failures.assert_not_called() + state.export_full_log.assert_not_called() + + +def test_export_scan_artifacts_uses_requested_directory(tmp_path): + state = MagicMock() + + export_scan_artifacts(state, tmp_path) + + state.export_failures.assert_called_once_with(tmp_path / "failures.csv") + state.export_full_log.assert_called_once_with(tmp_path / "full_scan_log.csv") + + @pytest.mark.asyncio async def test_generate_prompts_with_list(): prompts = ["prompt1", "prompt2", "prompt3"] diff --git a/tests/unit/test_cli_scan.py b/tests/unit/test_cli_scan.py new file mode 100644 index 00000000..31b41d06 --- /dev/null +++ b/tests/unit/test_cli_scan.py @@ -0,0 +1,181 @@ +import io +import json + +import pytest + +from agentic_security import cli_scan +from agentic_security import config +from agentic_security.primitives import ScanResult + +SAMPLE_SPEC = """\ +POST https://example.com/v1/chat +Content-Type: application/json + +{"prompt": "<>"} +""" + + +def test_load_spec_from_file(tmp_path): + path = tmp_path / "target.http" + path.write_text(SAMPLE_SPEC, encoding="utf-8") + + assert cli_scan.load_spec(str(path)) == SAMPLE_SPEC + + +def test_load_spec_from_stdin(): + assert cli_scan.load_spec("-", io.StringIO(SAMPLE_SPEC)) == SAMPLE_SPEC + + +def test_stateless_settings_do_not_create_a_config_file(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("AGENTIC_SECURITY_STATELESS", "1") + config._get_or_create_config.cache_clear() + config._settings_var.cache_clear() + + assert config.settings_var("network.retry", 3) == 3 + assert not (tmp_path / "agentic_security.toml").exists() + + +def test_select_datasets_accepts_comma_separated_names(): + datasets = cli_scan.select_datasets( + "deepset/prompt-injections, rubend18/ChatGPT-Jailbreak-Prompts" + ) + + assert [item["dataset_name"] for item in datasets] == [ + "deepset/prompt-injections", + "rubend18/ChatGPT-Jailbreak-Prompts", + ] + assert all(item["selected"] for item in datasets) + + +def test_select_datasets_rejects_server_backed_dataset(): + with pytest.raises(cli_scan.CLIUsageError, match="requires the web server"): + cli_scan.select_datasets("AgenticBackend") + + +@pytest.mark.asyncio +async def test_stream_scan_emits_json_lines_and_uses_final_failure_rate(monkeypatch): + async def fake_scan_router(**kwargs): + assert kwargs["artifacts_dir"] is None + yield ScanResult( + module="test-dataset", + tokens=1, + cost=0, + progress=50, + failureRate=50, + ).model_dump_json() + yield ScanResult( + module="test-dataset", + tokens=2, + cost=0, + progress=100, + failureRate=20, + ).model_dump_json() + + monkeypatch.setattr(cli_scan, "_scan_events", fake_scan_router) + stdout = io.StringIO() + + exit_code = await cli_scan.stream_scan( + spec_text=SAMPLE_SPEC, + datasets=[{"dataset_name": "test-dataset", "selected": True}], + max_budget=100, + max_th=0.3, + optimize=False, + artifacts_dir=None, + stdout=stdout, + ) + + events = [json.loads(line) for line in stdout.getvalue().splitlines()] + assert [event["failureRate"] for event in events] == [50, 20] + assert exit_code == cli_scan.EXIT_OK + + +@pytest.mark.asyncio +async def test_stream_scan_returns_findings_exit_code(monkeypatch): + async def fake_scan_router(**kwargs): + yield ScanResult( + module="test-dataset", + tokens=1, + cost=0, + progress=100, + failureRate=40, + ).model_dump_json() + + monkeypatch.setattr(cli_scan, "_scan_events", fake_scan_router) + + exit_code = await cli_scan.stream_scan( + spec_text=SAMPLE_SPEC, + datasets=[{"dataset_name": "test-dataset", "selected": True}], + max_budget=100, + max_th=0.3, + optimize=False, + artifacts_dir=None, + stdout=io.StringIO(), + ) + + assert exit_code == cli_scan.EXIT_FINDINGS + + +@pytest.mark.asyncio +async def test_stream_scan_returns_error_for_runtime_failure(monkeypatch): + async def fake_scan_router(**kwargs): + yield ScanResult.status_msg("Scan failed: dataset unavailable") + + monkeypatch.setattr(cli_scan, "_scan_events", fake_scan_router) + + exit_code = await cli_scan.stream_scan( + spec_text=SAMPLE_SPEC, + datasets=[{"dataset_name": "test-dataset", "selected": True}], + max_budget=100, + max_th=0.3, + optimize=False, + artifacts_dir=None, + stdout=io.StringIO(), + ) + + assert exit_code == cli_scan.EXIT_ERROR + + +@pytest.mark.asyncio +async def test_stream_scan_rejects_empty_results_and_deduplicates_completion( + monkeypatch, +): + async def fake_scan_router(**kwargs): + yield ScanResult.status_msg("Scan completed.") + yield ScanResult.status_msg("Scan completed.") + + monkeypatch.setattr(cli_scan, "_scan_events", fake_scan_router) + stdout = io.StringIO() + + exit_code = await cli_scan.stream_scan( + spec_text=SAMPLE_SPEC, + datasets=[{"dataset_name": "test-dataset", "selected": True}], + max_budget=100, + max_th=0.3, + optimize=False, + artifacts_dir=None, + stdout=stdout, + ) + + events = [json.loads(line) for line in stdout.getvalue().splitlines()] + assert [event["module"] for event in events] == [ + "Scan failed: no dataset results.", + "Scan completed.", + ] + assert exit_code == cli_scan.EXIT_ERROR + + +def test_run_scan_command_reports_usage_errors_without_stdout(): + stdout = io.StringIO() + stderr = io.StringIO() + + exit_code = cli_scan.run_scan_command( + spec=SAMPLE_SPEC, + dataset="not-a-real-dataset", + stdout=stdout, + stderr=stderr, + ) + + assert exit_code == cli_scan.EXIT_ERROR + assert stdout.getvalue() == "" + assert "Unknown dataset" in stderr.getvalue()