Spaces:
Sleeping
Sleeping
File size: 3,573 Bytes
f9adcbf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """
Tests for FastAPI endpoints
"""
import pytest
from fastapi.testclient import TestClient
import app.main as main_module
from app.core.orchestrator import Orchestrator
@pytest.fixture(scope="module")
def test_orchestrator():
"""Create orchestrator for testing"""
return Orchestrator()
@pytest.fixture(scope="module", autouse=True)
async def setup_orchestrator(test_orchestrator):
"""Initialize orchestrator before tests"""
await test_orchestrator.initialize()
# Inject into main module
main_module.orchestrator = test_orchestrator
yield
await test_orchestrator.shutdown()
client = TestClient(main_module.app)
def test_health_endpoint():
"""Test health check endpoint"""
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "version" in data
def test_query_format_endpoint(unformatted_python_code):
"""Test query endpoint for format task"""
response = client.post(
"/api/v1/query",
json={
"task": "format",
"code": unformatted_python_code,
"language": "python"
}
)
assert response.status_code == 200
data = response.json()
assert data["success"] is True
assert data["used_automata"] is True
assert "result" in data
assert "pipeline" in data
def test_query_fix_endpoint(buggy_python_code):
"""Test query endpoint for fix task"""
response = client.post(
"/api/v1/query",
json={
"task": "fix",
"code": buggy_python_code,
"language": "python"
}
)
assert response.status_code == 200
data = response.json()
assert data["success"] is True
assert "result" in data
def test_query_invalid_task():
"""Test query endpoint with invalid task"""
response = client.post(
"/api/v1/query",
json={
"task": "invalid_task",
"code": "def foo(): pass",
"language": "python"
}
)
# Should return validation error
assert response.status_code == 422
def test_query_missing_code():
"""Test query endpoint with missing code"""
response = client.post(
"/api/v1/query",
json={
"task": "format",
"language": "python"
}
)
# Should return validation error
assert response.status_code == 422
def test_query_with_context():
"""Test query endpoint with context parameter"""
response = client.post(
"/api/v1/query",
json={
"task": "boilerplate",
"code": "",
"language": "python",
"context": "Create a function to validate email addresses"
}
)
assert response.status_code == 200
data = response.json()
# May use SLM or fail gracefully
assert "success" in data
def test_query_with_trace(sample_trace):
"""Test query endpoint with trace parameter"""
response = client.post(
"/api/v1/query",
json={
"task": "explain",
"code": "def divide(a, b): return a / b",
"language": "python",
"trace": sample_trace
}
)
assert response.status_code == 200
data = response.json()
assert "success" in data
def test_stats_endpoint():
"""Test stats endpoint"""
response = client.get("/api/v1/stats")
# Stats endpoint may not exist yet in V1
# Just check it doesn't crash
assert response.status_code in [200, 404]
|