Merge pull request #52 from wayfair-incubator/fix_stop_on_terminal_dag

fix for stopping dags in terminal state

dagger/tasks/task.py

@@ -1177,6 +1177,11 @@ async def stop(
         :param runtime_parameters: The runtime parameters of the task
         :param workflow_instance: The workflow object
         """
+        if self.status.code in TERMINAL_STATUSES:
+            logger.info(
+                f"Stop called in workflow in terminal state {workflow_instance.id}"
+            )
+            return
         remaining_tasks: Optional[List[ITask]] = await self.get_remaining_tasks(
             next_dag_id=self.root_dag, workflow_instance=self, tasks=[]  # type: ignore
         )  # type: ignore

tests/tasks/test_task.py

@@ -326,6 +326,18 @@ async def test_parallel_composite_task_notify_atleast_one_one_complete(
         )
         assert parallel_composite_task_fixture.on_complete.called
 
+    @pytest.mark.asyncio
+    async def test_stop_workflow_already_stopped(self, template_fixture):
+        template_fixture.on_complete = CoroutineMock()
+        template_fixture.status = TaskStatus(
+            code=TaskStatusEnum.COMPLETED.name, value=TaskStatusEnum.COMPLETED.value
+        )
+        await template_fixture.stop(
+            runtime_parameters=template_fixture.runtime_parameters,
+            workflow_instance=template_fixture,
+        )
+        template_fixture.on_complete.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_stop_workflow(
         self, template_fixture, executor_fixture, sensor_fixture, decision_fixture