fix(typing): Type tests.sentry.workflow_engine.processors (#103174)

Some EZPZ missing test types.

Author: thetruecpaul

pyproject.toml

@@ -866,7 +866,7 @@ module = [
     "tests.sentry.workflow_engine.endpoints.*",
     "tests.sentry.workflow_engine.handlers.action.*",
     "tests.sentry.workflow_engine.models.*",
-    "tests.sentry.workflow_engine.processors.contexts.*",
+    "tests.sentry.workflow_engine.processors.*",
     "tests.sentry.workflow_engine.service.*",
     "tests.sentry.workflow_engine.utils.*",
     "tests.sentry_plugins.*",

tests/sentry/workflow_engine/processors/test_data_sources.py

@@ -37,8 +37,10 @@ def setUp(self) -> None:
         self.ds2 = self.create_data_source(source_id=source_id_2, type="test")
         self.ds2.detectors.set([self.detector_one, self.detector_two])
 
-        self.packet = DataPacket[dict](source_id_1, {"source_id": source_id_1, "foo": "bar"})
-        self.two_detector_packet = DataPacket[dict](
+        self.packet = DataPacket[dict[str, str]](
+            source_id_1, {"source_id": source_id_1, "foo": "bar"}
+        )
+        self.two_detector_packet = DataPacket[dict[str, str]](
             source_id_2, {"source_id": source_id_2, "foo": "baz"}
         )
 

tests/sentry/workflow_engine/processors/test_delayed_workflow.py

@@ -175,7 +175,9 @@ def create_project_event_freq_workflow(
 
         return workflow, [workflow_action_slow_filter_group, workflow_action_filter_group]
 
-    def setup_event(self, project, environment, name) -> tuple[Event, Group]:
+    def setup_event(
+        self, project: Project, environment: Environment, name: str
+    ) -> tuple[Event, Group]:
         event = self.create_event(project.id, FROZEN_TIME, name, environment.name)
         assert event.group
         return event, event.group

tests/sentry/workflow_engine/processors/test_detector.py

@@ -1,5 +1,6 @@
 import unittest
 import uuid
+from typing import Any
 from unittest import mock
 from unittest.mock import MagicMock, call
 
@@ -72,9 +73,9 @@ class TestProcessDetectors(BaseDetectorHandlerTest):
     def setUp(self) -> None:
         super().setUp()
 
-    def build_data_packet(self, **kwargs):
+    def build_data_packet(self, **kwargs: Any) -> DataPacket[dict[str, Any]]:
         source_id = "1234"
-        return DataPacket[dict](
+        return DataPacket[dict[str, Any]](
             source_id, {"source_id": source_id, "group_vals": {"group_1": 6}, **kwargs}
         )
 
@@ -228,8 +229,11 @@ def test_sending_metric_before_evaluating(self) -> None:
     @mock.patch("sentry.workflow_engine.processors.detector.metrics")
     @mock.patch("sentry.workflow_engine.processors.detector.logger")
     def test_metrics_and_logs_fire(
-        self, mock_logger, mock_metrics, mock_produce_occurrence_to_kafka
-    ):
+        self,
+        mock_logger: mock.MagicMock,
+        mock_metrics: mock.MagicMock,
+        mock_produce_occurrence_to_kafka: mock.MagicMock,
+    ) -> None:
         detector, _ = self.create_detector_and_condition(type=self.handler_state_type.slug)
         data_packet = DataPacket("1", {"dedupe": 2, "group_vals": {None: 6}})
         results = process_detectors(data_packet, [detector])
@@ -281,8 +285,11 @@ def test_metrics_and_logs_fire(
     @mock.patch("sentry.workflow_engine.processors.detector.metrics")
     @mock.patch("sentry.workflow_engine.processors.detector.logger")
     def test_metrics_and_logs_resolve(
-        self, mock_logger, mock_metrics, mock_produce_occurrence_to_kafka
-    ):
+        self,
+        mock_logger: mock.MagicMock,
+        mock_metrics: mock.MagicMock,
+        mock_produce_occurrence_to_kafka: mock.MagicMock,
+    ) -> None:
         detector, _ = self.create_detector_and_condition(type=self.handler_state_type.slug)
         data_packet = DataPacket("1", {"dedupe": 2, "group_vals": {None: 6}})
         process_detectors(data_packet, [detector])
@@ -751,7 +758,7 @@ def test_dedupe(self) -> None:
             handler.state_manager.enqueue_dedupe_update("group_key", 99)
             handler.state_manager.commit_state_updates()
 
-            data_packet = DataPacket[dict](
+            data_packet = DataPacket[dict[str, Any]](
                 source_id="1234",
                 packet={"id": "1234", "group_vals": {"group_key": 10}, "dedupe": 100},
             )
@@ -778,7 +785,7 @@ def test_dedupe__already_processed(self) -> None:
             handler.state_manager.commit_state_updates()
 
             handler.evaluate(
-                DataPacket[dict](
+                DataPacket[dict[str, Any]](
                     source_id="1234",
                     packet={"id": "1234", "group_vals": {"group_key": 10}, "dedupe": 100},
                 ),
@@ -789,7 +796,7 @@ def test_dedupe__already_processed(self) -> None:
 
     def test_status_change(self) -> None:
         handler = self.build_handler()
-        data_packet = DataPacket[dict](
+        data_packet = DataPacket[dict[str, Any]](
             source_id="1234", packet={"id": "1234", "group_vals": {"group_key": 10}, "dedupe": 100}
         )
 

tests/sentry/workflow_engine/processors/test_schedule.py

@@ -35,7 +35,14 @@ def setUp(self) -> None:
         super().setUp()
         self.batch_client = DelayedWorkflowClient()
 
-    def push_to_hash(self, project_id, rule_id, group_id, event_id=None, occurrence_id=None):
+    def push_to_hash(
+        self,
+        project_id: int,
+        rule_id: str | int,
+        group_id: str | int,
+        event_id: str | int | None = None,
+        occurrence_id: str | int | None = None,
+    ) -> None:
         value = json.dumps({"event_id": event_id, "occurrence_id": occurrence_id})
         self.batch_client.for_project(project_id).push_to_hash(
             batch_key=None,
@@ -81,7 +88,7 @@ def test_fetches_from_buffer_and_executes_with_conditional_delete(
 
     @patch("sentry.workflow_engine.processors.schedule.process_in_batches")
     @override_options({"delayed_workflow.rollout": False})
-    def test_skips_processing_with_option(self, mock_process_in_batches) -> None:
+    def test_skips_processing_with_option(self, mock_process_in_batches: MagicMock) -> None:
         project = self.create_project()
         self.batch_client.add_project_ids([project.id])
 
@@ -197,12 +204,12 @@ def run_to_timestamp(run: int, interval_sec: int, jitter: bool = True) -> float:
 
 class TestProjectChooser:
     @pytest.fixture
-    def mock_buffer(self):
+    def mock_buffer(self) -> Mock:
         mock_buffer = Mock(spec=DelayedWorkflowClient)
         return mock_buffer
 
     @pytest.fixture
-    def project_chooser(self, mock_buffer):
+    def project_chooser(self, mock_buffer: Mock) -> ProjectChooser:
         return ProjectChooser(mock_buffer, num_cohorts=6, min_scheduling_age=timedelta(seconds=50))
 
     def _find_projects_for_cohorts(self, chooser: ProjectChooser, num_cohorts: int) -> list[int]:
@@ -218,7 +225,7 @@ def _find_projects_for_cohorts(self, chooser: ProjectChooser, num_cohorts: int)
             project_id += 1
         return all_project_ids
 
-    def test_project_id_to_cohort_distribution(self, project_chooser):
+    def test_project_id_to_cohort_distribution(self, project_chooser: ProjectChooser) -> None:
         project_ids = list(range(1, 1001))  # 1000 project IDs
         cohorts = [project_chooser._project_id_to_cohort(pid) for pid in project_ids]
 
@@ -230,7 +237,7 @@ def test_project_id_to_cohort_distribution(self, project_chooser):
         assert all(count > 0 for count in cohort_counts)
         assert all(count < 1000 for count in cohort_counts)
 
-    def test_project_id_to_cohort_consistent(self, project_chooser):
+    def test_project_id_to_cohort_consistent(self, project_chooser: ProjectChooser) -> None:
         for project_id in [123, 999, 4, 193848493]:
             cohort1 = project_chooser._project_id_to_cohort(project_id)
             cohort2 = project_chooser._project_id_to_cohort(project_id)
@@ -239,7 +246,9 @@ def test_project_id_to_cohort_consistent(self, project_chooser):
             assert cohort1 == cohort2 == cohort3
             assert 0 <= cohort1 < 6
 
-    def test_project_ids_to_process_must_process_over_minute(self, project_chooser):
+    def test_project_ids_to_process_must_process_over_minute(
+        self, project_chooser: ProjectChooser
+    ) -> None:
         fetch_time = 1000.0
         cohort_updates = CohortUpdates(
             values={
@@ -260,7 +269,9 @@ def test_project_ids_to_process_must_process_over_minute(self, project_chooser):
         # cohort_updates should be updated with fetch_time for processed cohorts
         assert 0 in cohort_updates.values
 
-    def test_project_ids_to_process_may_process_fallback(self, project_chooser):
+    def test_project_ids_to_process_may_process_fallback(
+        self, project_chooser: ProjectChooser
+    ) -> None:
         fetch_time = 1000.0
         cohort_updates = CohortUpdates(
             values={
@@ -282,7 +293,9 @@ def test_project_ids_to_process_may_process_fallback(self, project_chooser):
         for cohort in processed_cohorts:
             assert cohort_updates.values[cohort] == fetch_time
 
-    def test_project_ids_to_process_no_processing_needed(self, project_chooser):
+    def test_project_ids_to_process_no_processing_needed(
+        self, project_chooser: ProjectChooser
+    ) -> None:
         fetch_time = 1000.0
         cohort_updates = CohortUpdates(
             values={
@@ -369,7 +382,7 @@ def test_scenario_six_times_per_minute(self, project_chooser: ProjectChooser) ->
                 ), f"Run {run} should process all cohorts, got {processed_in_last_cycle}"
             processed_cohorts_over_time.append(processed_cohorts)
 
-    def test_scenario_once_per_minute_cohort_count_1(self, mock_buffer) -> None:
+    def test_scenario_once_per_minute_cohort_count_1(self, mock_buffer: Mock) -> None:
         """
         Scenario test: Running once per minute with cohort count of 1 (production default).
         This demonstrates that all projects are processed together every minute.
@@ -411,7 +424,7 @@ def test_scenario_once_per_minute_cohort_count_1(self, mock_buffer) -> None:
                 f"but got {len(processed_projects)}: {sorted(processed_projects)}"
             )
 
-    def test_cohort_count_change_uses_eldest_freshness(self, mock_buffer) -> None:
+    def test_cohort_count_change_uses_eldest_freshness(self, mock_buffer: Mock) -> None:
         """
         Test that when num_cohorts changes, all new cohorts use the eldest stored cohort freshness,
         then cohorts that need processing are scheduled and updated to current time.
@@ -440,11 +453,11 @@ def test_cohort_count_change_uses_eldest_freshness(self, mock_buffer) -> None:
 
 class TestChosenProjects:
     @pytest.fixture
-    def mock_project_chooser(self):
+    def mock_project_chooser(self) -> Mock:
         """Create a mock ProjectChooser."""
         return Mock(spec=ProjectChooser)
 
-    def test_chosen_projects_context_manager(self, mock_project_chooser):
+    def test_chosen_projects_context_manager(self, mock_project_chooser: Mock) -> None:
         """Test chosen_projects as a context manager."""
         # Setup mocks
         mock_cohort_updates = Mock(spec=CohortUpdates)
@@ -476,7 +489,7 @@ def test_chosen_projects_context_manager(self, mock_project_chooser):
         # Verify persist_updates was called after context exit
         mock_buffer_client.persist_updates.assert_called_once_with(mock_cohort_updates)
 
-    def test_chosen_projects_fetch_updates_exception(self, mock_project_chooser):
+    def test_chosen_projects_fetch_updates_exception(self, mock_project_chooser: Mock) -> None:
         """Test that exception during fetch_updates is properly handled."""
         # Setup mocks
         mock_buffer_client = Mock(spec=DelayedWorkflowClient)
@@ -495,7 +508,7 @@ def test_chosen_projects_fetch_updates_exception(self, mock_project_chooser):
         # persist_updates should not be called if fetch_updates fails
         mock_buffer_client.persist_updates.assert_not_called()
 
-    def test_chosen_projects_exception_during_processing(self, mock_project_chooser):
+    def test_chosen_projects_exception_during_processing(self, mock_project_chooser: Mock) -> None:
         mock_buffer_client = Mock(spec=DelayedWorkflowClient)
         mock_project_chooser.client = mock_buffer_client
         mock_buffer_client.fetch_updates.return_value = Mock(spec=CohortUpdates)

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -439,7 +439,7 @@ def test_many_workflows(self) -> None:
         assert triggered_workflows == {self.workflow, workflow_two}
 
     @patch.object(get_data_conditions_for_group, "batch")
-    def test_batched_data_condition_lookup_is_used(self, mock_batch):
+    def test_batched_data_condition_lookup_is_used(self, mock_batch: MagicMock) -> None:
         """Test that batch lookup is used when evaluating multiple workflows."""
         workflow_two, _, _, _ = self.create_detector_and_workflow(name_prefix="two")
 
@@ -816,7 +816,7 @@ def test_with_slow_conditions(self) -> None:
         assert not triggered_action_filters
 
     @patch.object(get_data_conditions_for_group, "batch")
-    def test_batched_data_condition_lookup_for_action_filters(self, mock_batch):
+    def test_batched_data_condition_lookup_for_action_filters(self, mock_batch: MagicMock) -> None:
         """Test that batch lookup is used when evaluating action filters."""
         # Create a second workflow with action filters
         workflow_two, _, _, _ = self.create_detector_and_workflow(name_prefix="two")
@@ -878,7 +878,7 @@ def test_activity__with_slow_conditions(self) -> None:
         # ensure we do not enqueue slow condition evaluation
         assert not queue_items
 
-    def test_enqueues_when_slow_conditions(self):
+    def test_enqueues_when_slow_conditions(self) -> None:
         assert isinstance(self.event_data.event, GroupEvent)
         queue_items_by_workflow_id = {
             self.workflow: DelayedWorkflowItem(