dograh/api/tests/test_workflow_create_route.py

from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

from fastapi import FastAPI
from fastapi.testclient import TestClient

from api.routes.workflow import router
from api.services.auth.depends import get_user


def _make_test_app() -> FastAPI:
    app = FastAPI()
    app.include_router(router)
    app.dependency_overrides[get_user] = lambda: SimpleNamespace(
        id=1,
        provider_id="provider-1",
        selected_organization_id=11,
    )
    return app


def test_create_workflow_rejects_invalid_trigger_path_before_db_write():
    app = _make_test_app()
    client = TestClient(app)

    with patch("api.routes.workflow.db_client") as mock_db:
        response = client.post(
            "/workflow/create/definition",
            json={
                "name": "Support Agent",
                "workflow_definition": {
                    "nodes": [
                        {
                            "id": "trigger-1",
                            "type": "trigger",
                            "data": {"trigger_path": "support/west"},
                        }
                    ],
                    "edges": [],
                },
            },
        )

    assert response.status_code == 422
    detail = response.json()["detail"]
    assert detail["is_valid"] is False
    assert detail["errors"][0]["field"] == "data.trigger_path"
    assert "single URL path segment" in detail["errors"][0]["message"]
    assert mock_db.mock_calls == []


def test_create_workflow_rejects_duplicate_api_triggers_before_db_write():
    app = _make_test_app()
    client = TestClient(app)

    with patch("api.routes.workflow.db_client") as mock_db:
        response = client.post(
            "/workflow/create/definition",
            json={
                "name": "Support Agent",
                "workflow_definition": {
                    "nodes": [
                        {
                            "id": "trigger-1",
                            "type": "trigger",
                            "data": {"trigger_path": "support_west"},
                        },
                        {
                            "id": "trigger-2",
                            "type": "trigger",
                            "data": {"trigger_path": "support_east"},
                        },
                    ],
                    "edges": [],
                },
            },
        )

    assert response.status_code == 422
    detail = response.json()["detail"]
    assert detail["is_valid"] is False
    assert detail["errors"][0]["kind"] == "workflow"
    assert "at most one API Trigger" in detail["errors"][0]["message"]
    assert mock_db.mock_calls == []


def test_create_workflow_run_uses_draft_and_template_context():
    app = _make_test_app()
    client = TestClient(app)

    workflow = SimpleNamespace(
        id=33,
        released_definition=SimpleNamespace(
            id=77,
            template_context_variables={"name": "published"},
        ),
        current_definition=None,
        template_context_variables={"name": "workflow"},
    )
    draft = SimpleNamespace(
        id=88,
        template_context_variables={"name": "draft", "draft_only": "kept"},
    )
    run = SimpleNamespace(
        id=501,
        workflow_id=workflow.id,
        name="WR-test",
        mode="smallwebrtc",
        created_at=datetime.now(UTC),
        definition_id=draft.id,
        initial_context={"name": "draft", "draft_only": "kept"},
        gathered_context={},
    )

    with patch("api.routes.workflow.db_client") as mock_db:
        mock_db.get_workflow = AsyncMock(return_value=workflow)
        mock_db.get_draft_version = AsyncMock(return_value=draft)
        mock_db.create_workflow_run = AsyncMock(return_value=run)

        response = client.post(
            f"/workflow/{workflow.id}/runs",
            json={"name": "WR-test", "mode": "smallwebrtc"},
        )

    assert response.status_code == 200
    mock_db.get_draft_version.assert_awaited_once_with(workflow.id)
    create_kwargs = mock_db.create_workflow_run.await_args.kwargs
    assert create_kwargs["definition_id"] == draft.id
    assert create_kwargs["initial_context"] == {
        "name": "draft",
        "draft_only": "kept",
    }