fixup

Author: kcons

src/sentry/workflow_engine/buffer/__init__.py

@@ -5,16 +5,16 @@
 from sentry.buffer.base import Buffer
 from sentry.utils.services import LazyServiceWrapper
 
+# Workflows-specific Buffer that can be configured separately from the default Buffer.
 _backend = LazyServiceWrapper(
     Buffer, settings.SENTRY_WORKFLOW_BUFFER, settings.SENTRY_WORKFLOW_BUFFER_OPTIONS
 )
 
 
-def validate_new_backend() -> None:
-    pass
-
-
 def get_backend() -> LazyServiceWrapper[Buffer]:
+    """
+    Retrieve the appropriate Buffer to use for the workflow engine.
+    """
     if options.get("workflow_engine.buffer.use_new_buffer"):
         return _backend
     else:

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -4,18 +4,20 @@
 import pytest
 from django.utils import timezone
 
-from sentry import buffer
+from sentry.eventstore.models import GroupEvent
 from sentry.eventstream.base import GroupState
 from sentry.grouping.grouptype import ErrorGroupType
 from sentry.models.activity import Activity
 from sentry.models.environment import Environment
 from sentry.services.eventstore.models import GroupEvent
 from sentry.testutils.factories import Factories
 from sentry.testutils.helpers.datetime import before_now, freeze_time
+from sentry.testutils.helpers.options import override_options
 from sentry.testutils.helpers.redis import mock_redis_buffer
 from sentry.testutils.pytest.fixtures import django_db_all
 from sentry.types.activity import ActivityType
 from sentry.utils import json
