diff --git a/pyrit/score/scorer_evaluation/scorer_evaluator.py b/pyrit/score/scorer_evaluation/scorer_evaluator.py index 0036e0d147..bb5478073f 100644 --- a/pyrit/score/scorer_evaluation/scorer_evaluator.py +++ b/pyrit/score/scorer_evaluation/scorer_evaluator.py @@ -481,23 +481,19 @@ async def _score_responses_grouped_async( @staticmethod def _score_matches_harm_category(*, score: Score, harm_category: str) -> bool: """Return whether a score category matches a canonical or aliased harm category.""" - labeled_categories = set(HarmCategory.parse_many(harm_category)) - if labeled_categories == {HarmCategory.OTHER} and harm_category.casefold() not in { - HarmCategory.OTHER.name.casefold(), - HarmCategory.OTHER.value.casefold(), - }: - labeled_categories = set() + target_casefold = harm_category.casefold() + if any(c.casefold() == target_casefold for c in score.score_category or []): + return True + + labeled_category = HarmCategory.parse(harm_category) + if labeled_category == HarmCategory.OTHER and target_casefold != "other": + return False for score_category in score.score_category or []: - if score_category == harm_category: - return True - score_categories = set(HarmCategory.parse_many(score_category)) - if score_categories == {HarmCategory.OTHER} and score_category.casefold() not in { - HarmCategory.OTHER.name.casefold(), - HarmCategory.OTHER.value.casefold(), - }: + score_category_parsed = HarmCategory.parse(score_category) + if score_category_parsed == HarmCategory.OTHER and score_category.casefold() != "other": continue - if score_categories & labeled_categories: + if score_category_parsed == labeled_category: return True return False diff --git a/tests/unit/score/test_scorer_evaluator.py b/tests/unit/score/test_scorer_evaluator.py index 7f678c3921..0a77689145 100644 --- a/tests/unit/score/test_scorer_evaluator.py +++ b/tests/unit/score/test_scorer_evaluator.py @@ -1,12 +1,13 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +import csv from unittest.mock import AsyncMock, MagicMock, patch import numpy as np import pytest -from azure.ai.contentsafety.models import TextCategory +from pyrit.common.path import SCORER_EVALS_PATH from pyrit.memory import MemoryInterface from pyrit.models import Message, MessagePiece, Score, ScoreStatus from pyrit.score import ( @@ -16,6 +17,7 @@ HarmScorerEvaluator, HarmScorerMetrics, HumanLabeledDataset, + LikertScalePaths, MetricsType, ObjectiveHumanLabeledEntry, ObjectiveScorerEvaluator, @@ -958,20 +960,60 @@ class TestSelectEvaluationScore: def _score(*, category: list[str] | None) -> Score: return Score(score_type="float_scale", score_value="0.5", score_category=category) - @pytest.mark.parametrize("category", list(AzureContentFilterScorer._CATEGORY_EVAL_FILES)) + @pytest.mark.parametrize( + ("emitted_category", "csv_relative_path", "registered_category"), + [ + *( + (cat.value, files[0][0], files[2]) + for cat, files in AzureContentFilterScorer._CATEGORY_EVAL_FILES.items() + ), + *( + ( + preset.load().category, + preset.evaluation_files.human_labeled_datasets_files[0], + preset.evaluation_files.harm_category, + ) + for preset in LikertScalePaths + if preset.evaluation_files is not None + ), + ], + ) @pytest.mark.parametrize("multiple_scores", [False, True]) - def test_azure_categories_match_registered_evaluation(self, category: TextCategory, multiple_scores: bool) -> None: - config = AzureContentFilterScorer._get_eval_files_for_category(category) - assert config is not None - selected = self._score(category=[category.value]) + def test_shipped_pairings_match_csv_evaluation( + self, + emitted_category: str, + csv_relative_path: str, + registered_category: str, + multiple_scores: bool, + ) -> None: + csv_path = SCORER_EVALS_PATH / csv_relative_path + with open(csv_path, encoding="utf-8") as f: + reader = csv.DictReader(line for line in f if not line.startswith("#")) + csv_categories = {row["harm_category"] for row in reader if "harm_category" in row} + assert len(csv_categories) == 1, f"Expected exactly one harm_category in {csv_path}, got {csv_categories}" + csv_harm_category = csv_categories.pop() + + assert registered_category == csv_harm_category + + selected = self._score(category=[emitted_category]) scores = [selected] if multiple_scores: - scores = [ - self._score(category=[other.value]) - for other in AzureContentFilterScorer._CATEGORY_EVAL_FILES - if other != category - ] + scores - assert ScorerEvaluator._select_evaluation_score(scores=scores, harm_category=config.harm_category) is selected + scores = [self._score(category=["unrelated_harm_category"]), selected] + assert ScorerEvaluator._select_evaluation_score(scores=scores, harm_category=csv_harm_category) is selected + + def test_rejects_hate_score_for_representational_dataset(self): + score = self._score(category=["Hate"]) + with pytest.raises(ValueError, match="requires a score for harm category 'REPRESENTATIONAL'"): + ScorerEvaluator._select_evaluation_score(scores=[score], harm_category="REPRESENTATIONAL") + + def test_rejects_bias_score_for_hate_speech_dataset(self): + score = self._score(category=["bias"]) + with pytest.raises(ValueError, match="requires a score for harm category 'hate_speech'"): + ScorerEvaluator._select_evaluation_score(scores=[score], harm_category="hate_speech") + + def test_accepts_case_insensitive_unrecognized_category(self): + score = self._score(category=["Jailbreak"]) + assert ScorerEvaluator._select_evaluation_score(scores=[score], harm_category="jailbreak") is score def test_returns_none_when_the_scorer_returned_nothing(self): assert ScorerEvaluator._select_evaluation_score(scores=[], harm_category="hate_speech") is None