MAINT Enable ruff PGH rule for pygrep-hooks linting

pyproject.toml

@@ -259,6 +259,7 @@ select = [
     "DOC",  # https://docs.astral.sh/ruff/rules/#pydoclint-doc
     "F401",  # unused-import
     "I",  # isort
+    "PGH",  # https://docs.astral.sh/ruff/rules/#pygrep-hooks-pgh
     "PIE",  # https://docs.astral.sh/ruff/rules/#flake8-pie-pie
     "RET",  # https://docs.astral.sh/ruff/rules/#flake8-return-ret
     "SIM",  # https://docs.astral.sh/ruff/rules/#flake8-simplify-sim

pyrit/auth/copilot_authenticator.py

@@ -233,7 +233,7 @@ async def _get_cached_token_if_available_and_valid(self) -> Optional[dict[str, A
                 minutes_left = (expiry_time - current_time).total_seconds() / 60
                 logger.info(f"Cached token is valid for another {minutes_left:.2f} minutes")
 
-            return token_data  # type: ignore
+            return token_data  # type: ignore[no-any-return]
 
         except Exception as e:
             error_name = type(e).__name__
@@ -450,7 +450,7 @@ async def response_handler(response: Any) -> None:
                 else:
                     logger.error(f"Failed to retrieve bearer token within {self._token_capture_timeout} seconds.")
 
-                return bearer_token  # type: ignore
+                return bearer_token  # type: ignore[no-any-return]
             except Exception as e:
                 logger.error("Failed to retrieve access token using Playwright.")
 

pyrit/cli/frontend_core.py

@@ -32,11 +32,11 @@
     HAS_TERMCOLOR = False
 
     # Create a dummy termcolor module for fallback
-    class termcolor:  # type: ignore
+    class termcolor:  # type: ignore[no-redef]
         """Dummy termcolor fallback for colored printing if termcolor is not installed."""
 
         @staticmethod
-        def cprint(text: str, color: str = None, attrs: list = None) -> None:  # type: ignore
+        def cprint(text: str, color: str = None, attrs: list = None) -> None:  # type: ignore[type-arg]
             """Print text without color."""
             print(text)
 

pyrit/common/display_response.py

@@ -47,6 +47,6 @@ async def display_image_response(response_piece: MessagePiece) -> None:
         image = Image.open(image_stream)
 
         # Jupyter built-in display function only works in notebooks.
-        display(image)  # type: ignore # noqa: F821
+        display(image)  # type: ignore[name-defined] # noqa: F821
     if response_piece.response_error == "blocked":
         logger.info("---\nContent blocked, cannot show a response.\n---")

pyrit/common/notebook_utils.py

@@ -13,7 +13,7 @@ def is_in_ipython_session() -> bool:
         bool: True if the code is running in an IPython session, False otherwise.
     """
     try:
-        __IPYTHON__  # type: ignore  # noqa: B018
+        __IPYTHON__  # type: ignore[name-defined]  # noqa: B018
         return True
     except NameError:
         return False

pyrit/common/yaml_loadable.py

@@ -41,5 +41,5 @@ def from_yaml_file(cls: type[T], file: Union[Path | str]) -> T:
         # If this class provides a from_dict factory, use it;
         # otherwise, just instantiate directly with **yaml_data
         if hasattr(cls, "from_dict") and callable(getattr(cls, "from_dict")):  # noqa: B009
-            return cls.from_dict(yaml_data)  # type: ignore
+            return cls.from_dict(yaml_data)  # type: ignore[attr-defined, no-any-return]
         return cls(**yaml_data)

pyrit/datasets/seed_datasets/remote/red_team_social_bias_dataset.py

@@ -116,7 +116,7 @@ async def fetch_dataset(self, *, cache: bool = True) -> SeedDataset:
                             data_type="text",
                             prompt_group_id=group_id,
                             sequence=i,
-                            **prompt_metadata,  # type: ignore
+                            **prompt_metadata,  # type: ignore[arg-type]
                         )
                     )
             else:
@@ -133,7 +133,7 @@ async def fetch_dataset(self, *, cache: bool = True) -> SeedDataset:
                     SeedPrompt(
                         value=escaped_cleaned_value,
                         data_type="text",
-                        **prompt_metadata,  # type: ignore
+                        **prompt_metadata,  # type: ignore[arg-type]
                     )
                 )
 

pyrit/memory/azure_sql_memory.py

@@ -650,7 +650,7 @@ def _query_entries(
                     return query.distinct().all()
                 return query.all()
             except SQLAlchemyError as e:
-                logger.exception(f"Error fetching data from table {model_class.__tablename__}: {e}")  # type: ignore
+                logger.exception(f"Error fetching data from table {model_class.__tablename__}: {e}")  # type: ignore[attr-defined]
                 raise
 
     def _update_entries(self, *, entries: MutableSequence[Base], update_fields: dict[str, Any]) -> bool:

pyrit/memory/memory_interface.py

@@ -1050,7 +1050,7 @@ def get_seed_dataset_names(self) -> Sequence[str]:
         try:
             entries: Sequence[SeedEntry] = self._query_entries(
                 SeedEntry,
-                conditions=and_(SeedEntry.dataset_name is not None, SeedEntry.dataset_name != ""),  # type: ignore
+                conditions=and_(SeedEntry.dataset_name is not None, SeedEntry.dataset_name != ""),  # type: ignore[arg-type]
                 distinct=True,
             )
             # Extract unique dataset names from the entries
@@ -1484,7 +1484,7 @@ def update_scenario_run_state(self, *, scenario_result_id: str, scenario_run_sta
             scenario_result = scenario_results[0]
 
             # Update the scenario run state
-            scenario_result.scenario_run_state = scenario_run_state  # type: ignore
+            scenario_result.scenario_run_state = scenario_run_state  # type: ignore[assignment]
 
             # Save updated result back to memory using update
             entry = ScenarioResultEntry(entry=scenario_result)

pyrit/memory/memory_models.py

@@ -232,14 +232,14 @@ def __init__(self, *, entry: MessagePiece):
         )
 
         self.original_value = entry.original_value
-        self.original_value_data_type = entry.original_value_data_type  # type: ignore
+        self.original_value_data_type = entry.original_value_data_type  # type: ignore[assignment]
         self.original_value_sha256 = entry.original_value_sha256
 
         self.converted_value = entry.converted_value
-        self.converted_value_data_type = entry.converted_value_data_type  # type: ignore
+        self.converted_value_data_type = entry.converted_value_data_type  # type: ignore[assignment]
         self.converted_value_sha256 = entry.converted_value_sha256
 
-        self.response_error = entry.response_error  # type: ignore
+        self.response_error = entry.response_error  # type: ignore[assignment]
 
         self.original_prompt_id = entry.original_prompt_id
         self.pyrit_version = pyrit.__version__
@@ -562,7 +562,7 @@ def __init__(self, *, entry: Seed):
         self.data_type = entry.data_type
         self.name = entry.name
         self.dataset_name = entry.dataset_name
-        self.harm_categories = entry.harm_categories  # type: ignore
+        self.harm_categories = entry.harm_categories  # type: ignore[assignment]
         self.description = entry.description
         self.authors = list(entry.authors) if entry.authors else None
         self.groups = list(entry.groups) if entry.groups else None

pyrit/memory/sqlite_memory.py

@@ -240,7 +240,7 @@ def _query_entries(
                     return query.distinct().all()
                 return query.all()
             except SQLAlchemyError as e:
-                logger.exception(f"Error fetching data from table {model_class.__tablename__}: {e}")  # type: ignore
+                logger.exception(f"Error fetching data from table {model_class.__tablename__}: {e}")  # type: ignore[attr-defined]
                 raise
 
     def _insert_entry(self, entry: Base) -> None:
@@ -448,7 +448,7 @@ def export_all_tables(self, *, export_type: str = "json") -> None:
             file_extension = f".{export_type}"
             file_path = DB_DATA_PATH / f"{table_name}{file_extension}"
             # Convert to list for exporter compatibility
-            self.exporter.export_data(list(data), file_path=file_path, export_type=export_type)  # type: ignore
+            self.exporter.export_data(list(data), file_path=file_path, export_type=export_type)  # type: ignore[arg-type]
 
     def _get_attack_result_harm_category_condition(self, *, targeted_harm_categories: Sequence[str]) -> Any:
         """

pyrit/models/message_piece.py

@@ -91,7 +91,7 @@ def __init__(
         """
         self.id = id if id else uuid4()
 
-        if role not in ChatMessageRole.__args__:  # type: ignore
+        if role not in ChatMessageRole.__args__:  # type: ignore[attr-defined]
             raise ValueError(f"Role {role} is not a valid role.")
 
         self._role: ChatMessageRole = role
@@ -251,14 +251,20 @@ def role(self, value: ChatMessageRole) -> None:
             ValueError: If the role is not a valid ChatMessageRole.
 
         """
-        if value not in ChatMessageRole.__args__:  # type: ignore
+        if value not in ChatMessageRole.__args__:  # type: ignore[attr-defined]
             raise ValueError(f"Role {value} is not a valid role.")
         self._role = value
 
-    def to_message(self) -> Message:  # type: ignore # noqa F821
+    def to_message(self) -> Message:  # type: ignore[name-defined] # noqa: F821
+        """
+        Convert this message piece into a Message.
+
+        Returns:
+            Message: A Message containing this piece.
+        """
         from pyrit.models.message import Message
 
-        return Message([self])  # noqa F821
+        return Message([self])  # noqa: F821
 
     def has_error(self) -> bool:
         """

pyrit/models/seeds/seed.py

@@ -194,7 +194,7 @@ def render_template_value_silent(self, **kwargs: Any) -> str:
                 return self.value
 
         # Create a Jinja template with PartialUndefined placeholders
-        env = Environment(loader=BaseLoader, undefined=PartialUndefined)  # type: ignore
+        env = Environment(loader=BaseLoader, undefined=PartialUndefined)  # type: ignore[arg-type]
         jinja_template = env.from_string(self.value)
 
         try:

pyrit/prompt_converter/selective_text_converter.py

@@ -196,7 +196,7 @@ async def _convert_word_level_async(self, *, prompt: str) -> ConverterResult:
         words = prompt.split(self._word_separator)
 
         # Get selected word indices
-        selected_indices = self._selection_strategy.select_words(words=words)  # type: ignore
+        selected_indices = self._selection_strategy.select_words(words=words)  # type: ignore[attr-defined]
 
         # If no words selected, return original prompt
         if not selected_indices:

pyrit/prompt_target/azure_ml_chat_target.py

@@ -221,7 +221,7 @@ async def _complete_chat_async(
                 raise EmptyResponseException(message="The chat returned an empty response.") from e
             raise e(
                 f"Exception obtaining response from the target. Returned response: {response.json()}. "
-                + f"Exception: {str(e)}"  # type: ignore
+                + f"Exception: {str(e)}"  # type: ignore[operator]
             ) from e
 
     async def _construct_http_body_async(

pyrit/prompt_target/openai/openai_chat_target.py

@@ -587,7 +587,7 @@ async def _build_chat_messages_for_multi_modal_async(
                 elif message_piece.converted_value_data_type == "image_path":
                     data_base64_encoded_url = await convert_local_image_to_data_url(message_piece.converted_value)
                     image_url_entry = {"url": data_base64_encoded_url}
-                    entry = {"type": "image_url", "image_url": image_url_entry}  # type: ignore
+                    entry = {"type": "image_url", "image_url": image_url_entry}  # type: ignore[dict-item]
                     content.append(entry)
                 elif message_piece.converted_value_data_type == "audio_path":
                     ext = DataTypeSerializer.get_extension(message_piece.converted_value)
@@ -608,7 +608,7 @@ async def _build_chat_messages_for_multi_modal_async(
                     base64_data = await audio_serializer.read_data_base64()
                     audio_format = ext.lower().lstrip(".")
                     input_audio_entry = {"data": base64_data, "format": audio_format}
-                    entry = {"type": "input_audio", "input_audio": input_audio_entry}  # type: ignore
+                    entry = {"type": "input_audio", "input_audio": input_audio_entry}  # type: ignore[dict-item]
                     content.append(entry)
                 else:
                     raise ValueError(

pyrit/prompt_target/openai/openai_target.py

@@ -73,7 +73,7 @@ async def async_token_provider() -> str:
         Returns:
             str: The token string from the synchronous provider.
         """
-        return api_key()  # type: ignore
+        return api_key()  # type: ignore[return-value]
 
     return async_token_provider
 

pyrit/prompt_target/text_target.py

@@ -72,13 +72,13 @@ def import_scores_from_csv(self, csv_file_path: Path) -> list[MessagePiece]:
                 labels = json.loads(labels_str) if labels_str else None
 
                 message_piece = MessagePiece(
-                    role=row["role"],  # type: ignore
+                    role=row["role"],  # type: ignore[arg-type]
                     original_value=row["value"],
-                    original_value_data_type=row.get["data_type", None],  # type: ignore
+                    original_value_data_type=row.get("data_type", None),  # type: ignore[arg-type]
                     conversation_id=row.get("conversation_id", None),
                     sequence=int(sequence_str) if sequence_str else None,
                     labels=labels,
-                    response_error=row.get("response_error", None),  # type: ignore
+                    response_error=row.get("response_error", None),  # type: ignore[arg-type]
                     prompt_target_identifier=self.get_identifier(),
                 )
                 message_pieces.append(message_piece)

pyrit/scenario/core/scenario.py

@@ -175,7 +175,7 @@ def default_dataset_config(cls) -> DatasetConfiguration:
     async def initialize_async(
         self,
         *,
-        objective_target: PromptTarget = REQUIRED_VALUE,  # type: ignore
+        objective_target: PromptTarget = REQUIRED_VALUE,  # type: ignore[assignment]
         scenario_strategies: Optional[Sequence[ScenarioStrategy | ScenarioCompositeStrategy]] = None,
         dataset_config: Optional[DatasetConfiguration] = None,
         max_concurrency: int = 10,

pyrit/scenario/scenarios/airt/psychosocial_scenario.py

@@ -438,7 +438,7 @@ def _get_scorer(self, subharm: Optional[str] = None) -> FloatScaleThresholdScore
         )
 
         # Wrap with conversation scorer to evaluate full conversation history
-        conversation_scorer: FloatScaleScorer = create_conversation_scorer(scorer=psych_scorer)  # type: ignore
+        conversation_scorer: FloatScaleScorer = create_conversation_scorer(scorer=psych_scorer)  # type: ignore[assignment]
 
         # Wrap with threshold scorer
         # threshold=1.0 means only raw score 5 counts as attack success (extremely harmful)

pyrit/score/conversation_scorer.py

@@ -183,7 +183,7 @@ def create_conversation_scorer(
         )
 
     # Dynamically create a class that inherits from both ConversationScorer and the scorer's base class
-    class DynamicConversationScorer(ConversationScorer, scorer_base_class):  # type: ignore
+    class DynamicConversationScorer(ConversationScorer, scorer_base_class):  # type: ignore[misc, valid-type]
         """Dynamic ConversationScorer that inherits from both ConversationScorer and the wrapped scorer's base class."""
 
         def __init__(self) -> None:

tests/integration/ai_recruiter/test_ai_recruiter.py

@@ -218,7 +218,7 @@ async def test_ai_recruiter_workflow():
     # Execute the XPIA flow.
     # Step 1: PDF with hidden text is uploaded to /upload/
     # Step 2: /search_candidates/ is called automatically afterward.
-    final_result = await xpia.execute_async(  # type: ignore
+    final_result = await xpia.execute_async(  # type: ignore[arg-type]
         attack_content='{"description": "Hidden PDF Attack"}',
         processing_prompt="Evaluate all uploaded resumes and pick the best candidate.",
     )

tests/unit/common/test_helper_functions.py

@@ -358,4 +358,4 @@ def test_verify_and_resolve_path_with_path_object(self) -> None:
     def test_verify_and_resolve_path_invalid_type(self) -> None:
         """Test that the function raises ValueError for invalid types."""
         with pytest.raises(ValueError, match="Path must be a string or Path object"):
-            verify_and_resolve_path(123)  # type: ignore
+            verify_and_resolve_path(123)  # type: ignore[arg-type]

tests/unit/converter/test_add_image_text_converter.py

@@ -82,15 +82,15 @@ def test_image_text_converter_add_text_to_image(image_text_converter_sample_imag
 async def test_add_image_text_converter_invalid_input_text(image_text_converter_sample_image) -> None:
     converter = AddImageTextConverter(img_to_add=image_text_converter_sample_image)
     with pytest.raises(ValueError):
-        assert await converter.convert_async(prompt="", input_type="text")  # type: ignore
+        assert await converter.convert_async(prompt="", input_type="text")  # type: ignore[arg-type]
     os.remove("test.png")
 
 
 @pytest.mark.asyncio
 async def test_add_image_text_converter_invalid_file_path():
     converter = AddImageTextConverter(img_to_add="nonexistent_image.png", font_name="helvetica.ttf")
     with pytest.raises(FileNotFoundError):
-        assert await converter.convert_async(prompt="Sample Text!", input_type="text")  # type: ignore
+        assert await converter.convert_async(prompt="Sample Text!", input_type="text")  # type: ignore[arg-type]
 
 
 @pytest.mark.asyncio

tests/unit/converter/test_add_text_image_converter.py

@@ -77,7 +77,7 @@ def test_text_image_converter_add_text_to_image(text_image_converter_sample_imag
 async def test_add_text_image_converter_invalid_input_image() -> None:
     converter = AddTextImageConverter(text_to_add="test")
     with pytest.raises(FileNotFoundError):
-        assert await converter.convert_async(prompt="mock_image.png", input_type="image_path")  # type: ignore
+        assert await converter.convert_async(prompt="mock_image.png", input_type="image_path")  # type: ignore[arg-type]
 
 
 @pytest.mark.asyncio

tests/unit/converter/test_azure_speech_converter.py

@@ -67,7 +67,7 @@ async def test_send_prompt_to_audio_file_raises_value_error(self) -> None:
         # testing empty space string
         prompt = "     "
         with pytest.raises(ValueError):
-            await converter.convert_async(prompt=prompt, input_type="text")  # type: ignore
+            await converter.convert_async(prompt=prompt, input_type="text")  # type: ignore[arg-type]
 
     def test_azure_speech_audio_text_converter_input_supported(self):
         converter = AzureSpeechTextToAudioConverter()

tests/unit/converter/test_bin_ascii_converter.py

@@ -173,7 +173,7 @@ class TestBinAsciiConverterErrorHandling:
     async def test_invalid_encoding_function(self) -> None:
         """Test that invalid encoding function raises ValueError."""
         converter = BinAsciiConverter(encoding_func="hex")
-        converter._encoding_func = "invalid"  # type: ignore
+        converter._encoding_func = "invalid"  # type: ignore[arg-type]
 
         with pytest.raises(ValueError, match="Unsupported encoding function"):
             await converter.convert_async(prompt="test")
@@ -182,4 +182,4 @@ async def test_invalid_encoding_function(self) -> None:
     async def test_invalid_encoding_function_at_init(self) -> None:
         """Test that invalid encoding function at initialization raises ValueError."""
         with pytest.raises(ValueError, match="Invalid encoding_func"):
-            BinAsciiConverter(encoding_func="invalid")  # type: ignore
+            BinAsciiConverter(encoding_func="invalid")  # type: ignore[arg-type]