agent-platform/tests/test_workflow_designer_debugger.py

from __future__ import annotations

from tests.conftest import build_fastapi_test_client, prepare_known_service_import


prepare_known_service_import("workflow-service")

from app.bootstrap.app import create_app
from app.api.routes import get_workflow_application_service
from app.application.designer import WorkflowDebugPlan, build_debug_plan


def test_workflow_designer_validate_returns_graph_diagnostics() -> None:
    client = build_fastapi_test_client(create_app())

    response = client.post(
        "/workflows/designer/validate",
        json={
            "dsl_json": {
                "code": "order_flow",
                "name": "Order Flow",
                "nodes": [
                    {"id": "start", "type": "llm", "name": "Start"},
                    {"id": "answer", "type": "answer", "name": "Answer"},
                    {"id": "orphan", "type": "tool", "name": "Orphan Tool"},
                ],
                "edges": [{"source": "start", "target": "answer"}],
            }
        },
    )

    assert response.status_code == 200
    payload = response.json()
    assert payload["valid"] is True
    assert payload["node_count"] == 3
    assert payload["edge_count"] == 1
    assert payload["entry_node_ids"] == ["start", "orphan"]
    assert payload["terminal_node_ids"] == ["answer", "orphan"]
    assert payload["isolated_node_ids"] == ["orphan"]
    assert payload["unreachable_node_ids"] == []
    assert payload["diagnostics"][0]["code"] == "workflow.multiple_entry_nodes"
    assert payload["normalized_dsl_json"]["nodes"][0]["config"] == {}


def test_workflow_designer_validate_reports_blocking_errors() -> None:
    client = build_fastapi_test_client(create_app())

    response = client.post(
        "/workflows/designer/validate",
        json={
            "dsl_json": {
                "code": "broken_flow",
                "nodes": [{"id": "start", "type": "llm"}],
                "edges": [{"source": "start", "target": "missing"}],
            }
        },
    )

    assert response.status_code == 200
    payload = response.json()
    assert payload["valid"] is False
    assert payload["edges"][0]["valid_source"] is True
    assert payload["edges"][0]["valid_target"] is False
    assert any(item["code"] == "edge.target_missing" for item in payload["diagnostics"])


def test_workflow_debugger_returns_execution_preview() -> None:
    client = build_fastapi_test_client(create_app())

    response = client.post(
        "/workflows/designer/debug",
        json={
            "dsl_json": {
                "code": "debug_flow",
                "nodes": [
                    {"id": "start", "type": "llm"},
                    {"id": "tool", "type": "tool"},
                    {"id": "answer", "type": "answer"},
                ],
                "edges": [
                    {"source": "start", "target": "tool"},
                    {"source": "tool", "target": "answer"},
                ],
            },
            "max_preview_steps": 10,
        },
    )

    assert response.status_code == 200
    payload = response.json()
    assert payload["valid"] is True
    assert payload["truncated"] is False
    assert [item["node_id"] for item in payload["execution_preview"]] == [
        "start",
        "tool",
        "answer",
    ]
    assert payload["execution_preview"][0]["next_node_ids"] == ["tool"]


class FakeWorkflowDebugService:
    def build_version_debug_plan(
        self,
        *,
        tenant_id: str,
        workflow_version_id: str,
        max_preview_steps: int,
    ) -> WorkflowDebugPlan | None:
        assert tenant_id == "t1"
        assert workflow_version_id == "version-1"
        return build_debug_plan(
            {
                "code": "persisted_flow",
                "nodes": [
                    {"id": "start", "type": "llm"},
                    {"id": "answer", "type": "answer"},
                ],
                "edges": [{"source": "start", "target": "answer"}],
            },
            max_preview_steps=max_preview_steps,
        )


def test_workflow_version_debugger_api_uses_persisted_dsl_service() -> None:
    app = create_app()
    app.dependency_overrides[get_workflow_application_service] = lambda: FakeWorkflowDebugService()
    client = build_fastapi_test_client(app)

    debug_response = client.get(
        "/workflows/versions/version-1/debug",
        params={"tenant_id": "t1"},
    )
    assert debug_response.status_code == 200
    debug_payload = debug_response.json()
    assert debug_payload["valid"] is True
    assert [item["node_id"] for item in debug_payload["execution_preview"]] == [
        "start",
        "answer",
    ]