fix(workflow-tags): handle duplicate links and harden tests

Author: daryllimyt

tests/unit/api/test_api_workflow_tags.py

@@ -0,0 +1,64 @@
+"""HTTP-level tests for workflow tags API endpoints."""
+
+import uuid
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import status
+from fastapi.testclient import TestClient
+from sqlalchemy.exc import NoResultFound
+
+from tracecat.auth.types import Role
+from tracecat.workflow.tags import router as workflow_tags_router
+
+
+@pytest.mark.anyio
+async def test_add_tag_conflict_on_duplicate_assignment(
+    client: TestClient,
+    test_admin_role: Role,
+) -> None:
+    """POST /workflows/{workflow_id}/tags returns 409 on duplicate assignment."""
+    workflow_id = uuid.UUID("aaaaaaaa-aaaa-4aaa-aaaa-aaaaaaaaaaaa")
+    tag_id = uuid.UUID("bbbbbbbb-bbbb-4bbb-bbbb-bbbbbbbbbbbb")
+
+    with patch.object(workflow_tags_router, "WorkflowTagsService") as mock_service_cls:
+        mock_service = AsyncMock()
+        mock_service.add_workflow_tag.side_effect = ValueError(
+            "Tag already assigned to workflow"
+        )
+        mock_service_cls.return_value = mock_service
+
+        response = client.post(
+            f"/workflows/{workflow_id}/tags",
+            params={"workspace_id": str(test_admin_role.workspace_id)},
+            json={"tag_id": str(tag_id)},
+        )
+
+    assert response.status_code == status.HTTP_409_CONFLICT
+    assert response.json()["detail"] == "Tag already assigned to workflow"
+
+
+@pytest.mark.anyio
+async def test_add_tag_not_found_when_workflow_or_tag_missing(
+    client: TestClient,
+    test_admin_role: Role,
+) -> None:
+    """POST /workflows/{workflow_id}/tags returns 404 when resources are missing."""
+    workflow_id = uuid.UUID("aaaaaaaa-aaaa-4aaa-aaaa-aaaaaaaaaaaa")
+    tag_id = uuid.UUID("bbbbbbbb-bbbb-4bbb-bbbb-bbbbbbbbbbbb")
+
+    with patch.object(workflow_tags_router, "WorkflowTagsService") as mock_service_cls:
+        mock_service = AsyncMock()
+        mock_service.add_workflow_tag.side_effect = NoResultFound(
+            "Workflow or tag not found"
+        )
+        mock_service_cls.return_value = mock_service
+
+        response = client.post(
+            f"/workflows/{workflow_id}/tags",
+            params={"workspace_id": str(test_admin_role.workspace_id)},
+            json={"tag_id": str(tag_id)},
+        )
+
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json()["detail"] == "Workflow or tag not found"

tests/unit/test_tags_service.py

@@ -43,14 +43,14 @@ async def other_workspace(session: AsyncSession, svc_workspace: Workspace) -> Wo
 
 
 @pytest.fixture
-async def other_role(other_workspace: Workspace) -> Role:
-    """Role scoped to the secondary workspace."""
-    return Role(
-        type="user",
-        workspace_id=other_workspace.id,
-        organization_id=other_workspace.organization_id,
-        user_id=uuid4(),
-        service_id="tracecat-api",
+async def other_role(svc_role: Role, other_workspace: Workspace) -> Role:
+    """Clone base service role for the secondary workspace."""
+    return svc_role.model_copy(
+        update={
+            "workspace_id": other_workspace.id,
+            "organization_id": other_workspace.organization_id,
+            "user_id": uuid4(),
+        }
     )
 
 
@@ -372,6 +372,20 @@ async def test_remove_workflow_tag(
         workflow_tags = await workflow_tags_service.list_tags_for_workflow(workflow_id)
         assert len(workflow_tags) == 0
 
+    async def test_add_workflow_tag_duplicate_raises_value_error(
+        self,
+        workflow_tags_service: WorkflowTagsService,
+        tags_service: TagsService,
+        tag_create_params: TagCreate,
+        workflow_id: WorkflowID,
+    ) -> None:
+        """Adding the same tag twice to a workflow should raise a conflict error."""
+        tag = await tags_service.create_tag(tag_create_params)
+        await workflow_tags_service.add_workflow_tag(workflow_id, tag.id)
+
+        with pytest.raises(ValueError, match="already assigned"):
+            await workflow_tags_service.add_workflow_tag(workflow_id, tag.id)
+
     async def test_list_tags_for_workflow_enforces_workspace_scope(
         self,
         workflow_tags_service: WorkflowTagsService,

tracecat/workflow/tags/router.py

@@ -39,6 +39,11 @@ async def add_tag(
     service = WorkflowTagsService(session, role=role)
     try:
         await service.add_workflow_tag(workflow_id, params.tag_id)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(e),
+        ) from e
     except NoResultFound as e:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,

tracecat/workflow/tags/service.py

@@ -1,7 +1,7 @@
 from collections.abc import Sequence
 
 from sqlalchemy import exists, select
-from sqlalchemy.exc import NoResultFound
+from sqlalchemy.exc import IntegrityError, NoResultFound
 
 from tracecat.authz.controls import require_scope
 from tracecat.db.models import Workflow, WorkflowTag, WorkflowTagLink
@@ -72,7 +72,11 @@ async def add_workflow_tag(
         await self._require_workflow_and_tag_in_workspace(wf_id, tag_id)
         wf_tag = WorkflowTagLink(workflow_id=wf_id, tag_id=tag_id)
         self.session.add(wf_tag)
-        await self.session.commit()
+        try:
+            await self.session.commit()
+        except IntegrityError as e:
+            await self.session.rollback()
+            raise ValueError("Tag already assigned to workflow") from e
         return wf_tag
 
     @require_scope("workflow:update")