From d82db4cc5fcb9fe67a356dea7ea6023187d1dbf8 Mon Sep 17 00:00:00 2001 From: feiiiiii5 <204683769+feiiiiii5@users.noreply.github.com> Date: Sun, 23 Aug 2026 01:55:33 +0800 Subject: [PATCH 1/6] fix(scenario): warn when selected attack techniques have no registered factory resolve_technique_factories silently dropped techniques whose factory was not registered, so a typo or an unregistered custom technique shrank the run without any signal. Emit one warning naming the missing technique(s) in selection order (deduplicated), and update the docstring accordingly. Fixes #2461 --- .../core/matrix_atomic_attack_builder.py | 28 +++++++++--- .../core/test_matrix_atomic_attack_builder.py | 44 +++++++++++++++++++ 2 files changed, 66 insertions(+), 6 deletions(-) diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 40a35ca474..053dcd38b9 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -153,18 +153,34 @@ def resolve_technique_factories( Returns: dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by - the selected techniques. + the selected techniques. Techniques with no registered factory are skipped with a + warning naming them, so the caller can proceed with whatever techniques exist. """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry all_factories = dict(AttackTechniqueRegistry.get_registry_singleton().get_factories_or_raise()) if extra_factories: all_factories.update(extra_factories) - return { - technique.value: all_factories[technique.value] - for technique in context.scenario_techniques - if technique.value in all_factories - } + + resolved: dict[str, AttackTechniqueFactory] = {} + missing: list[str] = [] + seen_missing: set[str] = set() + for technique in context.scenario_techniques: + if technique.value in all_factories: + resolved[technique.value] = all_factories[technique.value] + elif technique.value not in seen_missing: + missing.append(technique.value) + seen_missing.add(technique.value) + + if missing: + logger.warning( + "Skipping %d selected attack technique(s) with no registered factory: %s. " + "Register the technique(s) (or pass them via extra_factories) to include them in the run.", + len(missing), + ", ".join(missing), + ) + + return resolved def build_matrix_atomic_attacks( diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index 044ce78512..4c60be3c6b 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -15,6 +15,7 @@ * optional baseline emission from the flattened seed groups. """ +import logging from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -408,6 +409,49 @@ def test_drops_techniques_without_factory(self): resolved = resolve_technique_factories(context=context) assert list(resolved.keys()) == ["alpha"] + def test_warns_when_dropping_techniques_without_factory(self, caplog): + factories = {"alpha": _mock_factory(name="alpha")} + context = _context(techniques=[_technique("alpha"), _technique("missing")]) + with ( + _patch_registry(factories), + caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), + ): + resolve_technique_factories(context=context) + assert any("missing" in record.message for record in caplog.records) + + def test_no_warning_when_all_techniques_resolve(self, caplog): + factories = { + "alpha": _mock_factory(name="alpha"), + "beta": _mock_factory(name="beta"), + } + context = _context(techniques=[_technique("alpha"), _technique("beta")]) + with ( + _patch_registry(factories), + caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), + ): + resolve_technique_factories(context=context) + assert not [record for record in caplog.records if record.levelno == logging.WARNING] + + def test_warning_lists_each_missing_technique_once_in_selection_order(self, caplog): + factories = {"alpha": _mock_factory(name="alpha")} + context = _context( + techniques=[ + _technique("missing_a"), + _technique("alpha"), + _technique("missing_b"), + _technique("missing_a"), + ] + ) + with ( + _patch_registry(factories), + caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), + ): + resolve_technique_factories(context=context) + messages = [record.message for record in caplog.records if record.levelno == logging.WARNING] + assert len(messages) == 1 + assert messages[0].index("missing_a") < messages[0].index("missing_b") + assert messages[0].count("missing_a") == 1 # duplicates are deduplicated + def test_extra_factories_merged_and_override_registry(self): registry_factories = {"alpha": _mock_factory(name="alpha")} local_alpha = _mock_factory(name="alpha") From 1b3a4811bf2d29a27a5f95aaad17c1654b39e8ed Mon Sep 17 00:00:00 2001 From: fei <204683769+feiiiiii5@users.noreply.github.com> Date: Mon, 24 Aug 2026 19:27:07 +0800 Subject: [PATCH 2/6] fix(scenario): surface skipped techniques and fail fast when no selection resolves MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the silent-technique-drop warning: - A nonempty technique selection that resolves to zero factories now raises TechniqueResolutionError instead of running baseline-only with a success status — a silently empty evaluation. Partial misses keep the warn-and-continue behavior. The error subclasses ValueError so existing handlers keep working, mirroring DatasetConstraintError. - Skipped selections are now visible in normal flows, not just pyrit_backend.log: ScenarioRunSummary gains skipped_techniques, populated by comparing the scenario identifier's resolved techniques against built display groups (execution-progress independent), and pyrit_scan renders a "Skipped:" line so users see when selected techniques were left out. Signed-off-by: fei <204683769+feiiiiii5@users.noreply.github.com> --- .../backend/services/scenario_run_service.py | 13 ++++ pyrit/cli/_output.py | 3 + pyrit/models/catalog/scenario.py | 4 ++ .../core/matrix_atomic_attack_builder.py | 24 +++++++ .../test_scenario_run_service_summary.py | 68 +++++++++++++++++++ tests/unit/cli/test_output.py | 18 +++++ .../core/test_matrix_atomic_attack_builder.py | 23 +++++++ 7 files changed, 153 insertions(+) create mode 100644 tests/unit/backend/services/test_scenario_run_service_summary.py diff --git a/pyrit/backend/services/scenario_run_service.py b/pyrit/backend/services/scenario_run_service.py index 7ba66d0f43..6896eada19 100644 --- a/pyrit/backend/services/scenario_run_service.py +++ b/pyrit/backend/services/scenario_run_service.py @@ -599,6 +599,18 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari completed_attacks = total_attacks techniques_used = scenario_result.get_techniques_used() + # Techniques the user selected but that never produced an attack cell were + # skipped during resolution (no registered factory for them). Compare against + # the built display groups rather than executed results so in-progress runs + # don't report not-yet-run techniques as skipped. + selected = set(scenario_result.scenario_identifier.techniques or []) + built_labels = set(scenario_result.display_group_map.values()) + skipped_techniques = sorted( + technique + for technique in selected + if technique not in built_labels and not any(technique in label for label in built_labels) + ) + # Surface per-attack errors and retry pressure regardless of overall run status: # a COMPLETED scenario can still hide errored objectives or rate-limit retries. failed_attacks: list[AttackErrorSummary] = [] @@ -641,6 +653,7 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari error=error, error_type=error_type, techniques_used=techniques_used, + skipped_techniques=skipped_techniques, total_attacks=total_attacks, completed_attacks=completed_attacks, objective_achieved_rate=scenario_result.objective_achieved_rate(), diff --git a/pyrit/cli/_output.py b/pyrit/cli/_output.py index 53136e15c8..9c8b994ea1 100644 --- a/pyrit/cli/_output.py +++ b/pyrit/cli/_output.py @@ -386,6 +386,9 @@ def print_scenario_run_summary(*, run: ScenarioRunSummary) -> None: if run.techniques_used: print(f" Techniques: {', '.join(run.techniques_used)}") + if run.skipped_techniques: + print(f" Skipped: {', '.join(run.skipped_techniques)} (no registered factory)") + if run.failed_attacks: print(f"\n Failed Attacks ({len(run.failed_attacks)}):") for failed in run.failed_attacks: diff --git a/pyrit/models/catalog/scenario.py b/pyrit/models/catalog/scenario.py index 488ccf8c78..1218e7d061 100644 --- a/pyrit/models/catalog/scenario.py +++ b/pyrit/models/catalog/scenario.py @@ -148,6 +148,10 @@ class ScenarioRunSummary(BaseModel): error: str | None = Field(None, description="Error message if status is FAILED") error_type: str | None = Field(None, description="Exception class name if status is FAILED") techniques_used: list[str] = Field(default_factory=list, description="Technique names that were executed") + skipped_techniques: list[str] = Field( + default_factory=list, + description="Selected techniques that were skipped because no factory was registered for them", + ) total_attacks: int = Field(0, ge=0, description="Total number of attack results persisted for this run") completed_attacks: int = Field(0, ge=0, description="Number of attacks that reached a terminal outcome") objective_achieved_rate: int = Field(0, ge=0, le=100, description="Success rate as percentage (0-100)") diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 053dcd38b9..5224f409b6 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -29,6 +29,17 @@ from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique + +class TechniqueResolutionError(ValueError): + """ + Raised when a scenario selects techniques but none of them resolve to a factory. + + Subclasses ``ValueError`` so existing ``except ValueError`` handlers keep working, + mirroring ``DatasetConstraintError``. Partial misses (some techniques resolve) + only warn, so a run still proceeds with the techniques that do exist. + """ + + if TYPE_CHECKING: from collections.abc import Callable, Mapping, Sequence @@ -155,6 +166,10 @@ def resolve_technique_factories( dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by the selected techniques. Techniques with no registered factory are skipped with a warning naming them, so the caller can proceed with whatever techniques exist. + + Raises: + TechniqueResolutionError: If the selection is nonempty but no technique resolves, + since running only the baseline would silently defeat the selection. """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry @@ -180,6 +195,15 @@ def resolve_technique_factories( ", ".join(missing), ) + if context.scenario_techniques and not resolved: + # A nonempty selection that resolves to nothing would otherwise run the + # baseline only while reporting success — a silently empty evaluation. + raise TechniqueResolutionError( + f"All {len(context.scenario_techniques)} selected attack technique(s) have no registered " + f"factory: {', '.join(missing)}. Register the technique(s) (or pass them via " + "extra_factories) so the run has at least one technique to execute." + ) + return resolved diff --git a/tests/unit/backend/services/test_scenario_run_service_summary.py b/tests/unit/backend/services/test_scenario_run_service_summary.py new file mode 100644 index 0000000000..ba56157493 --- /dev/null +++ b/tests/unit/backend/services/test_scenario_run_service_summary.py @@ -0,0 +1,68 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for skipped-technique diagnostics in ``ScenarioRunService`` run summaries.""" + +import pytest + +from pyrit.backend.services.scenario_run_service import ScenarioRunService +from pyrit.models.identifiers.scenario_identifier import ScenarioIdentifier +from pyrit.models.results.scenario_result import ScenarioResult + + +def _service() -> ScenarioRunService: + from unittest.mock import MagicMock + + service = object.__new__(ScenarioRunService) + service._active_tasks = {} + # The error fallback path queries persisted error results; none exist here. + service._memory = MagicMock() + service._memory.get_attack_results.return_value = [] + return service + + +def _result(*, techniques, display_groups) -> ScenarioResult: + return ScenarioResult( + scenario_identifier=ScenarioIdentifier(name="scenario", techniques=techniques), + attack_results={}, + display_group_map=display_groups, + ) + + +@pytest.mark.usefixtures("patch_central_database") +class TestBuildResponseSkippedTechniques: + """Selected techniques with no built attack cell surface as ``skipped_techniques``.""" + + def test_selected_without_built_label_is_reported_skipped(self): + result = _result(techniques=["alpha", "ghost"], display_groups={"alpha::ds": "alpha"}) + + summary = _service()._build_response_from_db(scenario_result=result) + + assert summary.skipped_techniques == ["ghost"] + assert summary.techniques_used is not None + + def test_decorated_display_label_still_counts_as_built(self): + # Custom ``display_group_fn`` may decorate technique names; a label that + # contains the technique name must not be reported as skipped. + result = _result(techniques=["alpha"], display_groups={"cell-1": "alpha (hard mode)"}) + + summary = _service()._build_response_from_db(scenario_result=result) + + assert summary.skipped_techniques == [] + + def test_no_selection_reports_no_skips(self): + result = _result(techniques=None, display_groups={}) + + summary = _service()._build_response_from_db(scenario_result=result) + + assert summary.skipped_techniques == [] + + def test_skips_are_sorted_and_deduplicated(self): + result = _result( + techniques=["zeta", "alpha", "alpha"], + display_groups={"mid::ds": "mid"}, + ) + + summary = _service()._build_response_from_db(scenario_result=result) + + assert summary.skipped_techniques == ["alpha", "zeta"] diff --git a/tests/unit/cli/test_output.py b/tests/unit/cli/test_output.py index 84c18fa05b..04bbb38474 100644 --- a/tests/unit/cli/test_output.py +++ b/tests/unit/cli/test_output.py @@ -541,6 +541,24 @@ def test_print_scenario_run_summary_completed(capsys): assert "Completed:" not in captured.out +def test_print_scenario_run_summary_lists_skipped_techniques(capsys): + run = _make_run( + scenario_name="partial", + scenario_result_id="id", + status=ScenarioRunState.COMPLETED, + total_attacks=2, + completed_attacks=2, + objective_achieved_rate=50, + techniques_used=["s1"], + skipped_techniques=["ghost_tech"], + ) + _output.print_scenario_run_summary(run=run) + captured = capsys.readouterr() + assert "Skipped:" in captured.out + assert "ghost_tech" in captured.out + assert "no registered factory" in captured.out + + def test_print_scenario_run_summary_with_error(capsys): run = _make_run( scenario_name="failing", diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index 4c60be3c6b..e3e9cef5fd 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -27,6 +27,7 @@ from pyrit.scenario.core.matrix_atomic_attack_builder import ( MatrixAtomicAttackBuilder, MatrixCombo, + TechniqueResolutionError, build_baseline_atomic_attack, build_matrix_atomic_attacks, resolve_technique_factories, @@ -432,6 +433,28 @@ def test_no_warning_when_all_techniques_resolve(self, caplog): resolve_technique_factories(context=context) assert not [record for record in caplog.records if record.levelno == logging.WARNING] + def test_raises_when_all_selected_techniques_missing(self): + """A nonempty selection resolving to nothing must fail loudly, not run baseline-only.""" + factories = {"alpha": _mock_factory(name="alpha")} + context = _context(techniques=[_technique("missing_a"), _technique("missing_b")]) + with _patch_registry(factories), pytest.raises(TechniqueResolutionError, match="missing_a"): + resolve_technique_factories(context=context) + + def test_empty_selection_resolves_without_error(self): + context = _context(techniques=[]) + with _patch_registry({}): + assert resolve_technique_factories(context=context) == {} + + def test_partial_miss_still_warns_and_continues(self, caplog): + factories = {"alpha": _mock_factory(name="alpha")} + context = _context(techniques=[_technique("alpha"), _technique("missing")]) + with ( + _patch_registry(factories), + caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), + ): + resolved = resolve_technique_factories(context=context) + assert list(resolved.keys()) == ["alpha"] + def test_warning_lists_each_missing_technique_once_in_selection_order(self, caplog): factories = {"alpha": _mock_factory(name="alpha")} context = _context( From a1600c20da6c9b8357d67d8b40eae7320bfad297 Mon Sep 17 00:00:00 2001 From: fei <204683769+feiiiiii5@users.noreply.github.com> Date: Tue, 25 Aug 2026 11:34:11 +0800 Subject: [PATCH 3/6] fix(scenario): persist authoritative skipped-technique names from resolution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round two on the skipped-techniques summary: - resolve_technique_factories() now returns a TechniqueResolution (resolved factories + skipped names) instead of a bare dict; jailbreak/adversarial record resolution.skipped and Scenario persists it into ScenarioResult.metadata["skipped_techniques"] at result creation. The run summary reads that authoritative record — display groups are presentation data (grouped by dataset/target/template depending on scenario) and cannot reveal which factories resolved. - tests/unit/backend/test_scenario_run_service.py's shared fixture now builds a real ScenarioResult (model_construct) carrying scenario_identifier and metadata, fixing the 28 spec-mock failures; added a custom-display-group regression proving labels don't affect reported skips. Signed-off-by: fei <204683769+feiiiiii5@users.noreply.github.com> --- .../backend/services/scenario_run_service.py | 14 ++---- .../core/matrix_atomic_attack_builder.py | 29 +++++++++--- pyrit/scenario/core/scenario.py | 3 ++ pyrit/scenario/scenarios/airt/jailbreak.py | 5 +- .../scenarios/benchmark/adversarial.py | 4 +- .../test_scenario_run_service_summary.py | 47 ++++++++++++------- .../unit/backend/test_scenario_run_service.py | 45 +++++++++++------- .../core/test_matrix_atomic_attack_builder.py | 15 ++++-- 8 files changed, 105 insertions(+), 57 deletions(-) diff --git a/pyrit/backend/services/scenario_run_service.py b/pyrit/backend/services/scenario_run_service.py index 6896eada19..dc7a4aa9d5 100644 --- a/pyrit/backend/services/scenario_run_service.py +++ b/pyrit/backend/services/scenario_run_service.py @@ -599,16 +599,12 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari completed_attacks = total_attacks techniques_used = scenario_result.get_techniques_used() - # Techniques the user selected but that never produced an attack cell were - # skipped during resolution (no registered factory for them). Compare against - # the built display groups rather than executed results so in-progress runs - # don't report not-yet-run techniques as skipped. - selected = set(scenario_result.scenario_identifier.techniques or []) - built_labels = set(scenario_result.display_group_map.values()) + # Authoritative skip record persisted at resolution time by the scenario + # (see resolve_technique_factories). Display groups are presentation data — + # they group by dataset/target/template depending on the scenario, so they + # cannot be used to derive which factories resolved. skipped_techniques = sorted( - technique - for technique in selected - if technique not in built_labels and not any(technique in label for label in built_labels) + set(scenario_result.metadata.get("skipped_techniques", []) or []) ) # Surface per-attack errors and retry pressure regardless of overall run status: diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 5224f409b6..9179049759 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -40,6 +40,23 @@ class TechniqueResolutionError(ValueError): """ +@dataclass(frozen=True) +class TechniqueResolution: + """ + Outcome of resolving a scenario's selected techniques to factories. + + Attributes: + resolved (dict[str, AttackTechniqueFactory]): Factories keyed by technique + name, ordered by the selection. + skipped (list[str]): Selected techniques with no registered factory, in + selection order. This is the authoritative record for surfacing skips — + display groups are presentation data and cannot be used to derive it. + """ + + resolved: dict[str, AttackTechniqueFactory] + skipped: list[str] + + if TYPE_CHECKING: from collections.abc import Callable, Mapping, Sequence @@ -147,7 +164,7 @@ def resolve_technique_factories( *, context: ScenarioContext, extra_factories: dict[str, AttackTechniqueFactory] | None = None, -) -> dict[str, AttackTechniqueFactory]: +) -> TechniqueResolution: """ Resolve a run's selected techniques to their registered ``AttackTechniqueFactory`` instances. @@ -163,9 +180,9 @@ def resolve_technique_factories( name. Returns: - dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by - the selected techniques. Techniques with no registered factory are skipped with a - warning naming them, so the caller can proceed with whatever techniques exist. + TechniqueResolution: ``resolved`` maps technique name to factory ordered by the + selection; ``skipped`` lists selected techniques with no registered factory in + selection order (also emitted as a warning so callers proceed knowingly). Raises: TechniqueResolutionError: If the selection is nonempty but no technique resolves, @@ -204,7 +221,7 @@ def resolve_technique_factories( "extra_factories) so the run has at least one technique to execute." ) - return resolved + return TechniqueResolution(resolved=resolved, skipped=missing) def build_matrix_atomic_attacks( @@ -254,7 +271,7 @@ def build_matrix_atomic_attacks( memory_labels=context.memory_labels, ) return builder.build( - technique_factories=resolve_technique_factories(context=context, extra_factories=extra_factories), + technique_factories=resolve_technique_factories(context=context, extra_factories=extra_factories).resolved, dataset_groups=context.seed_groups_by_dataset, display_group_fn=display_group_fn, technique_converters=technique_converters, diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index b607c1dee0..fe5cc990fa 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -687,6 +687,9 @@ def _build_initial_scenario_metadata(self) -> dict[str, Any]: dict[str, Any]: Metadata payload for the new ScenarioResult. """ metadata: dict[str, Any] = {} + # Authoritative record of selected techniques that had no registered factory, + # captured during attack construction (see resolve_technique_factories). + metadata["skipped_techniques"] = sorted(set(getattr(self, "_skipped_techniques", []) or [])) if getattr(self._dataset_config, "max_dataset_size", None) is None: return metadata hashes: list[str] = [] diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index 710c7eba7c..f4bc3603a2 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -313,7 +313,10 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list self._resolved_jailbreaks = self._resolve_templates() num_attempts = self.params.get("num_jailbreak_attempts", 1) - technique_factories = resolve_technique_factories(context=context, extra_factories=_extra_default_factories()) + resolution = resolve_technique_factories(context=context, extra_factories=_extra_default_factories()) + technique_factories = resolution.resolved + # Authoritative skip record for the run summary (display groups are presentation data). + self._skipped_techniques = resolution.skipped # ``jailbreak_system_prompt`` is delivered separately (native system prompt, no converter); # every other technique goes through the inline converter path. diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 006e695c75..72bb3f9b21 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -222,7 +222,9 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list ) resolved_targets = self._resolve_adversarial_targets(target_names=target_names) - technique_factories = resolve_technique_factories(context=context) + resolution = resolve_technique_factories(context=context) + technique_factories = resolution.resolved + self._skipped_techniques = resolution.skipped builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, diff --git a/tests/unit/backend/services/test_scenario_run_service_summary.py b/tests/unit/backend/services/test_scenario_run_service_summary.py index ba56157493..eca63ce8ce 100644 --- a/tests/unit/backend/services/test_scenario_run_service_summary.py +++ b/tests/unit/backend/services/test_scenario_run_service_summary.py @@ -21,48 +21,59 @@ def _service() -> ScenarioRunService: return service -def _result(*, techniques, display_groups) -> ScenarioResult: - return ScenarioResult( - scenario_identifier=ScenarioIdentifier(name="scenario", techniques=techniques), +def _result(*, techniques, skipped, display_group_map=None) -> ScenarioResult: + return ScenarioResult.model_construct( + scenario_identifier=ScenarioIdentifier(class_name="scenario", techniques=techniques), attack_results={}, - display_group_map=display_groups, + display_group_map=display_group_map or {}, + metadata={"skipped_techniques": sorted(skipped)}, ) @pytest.mark.usefixtures("patch_central_database") class TestBuildResponseSkippedTechniques: - """Selected techniques with no built attack cell surface as ``skipped_techniques``.""" + """``skipped_techniques`` comes from the resolution-time record the scenario persists.""" - def test_selected_without_built_label_is_reported_skipped(self): - result = _result(techniques=["alpha", "ghost"], display_groups={"alpha::ds": "alpha"}) + def test_persisted_skips_are_reported_verbatim(self): + result = _result(techniques=["alpha", "ghost"], skipped=["ghost"], display_group_map={"cell": "alpha"}) summary = _service()._build_response_from_db(scenario_result=result) assert summary.skipped_techniques == ["ghost"] - assert summary.techniques_used is not None - def test_decorated_display_label_still_counts_as_built(self): - # Custom ``display_group_fn`` may decorate technique names; a label that - # contains the technique name must not be reported as skipped. - result = _result(techniques=["alpha"], display_groups={"cell-1": "alpha (hard mode)"}) + def test_display_group_labels_do_not_affect_reporting(self): + """Custom display-group functions rename cells; skips must still surface verbatim.""" + result = _result( + techniques=["alpha", "ghost"], + skipped=["ghost"], + display_group_map={"cell-a": "alpha (hard mode)", "cell-b": "ghost (hard mode)"}, + ) summary = _service()._build_response_from_db(scenario_result=result) - assert summary.skipped_techniques == [] + assert summary.skipped_techniques == ["ghost"] def test_no_selection_reports_no_skips(self): - result = _result(techniques=None, display_groups={}) + result = _result(techniques=None, skipped=[]) summary = _service()._build_response_from_db(scenario_result=result) assert summary.skipped_techniques == [] def test_skips_are_sorted_and_deduplicated(self): - result = _result( - techniques=["zeta", "alpha", "alpha"], - display_groups={"mid::ds": "mid"}, - ) + result = _result(techniques=["zeta", "alpha", "alpha"], skipped=["zeta", "alpha", "alpha"]) summary = _service()._build_response_from_db(scenario_result=result) assert summary.skipped_techniques == ["alpha", "zeta"] + + def test_legacy_results_without_metadata_report_no_skips(self): + result = ScenarioResult.model_construct( + scenario_identifier=ScenarioIdentifier(class_name="scenario", techniques=["x"]), + attack_results={}, + metadata={}, + ) + + summary = _service()._build_response_from_db(scenario_result=result) + + assert summary.skipped_techniques == [] diff --git a/tests/unit/backend/test_scenario_run_service.py b/tests/unit/backend/test_scenario_run_service.py index 8ce480c62f..df0b6ac8b1 100644 --- a/tests/unit/backend/test_scenario_run_service.py +++ b/tests/unit/backend/test_scenario_run_service.py @@ -19,6 +19,7 @@ from pyrit.converter import Converter from pyrit.models import AttackOutcome, ScenarioResult, ScenarioRunState from pyrit.models.catalog.scenario import RunScenarioRequest +from pyrit.models.identifiers.scenario_identifier import ScenarioIdentifier from pyrit.scenario.core import DatasetAttackConfiguration, DatasetConfiguration from pyrit.scenario.core.scenario_technique import ScenarioTechnique from unit.mocks import make_scenario_result @@ -88,23 +89,33 @@ def _make_db_scenario_result( run_state: ScenarioRunState = ScenarioRunState.IN_PROGRESS, attack_results: dict | None = None, ) -> MagicMock: - """Create a mock ScenarioResult as returned by CentralMemory.""" - sr = MagicMock(spec=ScenarioResult) - sr.id = result_id - sr.scenario_name = scenario_name - sr.scenario_version = 1 - sr.scenario_run_state = run_state - sr.get_techniques_used.return_value = [] - sr.attack_results = attack_results or {} - sr.number_tries = 1 - sr.creation_time = datetime(2025, 1, 1, tzinfo=timezone.utc) - sr.completion_time = datetime(2025, 1, 1, 0, 5, tzinfo=timezone.utc) - sr.labels = {} - sr.objective_achieved_rate.return_value = 0 - sr.get_display_groups.return_value = {} - sr.display_group_map = {} - sr.error_message = None - sr.error_type = None + """Create a real ``ScenarioResult`` as CentralMemory would return it. + + A real object (not a spec'd mock) because the summary builder reads + ``scenario_identifier`` / ``metadata``, which a spec'd mock does not define. + """ + sr = ScenarioResult.model_construct( + id=result_id, + scenario_identifier=ScenarioIdentifier(class_name=scenario_name, techniques=[]), + attack_results=attack_results or {}, + scenario_run_state=run_state, + number_tries=1, + creation_time=datetime(2025, 1, 1, tzinfo=timezone.utc), + completion_time=datetime(2025, 1, 1, 0, 5, tzinfo=timezone.utc), + labels={}, + display_group_map={}, + error_message=None, + error_type=None, + metadata={"skipped_techniques": []}, + ) + + # Individual tests configure these; real methods would need live data. + from unittest.mock import MagicMock as MockFactory + + object.__setattr__(sr, "get_techniques_used", MockFactory(return_value=[])) + object.__setattr__(sr, "objective_achieved_rate", MockFactory(return_value=0)) + object.__setattr__(sr, "get_display_groups", MockFactory(return_value={})) + return sr diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index e3e9cef5fd..d2756e3bee 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -400,14 +400,16 @@ def test_keeps_only_selected_in_order(self): } context = _context(techniques=[_technique("beta"), _technique("alpha")]) with _patch_registry(factories): - resolved = resolve_technique_factories(context=context) + resolved = resolve_technique_factories(context=context).resolved + resolved = resolve_technique_factories(context=context).resolved assert list(resolved.keys()) == ["beta", "alpha"] def test_drops_techniques_without_factory(self): factories = {"alpha": _mock_factory(name="alpha")} context = _context(techniques=[_technique("alpha"), _technique("missing")]) with _patch_registry(factories): - resolved = resolve_technique_factories(context=context) + resolved = resolve_technique_factories(context=context).resolved + resolved = resolve_technique_factories(context=context).resolved assert list(resolved.keys()) == ["alpha"] def test_warns_when_dropping_techniques_without_factory(self, caplog): @@ -443,7 +445,7 @@ def test_raises_when_all_selected_techniques_missing(self): def test_empty_selection_resolves_without_error(self): context = _context(techniques=[]) with _patch_registry({}): - assert resolve_technique_factories(context=context) == {} + assert resolve_technique_factories(context=context).resolved == {} def test_partial_miss_still_warns_and_continues(self, caplog): factories = {"alpha": _mock_factory(name="alpha")} @@ -452,7 +454,8 @@ def test_partial_miss_still_warns_and_continues(self, caplog): _patch_registry(factories), caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), ): - resolved = resolve_technique_factories(context=context) + resolved = resolve_technique_factories(context=context).resolved + resolved = resolve_technique_factories(context=context).resolved assert list(resolved.keys()) == ["alpha"] def test_warning_lists_each_missing_technique_once_in_selection_order(self, caplog): @@ -481,11 +484,13 @@ def test_extra_factories_merged_and_override_registry(self): local_only = _mock_factory(name="local") context = _context(techniques=[_technique("alpha"), _technique("local")]) with _patch_registry(registry_factories): - resolved = resolve_technique_factories( + resolution = resolve_technique_factories( context=context, extra_factories={"alpha": local_alpha, "local": local_only}, ) + resolved = resolution.resolved assert list(resolved.keys()) == ["alpha", "local"] + assert resolution.skipped == [] assert resolved["alpha"] is local_alpha # extra overrides the registry factory of the same name assert resolved["local"] is local_only # local-only factory is selectable without global registration From 6b6755178492bbd821c2dba206fa93a5aec04413 Mon Sep 17 00:00:00 2001 From: fei <204683769+feiiiiii5@users.noreply.github.com> Date: Tue, 25 Aug 2026 13:41:45 +0800 Subject: [PATCH 4/6] fix(scenario): propagate skipped techniques through build_matrix_atomic_attacks and multilingual Review round two follow-ups: build_matrix_atomic_attacks now returns (attacks, skipped) so Cyber/Leakage/RapidResponse persist the authoritative skip record before ScenarioResult creation, and Multilingual consumes resolution.resolved while recording resolution.skipped (its dict-style use of the old return type caused 9 TypeErrors after the TechniqueResolution change). Jailbreak's mocked resolver updated to the new contract. Signed-off-by: fei <204683769+feiiiiii5@users.noreply.github.com> --- .../core/matrix_atomic_attack_builder.py | 12 +++-- pyrit/scenario/scenarios/airt/cyber.py | 4 +- pyrit/scenario/scenarios/airt/leakage.py | 4 +- pyrit/scenario/scenarios/airt/multilingual.py | 4 +- .../scenario/scenarios/airt/rapid_response.py | 4 +- tests/unit/scenario/airt/test_jailbreak.py | 6 ++- .../core/test_matrix_atomic_attack_builder.py | 54 +++++++++++-------- 7 files changed, 57 insertions(+), 31 deletions(-) diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 9179049759..1b970d5853 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -262,21 +262,25 @@ def build_matrix_atomic_attacks( can offer techniques without registering them globally. Returns: - list[AtomicAttack]: The generated atomic attacks, baseline first when - ``context.include_baseline`` is set. + tuple[list[AtomicAttack], list[str]]: The generated atomic attacks (baseline + first when ``context.include_baseline`` is set) and the names of selected + techniques that had no registered factory, so callers can persist the + authoritative skip record before creating the ``ScenarioResult``. """ builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, objective_scorer=objective_scorer, memory_labels=context.memory_labels, ) - return builder.build( - technique_factories=resolve_technique_factories(context=context, extra_factories=extra_factories).resolved, + resolution = resolve_technique_factories(context=context, extra_factories=extra_factories) + attacks = builder.build( + technique_factories=resolution.resolved, dataset_groups=context.seed_groups_by_dataset, display_group_fn=display_group_fn, technique_converters=technique_converters, include_baseline=context.include_baseline, ) + return attacks, resolution.skipped class MatrixAtomicAttackBuilder: diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index 2f622c7b41..1c5af7c6af 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -123,8 +123,10 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - return build_matrix_atomic_attacks( + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, technique_converters=self._technique_converters, ) + self._skipped_techniques = skipped + return attacks diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 264035f0ae..5113b3c5d3 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -140,9 +140,11 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - return build_matrix_atomic_attacks( + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, technique_converters=self._technique_converters, extra_factories={factory.name: factory for factory in _leakage_factories()}, ) + self._skipped_techniques = skipped + return attacks diff --git a/pyrit/scenario/scenarios/airt/multilingual.py b/pyrit/scenario/scenarios/airt/multilingual.py index bc70be31e6..a1c48afb0d 100644 --- a/pyrit/scenario/scenarios/airt/multilingual.py +++ b/pyrit/scenario/scenarios/airt/multilingual.py @@ -308,10 +308,12 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list self._resolved_languages = self._resolve_languages() adversarial_chat = self._adversarial_chat or get_default_adversarial_target() strategies = set(self.params.get("translation_strategies") or [_TRANSLATION, _RANDOM_TRANSLATION]) - technique_factories = resolve_technique_factories( + resolution = resolve_technique_factories( context=context, extra_factories=_extra_default_factories(), ) + technique_factories = resolution.resolved + self._skipped_techniques = resolution.skipped builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, objective_scorer=self._objective_scorer, diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 4fd292bbe8..d11ead6b2f 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -124,9 +124,11 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - return build_matrix_atomic_attacks( + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, display_group_fn=lambda combo: combo.dataset_name, technique_converters=self._technique_converters, ) + self._skipped_techniques = skipped + return attacks diff --git a/tests/unit/scenario/airt/test_jailbreak.py b/tests/unit/scenario/airt/test_jailbreak.py index b055e087aa..93aa1ce2de 100644 --- a/tests/unit/scenario/airt/test_jailbreak.py +++ b/tests/unit/scenario/airt/test_jailbreak.py @@ -19,6 +19,7 @@ from pyrit.registry.components.scenario_registry import ScenarioRegistry from pyrit.scenario.core import BaselineAttackPolicy from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory +from pyrit.scenario.core.matrix_atomic_attack_builder import TechniqueResolution from pyrit.scenario.scenarios.airt.jailbreak import ( _DEFAULT_NUM_JAILBREAKS, _DEFAULT_TECHNIQUES, @@ -392,7 +393,10 @@ async def test_missing_runtime_factory_is_rejected( self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups ): with _patch_seed_groups(mock_memory_seed_groups): - with patch("pyrit.scenario.scenarios.airt.jailbreak.resolve_technique_factories", return_value={}): + with patch( + "pyrit.scenario.scenarios.airt.jailbreak.resolve_technique_factories", + return_value=TechniqueResolution(resolved={}, skipped=[]), + ): scenario = Jailbreak(objective_scorer=mock_objective_scorer) scenario.set_params_from_args(args=_default_args(mock_objective_target, jailbreak_names=["aim.yaml"])) with pytest.raises(ValueError, match="no longer available.*prompt_sending"): diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index d2756e3bee..b52331d271 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -495,6 +495,7 @@ def test_extra_factories_merged_and_override_registry(self): assert resolved["local"] is local_only # local-only factory is selectable without global registration +@pytest.mark.usefixtures("patch_central_database") @pytest.mark.usefixtures("patch_central_database") class TestBuildMatrixAtomicAttacks: """``build_matrix_atomic_attacks`` wires the context into the builder in one call.""" @@ -502,46 +503,53 @@ class TestBuildMatrixAtomicAttacks: def test_builds_cross_product_grouped_by_technique(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) - assert [a.atomic_attack_name for a in result] == ["tech_ds"] - assert result[0].display_group == "tech" + attacks, skipped = build_matrix_atomic_attacks( + context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) + ) + assert [a.atomic_attack_name for a in attacks] == ["tech_ds"] + assert attacks[0].display_group == "tech" + assert skipped == [] def test_custom_display_group_fn(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - result = build_matrix_atomic_attacks( + attacks, _ = build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), display_group_fn=lambda combo: combo.dataset_name, ) - assert result[0].display_group == "ds" + assert attacks[0].display_group == "ds" def test_no_baseline_emitted_when_context_disables_it(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, include_baseline=False, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) - assert all(a.atomic_attack_name != "baseline" for a in result) + attacks, _ = build_matrix_atomic_attacks( + context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) + ) + assert all(a.atomic_attack_name != "baseline" for a in attacks) def test_baseline_emitted_when_context_enables_it(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, include_baseline=True, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) - assert result[0].atomic_attack_name == "baseline" - assert [a.atomic_attack_name for a in result] == ["baseline", "tech_ds"] + attacks, _ = build_matrix_atomic_attacks( + context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) + ) + assert attacks[0].atomic_attack_name == "baseline" + assert [a.atomic_attack_name for a in attacks] == ["baseline", "tech_ds"] def test_technique_converters_forwarded(self): from pyrit.converter import Converter @@ -553,7 +561,7 @@ def test_technique_converters_forwarded(self): factory = _mock_factory(name="tech") converter = MagicMock(spec=Converter) with _patch_registry({"tech": factory}): - build_matrix_atomic_attacks( + attacks, _ = build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), technique_converters={"tech": [converter]}, @@ -563,15 +571,17 @@ def test_technique_converters_forwarded(self): assert len(extra) == 1 def test_extra_factories_used_for_selection(self): + local_alpha = _mock_factory(name="alpha") + local_only = _mock_factory(name="local") context = _context( - techniques=[_technique("local")], - seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + techniques=[_technique("alpha"), _technique("local")], + seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, ) - # The selected technique exists only in extra_factories, not the registry. - with _patch_registry({"other": _mock_factory(name="other")}): - result = build_matrix_atomic_attacks( + with _patch_registry({}): + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), - extra_factories={"local": _mock_factory(name="local")}, + extra_factories={"alpha": local_alpha, "local": local_only}, ) - assert [a.atomic_attack_name for a in result] == ["local_ds"] + assert [a.atomic_attack_name for a in attacks] == ["alpha_ds", "local_ds"] + assert skipped == [] From 1b30e56c20c0f6cd920ddcd34a08463fa3fb28c6 Mon Sep 17 00:00:00 2001 From: fei <204683769+feiiiiii5@users.noreply.github.com> Date: Fri, 28 Aug 2026 21:50:10 +0800 Subject: [PATCH 5/6] FIX: annotate the matrix builder's real return contract and sync docs build_matrix_atomic_attacks returns (attacks, skipped technique names) but was still annotated list[AtomicAttack], so ty reported an invalid return type in the helper plus one in each Cyber/Leakage/RapidResponse override that unpacks it. The annotation now matches the code. The documented custom-scenario pattern showed the helper being returned directly from _build_atomic_attacks_async, which no longer type-checks. Updated the percent-format example, its .ipynb twin, and the scenario contributor instructions to unpack the tuple and record the skip list on self._skipped_techniques, since that is what makes a partial technique miss visible in ScenarioResult metadata and the CLI summary. Also drops a duplicated patch_central_database marker and the two call sites ruff-format wants joined. Signed-off-by: fei <204683769+feiiiiii5@users.noreply.github.com> --- .github/instructions/scenarios.instructions.md | 16 +++++++++++----- doc/code/scenarios/0_scenarios.ipynb | 12 ++++++++---- doc/code/scenarios/0_scenarios.py | 10 +++++++--- pyrit/backend/services/scenario_run_service.py | 4 +--- .../core/matrix_atomic_attack_builder.py | 2 +- .../core/test_matrix_atomic_attack_builder.py | 9 ++------- 6 files changed, 30 insertions(+), 23 deletions(-) diff --git a/.github/instructions/scenarios.instructions.md b/.github/instructions/scenarios.instructions.md index 59a18dd835..7b7a61f636 100644 --- a/.github/instructions/scenarios.instructions.md +++ b/.github/instructions/scenarios.instructions.md @@ -44,7 +44,8 @@ the constructor — no classmethod indirection required. 4. **Implement `_build_atomic_attacks_async(self, *, context)`** — this is the single abstract extension point every scenario must define (see "AtomicAttack Construction" below). - Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line. + Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns + the attacks plus the names of selected techniques that had no registered factory. ## Constructor Pattern @@ -191,7 +192,7 @@ via `build_matrix_atomic_attacks`/`MatrixAtomicAttackBuilder`, pass a `display_g callback that maps each `MatrixCombo` to a group string: ```python -build_matrix_atomic_attacks( +attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, display_group_fn=lambda combo: combo.technique_name, # default: group by technique @@ -221,23 +222,28 @@ half-initialized `self._*` state to build attacks — read everything from `cont ### Zero-boilerplate matrix scenarios Scenarios whose construction is the plain technique × dataset cross-product delegate to the -`build_matrix_atomic_attacks` helper in one line (see `Cyber`, `RapidResponse`): +`build_matrix_atomic_attacks` helper (see `Cyber`, `RapidResponse`). It returns the attacks +**and** the names of selected techniques that had no registered factory; record that second +value on `self._skipped_techniques` so the run reports the drop instead of hiding it: ```python from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: - return build_matrix_atomic_attacks( + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, technique_converters=self._technique_converters, # optional CLI converter stacks ) + self._skipped_techniques = skipped + return attacks ``` `build_matrix_atomic_attacks`: 1. Calls `resolve_technique_factories(context=context)` to map the selected techniques to their registered `AttackTechniqueFactory` instances (reads the `AttackTechniqueRegistry` singleton; - techniques with no registered factory are dropped). + techniques with no registered factory are dropped, warned about, and returned as the second + value, so a partial miss is visible in the run summary rather than silent). 2. Iterates every (technique × dataset) pair from `context.seed_groups_by_dataset`. 3. Calls `factory.create()` with the objective target, conditional scorer override, and any per-technique converters (from `--techniques :converter.`) as diff --git a/doc/code/scenarios/0_scenarios.ipynb b/doc/code/scenarios/0_scenarios.ipynb index 4cb38c0b58..f5aa48073b 100644 --- a/doc/code/scenarios/0_scenarios.ipynb +++ b/doc/code/scenarios/0_scenarios.ipynb @@ -62,7 +62,7 @@ "2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`:\n", " - `technique_class`: Your technique enum class\n", " - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point.\n", - " Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line.\n", + " Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns the attacks plus the names of any selected techniques that had no registered factory.\n", "\n", "3. **Default Dataset**: Pass `default_dataset_config=` to `super().__init__()` to specify the datasets your scenario uses out of the box.\n", " - Returns a `DatasetConfiguration` with one or more named datasets (e.g., `DatasetConfiguration(dataset_names=[\"my_dataset\"])`)\n", @@ -171,13 +171,17 @@ "\n", " # Implement the single abstract extension point. Matrix-shaped scenarios delegate\n", " # to build_matrix_atomic_attacks; pass display_group_fn to customize result grouping\n", - " # (default groups by technique; here we group by dataset instead).\n", + " # (default groups by technique; here we group by dataset instead). The helper also\n", + " # reports which selected techniques had no registered factory, so record that skip\n", + " # list -- the base Scenario persists it into ScenarioResult.metadata and the CLI summary.\n", " async def _build_atomic_attacks_async(self, *, context):\n", - " return build_matrix_atomic_attacks(\n", + " attacks, skipped = build_matrix_atomic_attacks(\n", " context=context,\n", " objective_scorer=self._objective_scorer,\n", " display_group_fn=lambda combo: combo.dataset_name,\n", - " )" + " )\n", + " self._skipped_techniques = skipped\n", + " return attacks" ] }, { diff --git a/doc/code/scenarios/0_scenarios.py b/doc/code/scenarios/0_scenarios.py index aa01af4930..65e44aabc1 100644 --- a/doc/code/scenarios/0_scenarios.py +++ b/doc/code/scenarios/0_scenarios.py @@ -64,7 +64,7 @@ # 2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`: # - `technique_class`: Your technique enum class # - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point. -# Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line. +# Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns the attacks plus the names of any selected techniques that had no registered factory. # # 3. **Default Dataset**: Pass `default_dataset_config=` to `super().__init__()` to specify the datasets your scenario uses out of the box. # - Returns a `DatasetConfiguration` with one or more named datasets (e.g., `DatasetConfiguration(dataset_names=["my_dataset"])`) @@ -148,13 +148,17 @@ def __init__( # Implement the single abstract extension point. Matrix-shaped scenarios delegate # to build_matrix_atomic_attacks; pass display_group_fn to customize result grouping - # (default groups by technique; here we group by dataset instead). + # (default groups by technique; here we group by dataset instead). The helper also + # reports which selected techniques had no registered factory, so record that skip + # list -- the base Scenario persists it into ScenarioResult.metadata and the CLI summary. async def _build_atomic_attacks_async(self, *, context): - return build_matrix_atomic_attacks( + attacks, skipped = build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, display_group_fn=lambda combo: combo.dataset_name, ) + self._skipped_techniques = skipped + return attacks # %% [markdown] diff --git a/pyrit/backend/services/scenario_run_service.py b/pyrit/backend/services/scenario_run_service.py index dc7a4aa9d5..d221429a12 100644 --- a/pyrit/backend/services/scenario_run_service.py +++ b/pyrit/backend/services/scenario_run_service.py @@ -603,9 +603,7 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari # (see resolve_technique_factories). Display groups are presentation data — # they group by dataset/target/template depending on the scenario, so they # cannot be used to derive which factories resolved. - skipped_techniques = sorted( - set(scenario_result.metadata.get("skipped_techniques", []) or []) - ) + skipped_techniques = sorted(set(scenario_result.metadata.get("skipped_techniques", []) or [])) # Surface per-attack errors and retry pressure regardless of overall run status: # a COMPLETED scenario can still hide errored objectives or rate-limit retries. diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 1b970d5853..ea1fe3dd71 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -231,7 +231,7 @@ def build_matrix_atomic_attacks( display_group_fn: Callable[[MatrixCombo], str] | None = None, technique_converters: dict[str, list[Converter]] | None = None, extra_factories: dict[str, AttackTechniqueFactory] | None = None, -) -> list[AtomicAttack]: +) -> tuple[list[AtomicAttack], list[str]]: """ Build a matrix-shaped scenario's atomic attacks from its resolved context in one call. diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index b52331d271..4f8cd9869c 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -495,7 +495,6 @@ def test_extra_factories_merged_and_override_registry(self): assert resolved["local"] is local_only # local-only factory is selectable without global registration -@pytest.mark.usefixtures("patch_central_database") @pytest.mark.usefixtures("patch_central_database") class TestBuildMatrixAtomicAttacks: """``build_matrix_atomic_attacks`` wires the context into the builder in one call.""" @@ -533,9 +532,7 @@ def test_no_baseline_emitted_when_context_disables_it(self): include_baseline=False, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - attacks, _ = build_matrix_atomic_attacks( - context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) - ) + attacks, _ = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) assert all(a.atomic_attack_name != "baseline" for a in attacks) def test_baseline_emitted_when_context_enables_it(self): @@ -545,9 +542,7 @@ def test_baseline_emitted_when_context_enables_it(self): include_baseline=True, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - attacks, _ = build_matrix_atomic_attacks( - context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) - ) + attacks, _ = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) assert attacks[0].atomic_attack_name == "baseline" assert [a.atomic_attack_name for a in attacks] == ["baseline", "tech_ds"] From 0ea576b9e413ea88abda68c04378c6b0177568d3 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Fri, 28 Aug 2026 12:36:25 -0700 Subject: [PATCH 6/6] fix(scenario): reject unregistered selected techniques Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../instructions/scenarios.instructions.md | 16 +-- doc/code/scenarios/0_scenarios.ipynb | 12 +- doc/code/scenarios/0_scenarios.py | 10 +- .../backend/services/scenario_run_service.py | 7 -- pyrit/cli/_output.py | 3 - pyrit/models/catalog/scenario.py | 4 - .../core/matrix_atomic_attack_builder.py | 78 +++--------- pyrit/scenario/core/scenario.py | 3 - pyrit/scenario/scenarios/airt/cyber.py | 4 +- pyrit/scenario/scenarios/airt/jailbreak.py | 13 +- pyrit/scenario/scenarios/airt/leakage.py | 4 +- pyrit/scenario/scenarios/airt/multilingual.py | 4 +- .../scenario/scenarios/airt/rapid_response.py | 4 +- .../scenarios/benchmark/adversarial.py | 4 +- .../test_scenario_run_service_summary.py | 79 ------------ .../unit/backend/test_scenario_run_service.py | 45 +++---- tests/unit/cli/test_output.py | 18 --- tests/unit/scenario/airt/test_jailbreak.py | 14 --- .../core/test_matrix_atomic_attack_builder.py | 113 +++++------------- 19 files changed, 85 insertions(+), 350 deletions(-) delete mode 100644 tests/unit/backend/services/test_scenario_run_service_summary.py diff --git a/.github/instructions/scenarios.instructions.md b/.github/instructions/scenarios.instructions.md index 7b7a61f636..9089b3f6ad 100644 --- a/.github/instructions/scenarios.instructions.md +++ b/.github/instructions/scenarios.instructions.md @@ -44,8 +44,7 @@ the constructor — no classmethod indirection required. 4. **Implement `_build_atomic_attacks_async(self, *, context)`** — this is the single abstract extension point every scenario must define (see "AtomicAttack Construction" below). - Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns - the attacks plus the names of selected techniques that had no registered factory. + Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line. ## Constructor Pattern @@ -192,7 +191,7 @@ via `build_matrix_atomic_attacks`/`MatrixAtomicAttackBuilder`, pass a `display_g callback that maps each `MatrixCombo` to a group string: ```python -attacks, skipped = build_matrix_atomic_attacks( +build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, display_group_fn=lambda combo: combo.technique_name, # default: group by technique @@ -222,28 +221,23 @@ half-initialized `self._*` state to build attacks — read everything from `cont ### Zero-boilerplate matrix scenarios Scenarios whose construction is the plain technique × dataset cross-product delegate to the -`build_matrix_atomic_attacks` helper (see `Cyber`, `RapidResponse`). It returns the attacks -**and** the names of selected techniques that had no registered factory; record that second -value on `self._skipped_techniques` so the run reports the drop instead of hiding it: +`build_matrix_atomic_attacks` helper in one line (see `Cyber`, `RapidResponse`): ```python from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: - attacks, skipped = build_matrix_atomic_attacks( + return build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, technique_converters=self._technique_converters, # optional CLI converter stacks ) - self._skipped_techniques = skipped - return attacks ``` `build_matrix_atomic_attacks`: 1. Calls `resolve_technique_factories(context=context)` to map the selected techniques to their registered `AttackTechniqueFactory` instances (reads the `AttackTechniqueRegistry` singleton; - techniques with no registered factory are dropped, warned about, and returned as the second - value, so a partial miss is visible in the run summary rather than silent). + raises ``TechniqueResolutionError`` if any selected technique has no registered factory). 2. Iterates every (technique × dataset) pair from `context.seed_groups_by_dataset`. 3. Calls `factory.create()` with the objective target, conditional scorer override, and any per-technique converters (from `--techniques :converter.`) as diff --git a/doc/code/scenarios/0_scenarios.ipynb b/doc/code/scenarios/0_scenarios.ipynb index f5aa48073b..4cb38c0b58 100644 --- a/doc/code/scenarios/0_scenarios.ipynb +++ b/doc/code/scenarios/0_scenarios.ipynb @@ -62,7 +62,7 @@ "2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`:\n", " - `technique_class`: Your technique enum class\n", " - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point.\n", - " Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns the attacks plus the names of any selected techniques that had no registered factory.\n", + " Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line.\n", "\n", "3. **Default Dataset**: Pass `default_dataset_config=` to `super().__init__()` to specify the datasets your scenario uses out of the box.\n", " - Returns a `DatasetConfiguration` with one or more named datasets (e.g., `DatasetConfiguration(dataset_names=[\"my_dataset\"])`)\n", @@ -171,17 +171,13 @@ "\n", " # Implement the single abstract extension point. Matrix-shaped scenarios delegate\n", " # to build_matrix_atomic_attacks; pass display_group_fn to customize result grouping\n", - " # (default groups by technique; here we group by dataset instead). The helper also\n", - " # reports which selected techniques had no registered factory, so record that skip\n", - " # list -- the base Scenario persists it into ScenarioResult.metadata and the CLI summary.\n", + " # (default groups by technique; here we group by dataset instead).\n", " async def _build_atomic_attacks_async(self, *, context):\n", - " attacks, skipped = build_matrix_atomic_attacks(\n", + " return build_matrix_atomic_attacks(\n", " context=context,\n", " objective_scorer=self._objective_scorer,\n", " display_group_fn=lambda combo: combo.dataset_name,\n", - " )\n", - " self._skipped_techniques = skipped\n", - " return attacks" + " )" ] }, { diff --git a/doc/code/scenarios/0_scenarios.py b/doc/code/scenarios/0_scenarios.py index 65e44aabc1..aa01af4930 100644 --- a/doc/code/scenarios/0_scenarios.py +++ b/doc/code/scenarios/0_scenarios.py @@ -64,7 +64,7 @@ # 2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`: # - `technique_class`: Your technique enum class # - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point. -# Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)`, which returns the attacks plus the names of any selected techniques that had no registered factory. +# Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line. # # 3. **Default Dataset**: Pass `default_dataset_config=` to `super().__init__()` to specify the datasets your scenario uses out of the box. # - Returns a `DatasetConfiguration` with one or more named datasets (e.g., `DatasetConfiguration(dataset_names=["my_dataset"])`) @@ -148,17 +148,13 @@ def __init__( # Implement the single abstract extension point. Matrix-shaped scenarios delegate # to build_matrix_atomic_attacks; pass display_group_fn to customize result grouping - # (default groups by technique; here we group by dataset instead). The helper also - # reports which selected techniques had no registered factory, so record that skip - # list -- the base Scenario persists it into ScenarioResult.metadata and the CLI summary. + # (default groups by technique; here we group by dataset instead). async def _build_atomic_attacks_async(self, *, context): - attacks, skipped = build_matrix_atomic_attacks( + return build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, display_group_fn=lambda combo: combo.dataset_name, ) - self._skipped_techniques = skipped - return attacks # %% [markdown] diff --git a/pyrit/backend/services/scenario_run_service.py b/pyrit/backend/services/scenario_run_service.py index d221429a12..7ba66d0f43 100644 --- a/pyrit/backend/services/scenario_run_service.py +++ b/pyrit/backend/services/scenario_run_service.py @@ -599,12 +599,6 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari completed_attacks = total_attacks techniques_used = scenario_result.get_techniques_used() - # Authoritative skip record persisted at resolution time by the scenario - # (see resolve_technique_factories). Display groups are presentation data — - # they group by dataset/target/template depending on the scenario, so they - # cannot be used to derive which factories resolved. - skipped_techniques = sorted(set(scenario_result.metadata.get("skipped_techniques", []) or [])) - # Surface per-attack errors and retry pressure regardless of overall run status: # a COMPLETED scenario can still hide errored objectives or rate-limit retries. failed_attacks: list[AttackErrorSummary] = [] @@ -647,7 +641,6 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari error=error, error_type=error_type, techniques_used=techniques_used, - skipped_techniques=skipped_techniques, total_attacks=total_attacks, completed_attacks=completed_attacks, objective_achieved_rate=scenario_result.objective_achieved_rate(), diff --git a/pyrit/cli/_output.py b/pyrit/cli/_output.py index 1b66e793f5..5b04d2611f 100644 --- a/pyrit/cli/_output.py +++ b/pyrit/cli/_output.py @@ -387,9 +387,6 @@ def print_scenario_run_summary(*, run: ScenarioRunSummary) -> None: if run.techniques_used: print(f" Techniques: {', '.join(run.techniques_used)}") - if run.skipped_techniques: - print(f" Skipped: {', '.join(run.skipped_techniques)} (no registered factory)") - if run.failed_attacks: print(f"\n Failed Attacks ({len(run.failed_attacks)}):") for failed in run.failed_attacks: diff --git a/pyrit/models/catalog/scenario.py b/pyrit/models/catalog/scenario.py index 1218e7d061..488ccf8c78 100644 --- a/pyrit/models/catalog/scenario.py +++ b/pyrit/models/catalog/scenario.py @@ -148,10 +148,6 @@ class ScenarioRunSummary(BaseModel): error: str | None = Field(None, description="Error message if status is FAILED") error_type: str | None = Field(None, description="Exception class name if status is FAILED") techniques_used: list[str] = Field(default_factory=list, description="Technique names that were executed") - skipped_techniques: list[str] = Field( - default_factory=list, - description="Selected techniques that were skipped because no factory was registered for them", - ) total_attacks: int = Field(0, ge=0, description="Total number of attack results persisted for this run") completed_attacks: int = Field(0, ge=0, description="Number of attacks that reached a terminal outcome") objective_achieved_rate: int = Field(0, ge=0, le=100, description="Success rate as percentage (0-100)") diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index ea1fe3dd71..1c67745cfb 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -32,31 +32,13 @@ class TechniqueResolutionError(ValueError): """ - Raised when a scenario selects techniques but none of them resolve to a factory. + Raised when a selected scenario technique has no registered factory. Subclasses ``ValueError`` so existing ``except ValueError`` handlers keep working, - mirroring ``DatasetConstraintError``. Partial misses (some techniques resolve) - only warn, so a run still proceeds with the techniques that do exist. + mirroring ``DatasetConstraintError``. """ -@dataclass(frozen=True) -class TechniqueResolution: - """ - Outcome of resolving a scenario's selected techniques to factories. - - Attributes: - resolved (dict[str, AttackTechniqueFactory]): Factories keyed by technique - name, ordered by the selection. - skipped (list[str]): Selected techniques with no registered factory, in - selection order. This is the authoritative record for surfacing skips — - display groups are presentation data and cannot be used to derive it. - """ - - resolved: dict[str, AttackTechniqueFactory] - skipped: list[str] - - if TYPE_CHECKING: from collections.abc import Callable, Mapping, Sequence @@ -164,13 +146,13 @@ def resolve_technique_factories( *, context: ScenarioContext, extra_factories: dict[str, AttackTechniqueFactory] | None = None, -) -> TechniqueResolution: +) -> dict[str, AttackTechniqueFactory]: """ Resolve a run's selected techniques to their registered ``AttackTechniqueFactory`` instances. Reads the ``AttackTechniqueRegistry`` singleton and keeps only the factories whose name - matches a selected technique, preserving selection order. Techniques with no registered - factory are silently dropped so the caller can proceed with whatever techniques exist. + matches a selected technique, preserving selection order. Raises if any selected + technique has no registered factory so the run cannot silently omit requested work. Args: context (ScenarioContext): The resolved runtime inputs for this run. @@ -180,13 +162,11 @@ def resolve_technique_factories( name. Returns: - TechniqueResolution: ``resolved`` maps technique name to factory ordered by the - selection; ``skipped`` lists selected techniques with no registered factory in - selection order (also emitted as a warning so callers proceed knowingly). + dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by + the selected techniques. Raises: - TechniqueResolutionError: If the selection is nonempty but no technique resolves, - since running only the baseline would silently defeat the selection. + TechniqueResolutionError: If any selected technique has no registered factory. """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry @@ -194,34 +174,16 @@ def resolve_technique_factories( if extra_factories: all_factories.update(extra_factories) - resolved: dict[str, AttackTechniqueFactory] = {} - missing: list[str] = [] - seen_missing: set[str] = set() - for technique in context.scenario_techniques: - if technique.value in all_factories: - resolved[technique.value] = all_factories[technique.value] - elif technique.value not in seen_missing: - missing.append(technique.value) - seen_missing.add(technique.value) + missing = list(dict.fromkeys(t.value for t in context.scenario_techniques if t.value not in all_factories)) if missing: - logger.warning( - "Skipping %d selected attack technique(s) with no registered factory: %s. " - "Register the technique(s) (or pass them via extra_factories) to include them in the run.", - len(missing), - ", ".join(missing), - ) - - if context.scenario_techniques and not resolved: - # A nonempty selection that resolves to nothing would otherwise run the - # baseline only while reporting success — a silently empty evaluation. raise TechniqueResolutionError( - f"All {len(context.scenario_techniques)} selected attack technique(s) have no registered " - f"factory: {', '.join(missing)}. Register the technique(s) (or pass them via " - "extra_factories) so the run has at least one technique to execute." + "The following selected attack techniques have no registered factory: " + f"{', '.join(missing)}. Register the techniques (or pass them via " + "extra_factories) before starting the run." ) - return TechniqueResolution(resolved=resolved, skipped=missing) + return {technique.value: all_factories[technique.value] for technique in context.scenario_techniques} def build_matrix_atomic_attacks( @@ -231,7 +193,7 @@ def build_matrix_atomic_attacks( display_group_fn: Callable[[MatrixCombo], str] | None = None, technique_converters: dict[str, list[Converter]] | None = None, extra_factories: dict[str, AttackTechniqueFactory] | None = None, -) -> tuple[list[AtomicAttack], list[str]]: +) -> list[AtomicAttack]: """ Build a matrix-shaped scenario's atomic attacks from its resolved context in one call. @@ -262,25 +224,21 @@ def build_matrix_atomic_attacks( can offer techniques without registering them globally. Returns: - tuple[list[AtomicAttack], list[str]]: The generated atomic attacks (baseline - first when ``context.include_baseline`` is set) and the names of selected - techniques that had no registered factory, so callers can persist the - authoritative skip record before creating the ``ScenarioResult``. + list[AtomicAttack]: The generated atomic attacks, baseline first when + ``context.include_baseline`` is set. """ builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, objective_scorer=objective_scorer, memory_labels=context.memory_labels, ) - resolution = resolve_technique_factories(context=context, extra_factories=extra_factories) - attacks = builder.build( - technique_factories=resolution.resolved, + return builder.build( + technique_factories=resolve_technique_factories(context=context, extra_factories=extra_factories), dataset_groups=context.seed_groups_by_dataset, display_group_fn=display_group_fn, technique_converters=technique_converters, include_baseline=context.include_baseline, ) - return attacks, resolution.skipped class MatrixAtomicAttackBuilder: diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index de2e96b440..9385c8c97b 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -695,9 +695,6 @@ def _build_initial_scenario_metadata(self) -> dict[str, Any]: dict[str, Any]: Metadata payload for the new ScenarioResult. """ metadata: dict[str, Any] = {} - # Authoritative record of selected techniques that had no registered factory, - # captured during attack construction (see resolve_technique_factories). - metadata["skipped_techniques"] = sorted(set(getattr(self, "_skipped_techniques", []) or [])) if getattr(self._dataset_config, "max_dataset_size", None) is None: return metadata hashes: list[str] = [] diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index 1c5af7c6af..2f622c7b41 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -123,10 +123,8 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - attacks, skipped = build_matrix_atomic_attacks( + return build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, technique_converters=self._technique_converters, ) - self._skipped_techniques = skipped - return attacks diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index cb028ec341..2439ca1b80 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -326,18 +326,7 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list self._resolved_jailbreaks = self._resolve_templates() num_attempts = self.params.get("num_jailbreak_attempts", 1) - resolution = resolve_technique_factories(context=context, extra_factories=_extra_default_factories()) - technique_factories = resolution.resolved - # Authoritative skip record for the run summary (display groups are presentation data). - self._skipped_techniques = resolution.skipped - - selected_names = {technique.value for technique in context.scenario_techniques} - missing = selected_names - set(technique_factories) - if missing: - raise ValueError( - "Jailbreak selected techniques that are no longer available: " - f"{sorted(missing)}. Refresh the plan and select a supported delivery method." - ) + technique_factories = resolve_technique_factories(context=context, extra_factories=_extra_default_factories()) prompt_sending_factory = technique_factories.get(_PROMPT_SENDING) system_selected = _JAILBREAK_SYSTEM_PROMPT in technique_factories diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 5113b3c5d3..264035f0ae 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -140,11 +140,9 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - attacks, skipped = build_matrix_atomic_attacks( + return build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, technique_converters=self._technique_converters, extra_factories={factory.name: factory for factory in _leakage_factories()}, ) - self._skipped_techniques = skipped - return attacks diff --git a/pyrit/scenario/scenarios/airt/multilingual.py b/pyrit/scenario/scenarios/airt/multilingual.py index a1c48afb0d..bc70be31e6 100644 --- a/pyrit/scenario/scenarios/airt/multilingual.py +++ b/pyrit/scenario/scenarios/airt/multilingual.py @@ -308,12 +308,10 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list self._resolved_languages = self._resolve_languages() adversarial_chat = self._adversarial_chat or get_default_adversarial_target() strategies = set(self.params.get("translation_strategies") or [_TRANSLATION, _RANDOM_TRANSLATION]) - resolution = resolve_technique_factories( + technique_factories = resolve_technique_factories( context=context, extra_factories=_extra_default_factories(), ) - technique_factories = resolution.resolved - self._skipped_techniques = resolution.skipped builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, objective_scorer=self._objective_scorer, diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index d11ead6b2f..4fd292bbe8 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -124,11 +124,9 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - attacks, skipped = build_matrix_atomic_attacks( + return build_matrix_atomic_attacks( context=context, objective_scorer=self._objective_scorer, display_group_fn=lambda combo: combo.dataset_name, technique_converters=self._technique_converters, ) - self._skipped_techniques = skipped - return attacks diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 48da9b62be..9270187920 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -226,9 +226,7 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list ) resolved_targets = self._resolve_adversarial_targets(target_names=target_names) - resolution = resolve_technique_factories(context=context) - technique_factories = resolution.resolved - self._skipped_techniques = resolution.skipped + technique_factories = resolve_technique_factories(context=context) builder = MatrixAtomicAttackBuilder( objective_target=context.objective_target, diff --git a/tests/unit/backend/services/test_scenario_run_service_summary.py b/tests/unit/backend/services/test_scenario_run_service_summary.py deleted file mode 100644 index eca63ce8ce..0000000000 --- a/tests/unit/backend/services/test_scenario_run_service_summary.py +++ /dev/null @@ -1,79 +0,0 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT license. - -"""Tests for skipped-technique diagnostics in ``ScenarioRunService`` run summaries.""" - -import pytest - -from pyrit.backend.services.scenario_run_service import ScenarioRunService -from pyrit.models.identifiers.scenario_identifier import ScenarioIdentifier -from pyrit.models.results.scenario_result import ScenarioResult - - -def _service() -> ScenarioRunService: - from unittest.mock import MagicMock - - service = object.__new__(ScenarioRunService) - service._active_tasks = {} - # The error fallback path queries persisted error results; none exist here. - service._memory = MagicMock() - service._memory.get_attack_results.return_value = [] - return service - - -def _result(*, techniques, skipped, display_group_map=None) -> ScenarioResult: - return ScenarioResult.model_construct( - scenario_identifier=ScenarioIdentifier(class_name="scenario", techniques=techniques), - attack_results={}, - display_group_map=display_group_map or {}, - metadata={"skipped_techniques": sorted(skipped)}, - ) - - -@pytest.mark.usefixtures("patch_central_database") -class TestBuildResponseSkippedTechniques: - """``skipped_techniques`` comes from the resolution-time record the scenario persists.""" - - def test_persisted_skips_are_reported_verbatim(self): - result = _result(techniques=["alpha", "ghost"], skipped=["ghost"], display_group_map={"cell": "alpha"}) - - summary = _service()._build_response_from_db(scenario_result=result) - - assert summary.skipped_techniques == ["ghost"] - - def test_display_group_labels_do_not_affect_reporting(self): - """Custom display-group functions rename cells; skips must still surface verbatim.""" - result = _result( - techniques=["alpha", "ghost"], - skipped=["ghost"], - display_group_map={"cell-a": "alpha (hard mode)", "cell-b": "ghost (hard mode)"}, - ) - - summary = _service()._build_response_from_db(scenario_result=result) - - assert summary.skipped_techniques == ["ghost"] - - def test_no_selection_reports_no_skips(self): - result = _result(techniques=None, skipped=[]) - - summary = _service()._build_response_from_db(scenario_result=result) - - assert summary.skipped_techniques == [] - - def test_skips_are_sorted_and_deduplicated(self): - result = _result(techniques=["zeta", "alpha", "alpha"], skipped=["zeta", "alpha", "alpha"]) - - summary = _service()._build_response_from_db(scenario_result=result) - - assert summary.skipped_techniques == ["alpha", "zeta"] - - def test_legacy_results_without_metadata_report_no_skips(self): - result = ScenarioResult.model_construct( - scenario_identifier=ScenarioIdentifier(class_name="scenario", techniques=["x"]), - attack_results={}, - metadata={}, - ) - - summary = _service()._build_response_from_db(scenario_result=result) - - assert summary.skipped_techniques == [] diff --git a/tests/unit/backend/test_scenario_run_service.py b/tests/unit/backend/test_scenario_run_service.py index df0b6ac8b1..8ce480c62f 100644 --- a/tests/unit/backend/test_scenario_run_service.py +++ b/tests/unit/backend/test_scenario_run_service.py @@ -19,7 +19,6 @@ from pyrit.converter import Converter from pyrit.models import AttackOutcome, ScenarioResult, ScenarioRunState from pyrit.models.catalog.scenario import RunScenarioRequest -from pyrit.models.identifiers.scenario_identifier import ScenarioIdentifier from pyrit.scenario.core import DatasetAttackConfiguration, DatasetConfiguration from pyrit.scenario.core.scenario_technique import ScenarioTechnique from unit.mocks import make_scenario_result @@ -89,33 +88,23 @@ def _make_db_scenario_result( run_state: ScenarioRunState = ScenarioRunState.IN_PROGRESS, attack_results: dict | None = None, ) -> MagicMock: - """Create a real ``ScenarioResult`` as CentralMemory would return it. - - A real object (not a spec'd mock) because the summary builder reads - ``scenario_identifier`` / ``metadata``, which a spec'd mock does not define. - """ - sr = ScenarioResult.model_construct( - id=result_id, - scenario_identifier=ScenarioIdentifier(class_name=scenario_name, techniques=[]), - attack_results=attack_results or {}, - scenario_run_state=run_state, - number_tries=1, - creation_time=datetime(2025, 1, 1, tzinfo=timezone.utc), - completion_time=datetime(2025, 1, 1, 0, 5, tzinfo=timezone.utc), - labels={}, - display_group_map={}, - error_message=None, - error_type=None, - metadata={"skipped_techniques": []}, - ) - - # Individual tests configure these; real methods would need live data. - from unittest.mock import MagicMock as MockFactory - - object.__setattr__(sr, "get_techniques_used", MockFactory(return_value=[])) - object.__setattr__(sr, "objective_achieved_rate", MockFactory(return_value=0)) - object.__setattr__(sr, "get_display_groups", MockFactory(return_value={})) - + """Create a mock ScenarioResult as returned by CentralMemory.""" + sr = MagicMock(spec=ScenarioResult) + sr.id = result_id + sr.scenario_name = scenario_name + sr.scenario_version = 1 + sr.scenario_run_state = run_state + sr.get_techniques_used.return_value = [] + sr.attack_results = attack_results or {} + sr.number_tries = 1 + sr.creation_time = datetime(2025, 1, 1, tzinfo=timezone.utc) + sr.completion_time = datetime(2025, 1, 1, 0, 5, tzinfo=timezone.utc) + sr.labels = {} + sr.objective_achieved_rate.return_value = 0 + sr.get_display_groups.return_value = {} + sr.display_group_map = {} + sr.error_message = None + sr.error_type = None return sr diff --git a/tests/unit/cli/test_output.py b/tests/unit/cli/test_output.py index d21aceae4e..8880c95466 100644 --- a/tests/unit/cli/test_output.py +++ b/tests/unit/cli/test_output.py @@ -541,24 +541,6 @@ def test_print_scenario_run_summary_completed(capsys): assert "Completed:" not in captured.out -def test_print_scenario_run_summary_lists_skipped_techniques(capsys): - run = _make_run( - scenario_name="partial", - scenario_result_id="id", - status=ScenarioRunState.COMPLETED, - total_attacks=2, - completed_attacks=2, - objective_achieved_rate=50, - techniques_used=["s1"], - skipped_techniques=["ghost_tech"], - ) - _output.print_scenario_run_summary(run=run) - captured = capsys.readouterr() - assert "Skipped:" in captured.out - assert "ghost_tech" in captured.out - assert "no registered factory" in captured.out - - def test_print_scenario_run_summary_with_error(capsys): run = _make_run( scenario_name="failing", diff --git a/tests/unit/scenario/airt/test_jailbreak.py b/tests/unit/scenario/airt/test_jailbreak.py index 93aa1ce2de..70d9e551f9 100644 --- a/tests/unit/scenario/airt/test_jailbreak.py +++ b/tests/unit/scenario/airt/test_jailbreak.py @@ -19,7 +19,6 @@ from pyrit.registry.components.scenario_registry import ScenarioRegistry from pyrit.scenario.core import BaselineAttackPolicy from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory -from pyrit.scenario.core.matrix_atomic_attack_builder import TechniqueResolution from pyrit.scenario.scenarios.airt.jailbreak import ( _DEFAULT_NUM_JAILBREAKS, _DEFAULT_TECHNIQUES, @@ -389,19 +388,6 @@ async def test_stale_incompatible_technique_is_rejected( with pytest.raises(ValueError, match="stale or incompatible"): await scenario.initialize_async() - async def test_missing_runtime_factory_is_rejected( - self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups - ): - with _patch_seed_groups(mock_memory_seed_groups): - with patch( - "pyrit.scenario.scenarios.airt.jailbreak.resolve_technique_factories", - return_value=TechniqueResolution(resolved={}, skipped=[]), - ): - scenario = Jailbreak(objective_scorer=mock_objective_scorer) - scenario.set_params_from_args(args=_default_args(mock_objective_target, jailbreak_names=["aim.yaml"])) - with pytest.raises(ValueError, match="no longer available.*prompt_sending"): - await scenario.initialize_async() - async def test_all_templates_produce_attacks( self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups ): diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index 4f8cd9869c..f5d46b65af 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -15,7 +15,6 @@ * optional baseline emission from the flattened seed groups. """ -import logging from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -400,40 +399,14 @@ def test_keeps_only_selected_in_order(self): } context = _context(techniques=[_technique("beta"), _technique("alpha")]) with _patch_registry(factories): - resolved = resolve_technique_factories(context=context).resolved - resolved = resolve_technique_factories(context=context).resolved + resolved = resolve_technique_factories(context=context) assert list(resolved.keys()) == ["beta", "alpha"] - def test_drops_techniques_without_factory(self): + def test_raises_when_any_selected_technique_is_missing(self): factories = {"alpha": _mock_factory(name="alpha")} context = _context(techniques=[_technique("alpha"), _technique("missing")]) - with _patch_registry(factories): - resolved = resolve_technique_factories(context=context).resolved - resolved = resolve_technique_factories(context=context).resolved - assert list(resolved.keys()) == ["alpha"] - - def test_warns_when_dropping_techniques_without_factory(self, caplog): - factories = {"alpha": _mock_factory(name="alpha")} - context = _context(techniques=[_technique("alpha"), _technique("missing")]) - with ( - _patch_registry(factories), - caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), - ): - resolve_technique_factories(context=context) - assert any("missing" in record.message for record in caplog.records) - - def test_no_warning_when_all_techniques_resolve(self, caplog): - factories = { - "alpha": _mock_factory(name="alpha"), - "beta": _mock_factory(name="beta"), - } - context = _context(techniques=[_technique("alpha"), _technique("beta")]) - with ( - _patch_registry(factories), - caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), - ): + with _patch_registry(factories), pytest.raises(TechniqueResolutionError, match="missing"): resolve_technique_factories(context=context) - assert not [record for record in caplog.records if record.levelno == logging.WARNING] def test_raises_when_all_selected_techniques_missing(self): """A nonempty selection resolving to nothing must fail loudly, not run baseline-only.""" @@ -445,20 +418,9 @@ def test_raises_when_all_selected_techniques_missing(self): def test_empty_selection_resolves_without_error(self): context = _context(techniques=[]) with _patch_registry({}): - assert resolve_technique_factories(context=context).resolved == {} + assert resolve_technique_factories(context=context) == {} - def test_partial_miss_still_warns_and_continues(self, caplog): - factories = {"alpha": _mock_factory(name="alpha")} - context = _context(techniques=[_technique("alpha"), _technique("missing")]) - with ( - _patch_registry(factories), - caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), - ): - resolved = resolve_technique_factories(context=context).resolved - resolved = resolve_technique_factories(context=context).resolved - assert list(resolved.keys()) == ["alpha"] - - def test_warning_lists_each_missing_technique_once_in_selection_order(self, caplog): + def test_error_lists_each_missing_technique_once_in_selection_order(self): factories = {"alpha": _mock_factory(name="alpha")} context = _context( techniques=[ @@ -468,15 +430,11 @@ def test_warning_lists_each_missing_technique_once_in_selection_order(self, capl _technique("missing_a"), ] ) - with ( - _patch_registry(factories), - caplog.at_level(logging.WARNING, logger="pyrit.scenario.core.matrix_atomic_attack_builder"), - ): + with _patch_registry(factories), pytest.raises(TechniqueResolutionError) as exc_info: resolve_technique_factories(context=context) - messages = [record.message for record in caplog.records if record.levelno == logging.WARNING] - assert len(messages) == 1 - assert messages[0].index("missing_a") < messages[0].index("missing_b") - assert messages[0].count("missing_a") == 1 # duplicates are deduplicated + message = str(exc_info.value) + assert message.index("missing_a") < message.index("missing_b") + assert message.count("missing_a") == 1 def test_extra_factories_merged_and_override_registry(self): registry_factories = {"alpha": _mock_factory(name="alpha")} @@ -484,13 +442,11 @@ def test_extra_factories_merged_and_override_registry(self): local_only = _mock_factory(name="local") context = _context(techniques=[_technique("alpha"), _technique("local")]) with _patch_registry(registry_factories): - resolution = resolve_technique_factories( + resolved = resolve_technique_factories( context=context, extra_factories={"alpha": local_alpha, "local": local_only}, ) - resolved = resolution.resolved assert list(resolved.keys()) == ["alpha", "local"] - assert resolution.skipped == [] assert resolved["alpha"] is local_alpha # extra overrides the registry factory of the same name assert resolved["local"] is local_only # local-only factory is selectable without global registration @@ -502,49 +458,46 @@ class TestBuildMatrixAtomicAttacks: def test_builds_cross_product_grouped_by_technique(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - attacks, skipped = build_matrix_atomic_attacks( - context=context, objective_scorer=MagicMock(spec=TrueFalseScorer) - ) - assert [a.atomic_attack_name for a in attacks] == ["tech_ds"] - assert attacks[0].display_group == "tech" - assert skipped == [] + result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) + assert [a.atomic_attack_name for a in result] == ["tech_ds"] + assert result[0].display_group == "tech" def test_custom_display_group_fn(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - attacks, _ = build_matrix_atomic_attacks( + result = build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), display_group_fn=lambda combo: combo.dataset_name, ) - assert attacks[0].display_group == "ds" + assert result[0].display_group == "ds" def test_no_baseline_emitted_when_context_disables_it(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, include_baseline=False, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - attacks, _ = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) - assert all(a.atomic_attack_name != "baseline" for a in attacks) + result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) + assert all(a.atomic_attack_name != "baseline" for a in result) def test_baseline_emitted_when_context_enables_it(self): context = _context( techniques=[_technique("tech")], - seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, + seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, include_baseline=True, ) with _patch_registry({"tech": _mock_factory(name="tech")}): - attacks, _ = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) - assert attacks[0].atomic_attack_name == "baseline" - assert [a.atomic_attack_name for a in attacks] == ["baseline", "tech_ds"] + result = build_matrix_atomic_attacks(context=context, objective_scorer=MagicMock(spec=TrueFalseScorer)) + assert result[0].atomic_attack_name == "baseline" + assert [a.atomic_attack_name for a in result] == ["baseline", "tech_ds"] def test_technique_converters_forwarded(self): from pyrit.converter import Converter @@ -556,7 +509,7 @@ def test_technique_converters_forwarded(self): factory = _mock_factory(name="tech") converter = MagicMock(spec=Converter) with _patch_registry({"tech": factory}): - attacks, _ = build_matrix_atomic_attacks( + build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), technique_converters={"tech": [converter]}, @@ -566,17 +519,15 @@ def test_technique_converters_forwarded(self): assert len(extra) == 1 def test_extra_factories_used_for_selection(self): - local_alpha = _mock_factory(name="alpha") - local_only = _mock_factory(name="local") context = _context( - techniques=[_technique("alpha"), _technique("local")], - seed_groups_by_dataset={"ds": [_seed_group(objective="q")]}, + techniques=[_technique("local")], + seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, ) - with _patch_registry({}): - attacks, skipped = build_matrix_atomic_attacks( + # The selected technique exists only in extra_factories, not the registry. + with _patch_registry({"other": _mock_factory(name="other")}): + result = build_matrix_atomic_attacks( context=context, objective_scorer=MagicMock(spec=TrueFalseScorer), - extra_factories={"alpha": local_alpha, "local": local_only}, + extra_factories={"local": _mock_factory(name="local")}, ) - assert [a.atomic_attack_name for a in attacks] == ["alpha_ds", "local_ds"] - assert skipped == [] + assert [a.atomic_attack_name for a in result] == ["local_ds"]