backend/test_engine.py

import pytest

from backend.engine import WorkflowEngine
from backend.models import StartRequest


def node(node_id, mode, **data):
    return {"id": node_id, "position": {"x": 0, "y": 0}, "data": {"label": data.pop("label", mode), "mode": mode, **data}}


def edge(edge_id, source, target, connection_type="flow", label=""):
    return {"id": edge_id, "source": source, "target": target, "label": label, "data": {"connectionType": connection_type}}


def test_ai_worker_calls_attached_simulated_tool_in_demo_mode(monkeypatch):
    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
    request = StartRequest.model_validate({"workflow": {"nodes": [
        node("i", "Input", value="Order 42 is damaged"),
        node("w", "AI Worker", prompt="Investigate and respond"),
        node("t", "Simulated Tool", label="Order lookup", description="Retrieves order details", input_guidance="Order ID", prompt="Return a plausible order"),
        node("o", "Output"),
    ], "edges": [edge("a", "i", "w"), edge("b", "w", "o"), edge("tool", "w", "t", "tool")]}})
    result = WorkflowEngine().start(request)
    assert result.status == "completed"
    assert "Order lookup" in result.result
    assert any(event.kind == "tool" and event.node_id == "t" for event in result.trace)


def test_human_review_pauses_and_resumes(monkeypatch):
    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
    request = StartRequest.model_validate({"workflow": {"nodes": [node("i", "Input", value="case"), node("h", "Human Review"), node("o", "Output")], "edges": [edge("a", "i", "h"), edge("b", "h", "o")]}})
    engine = WorkflowEngine()
    paused = engine.start(request)
    assert paused.status == "awaiting_review"
    finished = engine.resume(paused.run_id, "Approve")
    assert finished.status == "completed"
    assert finished.result == "Approve"


def test_tool_connection_must_target_simulated_tool(monkeypatch):
    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
    request = StartRequest.model_validate({"workflow": {"nodes": [node("i", "Input", value="case"), node("w", "AI Worker"), node("o", "Output")], "edges": [edge("a", "i", "w"), edge("b", "w", "o"), edge("bad", "w", "o", "tool")]}})
    with pytest.raises(ValueError, match="Simulated Tool"):
        WorkflowEngine().start(request)