diff --git a/pyrit/backend/services/scenario_run_service.py b/pyrit/backend/services/scenario_run_service.py index 7ba66d0f43..dc7a4aa9d5 100644 --- a/pyrit/backend/services/scenario_run_service.py +++ b/pyrit/backend/services/scenario_run_service.py @@ -599,6 +599,14 @@ 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] = [] @@ -641,6 +649,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 5b04d2611f..1b66e793f5 100644 --- a/pyrit/cli/_output.py +++ b/pyrit/cli/_output.py @@ -387,6 +387,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 40a35ca474..1b970d5853 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -29,6 +29,34 @@ 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. + """ + + +@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 @@ -136,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. @@ -152,19 +180,48 @@ def resolve_technique_factories( name. Returns: - dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by - the selected techniques. + 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, + since running only the baseline would silently defeat the selection. """ 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), + ) + + 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 TechniqueResolution(resolved=resolved, skipped=missing) def build_matrix_atomic_attacks( @@ -205,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), + 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/core/scenario.py b/pyrit/scenario/core/scenario.py index 9385c8c97b..de2e96b440 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -695,6 +695,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/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/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index 9bfcc63f19..cb028ec341 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -326,7 +326,11 @@ 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 + selected_names = {technique.value for technique in context.scenario_techniques} missing = selected_names - set(technique_factories) if missing: 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/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 9270187920..48da9b62be 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -226,7 +226,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 new file mode 100644 index 0000000000..eca63ce8ce --- /dev/null +++ b/tests/unit/backend/services/test_scenario_run_service_summary.py @@ -0,0 +1,79 @@ +# 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 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/cli/test_output.py b/tests/unit/cli/test_output.py index 8880c95466..d21aceae4e 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/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 044ce78512..b52331d271 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 @@ -26,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, @@ -398,31 +400,102 @@ 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): + 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_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).resolved == {} + + 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): + 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") 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 +@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.""" @@ -430,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 @@ -481,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]}, @@ -491,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 == []