From 591b46739c8963a9a424dd2f6884a0462e777f74 Mon Sep 17 00:00:00 2001 From: Artemouse Date: Wed, 30 Sep 2026 06:19:06 -0400 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20report=20triggered=20scanne?= =?UTF-8?q?r=20names=20in=20evaluation=20results?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports the scanner-mapping work from the GitLab-only branch reformat_input_and_output_csv_and_fetchscanners, which was never carried across to GitHub during the v2.0 refactor and would have been lost when the GitLab project is deleted. Scan responses identify scanners by opaque ID only. This resolves those IDs to human-readable names via the paged /scanners endpoint and records which scanners flagged each prompt in a new "scanners" field. Differences from the original implementation: - The lookup is lazy and cached rather than run at import time, so importing the module no longer performs network I/O. The original would have fired five requests on import and broken CI, which imports this module while mocking only requests.post. - Requests carry an explicit timeout, and a failing source is skipped rather than aborting the whole lookup. - Unknown scanner IDs fall back to the raw ID instead of a placeholder, so the information is preserved rather than discarded. - The field flows through all four output formats and the false positive and false negative exports. - Adds --no-scanner-names to skip the lookup entirely. The CSV input handling from the original commit was already superseded by tools/enhanced_dataset_reader.py and was not ported. Co-Authored-By: Allan Healy Co-Authored-By: Warp --- prompt_evaluator.py | 190 +++++++++++++-- tests/test_runner_simple.py | 5 + tests/test_scanner_resolution.py | 406 +++++++++++++++++++++++++++++++ 3 files changed, 586 insertions(+), 15 deletions(-) create mode 100644 tests/test_scanner_resolution.py diff --git a/prompt_evaluator.py b/prompt_evaluator.py index 5115195..70efd96 100644 --- a/prompt_evaluator.py +++ b/prompt_evaluator.py @@ -53,9 +53,132 @@ if not api_key: raise ValueError("CALYPSOAI_TOKEN environment variable not found") +# --------------------------------------------------------------------------- +# Scanner name resolution +# --------------------------------------------------------------------------- +# Scan responses identify scanners by opaque ID only. The /scanners endpoint +# exposes the ID -> name mapping, but it is partitioned by "source", so the +# full map is assembled by querying each known source once and caching the +# merged result for the lifetime of the process. +SCANNER_SOURCES = ("vendor", "system", "zip", "share", "internal") +SCANNER_REQUEST_TIMEOUT = 30 +SCANNER_PAGE_LIMIT = 100 + +# Cache sentinel. None means "never fetched"; a dict means "already fetched", +# including an empty dict, which records a failed lookup so the API is not +# re-queried once per prompt for the rest of the run. +_scanner_map_cache = None + + +def fetch_scanner_map(force_refresh=False): + """Build a mapping of scanner ID to human-readable scanner name. + + The mapping is cached at module level because it is identical for every + prompt in a run; without caching this would cost one request per source + per prompt. + + Args: + force_refresh (bool): Ignore any cached map and re-query the API. + + Returns: + dict: {scanner_id: scanner_name}. Empty when every source failed, + which callers treat as "fall back to raw IDs" rather than fatal. + + Edge cases: + - A failing source is warned about and skipped, so one bad source does + not discard scanners successfully read from the others. + - A total failure is cached as {} so the lookup is attempted only once. + - This is deliberately lazy. Calling it at import time would fire HTTP + requests simply by importing the module, which breaks CI and --help. + """ + global _scanner_map_cache + + if _scanner_map_cache is not None and not force_refresh: + return _scanner_map_cache + + scanner_map = {} + + with requests.Session() as session: + session.headers.update({'Authorization': f'Bearer {api_key}'}) + for source in SCANNER_SOURCES: + try: + response = session.get( + f"{base_url}/scanners", + params={'source': source, 'limit': SCANNER_PAGE_LIMIT}, + timeout=SCANNER_REQUEST_TIMEOUT, + ) + response.raise_for_status() + payload = response.json() + except (requests.RequestException, ValueError) as e: + print(f"Warning: could not fetch scanners for source '{source}': {e}") + continue + + for scanner in payload.get("scanners", []) or []: + scanner_id = scanner.get("id") + if scanner_id: + scanner_map[scanner_id] = scanner.get("name") or scanner_id + + if not scanner_map: + print("Warning: scanner name lookup returned no scanners; " + "results will report raw scanner IDs.") + + _scanner_map_cache = scanner_map + return scanner_map + + +def extract_triggered_scanners(result_payload, scanner_map=None): + """Return the names of the scanners that did not pass for a single scan. + + Args: + result_payload (dict): The 'result' object from a /scans response. + scanner_map (dict): Optional {id: name} mapping from + fetch_scanner_map(). When omitted, or when an ID is missing from + it, the raw scanner ID is used so the information is preserved + rather than replaced by a placeholder. + + Returns: + list[str]: Triggered scanner names (or IDs), in API order. + + Edge cases: + - Missing or non-list 'scannerResults' yields an empty list. + - Non-dict entries are skipped defensively. + - Entries with no 'outcome' count as triggered, since a result we + cannot classify is worth surfacing rather than silently dropping. + """ + scanner_map = scanner_map or {} + scanner_results = result_payload.get('scannerResults') or [] + + if not isinstance(scanner_results, list): + return [] + + triggered = [] + for scanner in scanner_results: + if not isinstance(scanner, dict): + continue + if str(scanner.get('outcome', '')).lower() == 'passed': + continue + scanner_id = scanner.get('scannerId') + if scanner_id is None: + continue + triggered.append(scanner_map.get(scanner_id, scanner_id)) + + return triggered + + # Utility function to scan a prompt using CalypsoAI -def calypsoai_scan(prompt): - """Scan a prompt using the CalypsoAI API.""" +def calypsoai_scan(prompt, scanner_map=None): + """Scan a prompt using the CalypsoAI API. + + Args: + prompt (str): The prompt text to scan. + scanner_map (dict): Optional {scanner_id: name} mapping used to label + the scanners that flagged the prompt. Raw IDs are reported when + this is absent. + + Returns: + dict: 'outcome' (str), 'response_time' (float seconds) and 'scanners' + (list of triggered scanner names). + """ headers = { 'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json', @@ -89,21 +212,38 @@ def calypsoai_scan(prompt): # Handle the new API response format # The response now has: { "result": { "outcome": "flagged" } } - if 'result' in outcome and 'outcome' in outcome['result']: - result_outcome = outcome['result']['outcome'] - else: - # Fallback to old format or other possible formats - result_outcome = outcome.get('result', {}).get('outcome', 'no_result') + result = outcome.get('result') or {} + if not isinstance(result, dict): + result = {} + result_outcome = result.get('outcome', 'no_result') return { 'outcome': result_outcome, - 'response_time': elapsed_time + 'response_time': elapsed_time, + 'scanners': extract_triggered_scanners(result, scanner_map) } # Import the enhanced dataset reader from tools.enhanced_dataset_reader import load_dataset_enhanced +def _format_scanners(value): + """Flatten a scanners list into one delimited cell for flat text formats. + + CSV and TSV cannot hold a list, so the names are joined with '; '. JSON and + Parquet keep the native list instead. + + Args: + value: The record's 'scanners' entry; normally a list, possibly None. + + Returns: + str: Delimited scanner names, or '' when there are none. + """ + if isinstance(value, (list, tuple)): + return '; '.join(str(v) for v in value) + return '' if value is None else str(value) + + def write_results_to_file(data, file_path, format_type): """Write results to file in the specified format.""" import os @@ -122,11 +262,11 @@ def write_results_to_file(data, file_path, format_type): # Write as CSV if not data: with open(file_path, 'w') as f: - f.write("prompt,expected,outcome,response_time,prompt_size,original_line,metadata\n") + f.write("prompt,expected,outcome,scanners,response_time,prompt_size,original_line,metadata\n") return with open(file_path, 'w', newline='', encoding='utf-8') as f: - fieldnames = ['prompt', 'expected', 'outcome', 'response_time', 'prompt_size', 'original_line', 'metadata'] + fieldnames = ['prompt', 'expected', 'outcome', 'scanners', 'response_time', 'prompt_size', 'original_line', 'metadata'] writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() @@ -136,6 +276,7 @@ def write_results_to_file(data, file_path, format_type): 'prompt': record.get('prompt', ''), 'expected': record.get('expected', ''), 'outcome': record.get('outcome', ''), + 'scanners': _format_scanners(record.get('scanners')), 'response_time': record.get('response_time', ''), 'prompt_size': record.get('prompt_size', ''), 'original_line': record.get('original_line', ''), @@ -147,12 +288,12 @@ def write_results_to_file(data, file_path, format_type): # Write as TSV if not data: with open(file_path, 'w') as f: - f.write("prompt\texpected\toutcome\tresponse_time\tprompt_size\toriginal_line\tmetadata\n") + f.write("prompt\texpected\toutcome\tscanners\tresponse_time\tprompt_size\toriginal_line\tmetadata\n") return with open(file_path, 'w', newline='', encoding='utf-8') as f: # Write header - f.write("prompt\texpected\toutcome\tresponse_time\tprompt_size\toriginal_line\tmetadata\n") + f.write("prompt\texpected\toutcome\tscanners\tresponse_time\tprompt_size\toriginal_line\tmetadata\n") for record in data: # Escape tabs in data @@ -164,6 +305,7 @@ def escape_tsv(value): f.write(f"{escape_tsv(record.get('prompt', ''))}\t") f.write(f"{escape_tsv(record.get('expected', ''))}\t") f.write(f"{escape_tsv(record.get('outcome', ''))}\t") + f.write(f"{escape_tsv(_format_scanners(record.get('scanners')))}\t") f.write(f"{escape_tsv(record.get('response_time', ''))}\t") f.write(f"{escape_tsv(record.get('prompt_size', ''))}\t") f.write(f"{escape_tsv(record.get('original_line', ''))}\t") @@ -223,6 +365,8 @@ def __init__(self, *args, **kwargs): self.response_times = [] self.prompt_sizes = [] self.start_time = None + # Resolved once in prep() so the lookup cost is paid per run, not per prompt. + self.scanner_map = {} def prep(self, shared): dataset = shared["dataset"] @@ -231,7 +375,14 @@ def prep(self, shared): if self.total_prompts == 0: print("\nNo prompts to scan. The dataset is empty.") return [] - + + # Resolve scanner IDs to names once up front. Opt-out avoids the extra + # requests when only pass/fail outcomes are needed. + if shared.get("resolve_scanner_names", True): + self.scanner_map = fetch_scanner_map() + else: + self.scanner_map = {} + print(f"\nStarting scan of {self.total_prompts} prompts...") self.start_time = time.time() return dataset @@ -239,9 +390,10 @@ def prep(self, shared): def exec(self, item): prompt = item["prompt"] try: - result = calypsoai_scan(prompt) + result = calypsoai_scan(prompt, self.scanner_map) outcome = result['outcome'] response_time = result['response_time'] + scanners = result['scanners'] # Calculate prompt size (in characters) prompt_size = len(prompt) @@ -260,6 +412,7 @@ def exec(self, item): "prompt": prompt, "expected": item["expected"], "outcome": outcome, + "scanners": scanners, "response_time": response_time, "prompt_size": prompt_size, "original_line": item["original_line"] @@ -275,6 +428,7 @@ def exec(self, item): "prompt": prompt, "expected": item["expected"], "outcome": f"error: {str(e)}", + "scanners": [], "response_time": None, "prompt_size": len(prompt), "original_line": item["original_line"] @@ -472,6 +626,7 @@ def exec(self, inputs): "prompt": result['prompt'], "expected": result['expected'], "outcome": result['outcome'], + "scanners": result.get('scanners', []), "response_time": result.get('response_time'), "prompt_size": result.get('prompt_size'), "original_line": result.get('original_line', ''), @@ -550,6 +705,7 @@ def exec(self, inputs): "prompt": result['prompt'], "expected": result['expected'], "outcome": result['outcome'], + "scanners": result.get('scanners', []), "response_time": result.get('response_time'), "prompt_size": result.get('prompt_size'), "original_line": result.get('original_line', ''), @@ -840,6 +996,9 @@ def main(): parser.add_argument('--results-format', '-r', type=str, default='json', choices=['json', 'csv', 'tsv', 'parquet'], help='Output format for main results file (default: json/jsonl)') + parser.add_argument('--no-scanner-names', action='store_true', + help='Skip the scanner ID to name lookup and report raw ' + 'scanner IDs instead (saves a few startup requests)') args = parser.parse_args() @@ -852,7 +1011,8 @@ def main(): "max_lines": args.lines, "output_file": args.output, "output_format": args.format, - "results_format": args.results_format + "results_format": args.results_format, + "resolve_scanner_names": not args.no_scanner_names } try: diff --git a/tests/test_runner_simple.py b/tests/test_runner_simple.py index e0459cd..1eebf3c 100644 --- a/tests/test_runner_simple.py +++ b/tests/test_runner_simple.py @@ -75,6 +75,11 @@ def main(): cmd = base_cmd + ["tests/test_core_functionality.py", "-v"] if run_command(cmd, "Core Functionality Tests"): success_count += 1 + + total_count += 1 + cmd = base_cmd + ["tests/test_scanner_resolution.py", "-v"] + if run_command(cmd, "Scanner Resolution Tests"): + success_count += 1 # Dataset converter tests if args.converter or args.all: diff --git a/tests/test_scanner_resolution.py b/tests/test_scanner_resolution.py new file mode 100644 index 0000000..d7b8380 --- /dev/null +++ b/tests/test_scanner_resolution.py @@ -0,0 +1,406 @@ +# Copyright 2025 Andy Ryan +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +#!/usr/bin/env python3 +"""Tests for scanner ID to name resolution in prompt_evaluator. + +Covers the triggered-scanner reporting feature: building the scanner map from +the paged /scanners endpoint, resolving scanner IDs on individual scans, and +serialising the resulting list into the flat output formats. +""" + +import csv +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import requests + +# The module reads credentials at import time, so provide throwaway values +# before importing it. No network calls are made by the import itself. +os.environ.setdefault("CALYPSOAI_URL", "https://calypsoai.test") +os.environ.setdefault("CALYPSOAI_TOKEN", "test_token_12345") + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import prompt_evaluator as pe # noqa: E402 + + +@pytest.fixture(autouse=True) +def reset_scanner_cache(): + """Clear the module-level scanner map cache around every test.""" + pe._scanner_map_cache = None + yield + pe._scanner_map_cache = None + + +def _mock_session(payload_by_source, failing_sources=()): + """Build a mock requests.Session whose GET returns per-source payloads. + + Args: + payload_by_source: {source_name: json payload} returned by the API. + failing_sources: Sources that should raise a RequestException. + + Returns: + A MagicMock usable as a context manager in place of requests.Session(). + """ + session = MagicMock() + + def _get(url, params=None, timeout=None): + source = (params or {}).get("source") + if source in failing_sources: + raise requests.RequestException(f"boom for {source}") + response = MagicMock() + response.json.return_value = payload_by_source.get(source, {"scanners": []}) + response.raise_for_status.return_value = None + return response + + session.get.side_effect = _get + session.__enter__.return_value = session + session.__exit__.return_value = False + return session + + +class TestFetchScannerMap: + """Behaviour of the cached scanner ID to name lookup.""" + + def test_merges_scanners_across_all_sources(self): + payloads = { + "vendor": {"scanners": [{"id": "s1", "name": "Prompt Injection"}]}, + "system": {"scanners": [{"id": "s2", "name": "PII"}]}, + "internal": {"scanners": [{"id": "s3", "name": "Custom"}]}, + } + session = _mock_session(payloads) + + with patch("prompt_evaluator.requests.Session", return_value=session): + result = pe.fetch_scanner_map() + + assert result == {"s1": "Prompt Injection", "s2": "PII", "s3": "Custom"} + # One request per known source. + assert session.get.call_count == len(pe.SCANNER_SOURCES) + + def test_every_request_sends_a_timeout(self): + """A request without a timeout can hang the whole evaluation run.""" + session = _mock_session({}) + + with patch("prompt_evaluator.requests.Session", return_value=session): + pe.fetch_scanner_map() + + assert session.get.call_count > 0 + for call in session.get.call_args_list: + assert call.kwargs.get("timeout") == pe.SCANNER_REQUEST_TIMEOUT + + def test_result_is_cached_across_calls(self): + payloads = {"vendor": {"scanners": [{"id": "s1", "name": "One"}]}} + session = _mock_session(payloads) + + with patch("prompt_evaluator.requests.Session", return_value=session): + first = pe.fetch_scanner_map() + second = pe.fetch_scanner_map() + + assert first == second + # Still only one pass over the sources; the second call hit the cache. + assert session.get.call_count == len(pe.SCANNER_SOURCES) + + def test_force_refresh_requeries_the_api(self): + payloads = {"vendor": {"scanners": [{"id": "s1", "name": "One"}]}} + session = _mock_session(payloads) + + with patch("prompt_evaluator.requests.Session", return_value=session): + pe.fetch_scanner_map() + pe.fetch_scanner_map(force_refresh=True) + + assert session.get.call_count == len(pe.SCANNER_SOURCES) * 2 + + def test_one_failing_source_does_not_discard_the_others(self): + payloads = { + "vendor": {"scanners": [{"id": "s1", "name": "Kept"}]}, + "system": {"scanners": [{"id": "s2", "name": "Also kept"}]}, + } + session = _mock_session(payloads, failing_sources=("system",)) + + with patch("prompt_evaluator.requests.Session", return_value=session): + result = pe.fetch_scanner_map() + + assert result == {"s1": "Kept"} + + def test_total_failure_returns_empty_map_and_caches_it(self): + session = _mock_session({}, failing_sources=pe.SCANNER_SOURCES) + + with patch("prompt_evaluator.requests.Session", return_value=session): + result = pe.fetch_scanner_map() + pe.fetch_scanner_map() + + assert result == {} + # The empty result is cached, so the failing sources are not retried. + assert session.get.call_count == len(pe.SCANNER_SOURCES) + + def test_scanner_without_a_name_falls_back_to_its_id(self): + payloads = {"vendor": {"scanners": [{"id": "s1"}, {"id": "s2", "name": None}]}} + session = _mock_session(payloads) + + with patch("prompt_evaluator.requests.Session", return_value=session): + result = pe.fetch_scanner_map() + + assert result == {"s1": "s1", "s2": "s2"} + + def test_entries_without_an_id_are_skipped(self): + payloads = {"vendor": {"scanners": [{"name": "No id"}, {"id": "s1", "name": "Good"}]}} + session = _mock_session(payloads) + + with patch("prompt_evaluator.requests.Session", return_value=session): + result = pe.fetch_scanner_map() + + assert result == {"s1": "Good"} + + +class TestExtractTriggeredScanners: + """Selection and labelling of the scanners that flagged a prompt.""" + + def test_returns_only_scanners_that_did_not_pass(self): + payload = { + "scannerResults": [ + {"scannerId": "s1", "outcome": "passed"}, + {"scannerId": "s2", "outcome": "failed"}, + ] + } + assert pe.extract_triggered_scanners(payload, {"s1": "A", "s2": "B"}) == ["B"] + + def test_passed_check_is_case_insensitive(self): + payload = {"scannerResults": [{"scannerId": "s1", "outcome": "PASSED"}]} + assert pe.extract_triggered_scanners(payload, {"s1": "A"}) == [] + + def test_unknown_id_falls_back_to_the_raw_id(self): + """Preserving the ID beats replacing it with a placeholder.""" + payload = {"scannerResults": [{"scannerId": "unmapped", "outcome": "failed"}]} + assert pe.extract_triggered_scanners(payload, {"s1": "A"}) == ["unmapped"] + + def test_missing_map_returns_raw_ids(self): + payload = {"scannerResults": [{"scannerId": "s1", "outcome": "failed"}]} + assert pe.extract_triggered_scanners(payload) == ["s1"] + + def test_scanner_without_an_outcome_counts_as_triggered(self): + payload = {"scannerResults": [{"scannerId": "s1"}]} + assert pe.extract_triggered_scanners(payload, {"s1": "A"}) == ["A"] + + def test_entry_without_a_scanner_id_is_skipped(self): + payload = {"scannerResults": [{"outcome": "failed"}]} + assert pe.extract_triggered_scanners(payload, {}) == [] + + def test_order_from_the_api_is_preserved(self): + payload = { + "scannerResults": [ + {"scannerId": "s3", "outcome": "failed"}, + {"scannerId": "s1", "outcome": "failed"}, + ] + } + mapping = {"s1": "One", "s3": "Three"} + assert pe.extract_triggered_scanners(payload, mapping) == ["Three", "One"] + + @pytest.mark.parametrize( + "payload", + [ + {}, + {"scannerResults": None}, + {"scannerResults": []}, + {"scannerResults": "not-a-list"}, + ], + ) + def test_malformed_payloads_yield_an_empty_list(self, payload): + assert pe.extract_triggered_scanners(payload, {"s1": "A"}) == [] + + def test_non_dict_entries_are_skipped(self): + payload = {"scannerResults": ["junk", {"scannerId": "s1", "outcome": "failed"}]} + assert pe.extract_triggered_scanners(payload, {"s1": "A"}) == ["A"] + + +class TestCalypsoaiScanScannerField: + """The scan helper surfaces triggered scanners alongside the outcome.""" + + def _mock_post(self, body): + response = MagicMock() + response.json.return_value = body + response.raise_for_status.return_value = None + return response + + def test_scan_reports_resolved_scanner_names(self): + body = { + "result": { + "outcome": "flagged", + "scannerResults": [ + {"scannerId": "s1", "outcome": "failed"}, + {"scannerId": "s2", "outcome": "passed"}, + ], + } + } + + with patch("prompt_evaluator.requests.post", return_value=self._mock_post(body)): + result = pe.calypsoai_scan("hello", {"s1": "Prompt Injection", "s2": "PII"}) + + assert result["outcome"] == "flagged" + assert result["scanners"] == ["Prompt Injection"] + assert result["response_time"] >= 0 + + def test_response_without_a_result_block_is_handled(self): + """Older or mocked responses must not raise.""" + with patch("prompt_evaluator.requests.post", return_value=self._mock_post({})): + result = pe.calypsoai_scan("hello") + + assert result["outcome"] == "no_result" + assert result["scanners"] == [] + + def test_non_dict_result_block_is_handled(self): + body = {"result": "unexpected"} + with patch("prompt_evaluator.requests.post", return_value=self._mock_post(body)): + result = pe.calypsoai_scan("hello") + + assert result["outcome"] == "no_result" + assert result["scanners"] == [] + + +class TestScannerSerialisation: + """Flat output formats need the scanner list collapsed into one cell.""" + + def test_format_scanners_joins_a_list(self): + assert pe._format_scanners(["A", "B"]) == "A; B" + + def test_format_scanners_handles_empty_and_none(self): + assert pe._format_scanners([]) == "" + assert pe._format_scanners(None) == "" + + def test_format_scanners_passes_through_a_plain_string(self): + assert pe._format_scanners("A") == "A" + + def test_csv_output_includes_the_scanners_column(self, tmp_path): + out = tmp_path / "results.csv" + records = [ + { + "prompt": "p1", + "expected": "true", + "outcome": "flagged", + "scanners": ["Prompt Injection", "PII"], + "response_time": 0.1, + "prompt_size": 2, + "original_line": 1, + } + ] + + pe.write_results_to_file(records, str(out), "csv") + + with open(out, newline="", encoding="utf-8") as f: + rows = list(csv.DictReader(f)) + + assert "scanners" in rows[0] + assert rows[0]["scanners"] == "Prompt Injection; PII" + + def test_tsv_output_includes_the_scanners_column(self, tmp_path): + out = tmp_path / "results.tsv" + records = [ + { + "prompt": "p1", + "expected": "true", + "outcome": "flagged", + "scanners": ["Prompt Injection"], + "response_time": 0.1, + "prompt_size": 2, + "original_line": 1, + } + ] + + pe.write_results_to_file(records, str(out), "tsv") + lines = out.read_text(encoding="utf-8").splitlines() + + header = lines[0].split("\t") + assert "scanners" in header + assert lines[1].split("\t")[header.index("scanners")] == "Prompt Injection" + + def test_empty_csv_still_writes_the_scanners_header(self, tmp_path): + out = tmp_path / "empty.csv" + pe.write_results_to_file([], str(out), "csv") + assert "scanners" in out.read_text(encoding="utf-8").splitlines()[0] + + def test_empty_tsv_still_writes_the_scanners_header(self, tmp_path): + out = tmp_path / "empty.tsv" + pe.write_results_to_file([], str(out), "tsv") + assert "scanners" in out.read_text(encoding="utf-8").splitlines()[0] + + def test_jsonl_output_keeps_the_native_list(self, tmp_path): + import json + + out = tmp_path / "results.jsonl" + records = [{"prompt": "p1", "outcome": "flagged", "scanners": ["A", "B"]}] + + pe.write_results_to_file(records, str(out), "json") + written = json.loads(out.read_text(encoding="utf-8").strip()) + + assert written["scanners"] == ["A", "B"] + + def test_record_missing_scanners_writes_an_empty_cell(self, tmp_path): + """Records produced before this feature must still serialise.""" + out = tmp_path / "legacy.csv" + records = [{"prompt": "p1", "expected": "true", "outcome": "cleared"}] + + pe.write_results_to_file(records, str(out), "csv") + + with open(out, newline="", encoding="utf-8") as f: + rows = list(csv.DictReader(f)) + + assert rows[0]["scanners"] == "" + + +class TestScanPromptsNodeIntegration: + """The batch node resolves the map once and attaches it to every record.""" + + def test_map_is_fetched_once_per_run(self): + node = pe.ScanPromptsNode() + shared = {"dataset": [{"prompt": "a", "expected": "true", "original_line": 1}]} + + with patch("prompt_evaluator.fetch_scanner_map", return_value={"s1": "A"}) as mock_fetch: + node.prep(shared) + + mock_fetch.assert_called_once() + assert node.scanner_map == {"s1": "A"} + + def test_opt_out_skips_the_lookup_entirely(self): + node = pe.ScanPromptsNode() + shared = { + "dataset": [{"prompt": "a", "expected": "true", "original_line": 1}], + "resolve_scanner_names": False, + } + + with patch("prompt_evaluator.fetch_scanner_map") as mock_fetch: + node.prep(shared) + + mock_fetch.assert_not_called() + assert node.scanner_map == {} + + def test_empty_dataset_does_not_trigger_a_lookup(self): + node = pe.ScanPromptsNode() + + with patch("prompt_evaluator.fetch_scanner_map") as mock_fetch: + assert node.prep({"dataset": []}) == [] + + mock_fetch.assert_not_called() + + def test_failed_scan_record_still_carries_a_scanners_key(self): + node = pe.ScanPromptsNode() + node.total_prompts = 1 + + with patch("prompt_evaluator.calypsoai_scan", side_effect=RuntimeError("api down")): + record = node.exec({"prompt": "a", "expected": "true", "original_line": 1}) + + assert record["scanners"] == [] + assert record["outcome"].startswith("error:")