Spaces:
Runtime error
Runtime error
| import pytest | |
| from fastapi.testclient import TestClient | |
| from app.main import app | |
| client = TestClient(app) | |
| def test_health_endpoint(): | |
| res = client.get("/health") | |
| assert res.status_code == 200 | |
| assert res.json() == {"status": "ok"} | |
| def test_session_flow(): | |
| res = client.post("/api/v1/sessions") | |
| assert res.status_code == 200 | |
| data = res.json() | |
| session_id = data["session_id"] | |
| assert session_id is not None | |
| # Send message prompting clarification | |
| res = client.post( | |
| f"/api/v1/sessions/{session_id}/messages", | |
| json={"message": "I want to create an agent that receives requests and posts data."} | |
| ) | |
| assert res.status_code == 200 | |
| data = res.json() | |
| assert data["session_id"] == session_id | |
| assert data["status"] == "awaiting_clarification" | |
| assert len(data["questions"]) > 0 | |
| # Retrieve session details and history | |
| res = client.get(f"/api/v1/sessions/{session_id}") | |
| assert res.status_code == 200 | |
| data = res.json() | |
| assert data["session_id"] == session_id | |
| assert data["status"] == "awaiting_clarification" | |
| assert len(data["messages"]) == 2 | |
| def test_agent_validation(): | |
| invalid_spec = { | |
| "name": "Test Agent", | |
| "connection": {"type": "api", "config": {}}, | |
| "trigger": {"type": "manual", "config": {}}, | |
| "mode": "workflow" | |
| } | |
| res = client.post("/api/v1/agents/validate", json=invalid_spec) | |
| assert res.status_code == 200 | |
| assert res.json()["valid"] is False | |
| valid_spec = { | |
| "name": "Test Agent", | |
| "connection": {"type": "api", "config": {"base_url": "http://api.example.com"}}, | |
| "trigger": {"type": "manual", "config": {}}, | |
| "mode": "workflow" | |
| } | |
| res = client.post("/api/v1/agents/validate", json=valid_spec) | |
| assert res.status_code == 200 | |
| assert res.json()["valid"] is True | |
| def test_agent_crud_and_versions(): | |
| spec = { | |
| "name": "CRUD Agent", | |
| "description": "Agent to verify database storage", | |
| "connection": {"type": "none", "config": {}}, | |
| "trigger": {"type": "manual", "config": {}}, | |
| "mode": "workflow" | |
| } | |
| res = client.post("/api/v1/agents/", json=spec) | |
| assert res.status_code == 200 | |
| data = res.json() | |
| agent_id = data["id"] | |
| assert agent_id is not None | |
| assert data["name"] == "CRUD Agent" | |
| assert data["version"] == "1.0.0" | |
| res = client.get(f"/api/v1/agents/{agent_id}") | |
| assert res.status_code == 200 | |
| assert res.json()["name"] == "CRUD Agent" | |
| updated_spec = dict(spec) | |
| updated_spec["name"] = "Updated CRUD Agent" | |
| res = client.put(f"/api/v1/agents/{agent_id}", json=updated_spec) | |
| assert res.status_code == 200 | |
| assert res.json()["name"] == "Updated CRUD Agent" | |
| assert res.json()["version"] == "1.0.1" | |
| res = client.get(f"/api/v1/agents/{agent_id}/versions") | |
| assert res.status_code == 200 | |
| versions = res.json() | |
| assert len(versions) == 2 | |
| assert versions[0]["version"] == "1.0.1" | |
| assert versions[1]["version"] == "1.0.0" | |