AI-VideoAssistant/api/tests/test_workflows.py

"""Tests for workflow graph schema and router behavior."""


class TestWorkflowAPI:
    """Workflow CRUD and graph validation test cases."""

    def _minimal_nodes(self):
        return [
            {
                "id": "start_1",
                "name": "start_1",
                "type": "start",
                "isStart": True,
                "metadata": {"position": {"x": 80, "y": 80}},
            },
            {
                "id": "assistant_1",
                "name": "assistant_1",
                "type": "assistant",
                "metadata": {"position": {"x": 280, "y": 80}},
                "prompt": "You are the first assistant node.",
            },
        ]

    def test_create_workflow_with_canonical_graph(self, client):
        payload = {
            "name": "Canonical Graph",
            "nodes": self._minimal_nodes(),
            "edges": [
                {
                    "id": "edge_start_assistant",
                    "fromNodeId": "start_1",
                    "toNodeId": "assistant_1",
                    "condition": {"type": "always"},
                }
            ],
        }

        resp = client.post("/api/workflows", json=payload)
        assert resp.status_code == 200
        data = resp.json()
        assert data["name"] == "Canonical Graph"
        assert data["nodeCount"] == 2
        assert data["nodes"][0]["id"] == "start_1"
        assert data["edges"][0]["fromNodeId"] == "start_1"
        assert data["edges"][0]["toNodeId"] == "assistant_1"

    def test_create_workflow_with_legacy_graph(self, client):
        payload = {
            "name": "Legacy Graph",
            "nodes": [
                {
                    "name": "legacy_start",
                    "type": "conversation",
                    "isStart": True,
                    "metadata": {"position": {"x": 100, "y": 100}},
                },
                {
                    "name": "legacy_human",
                    "type": "human",
                    "metadata": {"position": {"x": 300, "y": 100}},
                },
            ],
            "edges": [
                {
                    "from": "legacy_start",
                    "to": "legacy_human",
                    "label": "人工",
                }
            ],
        }

        resp = client.post("/api/workflows", json=payload)
        assert resp.status_code == 200
        data = resp.json()
        assert data["nodes"][0]["type"] == "assistant"
        assert data["nodes"][1]["type"] == "human_transfer"
        assert data["edges"][0]["fromNodeId"] == "legacy_start"
        assert data["edges"][0]["toNodeId"] == "legacy_human"
        assert data["edges"][0]["condition"]["type"] == "contains"

    def test_create_workflow_without_start_node_fails(self, client):
        payload = {
            "name": "No Start",
            "nodes": [
                {"id": "node_1", "name": "node_1", "type": "assistant", "metadata": {"position": {"x": 0, "y": 0}}},
            ],
            "edges": [],
        }
        resp = client.post("/api/workflows", json=payload)
        assert resp.status_code == 422

    def test_create_workflow_with_invalid_edge_fails(self, client):
        payload = {
            "name": "Bad Edge",
            "nodes": self._minimal_nodes(),
            "edges": [
                {"id": "edge_bad", "fromNodeId": "missing", "toNodeId": "assistant_1", "condition": {"type": "always"}},
            ],
        }
        resp = client.post("/api/workflows", json=payload)
        assert resp.status_code == 422

    def test_update_workflow_nodes_and_edges(self, client):
        create_payload = {
            "name": "Before Update",
            "nodes": self._minimal_nodes(),
            "edges": [
                {
                    "id": "edge_start_assistant",
                    "fromNodeId": "start_1",
                    "toNodeId": "assistant_1",
                    "condition": {"type": "always"},
                }
            ],
        }
        create_resp = client.post("/api/workflows", json=create_payload)
        assert create_resp.status_code == 200
        workflow_id = create_resp.json()["id"]

        update_payload = {
            "name": "After Update",
            "nodes": [
                {
                    "id": "start_1",
                    "name": "start_1",
                    "type": "start",
                    "isStart": True,
                    "metadata": {"position": {"x": 50, "y": 50}},
                },
                {
                    "id": "assistant_2",
                    "name": "assistant_2",
                    "type": "assistant",
                    "metadata": {"position": {"x": 250, "y": 50}},
                    "prompt": "new prompt",
                },
                {
                    "id": "end_1",
                    "name": "end_1",
                    "type": "end",
                    "metadata": {"position": {"x": 450, "y": 50}},
                },
            ],
            "edges": [
                {
                    "id": "edge_start_assistant2",
                    "fromNodeId": "start_1",
                    "toNodeId": "assistant_2",
                    "condition": {"type": "always"},
                },
                {
                    "id": "edge_assistant2_end",
                    "fromNodeId": "assistant_2",
                    "toNodeId": "end_1",
                    "condition": {"type": "contains", "source": "user", "value": "结束"},
                },
            ],
        }

        update_resp = client.put(f"/api/workflows/{workflow_id}", json=update_payload)
        assert update_resp.status_code == 200
        updated = update_resp.json()
        assert updated["name"] == "After Update"
        assert updated["nodeCount"] == 3
        assert len(updated["nodes"]) == 3
        assert len(updated["edges"]) == 2