feat(delete-worflow): move just not-empty statuses

python/apps/taiga/src/taiga/workflows/api/__init__.py

@@ -181,10 +181,7 @@ async def delete_workflow(
     workflow = await get_workflow_or_404(project_id=project_id, workflow_slug=workflow_slug)
     await check_permissions(permissions=DELETE_WORKFLOW, user=request.user, obj=workflow)
 
-    await workflows_services.delete_workflow(
-        workflow=workflow,
-        target_workflow_slug=query_params.move_to,
-    )
+    await workflows_services.delete_workflow(workflow=workflow, target_workflow_slug=query_params.move_to)
 
 
 ################################################

python/apps/taiga/src/taiga/workflows/repositories.py

@@ -15,7 +15,7 @@
 from uuid import UUID
 
 from asgiref.sync import sync_to_async
-from taiga.base.db.models import QuerySet
+from taiga.base.db.models import Count, QuerySet
 from taiga.base.repositories import neighbors as neighbors_repositories
 from taiga.base.repositories.neighbors import Neighbor
 from taiga.projects.projects.models import Project, ProjectTemplate
@@ -187,6 +187,7 @@ class WorkflowStatusFilters(TypedDict, total=False):
     workflow_id: UUID
     workflow_slug: str
     project_id: UUID
+    is_empty: bool
 
 
 def _apply_filters_to_workflow_status_queryset(
@@ -204,6 +205,13 @@ def _apply_filters_to_workflow_status_queryset(
     if "project_id" in filter_data:
         filter_data["workflow__project_id"] = filter_data.pop("project_id")
 
+    if "is_empty" in filter_data:
+        qs = qs.annotate(num_stories=Count("stories"))
+        if filter_data.pop("is_empty"):
+            filter_data["num_stories"] = 0
+        else:
+            filter_data["num_stories__gt"] = 0
+
     return qs.filter(**filter_data)
 
 

python/apps/taiga/src/taiga/workflows/services/__init__.py

@@ -188,7 +188,7 @@ async def delete_workflow(workflow: Workflow, target_workflow_slug: str | None =
             raise ex.SameMoveToWorkflow("The to-be-deleted workflow and the target-workflow cannot be the same")
 
         statuses_to_move = await workflows_repositories.list_workflow_statuses(
-            filters={"workflow_id": workflow.id},
+            filters={"workflow_id": workflow.id, "is_empty": False},
             order_by=["order"],
         )
 

python/apps/taiga/tests/integration/taiga/workflows/test_repositories.py

@@ -6,6 +6,7 @@
 # Copyright (c) 2023-present Kaleidos INC
 
 import uuid
+from unittest import IsolatedAsyncioTestCase
 
 import pytest
 from asgiref.sync import sync_to_async
@@ -131,18 +132,37 @@ async def test_create_workflow_status():
 ##########################################################
 
 
-async def test_list_workflows_statuses_ok() -> None:
-    workflow = await f.create_workflow()
-    statuses = await repositories.list_workflow_statuses(filters={"workflow_id": workflow.id})
-
-    assert len(statuses) > 0
-
-
-async def test_list_no_workflows_statuses() -> None:
-    workflow = await f.create_workflow(statuses=[])
-    statuses = await repositories.list_workflow_statuses(filters={"workflow_id": workflow.id})
-
-    assert len(statuses) == 0
+class ListWorkflowStatuses(IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.empty_workflow = await f.create_workflow(statuses=[])
+        self.workflow = await f.create_workflow(statuses=[])
+        self.workflow_status = await f.create_workflow_status(workflow=self.workflow)
+        await f.create_story(status=self.workflow_status, workflow=self.workflow)
+        self.empty_workflow_status = await f.create_workflow_status(workflow=self.workflow)
+
+    async def test_list_workflows_statuses_ok(self) -> None:
+        statuses = await repositories.list_workflow_statuses(filters={"workflow_id": self.workflow.id})
+        assert len(statuses) > 0
+
+    async def test_list_no_workflows_statuses(self) -> None:
+        statuses = await repositories.list_workflow_statuses(filters={"workflow_id": self.empty_workflow.id})
+        assert len(statuses) == 0
+
+    async def test_list_empty_statuses(self) -> None:
+        statuses = await repositories.list_workflow_statuses(
+            filters={"workflow_id": self.workflow.id, "is_empty": True}
+        )
+        assert self.empty_workflow_status in statuses
+        assert self.workflow_status not in statuses
+        assert len(statuses) == 1
+
+    async def test_list_not_empty_statuses(self) -> None:
+        statuses = await repositories.list_workflow_statuses(
+            filters={"workflow_id": self.workflow.id, "is_empty": False}
+        )
+        assert self.workflow_status in statuses
+        assert self.empty_workflow_status not in statuses
+        assert len(statuses) == 1
 
 
 ##########################################################

python/apps/taiga/tests/unit/taiga/workflows/test_services.py

@@ -361,7 +361,7 @@ async def test_delete_workflow_with_target_workflow_with_anchor_status_ok():
         # asserts
         fake_workflows_repo.list_workflow_statuses.assert_has_awaits(
             [
-                call(filters={"workflow_id": deleted_workflow.id}, order_by=["order"]),
+                call(filters={"workflow_id": deleted_workflow.id, "is_empty": False}, order_by=["order"]),
                 call(filters={"workflow_id": target_workflow.id}, order_by=["-order"], offset=0, limit=1),
             ]
         )
@@ -434,7 +434,7 @@ async def test_delete_workflow_with_target_workflow_with_no_anchor_status_ok():
         # asserts
         fake_workflows_repo.list_workflow_statuses.assert_has_awaits(
             [
-                call(filters={"workflow_id": deleted_workflow.id}, order_by=["order"]),
+                call(filters={"workflow_id": deleted_workflow.id, "is_empty": False}, order_by=["order"]),
                 call(filters={"workflow_id": target_workflow.id}, order_by=["-order"], offset=0, limit=1),
             ]
         )