diff --git a/pyrit/memory/memory_models.py b/pyrit/memory/memory_models.py index e70a421a6d..69c9306457 100644 --- a/pyrit/memory/memory_models.py +++ b/pyrit/memory/memory_models.py @@ -401,14 +401,17 @@ def __init__(self, *, entry: Score) -> None: self.score_rationale = entry.score_rationale self.score_metadata = entry.score_metadata or {} normalized_scorer = entry.scorer_class_identifier - # Ensure eval_hash is set before truncation so it survives the DB round-trip - if normalized_scorer.eval_hash is None: - normalized_scorer = normalized_scorer.with_eval_hash( - ScorerEvaluationIdentifier(normalized_scorer).eval_hash + if normalized_scorer is None: + self.scorer_class_identifier = {} + else: + # Ensure eval_hash is set before truncation so it survives the DB round-trip + if normalized_scorer.eval_hash is None: + normalized_scorer = normalized_scorer.with_eval_hash( + ScorerEvaluationIdentifier(normalized_scorer).eval_hash + ) + self.scorer_class_identifier = normalized_scorer.model_dump( + context={"max_value_length": MAX_IDENTIFIER_VALUE_LENGTH}, ) - self.scorer_class_identifier = normalized_scorer.model_dump( - context={"max_value_length": MAX_IDENTIFIER_VALUE_LENGTH}, - ) self.prompt_request_response_id = entry.message_piece_id if entry.message_piece_id else None self.timestamp = entry.timestamp # Store in both columns for backward compatibility @@ -439,7 +442,7 @@ def get_score(self) -> Score: score_category=self.score_category, score_rationale=self.score_rationale, score_metadata=self.score_metadata, - scorer_class_identifier=scorer_identifier, # type: ignore[ty:invalid-argument-type] + scorer_class_identifier=scorer_identifier, message_piece_id=self.prompt_request_response_id, timestamp=_ensure_utc(self.timestamp), objective=self.objective, @@ -975,8 +978,8 @@ class ScenarioResultEntry(Base): scenario_init_data (dict): Optional initialization parameters used to configure the scenario. objective_target_identifier (dict): Identifier for the target being evaluated in the scenario. objective_scorer_identifier (dict): Optional identifier for the scorer used to evaluate results. - scenario_run_state (Literal["CREATED", "IN_PROGRESS", "COMPLETED", "FAILED"]): Current execution state - of the scenario. + scenario_run_state (str): Current execution state of the scenario + (one of CREATED, IN_PROGRESS, COMPLETED, FAILED, CANCELLED). attack_results_json (str): JSON-serialized dictionary mapping attack names to conversation IDs. Format: {"attack_name": ["conversation_id1", "conversation_id2", ...]}. The full AttackResult objects are stored in AttackResultEntries and can be queried by conversation_id. @@ -1003,9 +1006,7 @@ class ScenarioResultEntry(Base): scenario_init_data: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) objective_target_identifier: Mapped[dict[str, str]] = mapped_column(JSON, nullable=False) objective_scorer_identifier: Mapped[dict[str, str] | None] = mapped_column(JSON, nullable=True) - scenario_run_state: Mapped[Literal["CREATED", "IN_PROGRESS", "COMPLETED", "FAILED", "CANCELLED"]] = mapped_column( - String, nullable=False, default="CREATED" - ) + scenario_run_state: Mapped[str] = mapped_column(String, nullable=False, default="CREATED") attack_results_json: Mapped[str] = mapped_column(Unicode, nullable=False) display_group_map_json: Mapped[str | None] = mapped_column(Unicode, nullable=True) labels: Mapped[dict[str, str] | None] = mapped_column(JSON, nullable=True) @@ -1072,7 +1073,7 @@ def __init__(self, *, entry: ScenarioResult) -> None: self.attack_results_json = json.dumps(serialized_attack_results) # Serialize display_group_map if present - self.display_group_map_json = json.dumps(entry._display_group_map) if entry._display_group_map else None + self.display_group_map_json = json.dumps(entry.display_group_map) if entry.display_group_map else None self.error_message = entry.error_message self.error_type = entry.error_type @@ -1126,14 +1127,14 @@ def get_scenario_result(self) -> ScenarioResult: attack_results=attack_results, objective_scorer_identifier=scorer_identifier, scenario_run_state=self.scenario_run_state, - labels=self.labels, + labels=self.labels or {}, creation_time=self.timestamp, number_tries=self.number_tries, completion_time=self.completion_time, - display_group_map=display_group_map, + display_group_map=display_group_map or {}, error_message=self.error_message, error_type=self.error_type, - metadata=dict(self.scenario_metadata) if self.scenario_metadata else None, + metadata=dict(self.scenario_metadata) if self.scenario_metadata else {}, ) def get_conversation_ids_by_attack_name(self) -> dict[str, list[str]]: diff --git a/pyrit/models/identifiers/component_identifier.py b/pyrit/models/identifiers/component_identifier.py index 948ad2eeec..d3d0c71933 100644 --- a/pyrit/models/identifiers/component_identifier.py +++ b/pyrit/models/identifiers/component_identifier.py @@ -168,7 +168,7 @@ class ComponentIdentifier(BaseModel): #: may have been truncated. hash: Optional[str] = None #: Version tag for storage. Not included in the content hash. - pyrit_version: str = Field(default_factory=lambda: pyrit.__version__) + pyrit_version: str = Field(default=pyrit.__version__) #: Evaluation hash. Computed by EvaluationIdentifier subclasses and attached #: to the identifier so it survives DB round-trips with truncated params. eval_hash: Optional[str] = None diff --git a/pyrit/models/scenario_result.py b/pyrit/models/scenario_result.py index 3bddaf776f..60227b96a4 100644 --- a/pyrit/models/scenario_result.py +++ b/pyrit/models/scenario_result.py @@ -6,171 +6,143 @@ import logging import uuid from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, Literal +from enum import Enum +from typing import Any -import pyrit -from pyrit.models.attack_result import AttackOutcome, AttackResult +from pydantic import BaseModel, ConfigDict, Field -if TYPE_CHECKING: - from pyrit.models.identifiers.component_identifier import ComponentIdentifier - from pyrit.score.scorer_evaluation.scorer_metrics import ScorerMetrics +import pyrit +from pyrit.common.deprecation import print_deprecation_message +from pyrit.models.identifiers.component_identifier import ( # noqa: TC001 (runtime-required by Pydantic field annotations) + ComponentIdentifier, +) +from pyrit.models.results.attack_result import AttackOutcome, AttackResult logger = logging.getLogger(__name__) -class ScenarioIdentifier: +class ScenarioIdentifier(BaseModel): """ - Scenario result class for aggregating results from multiple AtomicAttacks. + Identifier describing the executed scenario. """ - def __init__( - self, - name: str, - description: str = "", - scenario_version: int = 1, - init_data: dict[str, Any] | None = None, - pyrit_version: str | None = None, - ) -> None: - """ - Initialize a ScenarioIdentifier. + model_config = ConfigDict(extra="forbid", populate_by_name=True) - Args: - name (str): Name of the scenario. - description (str): Description of the scenario. - scenario_version (int): Version of the scenario. - init_data (Optional[dict]): Initialization data. - pyrit_version (Optional[str]): PyRIT version string. If None, uses current version. - - """ - self.name = name - self.description = description - self.version = scenario_version - self.pyrit_version = pyrit_version if pyrit_version is not None else pyrit.__version__ - self.init_data = init_data + #: Name of the scenario. + name: str + #: Description of the scenario. + description: str = "" + #: Version of the scenario. Accepts the legacy ``scenario_version`` kwarg/wire key. + version: int = Field(default=1, alias="scenario_version") + #: PyRIT version string. Defaults to the current installed version. + pyrit_version: str = Field(default=pyrit.__version__) + #: Optional initialization data. + init_data: dict[str, Any] | None = None def to_dict(self) -> dict[str, Any]: """ Serialize to a JSON-compatible dictionary. + Deprecated: use ``model_dump(by_alias=True)`` instead. + Returns: dict[str, Any]: Serialized payload. """ - return { - "name": self.name, - "description": self.description, - "scenario_version": self.version, - "pyrit_version": self.pyrit_version, - "init_data": self.init_data, - } + print_deprecation_message( + old_item="ScenarioIdentifier.to_dict()", + new_item="ScenarioIdentifier.model_dump(by_alias=True)", + removed_in="0.16.0", + ) + return self.model_dump(by_alias=True) @classmethod def from_dict(cls, data: dict[str, Any]) -> ScenarioIdentifier: """ Reconstruct a ScenarioIdentifier from a dictionary. + Deprecated: use ``model_validate(...)`` instead. + Args: - data (dict[str, Any]): Dictionary as produced by to_dict(). + data (dict[str, Any]): Dictionary as produced by ``model_dump(by_alias=True)``. Returns: ScenarioIdentifier: Reconstructed instance. """ - return cls( - name=data["name"], - description=data.get("description", ""), - scenario_version=data.get("scenario_version", 1), - init_data=data.get("init_data"), - pyrit_version=data.get("pyrit_version") or "unknown", + print_deprecation_message( + old_item="ScenarioIdentifier.from_dict(...)", + new_item="ScenarioIdentifier.model_validate(...)", + removed_in="0.16.0", ) + return cls.model_validate(data) -ScenarioRunState = Literal["CREATED", "IN_PROGRESS", "COMPLETED", "FAILED", "CANCELLED"] +class ScenarioRunState(str, Enum): + """ + Lifecycle state of a scenario run. + + Inherits from ``str`` so values serialize naturally in Pydantic models and + REST responses, and compare equal to their string form. + """ + CREATED = "CREATED" + IN_PROGRESS = "IN_PROGRESS" + COMPLETED = "COMPLETED" + FAILED = "FAILED" + CANCELLED = "CANCELLED" -class ScenarioResult: + +class ScenarioResult(BaseModel): """ Scenario result class for aggregating scenario results. """ - def __init__( - self, - *, - scenario_identifier: ScenarioIdentifier, - objective_target_identifier: ComponentIdentifier | None, - attack_results: dict[str, list[AttackResult]], - objective_scorer_identifier: ComponentIdentifier | None, - scenario_run_state: ScenarioRunState = "CREATED", - labels: dict[str, str] | None = None, - creation_time: datetime | None = None, - completion_time: datetime | None = None, - number_tries: int = 0, - id: uuid.UUID | None = None, # noqa: A002 - display_group_map: dict[str, str] | None = None, - error_message: str | None = None, - error_type: str | None = None, - error_attack_result_ids: list[str] | None = None, - metadata: dict[str, Any] | None = None, - ) -> None: - """ - Initialize a scenario result. - - Args: - scenario_identifier (ScenarioIdentifier): Identifier for the executed scenario. - objective_target_identifier (ComponentIdentifier): Target identifier. - attack_results (dict[str, List[AttackResult]]): Results grouped by atomic attack name. - objective_scorer_identifier (ComponentIdentifier | None): Objective scorer identifier, - or None if the scenario has no objective scorer. - scenario_run_state (ScenarioRunState): Current scenario run state. - labels (Optional[dict[str, str]]): Optional labels. - creation_time (datetime | None): When the scenario result was created. - completion_time (Optional[datetime]): Optional completion timestamp. - number_tries (int): Number of run attempts. - id (Optional[uuid.UUID]): Optional scenario result ID. - display_group_map (Optional[dict[str, str]]): Optional mapping of - atomic_attack_name → display group label. Used by the console - printer to aggregate results for user-facing output. - error_message (Optional[str]): Scenario-level error message when the run fails. - error_type (Optional[str]): Exception class name when the run fails. - error_attack_result_ids (Optional[list[str]]): IDs of attack results that - errored during the scenario run. Defaults to an empty list. - metadata (Optional[dict[str, Any]]): Free-form JSON metadata persisted - with the scenario result. Currently used to record - ``objective_hashes`` — the objective ``sha256`` set chosen - on the first run, replayed on resume so a fresh - ``random.sample`` can't silently change which objectives the - scenario operates on. Keys are not part of any public contract - and may evolve. - - """ - self.id = id if id is not None else uuid.uuid4() - self.scenario_identifier = scenario_identifier - - self.objective_target_identifier = objective_target_identifier - - self.objective_scorer_identifier = objective_scorer_identifier - - self.scenario_run_state = scenario_run_state - self.attack_results = attack_results - self.labels = labels if labels is not None else {} - self.creation_time = creation_time if creation_time is not None else datetime.now(timezone.utc) - self.completion_time = completion_time if completion_time is not None else datetime.now(timezone.utc) - self.number_tries = number_tries - self._display_group_map = display_group_map or {} - self.error_message = error_message - self.error_type = error_type - self.error_attack_result_ids: list[str] = list(error_attack_result_ids) if error_attack_result_ids else [] - self.metadata: dict[str, Any] = metadata if metadata is not None else {} - - @property - def display_group_map(self) -> dict[str, str]: - """Mapping of atomic_attack_name → display group label.""" - return self._display_group_map + model_config = ConfigDict( + arbitrary_types_allowed=True, + extra="forbid", + validate_assignment=False, + ) + + #: Scenario result ID. + id: uuid.UUID = Field(default_factory=uuid.uuid4) # noqa: A003 + #: Identifier for the executed scenario. + scenario_identifier: ScenarioIdentifier + #: Target identifier. + objective_target_identifier: ComponentIdentifier | None + #: Objective scorer identifier, or None if the scenario has no objective scorer. + objective_scorer_identifier: ComponentIdentifier | None + #: Results grouped by atomic attack name. + attack_results: dict[str, list[AttackResult]] + #: Current scenario run state. + scenario_run_state: ScenarioRunState = ScenarioRunState.CREATED + #: Optional labels. + labels: dict[str, str] = Field(default_factory=dict) + #: When the scenario result was created. + creation_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + #: Optional completion timestamp. + completion_time: datetime | None = Field(default_factory=lambda: datetime.now(timezone.utc)) + #: Number of run attempts. + number_tries: int = 0 + #: Mapping of ``atomic_attack_name`` -> display group label. Used by the console + #: printer to aggregate results for user-facing output. + display_group_map: dict[str, str] = Field(default_factory=dict) + #: Scenario-level error message when the run fails. + error_message: str | None = None + #: Exception class name when the run fails. + error_type: str | None = None + #: IDs of attack results that errored during the scenario run. + error_attack_result_ids: list[str] = Field(default_factory=list) + #: Free-form JSON metadata persisted with the scenario result. Currently used to record + #: ``objective_hashes`` — the objective ``sha256`` set chosen on the first run, replayed + #: on resume so a fresh ``random.sample`` can't silently change which objectives the + #: scenario operates on. Keys are not part of any public contract and may evolve. + metadata: dict[str, Any] = Field(default_factory=dict) def get_strategies_used(self) -> list[str]: """ Get the list of strategies used in this scenario. Returns: - List[str]: Atomic attack strategy names present in the results. + list[str]: Atomic attack strategy names present in the results. """ return list(self.attack_results.keys()) @@ -181,18 +153,18 @@ def get_display_groups(self) -> dict[str, list[AttackResult]]: When a ``display_group_map`` was provided, results from multiple ``atomic_attack_name`` keys that share the same display group are - merged into a single list. When no map was provided, this returns + merged into a single list. When no map was provided, this returns the same structure as ``attack_results`` (identity mapping). Returns: dict[str, list[AttackResult]]: Results grouped by display label. """ - if not self._display_group_map: + if not self.display_group_map: return dict(self.attack_results) grouped: dict[str, list[AttackResult]] = {} for attack_name, results in self.attack_results.items(): - group = self._display_group_map.get(attack_name, attack_name) + group = self.display_group_map.get(attack_name, attack_name) grouped.setdefault(group, []).extend(results) return grouped @@ -201,11 +173,11 @@ def get_objectives(self, *, atomic_attack_name: str | None = None) -> list[str]: Get the list of unique objectives for this scenario. Args: - atomic_attack_name (Optional[str]): Name of specific atomic attack to include. + atomic_attack_name (str | None): Name of specific atomic attack to include. If None, includes objectives from all atomic attacks. Defaults to None. Returns: - List[str]: Deduplicated list of objectives. + list[str]: Deduplicated list of objectives. """ objectives: list[str] = [] @@ -231,7 +203,7 @@ def objective_achieved_rate(self, *, atomic_attack_name: str | None = None) -> i Get the success rate of this scenario. Args: - atomic_attack_name (Optional[str]): Name of specific atomic attack to calculate rate for. + atomic_attack_name (str | None): Name of specific atomic attack to calculate rate for. If None, calculates rate across all atomic attacks. Defaults to None. Returns: @@ -257,6 +229,42 @@ def objective_achieved_rate(self, *, atomic_attack_name: str | None = None) -> i successful_results = sum(1 for result in all_results if result.outcome == AttackOutcome.SUCCESS) return int((successful_results / total_results) * 100) + def to_dict(self) -> dict[str, Any]: + """ + Serialize this scenario result to a JSON-compatible dictionary. + + Deprecated: use ``model_dump(mode="json", by_alias=True)`` instead. + + Returns: + dict[str, Any]: Serialized payload suitable for REST APIs or persistence. + """ + print_deprecation_message( + old_item="ScenarioResult.to_dict()", + new_item="ScenarioResult.model_dump(mode='json', by_alias=True)", + removed_in="0.16.0", + ) + return self.model_dump(mode="json", by_alias=True) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ScenarioResult: + """ + Reconstruct a ScenarioResult from a dictionary. + + Deprecated: use ``model_validate(...)`` instead. + + Args: + data (dict[str, Any]): Dictionary as produced by ``model_dump(mode="json")``. + + Returns: + ScenarioResult: Reconstructed instance. + """ + print_deprecation_message( + old_item="ScenarioResult.from_dict(...)", + new_item="ScenarioResult.model_validate(...)", + removed_in="0.16.0", + ) + return cls.model_validate(data) + @staticmethod def normalize_scenario_name(scenario_name: str) -> str: """ @@ -270,10 +278,10 @@ def normalize_scenario_name(scenario_name: str) -> str: This is the inverse of ScenarioRegistry._class_name_to_scenario_name(). Args: - scenario_name: The scenario name to normalize. + scenario_name (str): The scenario name to normalize. Returns: - The normalized scenario name suitable for database queries. + str: The normalized scenario name suitable for database queries. """ # Check if it looks like snake_case (contains underscore and is lowercase) @@ -284,98 +292,3 @@ def normalize_scenario_name(scenario_name: str) -> str: return "".join(part.capitalize() for part in parts) # Already PascalCase or other format, return as-is return scenario_name - - def get_scorer_evaluation_metrics(self) -> ScorerMetrics | None: - """ - Get the evaluation metrics for the scenario's scorer from the scorer evaluation registry. - - Returns: - ScorerMetrics: The evaluation metrics object, or None if not found. - - """ - # import here to avoid circular imports - from pyrit.models.identifiers.evaluation_identifier import ScorerEvaluationIdentifier - from pyrit.score.scorer_evaluation.scorer_metrics_io import ( - find_objective_metrics_by_eval_hash, - ) - - if not self.objective_scorer_identifier: - return None - - eval_hash = ScorerEvaluationIdentifier(self.objective_scorer_identifier).eval_hash - - return find_objective_metrics_by_eval_hash(eval_hash=eval_hash) - - def to_dict(self) -> dict[str, Any]: - """ - Serialize this scenario result to a JSON-compatible dictionary. - - Returns: - dict[str, Any]: Serialized payload suitable for REST APIs or persistence. - """ - return { - "id": str(self.id), - "scenario_identifier": self.scenario_identifier.to_dict(), - "objective_target_identifier": ( - self.objective_target_identifier.model_dump() if self.objective_target_identifier else None - ), - "objective_scorer_identifier": ( - self.objective_scorer_identifier.model_dump() if self.objective_scorer_identifier else None - ), - "scenario_run_state": self.scenario_run_state, - "attack_results": {name: [r.to_dict() for r in results] for name, results in self.attack_results.items()}, - "display_group_map": self._display_group_map, - "labels": self.labels, - "creation_time": self.creation_time.isoformat() if self.creation_time else None, - "completion_time": self.completion_time.isoformat() if self.completion_time else None, - "number_tries": self.number_tries, - "error_attack_result_ids": self.error_attack_result_ids, - "error_message": self.error_message, - "error_type": self.error_type, - } - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> ScenarioResult: - """ - Reconstruct a ScenarioResult from a dictionary. - - Args: - data (dict[str, Any]): Dictionary as produced by to_dict(). - - Returns: - ScenarioResult: Reconstructed instance. - """ - from pyrit.models.identifiers.component_identifier import ComponentIdentifier - - result = cls( - id=uuid.UUID(data["id"]) if data.get("id") else None, - scenario_identifier=ScenarioIdentifier.from_dict(data["scenario_identifier"]), - objective_target_identifier=( - ComponentIdentifier.model_validate(data["objective_target_identifier"]) - if data.get("objective_target_identifier") - else None - ), - objective_scorer_identifier=( - ComponentIdentifier.model_validate(data["objective_scorer_identifier"]) - if data.get("objective_scorer_identifier") - else None - ), - scenario_run_state=data.get("scenario_run_state", "CREATED"), - attack_results={ - name: [AttackResult.from_dict(r) for r in results] - for name, results in data.get("attack_results", {}).items() - }, - display_group_map=data.get("display_group_map"), - labels=data.get("labels"), - creation_time=(datetime.fromisoformat(data["creation_time"]) if data.get("creation_time") else None), - completion_time=(datetime.fromisoformat(data["completion_time"]) if data.get("completion_time") else None), - number_tries=data.get("number_tries", 0), - error_attack_result_ids=data.get("error_attack_result_ids"), - error_message=data.get("error_message"), - error_type=data.get("error_type"), - ) - # Preserve missing completion_time: __init__ defaults it to now(), but a - # still-running scenario shouldn't be marked as completed-at-load-time. - if not data.get("completion_time"): - result.completion_time = None # type: ignore[ty:invalid-assignment] - return result diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index f6d57d2d29..e08ef6d3d1 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -357,13 +357,13 @@ async def run_async(self) -> ScenarioResult: Returns: ScenarioResult: The scenario result with cached attack results merged into ``attack_results`` and cached display groups merged into - ``_display_group_map``. + ``display_group_map``. """ result = await super().run_async() if self._precomputed_cached_results: for attack_name, prior_results in self._precomputed_cached_results.items(): result.attack_results.setdefault(attack_name, []).extend(prior_results) - result._display_group_map.update(self._precomputed_cached_display_groups) + result.display_group_map.update(self._precomputed_cached_display_groups) return result def _collect_cached_completion_pairs(self, *, atomic_attacks: list[AtomicAttack]) -> set[str]: diff --git a/tests/unit/models/test_import_boundary.py b/tests/unit/models/test_import_boundary.py index 2c95c1e59d..981012711c 100644 --- a/tests/unit/models/test_import_boundary.py +++ b/tests/unit/models/test_import_boundary.py @@ -62,10 +62,6 @@ "pyrit.models.identifiers.evaluation_identifier": { "pyrit.executor.attack.core.attack_strategy": "phase-7", }, - "pyrit.models.scenario_result": { - "pyrit.score.scorer_evaluation.scorer_metrics": "phase-7", - "pyrit.score.scorer_evaluation.scorer_metrics_io": "phase-7", - }, "pyrit.models.storage_io": { "pyrit.auth": "phase-9", }, diff --git a/tests/unit/models/test_scenario_result.py b/tests/unit/models/test_scenario_result.py index fbf46e19da..e15abc53e7 100644 --- a/tests/unit/models/test_scenario_result.py +++ b/tests/unit/models/test_scenario_result.py @@ -2,9 +2,15 @@ # Licensed under the MIT license. import uuid +from datetime import datetime, timezone +import pytest + +import pyrit from pyrit.models import ComponentIdentifier +from pyrit.models.conversation_reference import ConversationReference, ConversationType from pyrit.models.results.attack_result import AttackOutcome, AttackResult +from pyrit.models.retry_event import RetryEvent from pyrit.models.scenario_result import ScenarioIdentifier, ScenarioResult @@ -47,8 +53,6 @@ def test_init_with_all_params(self): assert si.pyrit_version == "1.0.0" def test_init_default_pyrit_version(self): - import pyrit - si = ScenarioIdentifier(name="X") assert si.pyrit_version == pyrit.__version__ @@ -201,11 +205,6 @@ def test_scenario_identifier_to_dict_from_dict_roundtrip(): def test_scenario_result_to_dict_from_dict_roundtrip(): - from datetime import datetime, timezone - - from pyrit.models.conversation_reference import ConversationReference, ConversationType - from pyrit.models.retry_event import RetryEvent - scenario_id = ScenarioIdentifier( name="ContentHarms", description="Tests content harm scenarios", @@ -268,10 +267,13 @@ def test_scenario_result_to_dict_from_dict_roundtrip(): ) roundtripped = ScenarioResult.from_dict(original.to_dict()) assert original.to_dict() == roundtripped.to_dict() + # The nested identifier must preserve the legacy ``scenario_version`` wire key. + assert "scenario_version" in original.to_dict()["scenario_identifier"] + assert "version" not in original.to_dict()["scenario_identifier"] -def test_scenario_identifier_from_dict_missing_pyrit_version_yields_unknown(): - """from_dict should preserve 'unknown' rather than fabricating the current version when missing.""" +def test_scenario_identifier_from_dict_missing_pyrit_version_uses_current(): + """A payload missing pyrit_version now resolves to the current version via the Pydantic default.""" data = { "name": "Legacy", "description": "loaded from older payload", @@ -280,7 +282,7 @@ def test_scenario_identifier_from_dict_missing_pyrit_version_yields_unknown(): # pyrit_version intentionally absent } identifier = ScenarioIdentifier.from_dict(data) - assert identifier.pyrit_version == "unknown" + assert identifier.pyrit_version == pyrit.__version__ def test_scenario_result_from_dict_preserves_missing_completion_time(): @@ -300,3 +302,41 @@ def test_scenario_result_from_dict_preserves_missing_completion_time(): roundtripped = ScenarioResult.from_dict(original.to_dict()) assert roundtripped.completion_time is None assert roundtripped.scenario_run_state == "IN_PROGRESS" + + +def test_scenario_identifier_to_dict_from_dict_emit_deprecation_warnings(): + identifier = ScenarioIdentifier(name="Test", scenario_version=1, pyrit_version="0.14.0") + with pytest.warns(DeprecationWarning): + payload = identifier.to_dict() + with pytest.warns(DeprecationWarning): + ScenarioIdentifier.from_dict(payload) + + +def test_scenario_result_to_dict_from_dict_emit_deprecation_warnings(): + scenario_id = ScenarioIdentifier(name="Test", scenario_version=1, pyrit_version="0.14.0") + result = ScenarioResult( + scenario_identifier=scenario_id, + objective_target_identifier=ComponentIdentifier.from_dict({}), + objective_scorer_identifier=None, + attack_results={}, + ) + with pytest.warns(DeprecationWarning): + payload = result.to_dict() + with pytest.warns(DeprecationWarning): + ScenarioResult.from_dict(payload) + + +def test_scenario_result_display_group_map_is_public_field(): + scenario_id = ScenarioIdentifier(name="Test", scenario_version=1, pyrit_version="0.14.0") + result = ScenarioResult( + scenario_identifier=scenario_id, + objective_target_identifier=ComponentIdentifier.from_dict({}), + objective_scorer_identifier=None, + attack_results={"crescendo": []}, + display_group_map={"crescendo": "Crescendo Attack"}, + ) + assert "display_group_map" in ScenarioResult.model_fields + assert result.display_group_map == {"crescendo": "Crescendo Attack"} + # Mutable and writable (used by benchmark merge logic). + result.display_group_map["foundry"] = "Foundry" + assert result.display_group_map["foundry"] == "Foundry" diff --git a/tests/unit/output/scenario_result/test_pretty.py b/tests/unit/output/scenario_result/test_pretty.py index 3e5b10a5c2..b2f8cced9c 100644 --- a/tests/unit/output/scenario_result/test_pretty.py +++ b/tests/unit/output/scenario_result/test_pretty.py @@ -35,7 +35,7 @@ def _scenario_result( objective_target_identifier=_target_identifier(**(target_params or {})), attack_results=attack_results or {"strategy_a": [_attack_result()]}, objective_scorer_identifier=objective_scorer_identifier, - display_group_map=display_group_map, + display_group_map=display_group_map or {}, ) diff --git a/tests/unit/scenario/benchmark/test_adversarial.py b/tests/unit/scenario/benchmark/test_adversarial.py index 10e823d9d8..7355f53b31 100644 --- a/tests/unit/scenario/benchmark/test_adversarial.py +++ b/tests/unit/scenario/benchmark/test_adversarial.py @@ -1138,7 +1138,7 @@ async def test_precomputed_results_injected_into_attack_results(self): # Base run_async produced only the non-skipped attack's result base_scenario_result = MagicMock() base_scenario_result.attack_results = {"technique_c__adv_target_harmbench": [result_z]} - base_scenario_result._display_group_map = {} + base_scenario_result.display_group_map = {} with patch.object(Scenario, "run_async", new=AsyncMock(return_value=base_scenario_result)): result = await bench.run_async() @@ -1164,12 +1164,12 @@ async def test_display_group_map_updated_for_cached_attacks(self): base_scenario_result = MagicMock() base_scenario_result.attack_results = {} - base_scenario_result._display_group_map = {} + base_scenario_result.display_group_map = {} with patch.object(Scenario, "run_async", new=AsyncMock(return_value=base_scenario_result)): result = await bench.run_async() - assert result._display_group_map["technique_a__adv_target_harmbench"] == "adv_target" + assert result.display_group_map["technique_a__adv_target_harmbench"] == "adv_target" async def test_no_injection_when_no_cached_attacks(self): """When all attacks were executed freshly, attack_results is returned unchanged.""" @@ -1181,7 +1181,7 @@ async def test_no_injection_when_no_cached_attacks(self): result_z = MagicMock(spec=AttackResult) base_scenario_result = MagicMock() base_scenario_result.attack_results = {"technique_c__adv_target_harmbench": [result_z]} - base_scenario_result._display_group_map = {} + base_scenario_result.display_group_map = {} with patch.object(Scenario, "run_async", new=AsyncMock(return_value=base_scenario_result)): result = await bench.run_async()