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:")