diff --git a/frontend/src/components/Scenarios/ScenarioRunPage.test.tsx b/frontend/src/components/Scenarios/ScenarioRunPage.test.tsx index e703767b7f..7286e526d5 100644 --- a/frontend/src/components/Scenarios/ScenarioRunPage.test.tsx +++ b/frontend/src/components/Scenarios/ScenarioRunPage.test.tsx @@ -473,7 +473,7 @@ describe('ScenarioRunPage', () => { 'data-location', `/scanner-history/${SCENARIO_RESULT_ID}/attack-result-1`, )) - const dialog = await screen.findByRole('dialog', { name: 'attack-technique' }) + const dialog = await screen.findByRole('dialog', { name: 'attack-technique', hidden: true }) await user.click(within(dialog).getByRole('button', { name: 'Close', hidden: true })) await waitFor(() => expect(detailsRow).toHaveFocus()) diff --git a/pyrit/memory/alembic/versions/1b3d5f7a9c2e_persist_scored_expectation.py b/pyrit/memory/alembic/versions/1b3d5f7a9c2e_persist_scored_expectation.py new file mode 100644 index 0000000000..11f8eaba8e --- /dev/null +++ b/pyrit/memory/alembic/versions/1b3d5f7a9c2e_persist_scored_expectation.py @@ -0,0 +1,177 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Persist the full scoring expectation instead of a bare objective. + +``ScoreEntries.objective`` held only the objective string a score was judged against. +``ScoreEntries.scored_expectation`` records the complete versioned expectation (objective +plus any typed conditions), so a persisted score keeps the whole of what it was scored +for. On upgrade the legacy objective is folded into an objective-only expectation; on +downgrade only the objective survives and typed conditions are dropped. + +Revision ID: 1b3d5f7a9c2e +Revises: 0f2e4d6c8b1a +Create Date: 2026-09-03 10:00:00.000000 +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Sequence # noqa: TC003 + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "1b3d5f7a9c2e" +down_revision: str | None = "0f2e4d6c8b1a" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +logger = logging.getLogger(__name__) + +#: Version stamped onto every backfilled expectation. +_SCHEMA_VERSION = 1 + +#: Rows per page so a large score table migrates in bounded keyset batches, not one statement. +_BACKFILL_BATCH_SIZE = 500 + + +def upgrade() -> None: + """Add ``scored_expectation``, fold the legacy objective into it, then drop ``objective``.""" + with op.batch_alter_table("ScoreEntries") as batch_op: + batch_op.add_column(sa.Column("scored_expectation", sa.JSON(), nullable=True)) + + _backfill_scored_expectation() + + with op.batch_alter_table("ScoreEntries") as batch_op: + batch_op.drop_column("objective") + + +def downgrade() -> None: + """ + Re-add ``objective``, recover it from ``scored_expectation``, then drop the expectation. + + This is lossy by design: only the expectation's objective survives. Typed conditions + have no column in the old schema and are dropped. + """ + with op.batch_alter_table("ScoreEntries") as batch_op: + batch_op.add_column(sa.Column("objective", sa.String(), nullable=True)) + + _backfill_objective() + + with op.batch_alter_table("ScoreEntries") as batch_op: + batch_op.drop_column("scored_expectation") + + +def _backfill_scored_expectation() -> None: + """ + Fold every non-null legacy objective into an objective-only versioned expectation. + + Rows are read a page at a time, keyed on ``id``, so a large score table is never pulled + into memory at once. Scores with no objective keep a NULL expectation. + """ + connection = op.get_bind() + score_entries = sa.table( + "ScoreEntries", + sa.column("id"), + sa.column("objective"), + sa.column("scored_expectation"), + ) + statement = sa.text('UPDATE "ScoreEntries" SET scored_expectation = :scored_expectation WHERE id = :score_id') + + last_id = None + while True: + conditions = [ + score_entries.c.objective.isnot(None), + score_entries.c.scored_expectation.is_(None), + ] + if last_id is not None: + conditions.append(score_entries.c.id > last_id) + rows = connection.execute( + sa.select(score_entries.c.id, score_entries.c.objective) + .where(*conditions) + .order_by(score_entries.c.id) + .limit(_BACKFILL_BATCH_SIZE) + ).fetchall() + if not rows: + return + last_id = rows[-1][0] + + updates = [ + { + "score_id": score_id, + "scored_expectation": json.dumps( + {"schema_version": _SCHEMA_VERSION, "objective": objective, "conditions": []} + ), + } + for score_id, objective in rows + ] + connection.execute(statement, updates) + + +def _backfill_objective() -> None: + """ + Recover the objective string from every stored expectation. + + Typed conditions cannot be represented by the old ``objective`` column and are dropped. + Rows are read a page at a time, keyed on ``id``. + """ + connection = op.get_bind() + score_entries = sa.table( + "ScoreEntries", + sa.column("id"), + sa.column("objective"), + sa.column("scored_expectation"), + ) + statement = sa.text('UPDATE "ScoreEntries" SET objective = :objective WHERE id = :score_id') + + last_id = None + while True: + conditions = [ + score_entries.c.scored_expectation.isnot(None), + score_entries.c.objective.is_(None), + ] + if last_id is not None: + conditions.append(score_entries.c.id > last_id) + rows = connection.execute( + sa.select(score_entries.c.id, score_entries.c.scored_expectation) + .where(*conditions) + .order_by(score_entries.c.id) + .limit(_BACKFILL_BATCH_SIZE) + ).fetchall() + if not rows: + return + last_id = rows[-1][0] + + updates = [] + for score_id, scored_expectation in rows: + objective = _extract_objective(scored_expectation) + if objective is None: + continue + updates.append({"score_id": score_id, "objective": objective}) + if updates: + connection.execute(statement, updates) + + +def _extract_objective(scored_expectation: object) -> str | None: + """ + Read the objective out of a stored expectation, tolerating either a dict or JSON text. + + Args: + scored_expectation (object): The stored ``scored_expectation`` value. + + Returns: + str | None: The objective string, or ``None`` when absent or unparsable. + """ + if isinstance(scored_expectation, str): + try: + scored_expectation = json.loads(scored_expectation) + except (ValueError, TypeError): + return None + if isinstance(scored_expectation, dict): + objective = scored_expectation.get("objective") + return objective if isinstance(objective, str) else None + return None diff --git a/pyrit/memory/memory_models.py b/pyrit/memory/memory_models.py index 12a1dea54d..fe02ce8332 100644 --- a/pyrit/memory/memory_models.py +++ b/pyrit/memory/memory_models.py @@ -63,6 +63,7 @@ ScorerEvaluationIdentifier, ScorerIdentifier, ScoreStatus, + ScoringExpectation, Seed, SeedIdentifier, SeedObjective, @@ -1139,7 +1140,9 @@ class ScoreEntry(Base): ) prompt_request_response_id = mapped_column(CustomUUID, ForeignKey(f"{PromptMemoryEntry.__tablename__}.id")) timestamp = mapped_column(UTCDateTime, nullable=False) - objective = mapped_column(String, nullable=True) + # The full, versioned expectation this score was judged against (objective + conditions). + # Supersedes the legacy ``objective`` column. + scored_expectation: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) # Version of PyRIT used when this score was created # Nullable for backwards compatibility with existing databases pyrit_version = mapped_column(String, nullable=True) @@ -1152,7 +1155,7 @@ def __init__(self, *, entry: Score) -> None: Args: entry (Score): The score object to convert into a database entry. """ - entry = Score.model_validate(entry.model_dump()) + entry = Score.model_validate(entry.model_dump(exclude={"objective"})) self.id = entry.id self.score_value = entry.score_value self.score_value_description = entry.score_value_description @@ -1176,7 +1179,7 @@ def __init__(self, *, entry: Score) -> None: self.scorer_identifier_hash = normalized_scorer.hash if normalized_scorer else None self.prompt_request_response_id = entry.message_piece_id if entry.message_piece_id else None self.timestamp = entry.timestamp - self.objective = entry.objective + self.scored_expectation = entry.scored_expectation.model_dump(mode="json") if entry.scored_expectation else None self.pyrit_version = pyrit.__version__ def get_score(self) -> Score: @@ -1207,7 +1210,11 @@ def get_score(self) -> Score: message_piece_id=self.prompt_request_response_id, scorable=scorable_from_dict(self.scorable) if self.scorable else None, timestamp=self.timestamp, - objective=self.objective, + scored_expectation=( + ScoringExpectation.model_validate_persisted(self.scored_expectation) + if self.scored_expectation is not None + else None + ), ) def to_dict(self) -> dict[str, Any]: @@ -1231,7 +1238,8 @@ def to_dict(self) -> dict[str, Any]: "scorable_content_id": str(self.scorable_content_id) if self.scorable_content_id else None, "prompt_request_response_id": str(self.prompt_request_response_id), "timestamp": self.timestamp.isoformat() if self.timestamp else None, - "objective": self.objective, + "scored_expectation": self.scored_expectation, + "objective": self.scored_expectation.get("objective") if self.scored_expectation else None, } diff --git a/pyrit/models/__init__.py b/pyrit/models/__init__.py index 7a532f49df..2a4b2432e0 100644 --- a/pyrit/models/__init__.py +++ b/pyrit/models/__init__.py @@ -144,6 +144,7 @@ UndeterminedScoreError, UnvalidatedScore, scorable_from_dict, + scoring_expectation_fingerprint, ) from pyrit.models.seeds import ( AttackSeedGroup, @@ -320,6 +321,7 @@ "read_usage_int": "pyrit.models.target", "read_usage_value": "pyrit.models.target", "scorable_from_dict": "pyrit.models.score", + "scoring_expectation_fingerprint": "pyrit.models.score", "validate_registry_name": "pyrit.models.identifiers", "project_behavioral_identity": "pyrit.models.identifiers", "RetryEvent": "pyrit.models.retry_event", diff --git a/pyrit/models/score/__init__.py b/pyrit/models/score/__init__.py index b11db4c85b..2c6a6d5340 100644 --- a/pyrit/models/score/__init__.py +++ b/pyrit/models/score/__init__.py @@ -16,7 +16,10 @@ if TYPE_CHECKING: from pyrit.models.score.condition import Condition, MatchesObjective - from pyrit.models.score.expectation import ScoringExpectation + from pyrit.models.score.expectation import ( + ScoringExpectation, + scoring_expectation_fingerprint, + ) from pyrit.models.score.scorable import ( ContentEntryScorable, ContentScorable, @@ -50,6 +53,7 @@ "UndeterminedScoreError": "pyrit.models.score.score", "UnvalidatedScore": "pyrit.models.score.score", "scorable_from_dict": "pyrit.models.score.scorable", + "scoring_expectation_fingerprint": "pyrit.models.score.expectation", } __all__ = list(_LAZY_EXPORTS) diff --git a/pyrit/models/score/condition.py b/pyrit/models/score/condition.py index 25ddaeedad..7ea9311b7d 100644 --- a/pyrit/models/score/condition.py +++ b/pyrit/models/score/condition.py @@ -3,21 +3,144 @@ from __future__ import annotations -from abc import ABC -from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Literal, cast, get_args, get_origin +from pydantic import BaseModel, ConfigDict, model_validator -class Condition(ABC): # noqa: B024 root type; each scoring domain declares its own criterion +if TYPE_CHECKING: + from pydantic.config import ExtraValues + from typing_extensions import Self + +#: Maps each condition's stable discriminator to its type. A condition is persisted under +#: its ``condition_type`` discriminator rather than its import path, so a stored score survives +#: a class rename or a module move. Populated by ``Condition.__pydantic_init_subclass__``. +_CONDITION_TYPES: dict[str, type[Condition]] = {} + + +class Condition(BaseModel): """ What counts as satisfied. A condition is a neutral predicate about evidence: it says what to detect, never whether detecting it is good or bad. Polarity belongs to a scorer that wraps another, such as ``TrueFalseInverterScorer``. Each scoring domain adds its own subclass. + + A concrete subclass declares a ``condition_type`` field as a single-value ``Literal`` with + a matching default. That default is the stable discriminator persisted with the condition + and carried in REST payloads, so the type survives serialization without its import path. """ + model_config = ConfigDict(frozen=True, extra="forbid") + + condition_type: str + + @classmethod + def __pydantic_init_subclass__(cls, **kwargs: Any) -> None: + """ + Register the subclass under its stable discriminator. + + Args: + **kwargs (Any): Forwarded to ``super().__pydantic_init_subclass__``. + + Raises: + TypeError: If the discriminator declaration is invalid. + ValueError: If the discriminator already names a different condition. + """ + super().__pydantic_init_subclass__(**kwargs) + field = cls.model_fields.get("condition_type") + literal_values = ( + get_args(field.annotation) if field is not None and get_origin(field.annotation) is Literal else () + ) + if len(literal_values) != 1 or not isinstance(literal_values[0], str) or not literal_values[0]: + raise TypeError( + f"{cls.__name__}.condition_type must be a single non-empty string Literal with a matching default." + ) + discriminator = literal_values[0] + if field.default != discriminator: + raise TypeError(f"{cls.__name__}.condition_type must default to its Literal value {discriminator!r}.") + registered = _CONDITION_TYPES.get(discriminator) + if registered is not None and registered is not cls: + raise ValueError( + f"Condition discriminator {discriminator!r} is already registered to " + f"{registered.__name__}; give {cls.__name__} a distinct condition_type." + ) + _CONDITION_TYPES[discriminator] = cls + + @model_validator(mode="after") + def _reject_base_condition(self) -> Condition: + """ + Reject the untyped registry root as a concrete condition. + + Returns: + Condition: The validated concrete condition. + + Raises: + ValueError: If the registry root is instantiated directly. + """ + if type(self) is Condition: + raise ValueError("Condition is an abstract registry root and cannot be instantiated directly.") + return self + + @classmethod + def model_validate( + cls, + obj: Any, + *, + strict: bool | None = None, + extra: ExtraValues | None = None, + from_attributes: bool | None = None, + context: Any | None = None, + by_alias: bool | None = None, + by_name: bool | None = None, + ) -> Self: + """ + Validate a condition, dispatching the registry root to its concrete subtype. + + Args: + obj (Any): The condition instance or discriminator-tagged representation. + strict (bool | None): Whether Pydantic uses strict validation. + extra (ExtraValues | None): How Pydantic handles extra fields. + from_attributes (bool | None): Whether Pydantic reads object attributes. + context (Any | None): Context supplied to Pydantic validators. + by_alias (bool | None): Whether Pydantic accepts field aliases. + by_name (bool | None): Whether Pydantic accepts field names. + + Returns: + Self: The validated concrete condition. + + Raises: + ValueError: If the abstract root receives an invalid or unknown discriminator. + """ + if cls is Condition and isinstance(obj, dict): + discriminator = obj.get("condition_type") + if not isinstance(discriminator, str): + raise ValueError("Condition requires a string condition_type discriminator.") + condition_type = _CONDITION_TYPES.get(discriminator) + if condition_type is None: + raise ValueError(f"Unknown condition_type {discriminator!r}.") + return cast( + "Self", + condition_type.model_validate( + obj, + strict=strict, + extra=extra, + from_attributes=from_attributes, + context=context, + by_alias=by_alias, + by_name=by_name, + ), + ) + return super().model_validate( + obj, + strict=strict, + extra=extra, + from_attributes=from_attributes, + context=context, + by_alias=by_alias, + by_name=by_name, + ) + -@dataclass(frozen=True, kw_only=True) class MatchesObjective(Condition): """ The evidence satisfies the expectation's own objective, as a judge reads it. @@ -26,3 +149,5 @@ class MatchesObjective(Condition): so a scorer matching this condition reads it from there and the two can never disagree. """ + + condition_type: Literal["matches_objective"] = "matches_objective" diff --git a/pyrit/models/score/expectation.py b/pyrit/models/score/expectation.py index b795a17855..58d2258825 100644 --- a/pyrit/models/score/expectation.py +++ b/pyrit/models/score/expectation.py @@ -3,13 +3,23 @@ from __future__ import annotations -from dataclasses import dataclass, field +import hashlib +import json +from typing import TYPE_CHECKING, Any, Literal -from pyrit.models.score.condition import Condition # noqa: TC001 (runtime-required by dataclass field annotations) +from pydantic import BaseModel, ConfigDict, SerializeAsAny, TypeAdapter, field_validator, model_validator +from pyrit.models.score.condition import _CONDITION_TYPES, Condition -@dataclass(frozen=True, kw_only=True) -class ScoringExpectation: +if TYPE_CHECKING: + from pydantic import GetJsonSchemaHandler, ValidationInfo + from pydantic.json_schema import JsonSchemaValue + from pydantic_core import CoreSchema + +_PERSISTED_VALIDATION_CONTEXT = "require_scoring_expectation_schema_version" + + +class ScoringExpectation(BaseModel): """ What a scorer scores against. @@ -23,8 +33,141 @@ class ScoringExpectation: ``conditions`` carry the criteria: typed objects routed by type to the scorers that match them. Attacks forward them without inspecting them, and a scorer matches at - most one of them. + most one of them. ``SerializeAsAny`` keeps each condition serialized as its own + subtype, so subclass fields survive a round trip. """ + model_config = ConfigDict(frozen=True, extra="forbid") + + #: Version of the persisted shape. Bumped only when the serialized dict changes in a way an + #: older reader cannot understand; the validator rejects any other value on load. + schema_version: Literal[1] = 1 + objective: str | None = None - conditions: tuple[Condition, ...] = field(default_factory=tuple) + conditions: tuple[SerializeAsAny[Condition], ...] = () + + @model_validator(mode="before") + @classmethod + def _require_persisted_schema_version(cls, value: Any, info: ValidationInfo) -> Any: + """ + Require the version field when validating a persisted representation. + + Args: + value (Any): The incoming expectation representation. + info (ValidationInfo): Pydantic validation metadata. + + Returns: + Any: The unchanged representation. + + Raises: + ValueError: If a persisted representation omits its schema version. + """ + if ( + info.context + and info.context.get(_PERSISTED_VALIDATION_CONTEXT) + and (not isinstance(value, dict) or "schema_version" not in value) + ): + raise ValueError("Persisted ScoringExpectation requires an explicit schema_version.") + return value + + @field_validator("schema_version", mode="before") + @classmethod + def _check_schema_version(cls, value: Any) -> Any: + """ + Reject a schema version this model cannot read without coercion. + + Args: + value (Any): The incoming schema version. + + Returns: + Any: The validated version. + + Raises: + ValueError: If the version is not the exact integer this model understands. + """ + if type(value) is not int or value != 1: + raise ValueError(f"Unsupported ScoringExpectation schema_version {value!r}; expected 1.") + return value + + @field_validator("conditions", mode="before") + @classmethod + def _rebuild_conditions(cls, value: Any) -> Any: + """ + Rebuild serialized conditions into their concrete subtypes. + + Args: + value (Any): The incoming conditions: an iterable of ``Condition`` instances or of + serialized, discriminator-tagged dicts. + + Returns: + Any: A tuple of conditions, with any dict routed through the condition registry. + + Raises: + ValueError: If a serialized condition names an unknown discriminator. + """ + if value is None: + return () + return tuple(Condition.model_validate(item) if isinstance(item, dict) else item for item in value) + + @classmethod + def model_validate_persisted(cls, value: Any) -> ScoringExpectation: + """ + Validate an expectation loaded from its durable, versioned representation. + + Args: + value (Any): The persisted expectation representation. + + Returns: + ScoringExpectation: The validated expectation. + """ + return cls.model_validate(value, context={_PERSISTED_VALIDATION_CONTEXT: True}) + + @classmethod + def __get_pydantic_json_schema__( + cls, + core_schema: CoreSchema, + handler: GetJsonSchemaHandler, + ) -> JsonSchemaValue: + """ + Describe every registered condition subtype in the generated schema. + + Returns: + JsonSchemaValue: The expectation schema with a discriminated condition union. + """ + schema = handler(core_schema) + resolved_schema = handler.resolve_ref_schema(schema) + condition_schemas = [ + handler(TypeAdapter(condition_class).core_schema) for _, condition_class in sorted(_CONDITION_TYPES.items()) + ] + for condition_schema in condition_schemas: + required_fields = condition_schema.setdefault("required", []) + if "condition_type" not in required_fields: + required_fields.append("condition_type") + resolved_schema["properties"]["conditions"]["items"] = { + "discriminator": {"propertyName": "condition_type"}, + "oneOf": condition_schemas, + } + return schema + + +def scoring_expectation_fingerprint(exp: ScoringExpectation) -> str: + """ + Return a stable content fingerprint of an expectation. + + The fingerprint is the lowercase SHA-256 hex of the canonical JSON serialization + (sorted keys, compact separators), so two expectations with the same objective and + conditions hash identically regardless of construction order. + + Args: + exp (ScoringExpectation): The expectation to fingerprint. + + Returns: + str: The lowercase SHA-256 hex digest. + """ + serialized = json.dumps( + exp.model_dump(mode="json"), + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() diff --git a/pyrit/models/score/score.py b/pyrit/models/score/score.py index 01ec7d8795..04737a757a 100644 --- a/pyrit/models/score/score.py +++ b/pyrit/models/score/score.py @@ -22,6 +22,7 @@ ) from pyrit.models.identifiers.component_identifier import ComponentIdentifier +from pyrit.models.score.expectation import ScoringExpectation from pyrit.models.score.scorable import ( # noqa: TC001 (runtime-required by Pydantic field annotations) MessageScorable, ScorableUnion, @@ -111,12 +112,71 @@ class Score(BaseModel): # Timestamp of when the score was created timestamp: AwareDatetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc)) - # The task based on which the text is scored (the original attacker model's objective). - objective: str | None = None + # The full, versioned expectation this score was judged against (objective + conditions). + # This is the durable record of what the score was scored for. + scored_expectation: ScoringExpectation | None = Field(default=None, frozen=True) + + # Derived, read-only compatibility view over ``scored_expectation.objective``. Existing + # readers that expect a bare objective keep working; it is set from the expectation, and an + # explicit value that disagrees with the expectation is rejected. + objective: str | None = Field(default=None, frozen=True) # ------------------------------------------------------------------ # # Validators # ------------------------------------------------------------------ # + @model_validator(mode="before") + @classmethod + def _validate_compatibility_objective(cls, data: Any) -> Any: + """ + Fold a legacy ``objective`` input into ``scored_expectation``. + + ``objective`` is a read-only view now, so an incoming value is never set directly. It + becomes an objective-only expectation when none is supplied, and otherwise must agree + with the expectation it accompanies. + + Args: + data (Any): Raw input to the model. + + Returns: + Any: The input with ``objective`` removed and ``scored_expectation`` populated. + + Raises: + ValueError: If an explicit objective disagrees with ``scored_expectation``. + """ + if not isinstance(data, dict) or "objective" not in data: + return data + data = dict(data) + objective = data.pop("objective") + expectation = data.get("scored_expectation") + if expectation is None: + if objective is not None: + data["scored_expectation"] = ScoringExpectation(objective=objective) + else: + if isinstance(expectation, ScoringExpectation): + expectation_objective = expectation.objective + elif isinstance(expectation, dict): + expectation_objective = expectation.get("objective") + else: + expectation_objective = None + if objective != expectation_objective: + raise ValueError( + f"objective {objective!r} conflicts with scored_expectation.objective {expectation_objective!r}." + ) + return data + + @field_validator("scored_expectation", mode="before") + @classmethod + def _load_scored_expectation(cls, value: Any) -> Any: + """ + Rebuild a scored expectation from its persisted representation. + + Returns: + Any: The validated expectation or the unchanged input. + """ + if isinstance(value, dict): + return ScoringExpectation.model_validate_persisted(value) + return value + @field_validator("score_metadata", mode="before") @classmethod def _default_metadata(cls, value: Any) -> Any: @@ -170,6 +230,17 @@ def _reconcile_message_piece_id(self) -> Score: ) return self + @model_validator(mode="after") + def _sync_compatibility_objective(self) -> Score: + """ + Refresh the derived, read-only ``objective`` view from ``scored_expectation``. + + Returns: + Score: ``self`` with ``objective`` mirroring ``scored_expectation.objective``. + """ + object.__setattr__(self, "objective", self.scored_expectation.objective if self.scored_expectation else None) + return self + def _check_score_value(self) -> None: """ Validate ``score_value`` against ``status`` and ``score_type`` constraints. @@ -265,11 +336,30 @@ class UnvalidatedScore: score_metadata: dict[str, str | int | float] | None scorer_class_identifier: ComponentIdentifier message_piece_id: uuid.UUID | str | None - objective: str | None + objective: str | None = None + scored_expectation: ScoringExpectation | None = None id: uuid.UUID | str | None = None timestamp: datetime | None = None scorable: ScorableUnion | None = None + def __post_init__(self) -> None: + """ + Keep ``objective`` and ``scored_expectation`` consistent. + + Raises: + ValueError: If an explicit objective disagrees with ``scored_expectation``. + """ + if self.scored_expectation is None: + if self.objective is not None: + self.scored_expectation = ScoringExpectation(objective=self.objective) + elif self.objective is None: + self.objective = self.scored_expectation.objective + elif self.objective != self.scored_expectation.objective: + raise ValueError( + f"objective {self.objective!r} conflicts with scored_expectation.objective " + f"{self.scored_expectation.objective!r}." + ) + def to_score(self, *, score_value: str, score_type: ScoreType) -> Score: """ Convert this unvalidated score into a validated Score. @@ -294,5 +384,6 @@ def to_score(self, *, score_value: str, score_type: ScoreType) -> Score: message_piece_id=self.message_piece_id, scorable=self.scorable, timestamp=self.timestamp if self.timestamp else datetime.now(tz=timezone.utc), - objective=self.objective, + scored_expectation=self.scored_expectation + or (ScoringExpectation(objective=self.objective) if self.objective is not None else None), ) diff --git a/pyrit/score/message_scorer.py b/pyrit/score/message_scorer.py index 2dc68291d9..4763f4a6a1 100644 --- a/pyrit/score/message_scorer.py +++ b/pyrit/score/message_scorer.py @@ -791,7 +791,9 @@ async def _score_resolved_message_async( if scoring_message is None: scores = self._build_fallback_score(message=message, objective=objective) - self._finalize_message_scores(message=message, scores=scores, anchor=anchor) + self._finalize_message_scores( + message=message, scores=scores, anchor=anchor, expectation=effective_expectation + ) return scores self._validate_scoring_message(message=scoring_message, objective=objective) @@ -839,7 +841,9 @@ async def _score_resolved_message_async( if not scores and scoring_message.message_pieces and not self._get_supported_pieces(scoring_message): scores = self._build_fallback_score(message=message, objective=objective) - self._finalize_message_scores(message=scoring_message, scores=scores, anchor=anchor) + self._finalize_message_scores( + message=scoring_message, scores=scores, anchor=anchor, expectation=effective_expectation + ) return scores @@ -859,8 +863,9 @@ def _finalize_message_scores( message: Message, scores: list[Score], anchor: Scorable | None, + expectation: ScoringExpectation | None, ) -> None: - """Apply legacy and canonical evidence anchors to completed message scores.""" + """Apply evidence anchors and record the expectation on completed message scores.""" persisted_piece_ids = self._get_persisted_piece_ids(message=message) if anchor is None else None self._drop_ephemeral_score_links( message=message, @@ -873,6 +878,7 @@ def _finalize_message_scores( anchor=anchor, persisted_piece_ids=persisted_piece_ids, ) + self._stamp_scored_expectation(scores=scores, expectation=expectation) async def _score_prepared_message_async( self, diff --git a/pyrit/score/scorer.py b/pyrit/score/scorer.py index 6b148a6939..3215052b37 100644 --- a/pyrit/score/scorer.py +++ b/pyrit/score/scorer.py @@ -330,8 +330,28 @@ async def score_async( raise except Exception as e: raise RuntimeError(f"Error in scorer {self.__class__.__name__}: {str(e)}") from e + self._stamp_scored_expectation(scores=scores, expectation=expectation) return await self._validate_and_persist_scores_async(scores=scores) + @staticmethod + def _stamp_scored_expectation(*, scores: list[Score], expectation: ScoringExpectation | None) -> None: + """ + Record on each score the expectation it was judged against. + + The scorer, not the score, knows the expectation it used, so it stamps the finished + scores before they persist. ``objective`` is the derived view, so it is refreshed to + match. A ``None`` expectation leaves the scores unchanged. + + Args: + scores (list[Score]): The scores to stamp. + expectation (ScoringExpectation | None): The expectation the scorer used. + """ + if expectation is None: + return + for score in scores: + object.__setattr__(score, "scored_expectation", expectation) + object.__setattr__(score, "objective", expectation.objective) + def _validate_expectation( self, *, diff --git a/tests/unit/executor/attack/component/test_simulated_conversation.py b/tests/unit/executor/attack/component/test_simulated_conversation.py index e14909ca40..fcd7000e6e 100644 --- a/tests/unit/executor/attack/component/test_simulated_conversation.py +++ b/tests/unit/executor/attack/component/test_simulated_conversation.py @@ -290,7 +290,7 @@ async def test_returns_simulated_conversation_result( ): """Test that the function returns a list of SeedPrompts.""" conversation_id = str(uuid.uuid4()) - mock_score = MagicMock(spec=Score) + mock_score = MagicMock(spec=Score, scored_expectation=None) with patch("pyrit.executor.attack.multi_turn.simulated_conversation.RedTeamingAttack") as mock_attack_class: mock_attack = MagicMock() diff --git a/tests/unit/executor/attack/multi_turn/test_tree_of_attacks.py b/tests/unit/executor/attack/multi_turn/test_tree_of_attacks.py index ad88a9346b..62d7a21c5b 100644 --- a/tests/unit/executor/attack/multi_turn/test_tree_of_attacks.py +++ b/tests/unit/executor/attack/multi_turn/test_tree_of_attacks.py @@ -160,6 +160,7 @@ def create_node(config: NodeMockConfig | None = None) -> "_TreeOfAttacksNode": get_value=MagicMock(return_value=config.objective_score_value), is_undetermined=False, score_metadata=None, + scored_expectation=None, ) else: node.objective_score = None @@ -2881,12 +2882,14 @@ async def _send_prompt(objective: str) -> None: spec=Score, get_value=MagicMock(return_value=0.0), score_metadata=None, + scored_expectation=None, ) elif b.score is not None: node.objective_score = MagicMock( spec=Score, get_value=MagicMock(return_value=b.score), score_metadata=None, + scored_expectation=None, ) node = MagicMock() diff --git a/tests/unit/executor/workflow/test_xpia.py b/tests/unit/executor/workflow/test_xpia.py index 6badb1bae8..8a6beaac82 100644 --- a/tests/unit/executor/workflow/test_xpia.py +++ b/tests/unit/executor/workflow/test_xpia.py @@ -36,6 +36,13 @@ def _mock_target_id(name: str = "MockTarget") -> ComponentIdentifier: ) +def _mock_score(*, value: object, is_undetermined: bool = False) -> MagicMock: + """Create a Score mock with the model fields used during Pydantic validation.""" + score = MagicMock(spec=Score, scored_expectation=None, is_undetermined=is_undetermined) + score.get_value.return_value = value + return score + + @pytest.fixture def mock_attack_setup_target() -> MagicMock: """Create a mock attack setup target.""" @@ -197,8 +204,7 @@ async def test_perform_async_complete_workflow_with_scorer( mock_response.get_value.return_value = "Attack setup response" mock_prompt_normalizer.send_prompt_async.return_value = mock_response - mock_score = MagicMock(spec=Score) - mock_score.get_value.return_value = 0.8 + mock_score = _mock_score(value=0.8) mock_scorer.score_text_async.return_value = [mock_score] # Execute workflow @@ -539,8 +545,7 @@ class TestXPIAResult: def test_success_property_with_positive_score(self) -> None: """Test success property returns True for positive score.""" - mock_score = MagicMock(spec=Score) - mock_score.get_value.return_value = 0.8 + mock_score = _mock_score(value=0.8) result = XPIAResult(processing_conversation_id="test-id", processing_response="test response", score=mock_score) @@ -548,8 +553,7 @@ def test_success_property_with_positive_score(self) -> None: def test_success_property_with_zero_score(self) -> None: """Test success property returns False for zero score.""" - mock_score = MagicMock(spec=Score) - mock_score.get_value.return_value = 0.0 + mock_score = _mock_score(value=0.0) result = XPIAResult(processing_conversation_id="test-id", processing_response="test response", score=mock_score) @@ -557,8 +561,7 @@ def test_success_property_with_zero_score(self) -> None: def test_success_property_with_negative_score(self) -> None: """Test success property returns False for negative score.""" - mock_score = MagicMock(spec=Score) - mock_score.get_value.return_value = -0.5 + mock_score = _mock_score(value=-0.5) result = XPIAResult(processing_conversation_id="test-id", processing_response="test response", score=mock_score) @@ -572,8 +575,7 @@ def test_success_property_with_no_score(self) -> None: def test_success_property_with_non_numeric_score(self) -> None: """Test success property returns False for non-numeric score.""" - mock_score = MagicMock(spec=Score) - mock_score.get_value.return_value = "invalid" + mock_score = _mock_score(value="invalid") result = XPIAResult(processing_conversation_id="test-id", processing_response="test response", score=mock_score) @@ -581,9 +583,7 @@ def test_success_property_with_non_numeric_score(self) -> None: def test_status_property_success(self) -> None: """Test status property returns SUCCESS for successful attack.""" - mock_score = MagicMock(spec=Score) - mock_score.get_value.return_value = 0.8 - mock_score.is_undetermined = False + mock_score = _mock_score(value=0.8) result = XPIAResult(processing_conversation_id="test-id", processing_response="test response", score=mock_score) @@ -591,9 +591,7 @@ def test_status_property_success(self) -> None: def test_status_property_failure(self) -> None: """Test status property returns FAILURE for failed attack.""" - mock_score = MagicMock(spec=Score) - mock_score.get_value.return_value = 0.0 - mock_score.is_undetermined = False + mock_score = _mock_score(value=0.0) result = XPIAResult(processing_conversation_id="test-id", processing_response="test response", score=mock_score) diff --git a/tests/unit/memory/test_memory_models.py b/tests/unit/memory/test_memory_models.py index ba81e4f4a8..26060d480a 100644 --- a/tests/unit/memory/test_memory_models.py +++ b/tests/unit/memory/test_memory_models.py @@ -461,7 +461,7 @@ def test_init_from_score(self): assert entry.id == score.id assert entry.score_value == "0.9" assert entry.score_type == "float_scale" - assert entry.objective == "test objective" + assert entry.scored_expectation == {"schema_version": 1, "objective": "test objective", "conditions": []} def test_roundtrip_get_score(self): score = _make_score() diff --git a/tests/unit/memory/test_migration.py b/tests/unit/memory/test_migration.py index ccbbb0f500..a5973d6d95 100644 --- a/tests/unit/memory/test_migration.py +++ b/tests/unit/memory/test_migration.py @@ -2522,3 +2522,112 @@ def test_migrations_do_not_use_unbounded_string_primary_keys() -> None: assert not violations, "Found unbounded string primary keys in migrations (SQL Server incompatible):\n" + "\n".join( violations ) + + +# --------------------------------------------------------------------------- # +# scored_expectation migration (1b3d5f7a9c2e) +# --------------------------------------------------------------------------- # +_SCORED_EXPECTATION_REV = "1b3d5f7a9c2e" +_SCORED_EXPECTATION_PREV_REV = "0f2e4d6c8b1a" + + +def _seed_pre_scored_expectation_score(connection, *, score_id, objective): + connection.execute( + text( + 'INSERT INTO "ScoreEntries" ' + "(id, score_value, status, score_type, score_metadata, scorer_class_identifier, timestamp, objective) " + "VALUES (:id, 'true', 'complete', 'true_false', '{}', '{}', '2026-01-01 00:00:00', :objective)" + ), + {"id": score_id, "objective": objective}, + ) + + +def test_scored_expectation_migration_script_metadata(): + """The scored_expectation migration declares the expected revision chain.""" + import importlib + + mig = importlib.import_module("pyrit.memory.alembic.versions.1b3d5f7a9c2e_persist_scored_expectation") + + assert mig.revision == _SCORED_EXPECTATION_REV + assert mig.down_revision == _SCORED_EXPECTATION_PREV_REV + assert mig.branch_labels is None + assert mig.depends_on is None + + +def test_scored_expectation_upgrade_backfills_objective_into_expectation(): + """Upgrading folds a non-null objective into a versioned expectation and leaves NULLs NULL.""" + id_with = str(uuid.uuid4()) + id_without = str(uuid.uuid4()) + + with tempfile.TemporaryDirectory() as temp_dir: + engine = create_engine(f"sqlite:///{os.path.join(temp_dir, 'scored-exp-up.db')}") + try: + with engine.begin() as connection: + config = _config_for(connection) + command.upgrade(config, _SCORED_EXPECTATION_PREV_REV) + _seed_pre_scored_expectation_score(connection, score_id=id_with, objective="legacy objective") + _seed_pre_scored_expectation_score(connection, score_id=id_without, objective=None) + + command.upgrade(config, _SCORED_EXPECTATION_REV) + + columns = {column["name"] for column in inspect(connection).get_columns("ScoreEntries")} + assert "scored_expectation" in columns + assert "objective" not in columns + + rows = dict(connection.execute(text('SELECT id, scored_expectation FROM "ScoreEntries"')).fetchall()) + assert json.loads(rows[id_with]) == { + "schema_version": 1, + "objective": "legacy objective", + "conditions": [], + } + assert rows[id_without] is None + finally: + engine.dispose() + + +def test_scored_expectation_downgrade_recovers_objective_and_drops_conditions(): + """Downgrading recovers the objective string and drops condition-only expectations.""" + id_objective = str(uuid.uuid4()) + id_condition_only = str(uuid.uuid4()) + + with tempfile.TemporaryDirectory() as temp_dir: + engine = create_engine(f"sqlite:///{os.path.join(temp_dir, 'scored-exp-down.db')}") + try: + with engine.begin() as connection: + config = _config_for(connection) + command.upgrade(config, _SCORED_EXPECTATION_REV) + + objective_expectation = json.dumps({"schema_version": 1, "objective": "recover me", "conditions": []}) + condition_only_expectation = json.dumps( + { + "schema_version": 1, + "objective": None, + "conditions": [{"condition_type": "matches_objective"}], + } + ) + insert = ( + 'INSERT INTO "ScoreEntries" ' + "(id, score_value, status, score_type, score_metadata, scorer_class_identifier, timestamp, " + "scored_expectation) " + "VALUES (:id, 'true', 'complete', 'true_false', '{}', '{}', '2026-01-01 00:00:00', :exp)" + ) + connection.execute( + text(insert), + [ + {"id": id_objective, "exp": objective_expectation}, + {"id": id_condition_only, "exp": condition_only_expectation}, + ], + ) + + command.downgrade(config, _SCORED_EXPECTATION_PREV_REV) + + columns = {column["name"] for column in inspect(connection).get_columns("ScoreEntries")} + assert "objective" in columns + assert "scored_expectation" not in columns + + objectives = dict(connection.execute(text('SELECT id, objective FROM "ScoreEntries"')).fetchall()) + assert objectives[id_objective] == "recover me" + # A condition-only expectation has no objective the old schema can hold, so it is dropped. + assert objectives[id_condition_only] is None + finally: + engine.dispose() diff --git a/tests/unit/memory/test_score_entry.py b/tests/unit/memory/test_score_entry.py index 2e6a1d48a1..77f34ebb00 100644 --- a/tests/unit/memory/test_score_entry.py +++ b/tests/unit/memory/test_score_entry.py @@ -2,11 +2,13 @@ # Licensed under the MIT license. import uuid +from typing import Any import pytest +from pydantic import ValidationError from pyrit.memory.memory_models import ScoreEntry -from pyrit.models import ComponentIdentifier, Score +from pyrit.models import ComponentIdentifier, MatchesObjective, Score, ScoringExpectation @pytest.mark.usefixtures("patch_central_database") @@ -177,6 +179,67 @@ def test_score_entry_to_dict(self): assert result["scorer_class_identifier"][ComponentIdentifier.KEY_CLASS_NAME] == "TestScorer" assert result["objective"] == "objective" + def test_score_entry_roundtrip_scored_expectation_with_conditions(self): + """A ScoreEntry preserves the full versioned expectation, including typed conditions.""" + expectation = ScoringExpectation(objective="obj", conditions=(MatchesObjective(),)) + score = Score( + score_value="true", + score_type="true_false", + scorer_class_identifier=ComponentIdentifier(class_name="X", class_module="pyrit.score"), + message_piece_id=uuid.uuid4(), + scored_expectation=expectation, + ) + + entry = ScoreEntry(entry=score) + + assert entry.scored_expectation == { + "schema_version": 1, + "objective": "obj", + "conditions": [{"condition_type": "matches_objective"}], + } + retrieved = entry.get_score() + assert retrieved.scored_expectation == expectation + assert retrieved.objective == "obj" + + def test_score_entry_without_expectation_roundtrips_as_none(self): + """A score with no expectation stores and restores a NULL scored_expectation.""" + score = Score( + score_value="true", + score_type="true_false", + scorer_class_identifier=ComponentIdentifier(class_name="X", class_module="pyrit.score"), + message_piece_id=uuid.uuid4(), + ) + + entry = ScoreEntry(entry=score) + + assert entry.scored_expectation is None + assert entry.get_score().scored_expectation is None + assert entry.to_dict()["objective"] is None + + @pytest.mark.parametrize("stored_expectation", [{}, []]) + def test_score_entry_rejects_falsey_malformed_expectation(self, stored_expectation: Any): + """Any non-NULL stored expectation must pass strict schema validation.""" + entry = ScoreEntry(entry=Score(score_value="true", score_type="true_false")) + entry.scored_expectation = stored_expectation + + with pytest.raises(ValidationError, match="requires an explicit schema_version"): + entry.get_score() + + def test_score_entry_to_dict_derives_objective_from_expectation(self): + """to_dict emits both the stored expectation and the derived objective view.""" + score = Score( + score_value="true", + score_type="true_false", + scorer_class_identifier=ComponentIdentifier(class_name="X", class_module="pyrit.score"), + message_piece_id=uuid.uuid4(), + objective="derived-obj", + ) + + result = ScoreEntry(entry=score).to_dict() + + assert result["objective"] == "derived-obj" + assert result["scored_expectation"]["objective"] == "derived-obj" + def test_score_to_dict_serializes_scorer_identifier(self): """Test that Score.model_dump() properly serializes the ComponentIdentifier.""" scorer_identifier = ComponentIdentifier( diff --git a/tests/unit/models/test_condition.py b/tests/unit/models/test_condition.py new file mode 100644 index 0000000000..4e1eceada4 --- /dev/null +++ b/tests/unit/models/test_condition.py @@ -0,0 +1,101 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +from typing import Literal + +import pytest +from pydantic import ValidationError + +from pyrit.models import Condition, MatchesObjective +from pyrit.models.score.condition import _CONDITION_TYPES + + +class _KeywordCondition(Condition): + condition_type: Literal["test_keyword_condition"] = "test_keyword_condition" + keyword: str + + +def test_matches_objective_has_stable_discriminator(): + assert MatchesObjective().condition_type == "matches_objective" + assert _CONDITION_TYPES["matches_objective"] is MatchesObjective + + +def test_explicit_discriminator_is_registered(): + assert _KeywordCondition(keyword="k").condition_type == "test_keyword_condition" + assert _CONDITION_TYPES["test_keyword_condition"] is _KeywordCondition + + +def test_condition_subclass_requires_discriminator_field(): + with pytest.raises(TypeError, match="single non-empty string Literal"): + + class _MissingDiscriminatorCondition(Condition): + threshold: float = 0.5 + + +def test_condition_subclass_requires_single_literal_value(): + with pytest.raises(TypeError, match="single non-empty string Literal"): + + class _MultipleDiscriminatorCondition(Condition): + condition_type: Literal["first", "second"] = "first" + + +def test_condition_subclass_requires_matching_literal_default(): + with pytest.raises(TypeError, match="must default to its Literal value"): + + class _MismatchedDiscriminatorCondition(Condition): + condition_type: Literal["expected"] = "different" # type: ignore[assignment] + + +def test_condition_base_cannot_be_instantiated(): + with pytest.raises(ValidationError, match="abstract registry root"): + Condition(condition_type="base") + + +def test_condition_is_frozen(): + condition = MatchesObjective() + + with pytest.raises(ValidationError): + condition.condition_type = "something" + + +def test_matches_objective_serializes_to_discriminator_only(): + assert MatchesObjective().model_dump() == {"condition_type": "matches_objective"} + + +def test_condition_round_trip_preserves_subclass_fields(): + condition = _KeywordCondition(keyword="secret") + + serialized = condition.model_dump() + + assert serialized == {"condition_type": "test_keyword_condition", "keyword": "secret"} + assert Condition.model_validate(serialized) == condition + + +def test_condition_model_validate_rejects_unknown_type(): + with pytest.raises(ValueError, match="Unknown condition_type 'nope'"): + Condition.model_validate({"condition_type": "nope"}) + + +@pytest.mark.parametrize("value", [{}, {"condition_type": 1}]) +def test_condition_model_validate_rejects_missing_or_non_string_type(value): + with pytest.raises(ValueError, match="requires a string condition_type"): + Condition.model_validate(value) + + +def test_condition_model_validate_rejects_extra_fields(): + with pytest.raises(ValidationError, match="Extra inputs are not permitted"): + Condition.model_validate({"condition_type": "matches_objective", "unexpected": True}) + + +def test_matches_objective_instances_compare_equal(): + assert MatchesObjective() == MatchesObjective() + + +def test_duplicate_discriminator_is_rejected(): + with pytest.raises((ValueError, TypeError), match="already registered"): + + class _First(Condition): + condition_type: Literal["test_duplicate_discriminator"] = "test_duplicate_discriminator" + + class _Second(Condition): + condition_type: Literal["test_duplicate_discriminator"] = "test_duplicate_discriminator" diff --git a/tests/unit/models/test_expectation.py b/tests/unit/models/test_expectation.py index 2fc5a2e2df..6290f622ff 100644 --- a/tests/unit/models/test_expectation.py +++ b/tests/unit/models/test_expectation.py @@ -1,11 +1,26 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. -import dataclasses +from typing import Literal import pytest +from pydantic import ValidationError -from pyrit.models import Condition, MatchesObjective, ScoringExpectation +from pyrit.models import ( + Condition, + MatchesObjective, + ScoringExpectation, + scoring_expectation_fingerprint, +) + + +class _AlphaCondition(Condition): + condition_type: Literal["test_expectation_alpha"] = "test_expectation_alpha" + + +class _BetaCondition(Condition): + condition_type: Literal["test_expectation_beta"] = "test_expectation_beta" + label: str = "b" def test_expectation_defaults(): @@ -18,7 +33,7 @@ def test_expectation_defaults(): def test_expectation_is_frozen(): expectation = ScoringExpectation(objective="exfiltrate") - with pytest.raises(dataclasses.FrozenInstanceError): + with pytest.raises(ValidationError): expectation.objective = "something else" @@ -44,10 +59,6 @@ def test_expectations_differing_only_in_conditions_compare_unequal(): assert ScoringExpectation(objective="a") != ScoringExpectation(objective="a", conditions=(MatchesObjective(),)) -def test_matches_objective_carries_no_text_of_its_own(): - assert dataclasses.fields(MatchesObjective()) == () - - def test_matches_objective_is_a_condition(): assert isinstance(MatchesObjective(), Condition) @@ -56,8 +67,123 @@ def test_matches_objective_instances_compare_equal(): assert MatchesObjective() == MatchesObjective() -def test_matches_objective_is_frozen(): - condition = MatchesObjective() +# --------------------------------------------------------------------------- # +# Versioned serialization (native Pydantic model_dump / model_validate) +# --------------------------------------------------------------------------- # +def test_objective_only_round_trip(): + expectation = ScoringExpectation(objective="do x") + + serialized = expectation.model_dump(mode="json") + + assert serialized == {"schema_version": 1, "objective": "do x", "conditions": []} + assert ScoringExpectation.model_validate_persisted(serialized) == expectation + + +def test_condition_only_round_trip(): + expectation = ScoringExpectation(conditions=(MatchesObjective(),)) + + serialized = expectation.model_dump(mode="json") + + assert serialized == { + "schema_version": 1, + "objective": None, + "conditions": [{"condition_type": "matches_objective"}], + } + assert ScoringExpectation.model_validate_persisted(serialized) == expectation + + +def test_mixed_round_trip(): + expectation = ScoringExpectation(objective="do x", conditions=(MatchesObjective(),)) + + assert ScoringExpectation.model_validate_persisted(expectation.model_dump(mode="json")) == expectation + + +def test_conditions_serialize_in_order(): + expectation = ScoringExpectation(conditions=(_AlphaCondition(), _BetaCondition(label="x"))) + + serialized = expectation.model_dump(mode="json") + + assert [condition["condition_type"] for condition in serialized["conditions"]] == [ + "test_expectation_alpha", + "test_expectation_beta", + ] + assert ScoringExpectation.model_validate_persisted(serialized) == expectation + + +def test_validate_rejects_unknown_schema_version(): + with pytest.raises(ValidationError, match="Unsupported ScoringExpectation schema_version"): + ScoringExpectation.model_validate({"schema_version": 99, "objective": None, "conditions": []}) + + +@pytest.mark.parametrize("schema_version", ["1", 1.0, True]) +def test_validate_rejects_coerced_schema_version(schema_version): + with pytest.raises(ValidationError, match="Unsupported ScoringExpectation schema_version"): + ScoringExpectation.model_validate({"schema_version": schema_version, "objective": None, "conditions": []}) + + +def test_persisted_validation_requires_schema_version(): + with pytest.raises(ValidationError, match="requires an explicit schema_version"): + ScoringExpectation.model_validate_persisted({"objective": None, "conditions": []}) + + +def test_validate_rejects_unknown_top_level_field(): + with pytest.raises(ValidationError): + ScoringExpectation.model_validate({"schema_version": 1, "objective": None, "conditions": [], "extra": 1}) + + +def test_validate_rejects_non_string_objective(): + with pytest.raises(ValidationError): + ScoringExpectation.model_validate({"schema_version": 1, "objective": 5, "conditions": []}) + + +def test_validate_rejects_unknown_condition_type(): + with pytest.raises(ValidationError, match="Unknown condition_type"): + ScoringExpectation.model_validate( + {"schema_version": 1, "objective": None, "conditions": [{"condition_type": "ghost"}]} + ) + + +def test_construction_rejects_non_string_objective(): + with pytest.raises(ValidationError): + ScoringExpectation(objective=5) # type: ignore[arg-type] + + +def test_construction_rejects_non_condition(): + with pytest.raises(ValidationError): + ScoringExpectation(conditions=("not a condition",)) # type: ignore[arg-type] + + +def test_json_schema_describes_registered_condition_subtypes(): + condition_items = ScoringExpectation.model_json_schema()["properties"]["conditions"]["items"] + + assert condition_items["discriminator"] == {"propertyName": "condition_type"} + condition_schemas = condition_items["oneOf"] + condition_types = {schema["properties"]["condition_type"]["const"]: schema for schema in condition_schemas} + assert all("condition_type" in schema["required"] for schema in condition_schemas) + assert "matches_objective" in condition_types + assert "test_expectation_beta" in condition_types + assert condition_types["test_expectation_beta"]["properties"]["label"]["type"] == "string" + + +# --------------------------------------------------------------------------- # +# Fingerprint +# --------------------------------------------------------------------------- # +def test_fingerprint_ignores_construction_kwarg_order(): + first = ScoringExpectation(objective="o", conditions=(_BetaCondition(label="x"),)) + second = ScoringExpectation(conditions=(_BetaCondition(label="x"),), objective="o") + + assert scoring_expectation_fingerprint(first) == scoring_expectation_fingerprint(second) + + +def test_fingerprint_is_lowercase_sha256_hex(): + fingerprint = scoring_expectation_fingerprint(ScoringExpectation(objective="o")) + + assert len(fingerprint) == 64 + assert fingerprint == fingerprint.lower() + + +def test_fingerprint_depends_on_condition_order(): + forward = ScoringExpectation(conditions=(_AlphaCondition(), _BetaCondition(label="x"))) + reverse = ScoringExpectation(conditions=(_BetaCondition(label="x"), _AlphaCondition())) - with pytest.raises(dataclasses.FrozenInstanceError): - condition.objective = "something" + assert scoring_expectation_fingerprint(forward) != scoring_expectation_fingerprint(reverse) diff --git a/tests/unit/models/test_score.py b/tests/unit/models/test_score.py index d8d3936b93..94254bc57f 100644 --- a/tests/unit/models/test_score.py +++ b/tests/unit/models/test_score.py @@ -7,7 +7,7 @@ import pytest from pydantic import ValidationError -from pyrit.models import ComponentIdentifier, MessageScorable, Score +from pyrit.models import ComponentIdentifier, MatchesObjective, MessageScorable, Score, ScoringExpectation from pyrit.models.score import UnvalidatedScore @@ -228,3 +228,125 @@ def test_unvalidated_score_to_score(): assert score.score_type == "float_scale" assert score.score_category == ["hate"] assert score.objective == "obj" + + +def _make_unvalidated(**overrides) -> UnvalidatedScore: + defaults: dict = { + "raw_score_value": "3", + "score_value_description": "middle", + "score_category": ["hate"], + "score_rationale": "because", + "score_metadata": {"likert_value": 3}, + "scorer_class_identifier": ComponentIdentifier(class_name="LikertScorer", class_module="pyrit.score"), + "message_piece_id": str(uuid.uuid4()), + } + defaults.update(overrides) + return UnvalidatedScore(**defaults) + + +def test_unvalidated_score_syncs_objective_from_expectation(): + unvalidated = _make_unvalidated(scored_expectation=ScoringExpectation(objective="synced")) + + assert unvalidated.objective == "synced" + + +def test_unvalidated_score_carries_full_expectation_to_score(): + expectation = ScoringExpectation(objective="obj-u", conditions=(MatchesObjective(),)) + unvalidated = _make_unvalidated(scored_expectation=expectation) + + score = unvalidated.to_score(score_value="0.5", score_type="float_scale") + + assert score.scored_expectation == expectation + assert score.objective == "obj-u" + + +def test_unvalidated_score_conflicting_objective_rejected(): + with pytest.raises(ValueError, match="conflicts"): + _make_unvalidated(objective="a", scored_expectation=ScoringExpectation(objective="b")) + + +# --------------------------------------------------------------------------- # +# scored_expectation / derived objective +# --------------------------------------------------------------------------- # +def test_objective_input_becomes_objective_only_expectation(): + score = _make_score(objective="obj-a") + + assert score.objective == "obj-a" + assert score.scored_expectation == ScoringExpectation(objective="obj-a") + + +def test_objective_is_derived_from_scored_expectation(): + expectation = ScoringExpectation(objective="obj-b", conditions=(MatchesObjective(),)) + score = _make_score(scored_expectation=expectation) + + assert score.objective == "obj-b" + assert score.scored_expectation.conditions == (MatchesObjective(),) + + +def test_no_objective_leaves_expectation_none(): + score = _make_score() + + assert score.scored_expectation is None + assert score.objective is None + + +def test_score_objective_is_read_only(): + score = _make_score(objective="obj-a") + + with pytest.raises(ValidationError): + score.objective = "obj-z" + + +def test_score_scored_expectation_is_read_only(): + score = _make_score(scored_expectation=ScoringExpectation(objective="obj-a")) + + with pytest.raises(ValidationError): + score.scored_expectation = ScoringExpectation(objective="obj-z") + + assert score.objective == "obj-a" + + +def test_conflicting_objective_and_expectation_rejected(): + with pytest.raises(ValidationError, match="conflicts"): + _make_score(objective="a", scored_expectation=ScoringExpectation(objective="b")) + + +def test_explicit_null_objective_conflicting_with_expectation_rejected(): + with pytest.raises(ValidationError, match="conflicts"): + _make_score(objective=None, scored_expectation=ScoringExpectation(objective="non-null")) + + +def test_matching_objective_and_expectation_allowed(): + score = _make_score(objective="same", scored_expectation=ScoringExpectation(objective="same")) + + assert score.objective == "same" + + +def test_scored_expectation_round_trips_through_model_dump(): + expectation = ScoringExpectation(objective="obj", conditions=(MatchesObjective(),)) + score = _make_score(scored_expectation=expectation) + + dumped = score.model_dump(mode="json") + + assert dumped["scored_expectation"] == { + "schema_version": 1, + "objective": "obj", + "conditions": [{"condition_type": "matches_objective"}], + } + restored = Score.model_validate(dumped) + assert restored.scored_expectation == expectation + assert restored.objective == "obj" + + +def test_score_rejects_unversioned_persisted_expectation(): + with pytest.raises(ValidationError, match="requires an explicit schema_version"): + _make_score(scored_expectation={"objective": "obj", "conditions": []}) + + +def test_model_validate_does_not_mutate_input_dict(): + dumped = _make_score(objective="obj-a").model_dump() + + Score.model_validate(dumped) + + assert dumped["objective"] == "obj-a" + assert dumped["scored_expectation"]["objective"] == "obj-a" diff --git a/tests/unit/score/test_message_scorer.py b/tests/unit/score/test_message_scorer.py index 67f7c0ef29..3680df6e8e 100644 --- a/tests/unit/score/test_message_scorer.py +++ b/tests/unit/score/test_message_scorer.py @@ -1,10 +1,10 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. -import dataclasses import inspect import uuid from pathlib import Path +from typing import Literal from unittest.mock import MagicMock import pytest @@ -715,9 +715,8 @@ async def test_matches_objective_without_an_objective_raises(self): ) async def test_unconsumed_condition_raises_instead_of_being_dropped(self): - @dataclasses.dataclass(frozen=True) class UnroutedCondition(Condition): - pass + condition_type: Literal["test_unrouted"] = "test_unrouted" scorer = RecordingScorer()