Merge branch 'wa-fix-periodic-trigger-on-test' into 'develop'

Fix periodic trigger in simulation

See merge request baserow/baserow!3760

Author: jrmi

backend/src/baserow/contrib/automation/api/nodes/views.py

@@ -6,6 +6,7 @@
 from drf_spectacular.utils import OpenApiParameter, extend_schema
 from rest_framework.permissions import AllowAny, IsAuthenticated
 from rest_framework.response import Response
+from rest_framework.status import HTTP_202_ACCEPTED
 from rest_framework.views import APIView
 
 from baserow.api.decorators import (
@@ -434,15 +435,10 @@ class SimulateDispatchAutomationNodeView(APIView):
         }
     )
     def post(self, request, node_id: int):
-        updated_node = AutomationNodeService().simulate_dispatch_node(
-            request.user, node_id
+        AutomationWorkflowService().toggle_test_run(
+            request.user, simulate_until_node_id=node_id
         )
-
-        serializer = automation_node_type_registry.get_serializer(
-            updated_node, AutomationNodeSerializer
-        )
-
-        return Response(serializer.data)
+        return Response(status=HTTP_202_ACCEPTED)
 
 
 class MoveAutomationNodeView(APIView):

backend/src/baserow/contrib/automation/api/workflows/serializers.py

@@ -25,6 +25,7 @@ class Meta:
             "order",
             "automation_id",
             "allow_test_run_until",
+            "simulate_until_node_id",
             "published_on",
             "state",
         )

backend/src/baserow/contrib/automation/api/workflows/urls.py

@@ -3,6 +3,7 @@
 from baserow.contrib.automation.api.workflows.views import (
     AsyncAutomationDuplicateWorkflowView,
     AsyncPublishAutomationWorkflowView,
+    AutomationTestWorkflowView,
     AutomationWorkflowHistoryView,
     AutomationWorkflowsView,
     AutomationWorkflowView,
@@ -36,6 +37,11 @@
         AsyncPublishAutomationWorkflowView.as_view(),
         name="async_publish",
     ),
