Spaces:
Sleeping
Sleeping
| """ | |
| Web Console Tests | |
| ================== | |
| Tests for the FastAPI web application and developer console. | |
| """ | |
| import pytest | |
| import sys | |
| import os | |
| from pathlib import Path | |
| # Add src to path | |
| sys.path.insert(0, str(Path(__file__).parent.parent / 'src')) | |
| from httpx import AsyncClient, ASGITransport | |
| from web.app import app | |
| class TestWebConsole: | |
| """Tests for the web console endpoints.""" | |
| async def test_health_endpoint(self): | |
| """Test that health endpoint returns healthy status.""" | |
| transport = ASGITransport(app=app) | |
| async with AsyncClient(transport=transport, base_url="http://test") as client: | |
| response = await client.get("/health") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] == "healthy" | |
| async def test_console_page_returns_html(self): | |
| """Test that root endpoint returns HTML console page.""" | |
| transport = ASGITransport(app=app) | |
| async with AsyncClient(transport=transport, base_url="http://test") as client: | |
| response = await client.get("/") | |
| assert response.status_code == 200 | |
| assert "text/html" in response.headers["content-type"] | |
| assert "AgentMask Developer Console" in response.text | |
| async def test_run_endpoint_returns_steps(self): | |
| """ | |
| Test that POST /run returns 200 and JSON with 'steps' key. | |
| """ | |
| transport = ASGITransport(app=app) | |
| async with AsyncClient(transport=transport, base_url="http://test") as client: | |
| response = await client.post( | |
| "/run", | |
| json={"query": "test query for AI research"} | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| # Assert response structure | |
| assert "steps" in data, "Response should contain 'steps' key" | |
| assert isinstance(data["steps"], list), "Steps should be a list" | |
| assert len(data["steps"]) >= 2, "Should have at least 2 steps" | |
| # Assert merkle root is present and valid format | |
| assert "merkle_root" in data | |
| assert len(data["merkle_root"]) == 64 | |
| # Assert each step has required fields | |
| for step in data["steps"]: | |
| assert "agent" in step | |
| assert "output" in step | |
| assert "hash" in step | |
| async def test_run_endpoint_with_different_queries(self): | |
| """Test that different queries produce results.""" | |
| transport = ASGITransport(app=app) | |
| async with AsyncClient(transport=transport, base_url="http://test") as client: | |
| response = await client.post( | |
| "/run", | |
| json={"query": "diabetes treatment options"} | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["success"] is True | |
| assert "final_output" in data | |
| assert "summary" in data["final_output"] | |
| async def test_run_endpoint_returns_final_output(self): | |
| """Test that run endpoint includes final output from summarizer.""" | |
| transport = ASGITransport(app=app) | |
| async with AsyncClient(transport=transport, base_url="http://test") as client: | |
| response = await client.post( | |
| "/run", | |
| json={"query": "machine learning applications"} | |
| ) | |
| data = response.json() | |
| assert "final_output" in data | |
| assert "summary" in data["final_output"] | |
| assert len(data["final_output"]["summary"]) > 10 | |
| if __name__ == "__main__": | |
| pytest.main([__file__, "-v"]) | |