+from sentry.workflow_engine import buffer as workflow_buffer
 from sentry.workflow_engine.models import (
     Action,
     DataConditionGroup,
@@ -440,10 +442,15 @@ def test_enqueues_workflow_all_logic_type(self) -> None:
 
         process_workflows(self.event_data)
 
+<<<<<<< HEAD
         project_ids = buffer.backend.bulk_get_sorted_set(
             self.buffer_keys,
             min=0,
             max=self.buffer_timestamp,
+=======
+        project_ids = workflow_buffer.get_backend().get_sorted_set(
+            WORKFLOW_ENGINE_BUFFER_LIST_KEY, 0, self.buffer_timestamp
+>>>>>>> 72e32fd402c (fixup)
         )
         assert list(project_ids.keys()) == [self.project.id]
 
@@ -472,10 +479,15 @@ def test_enqueues_workflow_any_logic_type(self) -> None:
 
         process_workflows(self.event_data)
 
+<<<<<<< HEAD
         project_ids = buffer.backend.bulk_get_sorted_set(
             self.buffer_keys,
             min=0,
             max=self.buffer_timestamp,
+=======
+        project_ids = workflow_buffer.get_backend().get_sorted_set(
+            WORKFLOW_ENGINE_BUFFER_LIST_KEY, 0, self.buffer_timestamp
+>>>>>>> 72e32fd402c (fixup)
         )
         assert list(project_ids.keys()) == [self.project.id]
 
@@ -554,10 +566,15 @@ def test_enqueues_with_when_and_if_slow_conditions(self) -> None:
 
         process_workflows(self.event_data)
 
+<<<<<<< HEAD
         project_ids = buffer.backend.bulk_get_sorted_set(
             self.buffer_keys,
             min=0,
             max=self.buffer_timestamp,
+=======
+        project_ids = workflow_buffer.get_backend().get_sorted_set(
+            WORKFLOW_ENGINE_BUFFER_LIST_KEY, 0, self.buffer_timestamp
+>>>>>>> 72e32fd402c (fixup)
         )
         assert list(project_ids.keys()) == [self.project.id]
 
@@ -583,10 +600,15 @@ def test_enqueues_event_if_meets_fast_conditions(self) -> None:
 
         process_workflows(self.event_data)
 
+<<<<<<< HEAD
         project_ids = buffer.backend.bulk_get_sorted_set(
             self.buffer_keys,
             min=0,
             max=self.buffer_timestamp,
+=======
+        project_ids = workflow_buffer.get_backend().get_sorted_set(
+            WORKFLOW_ENGINE_BUFFER_LIST_KEY, 0, self.buffer_timestamp
+>>>>>>> 72e32fd402c (fixup)
         )
         assert not project_ids
 
@@ -595,10 +617,15 @@ def test_enqueues_event_if_meets_fast_conditions(self) -> None:
 
         process_workflows(self.event_data)
 
+<<<<<<< HEAD
         project_ids = buffer.backend.bulk_get_sorted_set(
             self.buffer_keys,
             min=0,
             max=self.buffer_timestamp,
+=======
+        project_ids = workflow_buffer.get_backend().get_sorted_set(
+            WORKFLOW_ENGINE_BUFFER_LIST_KEY, 0, self.buffer_timestamp
+>>>>>>> 72e32fd402c (fixup)
         )
         assert list(project_ids.keys()) == [self.project.id]
 
@@ -748,10 +775,15 @@ def test_enqueues_when_slow_conditions(self):
 
         enqueue_workflows(queue_items)
 
+<<<<<<< HEAD
         project_ids = buffer.backend.bulk_get_sorted_set(
             self.buffer_keys,
             min=0,
             max=timezone.now().timestamp(),
+=======
+        project_ids = workflow_buffer.get_backend().get_sorted_set(
+            WORKFLOW_ENGINE_BUFFER_LIST_KEY, 0, timezone.now().timestamp()
+>>>>>>> 72e32fd402c (fixup)
         )
         assert list(project_ids.keys()) == [self.project.id]
 
@@ -787,7 +819,11 @@ def setUp(self) -> None:
 
     @patch("sentry.buffer.backend.push_to_sorted_set")
     @patch("sentry.buffer.backend.push_to_hash_bulk")
+<<<<<<< HEAD
     @patch("random.choice")
+=======
+    @override_options({"workflow_engine.buffer.use_new_buffer": False})
+>>>>>>> 72e32fd402c (fixup)
     def test_enqueue_workflows__adds_to_workflow_engine_buffer(
         self, mock_randchoice, mock_push_to_hash_bulk, mock_push_to_sorted_set
     ):
@@ -810,8 +846,33 @@ def test_enqueue_workflows__adds_to_workflow_engine_buffer(
             value=[self.group_event.project_id],
         )
 
+    @patch("sentry.workflow_engine.buffer._backend.push_to_sorted_set")
+    @patch("sentry.workflow_engine.buffer._backend.push_to_hash_bulk")
+    @override_options({"workflow_engine.buffer.use_new_buffer": True})
+    def test_enqueue_workflows__adds_to_workflow_engine_buffer_new_buffer(
+        self, mock_push_to_hash_bulk, mock_push_to_sorted_set
+    ):
+        enqueue_workflows(
+            {
+                self.workflow: DelayedWorkflowItem(
+                    self.workflow,
+                    self.group_event,
+                    self.workflow.when_condition_group_id,
+                    [self.slow_workflow_filter_group.id],
+                    [self.workflow_filter_group.id],
+                    timestamp=timezone.now(),
+                )
+            }
+        )
+
+        mock_push_to_sorted_set.assert_called_once_with(
+            key=WORKFLOW_ENGINE_BUFFER_LIST_KEY,
+            value=[self.group_event.project_id],
+        )
+
     @patch("sentry.buffer.backend.push_to_sorted_set")
     @patch("sentry.buffer.backend.push_to_hash_bulk")
+    @override_options({"workflow_engine.buffer.use_new_buffer": False})
     def test_enqueue_workflow__adds_to_workflow_engine_set(
         self, mock_push_to_hash_bulk, mock_push_to_sorted_set
     ):