Spaces:
Sleeping
Sleeping
| 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) | |