+    re_path(
+        r"(?P<workflow_id>[0-9]+)/test/$",
+        AutomationTestWorkflowView.as_view(),
+        name="test",
+    ),
     re_path(
         r"(?P<workflow_id>[0-9]+)/history/$",
         AutomationWorkflowHistoryView.as_view(),

backend/src/baserow/contrib/automation/api/workflows/views.py

@@ -406,3 +406,45 @@ def post(self, request, workflow_id: int):
         serializer = job_type_registry.get_serializer(job, JobSerializer)
 
         return Response(serializer.data, status=HTTP_202_ACCEPTED)
+
+
+class AutomationTestWorkflowView(APIView):
+    permission_classes = (IsAuthenticated,)
+
+    @extend_schema(
+        parameters=[
+            OpenApiParameter(
+                name="workflow_id",
+                location=OpenApiParameter.PATH,
+                type=OpenApiTypes.INT,
+                description="The workflow id the user wants to test.",
+            ),
+            CLIENT_SESSION_ID_SCHEMA_PARAMETER,
+        ],
+        tags=[AUTOMATION_WORKFLOWS_TAG],
+        operation_id="test_automation_workflow",
+        description=(
+            "This endpoint plan the execution of a test run for this workflow."
+        ),
+        request=None,
+        responses={
+            202: HTTP_202_ACCEPTED,
+            404: get_error_schema(["ERROR_AUTOMATION_WORKFLOW_DOES_NOT_EXIST"]),
+        },
+    )
+    @transaction.atomic
+    @map_exceptions(
+        {
+            AutomationWorkflowDoesNotExist: ERROR_AUTOMATION_WORKFLOW_DOES_NOT_EXIST,
+        }
+    )
+    def post(self, request, workflow_id: int):
+        """
+        Start the workflow asynchronously in test mode.
+        """
+
+        AutomationWorkflowService().toggle_test_run(
+            request.user, workflow_id=workflow_id
+        )
+
+        return Response(status=HTTP_202_ACCEPTED)

backend/src/baserow/contrib/automation/automation_dispatch_context.py

@@ -35,7 +35,6 @@ def __init__(
         self.workflow = workflow
         self.previous_nodes_results: Dict[int, Any] = {}
         self.dispatch_history: List[int] = []
-        self.event_payload = event_payload
         self.simulate_until_node = simulate_until_node
 
         services = (
@@ -53,6 +52,7 @@ def __init__(
             update_sample_data_for=services,
             use_sample_data=bool(self.simulate_until_node),
             force_outputs=force_outputs,
+            event_payload=event_payload,
         )
 
     def clone(self, **kwargs):

backend/src/baserow/contrib/automation/nodes/handler.py

@@ -1,5 +1,5 @@
 from collections import defaultdict
-from typing import Any, Dict, Iterable, List, Optional, Union
+from typing import Any, Dict, Iterable, List, Optional, Type, Union
 
 from django.core.files.storage import Storage
 from django.db.models import QuerySet
@@ -10,29 +10,34 @@
 from baserow.contrib.automation.models import AutomationWorkflow
 from baserow.contrib.automation.nodes.exceptions import (
     AutomationNodeDoesNotExist,
-    AutomationNodeError,
+    AutomationNodeMisconfiguredService,
     AutomationNodeNotInWorkflow,
-    AutomationNodeSimulateDispatchError,
 )
 from baserow.contrib.automation.nodes.models import AutomationActionNode, AutomationNode
-from baserow.contrib.automation.nodes.node_types import AutomationNodeType
+from baserow.contrib.automation.nodes.node_types import (
+    AutomationNodeActionNodeType,
+    AutomationNodeType,
+)
 from baserow.contrib.automation.nodes.registries import automation_node_type_registry
 from baserow.contrib.automation.nodes.types import (
     AutomationNodeDict,
     AutomationNodeDuplication,
     AutomationNodeMove,
     NextAutomationNodeValues,
 )
-from baserow.contrib.automation.workflows.runner import AutomationWorkflowRunner
 from baserow.core.cache import local_cache
 from baserow.core.db import specific_iterator
 from baserow.core.exceptions import IdDoesNotExist
-from baserow.core.services.exceptions import UnexpectedDispatchException
+from baserow.core.services.exceptions import (
+    ServiceImproperlyConfiguredDispatchException,
+)
 from baserow.core.services.handler import ServiceHandler
 from baserow.core.services.models import Service
 from baserow.core.storage import ExportZipFile
 from baserow.core.utils import MirrorDict, extract_allowed
 
+from .signals import automation_node_updated
+
 
 class AutomationNodeHandler:
     allowed_fields = ["label", "service", "previous_node_id", "previous_node_output"]
@@ -612,33 +617,36 @@ def import_node_only(
 
         return node_instance
 
-    def simulate_dispatch_node(self, node: AutomationNode) -> AutomationNode:
+    def dispatch_node(
+        self, node: "AutomationNode", dispatch_context: AutomationDispatchContext
+    ):
         """
-        Simulates a dispatch of the provided node. This will cause the node's
-        `service.sample_data` to be populated.
+        Dispatch one node and recursively dispatch the next nodes.
 
-        :param node: The node to simulate the dispatch for.
-        :return: The updated node.
+        :param node: The node to start with.
+        :param dispatch_context: The context in which the workflow is being dispatched,
+            which contains the event payload and other relevant data.
         """
 
-        if node.get_type().is_workflow_trigger:
-            node.workflow.simulate_until_node = node
-            node.workflow.save()
-            return node
-
-        dispatch_context = AutomationDispatchContext(
-            node.workflow,
-            simulate_until_node=node.specific,
-        )
-
+        node_type: Type[AutomationNodeActionNodeType] = node.get_type()
         try:
-            AutomationWorkflowRunner().run(node.workflow, dispatch_context)
-        except (
-            AutomationNodeError,
-            UnexpectedDispatchException,
-        ) as e:
-            raise AutomationNodeSimulateDispatchError(str(e))
-
-        node.refresh_from_db()
-
-        return node
+            dispatch_result = node_type.dispatch(node, dispatch_context)
+            dispatch_context.after_dispatch(node, dispatch_result)
+
+            # Return early if this is a simulated dispatch
+            if until_node := dispatch_context.simulate_until_node:
+                if until_node.id == node.id:
+                    # sample_data was updated as it's a simulation we should tell to
+                    # the frontend
+                    node.service.refresh_from_db(fields=["sample_data"])
+                    automation_node_updated.send(self, user=None, node=node)
+                    return
+
+            next_nodes = node.get_next_nodes(dispatch_result.output_uid)
+
+            for next_node in next_nodes:
+                self.dispatch_node(next_node, dispatch_context)
+        except ServiceImproperlyConfiguredDispatchException as e:
+            raise AutomationNodeMisconfiguredService(
+                f"The node {node.id} has a misconfigured service."
+            ) from e

backend/src/baserow/contrib/automation/nodes/node_types.py

@@ -7,9 +7,6 @@
 from django.utils import timezone
 from django.utils.translation import gettext as _
 
-from baserow.contrib.automation.automation_dispatch_context import (
-    AutomationDispatchContext,
-)
 from baserow.contrib.automation.nodes.exceptions import (
     AutomationNodeMisconfiguredService,
     AutomationNodeNotDeletable,
@@ -53,32 +50,11 @@
 )
 from baserow.core.db import specific_iterator
 from baserow.core.registry import Instance
-from baserow.core.services.handler import ServiceHandler
 from baserow.core.services.models import Service
 from baserow.core.services.registries import service_type_registry
-from baserow.core.services.types import DispatchResult
-
-
-class ServiceDispatchableAutomationNodeTypeMixin:
-    """
-    A mixin for automation node types that can be dispatched using the service.
-    This mixin provides a default implementation of the `dispatch` method that uses
-    the `ServiceHandler` to dispatch the service associated with the node.
-    """
-
-    def dispatch(
-        self,
-        automation_node: AutomationNode,
-        dispatch_context: AutomationDispatchContext,
-    ) -> DispatchResult:
-        return ServiceHandler().dispatch_service(
-            automation_node.service.specific, dispatch_context
-        )
 
 
-class AutomationNodeActionNodeType(
-    ServiceDispatchableAutomationNodeTypeMixin, AutomationNodeType
-):
+class AutomationNodeActionNodeType(AutomationNodeType):
     is_workflow_action = True
 
 
@@ -238,19 +214,6 @@ class AutomationNodeTriggerType(AutomationNodeType):
 
     is_workflow_trigger = True
 
-    def dispatch(
-        self,
-        node: AutomationNode,
-        dispatch_context: AutomationDispatchContext,
-    ) -> DispatchResult:
-        if dispatch_context.use_sample_data:
-            if sample_data := node.service.get_type().get_sample_data(
-                node.service.specific
-            ):
-                return DispatchResult(**sample_data)
-
-        return DispatchResult(data=dispatch_context.event_payload)
-
     def after_register(self):
         service_type_registry.get(self.service_type).start_listening(self.on_event)
         return super().after_register()
@@ -298,31 +261,14 @@ def on_event(
 
         for trigger in triggers:
             workflow = trigger.workflow
-            simulate_until_node_id = (
-                trigger.workflow.simulate_until_node.id
-                if trigger.workflow.simulate_until_node
-                else None
-            )
-            AutomationWorkflowHandler().run_workflow(
+
+            AutomationWorkflowHandler().async_start_workflow(
                 workflow,
                 event_payload,
-                simulate_until_node_id,
             )
 
-            save_sample_data = False
-            if workflow.allow_test_run_until:
-                workflow.allow_test_run_until = None
-                workflow.save(update_fields=["allow_test_run_until"])
-                save_sample_data = True
-
-            if workflow.simulate_until_node and not workflow.is_published:
-                workflow.simulate_until_node = None
-                workflow.save(update_fields=["simulate_until_node"])
-                save_sample_data = True
-
-            if save_sample_data:
-                trigger.service.sample_data = {"data": event_payload}
-                trigger.service.save()
+            # We don't want subsequent events to trigger a new test run
+            AutomationWorkflowHandler().reset_workflow_temporary_states(workflow)
 
 
 class LocalBaserowRowsCreatedNodeTriggerType(AutomationNodeTriggerType):
@@ -343,32 +289,7 @@ class LocalBaserowRowsDeletedNodeTriggerType(AutomationNodeTriggerType):
     service_type = LocalBaserowRowsDeletedServiceType.type
 
 
-class AutomationNodeImmediateTriggerTypeMixin:
-    # On a workflow test run, this node type will be immediately dispatched,
-    # it does not wait for an initial internal or external event to occur.
-    immediate_dispatch = True
-
-    def dispatch(
-        self,
-        automation_node: AutomationNode,
-        dispatch_context: AutomationDispatchContext,
-    ) -> DispatchResult:
-        """
-        Immediate nodes are generally able to generate their own payload so unless
-        the context has an event payload, we directly try to generate one from
-        the service.
-        """
-
-        if dispatch_context.event_payload is not None:
-            return super().dispatch(automation_node, dispatch_context)
-
-        return ServiceHandler().dispatch_service(
-            automation_node.service.specific, dispatch_context
-        )
-
-
 class CorePeriodicTriggerNodeType(
-    AutomationNodeImmediateTriggerTypeMixin,
     AutomationNodeTriggerType,
 ):
     type = "periodic"

backend/src/baserow/contrib/automation/nodes/receivers.py

@@ -1,33 +1,10 @@
-from django.db import transaction
 from django.db.models.signals import post_delete
-from django.dispatch import receiver
 
 from baserow.contrib.automation.nodes.models import AutomationNode
-from baserow.contrib.automation.workflows.handler import AutomationWorkflowHandler
-from baserow.contrib.automation.workflows.models import AutomationWorkflow
-from baserow.contrib.automation.workflows.signals import automation_workflow_updated
 from baserow.core.services.handler import ServiceHandler
 from baserow.core.services.models import Service
 
 
-@receiver(automation_workflow_updated)
-def on_workflow_updated_test_run_dispatch_immediate_triggers(
-    sender, workflow: AutomationWorkflow, **kwargs
-):
-    def run_workflow():
-        if workflow.allow_test_run_until:
-            trigger = workflow.get_trigger()
-            # A subset of triggers support immediate test run dispatching, if this
-            # `node_type` supports it, we'll immediately run the workflow with
-            # the pre-defined data.
-            if trigger.get_type().immediate_dispatch:
-                AutomationWorkflowHandler().run_workflow(workflow, None)
-
-    # We need to wait the commit otherwise the celery task workflow instance doesn't
-    # have the allow_test_run_until property saved
-    transaction.on_commit(run_workflow)
-
-
 def after_permanently_deleted(sender, instance, **kwargs):
     """
     Delete the service related to the node.