MAINT Enable ruff PIE (flake8-pie) rules and fix all violations

build_scripts/check_links.py

@@ -89,8 +89,7 @@ def check_url(url, retries=2, delay=2):
         or any(url.endswith(reference) for reference in custom_myst_references)
         or os.path.isfile(url)
         or os.path.isdir(url)
-        or url.startswith("mailto:")
-        or url.startswith("attachment:")
+        or url.startswith(("mailto:", "attachment:"))
     ):
         return url, True
 

pyproject.toml

@@ -252,6 +252,7 @@ select = [
     "DOC",  # https://docs.astral.sh/ruff/rules/#pydoclint-doc
     "F401",  # unused-import
     "I",  # isort
+    "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
     "UP",  # https://docs.astral.sh/ruff/rules/#pyupgrade-up

pyrit/datasets/executors/question_answer/wmdp_dataset.py

@@ -42,10 +42,10 @@ def fetch_wmdp_dataset(category: Optional[str] = None) -> QuestionAnsweringDatas
     questions_answers = []
     for name in data_categories:
         ds = load_dataset("cais/wmdp", name)
-        for i in range(0, len(ds["test"])):
+        for i in range(len(ds["test"])):
             # For each question, save the 4 possible choices and their respective index
             choices = []
-            for j in range(0, 4):
+            for j in range(4):
                 c = QuestionChoice(index=j, text=ds["test"]["choices"][i][j])
                 choices.append(c)
 

pyrit/datasets/seed_datasets/seed_dataset_provider.py

@@ -50,7 +50,6 @@ def dataset_name(self) -> str:
         Returns:
             str: The dataset name (e.g., "HarmBench", "JailbreakBench JBB-Behaviors")
         """
-        pass
 
     @abstractmethod
     async def fetch_dataset(self, *, cache: bool = True) -> SeedDataset:
@@ -67,7 +66,6 @@ async def fetch_dataset(self, *, cache: bool = True) -> SeedDataset:
         Raises:
             Exception: If the dataset cannot be fetched or processed.
         """
-        pass
 
     @classmethod
     def get_all_providers(cls) -> dict[str, type["SeedDatasetProvider"]]:

pyrit/executor/attack/multi_turn/chunked_request.py

@@ -375,4 +375,3 @@ async def _teardown_async(self, *, context: ChunkedRequestAttackContext) -> None
         Args:
             context (ChunkedRequestAttackContext): The attack context containing conversation session.
         """
-        pass

pyrit/executor/attack/multi_turn/crescendo.py

@@ -403,7 +403,6 @@ async def _teardown_async(self, *, context: CrescendoAttackContext) -> None:
             context (CrescendoAttackContext): The attack context.
         """
         # Nothing to be done here, no-op
-        pass
 
     @pyrit_json_retry
     async def _get_attack_prompt_async(

pyrit/executor/attack/multi_turn/multi_prompt_sending.py

@@ -326,7 +326,6 @@ def _determine_attack_outcome(
     async def _teardown_async(self, *, context: MultiTurnAttackContext[Any]) -> None:
         """Clean up after attack execution."""
         # Nothing to be done here, no-op
-        pass
 
     async def _send_prompt_to_objective_target_async(
         self, *, current_message: Message, context: MultiTurnAttackContext[Any]

pyrit/executor/attack/multi_turn/red_teaming.py

@@ -324,7 +324,6 @@ async def _perform_async(self, *, context: MultiTurnAttackContext[Any]) -> Attac
     async def _teardown_async(self, *, context: MultiTurnAttackContext[Any]) -> None:
         """Clean up after attack execution."""
         # Nothing to be done here, no-op
-        pass
 
     async def _generate_next_prompt_async(self, context: MultiTurnAttackContext[Any]) -> Message:
         """

pyrit/executor/attack/multi_turn/tree_of_attacks.py

@@ -1546,7 +1546,6 @@ async def _teardown_async(self, *, context: TAPAttackContext) -> None:
                 state after execution.
         """
         # No specific teardown needed for TAP attack
-        pass
 
     async def _prepare_nodes_for_iteration_async(self, context: TAPAttackContext) -> None:
         """

pyrit/executor/attack/printer/attack_result_printer.py

@@ -37,7 +37,6 @@ async def print_result_async(
                 conversation (the red teaming LLM's reasoning). Only shown for successful
                 attacks to avoid overwhelming output. Defaults to False.
         """
-        pass
 
     @abstractmethod
     async def print_conversation_async(self, result: AttackResult, *, include_scores: bool = False) -> None:
@@ -49,7 +48,6 @@ async def print_conversation_async(self, result: AttackResult, *, include_scores
             include_scores (bool): Whether to include scores in the output.
                 Defaults to False.
         """
-        pass
 
     @abstractmethod
     async def print_summary_async(self, result: AttackResult) -> None:
@@ -59,7 +57,6 @@ async def print_summary_async(self, result: AttackResult) -> None:
         Args:
             result (AttackResult): The attack result to summarize
         """
-        pass
 
     @staticmethod
     def _get_outcome_icon(outcome: AttackOutcome) -> str:

pyrit/executor/attack/single_turn/prompt_sending.py

@@ -274,7 +274,6 @@ def _determine_attack_outcome(
     async def _teardown_async(self, *, context: SingleTurnAttackContext[Any]) -> None:
         """Clean up after attack execution."""
         # Nothing to be done here, no-op
-        pass
 
     def _get_message(self, context: SingleTurnAttackContext[Any]) -> Message:
         """

pyrit/executor/benchmark/question_answering.py

@@ -253,7 +253,6 @@ async def _teardown_async(self, *, context: QuestionAnsweringBenchmarkContext) -
         Args:
             context (QuestionAnsweringBenchmarkContext): The context for the strategy.
         """
-        pass
 
     @overload
     async def execute_async(

pyrit/executor/core/strategy.py

@@ -97,7 +97,6 @@ async def on_event(self, event_data: StrategyEventData[StrategyContextT, Strateg
         Args:
             event_data: Data about the event that occurred.
         """
-        pass
 
 
 class StrategyLogAdapter(logging.LoggerAdapter):  # type: ignore[type-arg]
@@ -199,7 +198,6 @@ def _validate_context(self, *, context: StrategyContextT) -> None:
         Raises:
             Exception: If the context is invalid for this strategy.
         """
-        pass
 
     @abstractmethod
     async def _setup_async(self, *, context: StrategyContextT) -> None:
@@ -211,7 +209,6 @@ async def _setup_async(self, *, context: StrategyContextT) -> None:
         Args:
             context (StrategyContextT): The context for the strategy.
         """
-        pass
 
     @abstractmethod
     async def _perform_async(self, *, context: StrategyContextT) -> StrategyResultT:
@@ -225,7 +222,6 @@ async def _perform_async(self, *, context: StrategyContextT) -> StrategyResultT:
         Returns:
             StrategyResultT: The result of the strategy execution.
         """
-        pass
 
     @abstractmethod
     async def _teardown_async(self, *, context: StrategyContextT) -> None:
@@ -237,7 +233,6 @@ async def _teardown_async(self, *, context: StrategyContextT) -> None:
         Args:
             context (StrategyContextT): The context for the strategy.
         """
-        pass
 
     async def _handle_event(
         self,

pyrit/executor/promptgen/anecdoctor.py

@@ -255,7 +255,6 @@ async def _teardown_async(self, *, context: AnecdoctorContext) -> None:
             context (AnecdoctorContext): The generation context.
         """
         # Nothing to clean up for this prompt generation
-        pass
 
     async def _prepare_examples_async(self, *, context: AnecdoctorContext) -> str:
         """

pyrit/executor/promptgen/fuzzer/fuzzer.py

@@ -862,7 +862,6 @@ async def _teardown_async(self, *, context: FuzzerContext) -> None:
             context (FuzzerContext): The generation context.
         """
         # No specific teardown needed
-        pass
 
     def _should_stop_generation(self, context: FuzzerContext) -> bool:
         """

pyrit/executor/promptgen/fuzzer/fuzzer_converter_base.py

@@ -61,7 +61,6 @@ def __init__(
 
     def update(self, **kwargs: Any) -> None:
         """Update the converter with new parameters."""
-        pass
 
     async def convert_async(self, *, prompt: str, input_type: PromptDataType = "text") -> ConverterResult:
         """

pyrit/executor/workflow/core/workflow_strategy.py

@@ -26,15 +26,11 @@
 class WorkflowContext(StrategyContext, ABC):
     """Base class for all workflow contexts."""
 
-    pass
-
 
 @dataclass
 class WorkflowResult(StrategyResult, ABC):
     """Base class for all workflow results."""
 
-    pass
-
 
 class _DefaultWorkflowEventHandler(StrategyEventHandler[WorkflowContextT, WorkflowResultT]):
     """

pyrit/executor/workflow/xpia.py

@@ -415,7 +415,6 @@ async def _teardown_async(self, *, context: XPIAContext) -> None:
             context (XPIAContext): The context for the workflow.
         """
         # No specific teardown operations required for base XPIA workflow
-        pass
 
     @overload
     async def execute_async(

pyrit/memory/memory_models.py

@@ -119,8 +119,6 @@ class Base(DeclarativeBase):
     Base class for all database models.
     """
 
-    pass
-
 
 class PromptMemoryEntry(Base):
     """

pyrit/models/data_type_serializer.py

@@ -423,7 +423,7 @@ def __init__(self, *, category: str, prompt_text: str, extension: Optional[str]
         self.value = prompt_text
         self.data_sub_directory = f"/{category}/urls"
         self.file_extension = extension if extension else "txt"
-        self.on_disk = not (prompt_text.startswith("http://") or prompt_text.startswith("https://"))
+        self.on_disk = not (prompt_text.startswith(("http://", "https://")))
 
     def data_on_disk(self) -> bool:
         """

pyrit/models/seeds/seed.py

@@ -95,16 +95,16 @@ class Seed(YamlLoadable):
     dataset_name: Optional[str] = None
 
     # Categories of harm associated with this prompt
-    harm_categories: Optional[Sequence[str]] = field(default_factory=lambda: [])
+    harm_categories: Optional[Sequence[str]] = field(default_factory=list)
 
     # Description of the prompt
     description: Optional[str] = None
 
     # Authors of the prompt
-    authors: Optional[Sequence[str]] = field(default_factory=lambda: [])
+    authors: Optional[Sequence[str]] = field(default_factory=list)
 
     # Groups affiliated with the prompt
-    groups: Optional[Sequence[str]] = field(default_factory=lambda: [])
+    groups: Optional[Sequence[str]] = field(default_factory=list)
 
     # Source of the prompt
     source: Optional[str] = None
@@ -116,7 +116,7 @@ class Seed(YamlLoadable):
     added_by: Optional[str] = None
 
     # Arbitrary metadata that can be attached to the prompt
-    metadata: Optional[dict[str, Union[str, int]]] = field(default_factory=lambda: {})
+    metadata: Optional[dict[str, Union[str, int]]] = field(default_factory=dict)
 
     # Unique identifier for the prompt group
     prompt_group_id: Optional[uuid.UUID] = None

pyrit/models/seeds/seed_prompt.py

@@ -44,7 +44,7 @@ class SeedPrompt(Seed):
     sequence: int = 0
 
     # Parameters that can be used in the prompt template
-    parameters: Optional[Sequence[str]] = field(default_factory=lambda: [])
+    parameters: Optional[Sequence[str]] = field(default_factory=list)
 
     def __post_init__(self) -> None:
         """

pyrit/prompt_converter/add_image_to_video_converter.py

@@ -162,7 +162,7 @@ async def _add_image_to_video(self, image_path: str, output_path: str) -> str:
                 # Blend overlay with frame
                 if overlay.shape[2] == 4:  # Check number of channels on image
                     alpha_overlay = overlay[:, :, 3] / 255.0
-                    for c in range(0, 3):
+                    for c in range(3):
                         frame[y : y + image_height, x : x + image_width, c] = (
                             alpha_overlay * overlay[:, :, c]
                             + (1 - alpha_overlay) * frame[y : y + image_height, x : x + image_width, c]

pyrit/prompt_converter/base2048_converter.py

@@ -26,7 +26,6 @@ class Base2048Converter(PromptConverter):
 
     def __init__(self) -> None:
         """Initialize the Base2048Converter."""
-        pass
 
     async def convert_async(self, *, prompt: str, input_type: PromptDataType = "text") -> ConverterResult:
         """

pyrit/prompt_converter/base64_converter.py

@@ -22,9 +22,6 @@ class Base64Converter(PromptConverter):
     SUPPORTED_INPUT_TYPES = ("text",)
     SUPPORTED_OUTPUT_TYPES = ("text",)
 
-    SUPPORTED_INPUT_TYPES = ("text",)
-    SUPPORTED_OUTPUT_TYPES = ("text",)
-
     EncodingFunc = Literal[
         "b64encode",
         "urlsafe_b64encode",

pyrit/prompt_converter/ecoji_converter.py

@@ -25,7 +25,6 @@ class EcojiConverter(PromptConverter):
 
     def __init__(self) -> None:
         """Initialize the Ecoji converter."""
-        pass
 
     async def convert_async(self, *, prompt: str, input_type: PromptDataType = "text") -> ConverterResult:
         """

pyrit/prompt_converter/insert_punctuation_converter.py

@@ -119,7 +119,7 @@ def _insert_punctuation(self, prompt: str, punctuation_list: list[str]) -> str:
         # Words list contains single spaces, single word without punctuations, single punctuations
         words = re.findall(r"\w+|[^\w\s]|\s", prompt)
         # Maintains indices for actual "words", i.e. letters and numbers not divided by punctuations
-        word_indices = [i for i in range(0, len(words)) if not re.match(r"\W", words[i])]
+        word_indices = [i for i in range(len(words)) if not re.match(r"\W", words[i])]
         # Calculate the number of insertions
         num_insertions = max(
             1, round(len(word_indices) * self._word_swap_ratio)
@@ -177,7 +177,7 @@ def _insert_within_words(self, prompt: str, num_insertions: int, punctuation_lis
         # Store random indices of prompt_list into insert_indices
         # If the prompt has only 0 or 1 chars, insert at the end of the prompt
         insert_indices = (
-            [1] if len(prompt_list) <= num_insertions else random.sample(range(0, len(prompt_list) - 1), num_insertions)
+            [1] if len(prompt_list) <= num_insertions else random.sample(range(len(prompt_list) - 1), num_insertions)
         )
 
         for index in insert_indices:

pyrit/prompt_converter/text_selection_strategy.py

@@ -26,7 +26,6 @@ def select_range(self, *, text: str) -> tuple[int, int]:
             tuple[int, int]: A tuple of (start_index, end_index) representing the character range.
                 The range is inclusive of start_index and exclusive of end_index.
         """
-        pass
 
 
 class TokenSelectionStrategy(TextSelectionStrategy):
@@ -87,7 +86,6 @@ def select_words(self, *, words: list[str]) -> list[int]:
         Returns:
             List[int]: A list of indices representing which words should be converted.
         """
-        pass
 
     def select_range(self, *, text: str, word_separator: str = " ") -> tuple[int, int]:
         """

pyrit/prompt_converter/word_level_converter.py

@@ -72,11 +72,9 @@ async def convert_word_async(self, word: str) -> str:
         Returns:
             str: The converted word.
         """
-        pass
 
     def validate_input(self, prompt: str) -> None:
         """Validate the input before processing (can be overridden by subclasses)."""
-        pass
 
     def join_words(self, words: list[str]) -> str:
         """

pyrit/prompt_target/common/prompt_chat_target.py

@@ -86,7 +86,6 @@ def is_json_response_supported(self) -> bool:
         Returns:
             bool: True if JSON response is supported, False otherwise.
         """
-        pass
 
     def is_response_format_json(self, message_piece: MessagePiece) -> bool:
         """

pyrit/prompt_target/openai/openai_target.py

@@ -257,7 +257,6 @@ def _get_target_api_paths(self) -> list[str]:
         Returns:
             List of API paths (e.g., ["/chat/completions", "/v1/chat/completions"])
         """
-        pass
 
     @abstractmethod
     def _get_provider_examples(self) -> dict[str, str]:
@@ -270,7 +269,6 @@ def _get_provider_examples(self) -> dict[str, str]:
             Dict mapping provider patterns to example URLs
             (e.g., {".openai.azure.com": "https://{resource}.openai.azure.com/openai/v1"})
         """
-        pass
 
     def _validate_url_for_target(self, endpoint_url: str) -> None:
         """
@@ -537,7 +535,6 @@ async def _construct_message_from_response(self, response: Any, request: Message
         Returns:
             Message: Constructed message with extracted content.
         """
-        pass
 
     def _check_content_filter(self, response: Any) -> bool:
         """
@@ -694,4 +691,3 @@ def is_json_response_supported(self) -> bool:
         Returns:
             bool: True if JSON response is supported, False otherwise.
         """
-        pass

pyrit/prompt_target/text_target.py

@@ -92,4 +92,3 @@ def _validate_request(self, *, message: Message) -> None:
 
     async def cleanup_target(self) -> None:
         """Target does not require cleanup."""
-        pass