Spaces:
Sleeping
Sleeping
File size: 5,082 Bytes
8807ee2 |
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 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 |
"""
Integration tests for FastAPI application.
"""
from fastapi.testclient import TestClient
class TestHealthEndpoint:
"""Tests for health check endpoint."""
def test_health_check(self):
"""Health endpoint should return 200."""
from app.main import app
client = TestClient(app)
response = client.get("/health")
assert response.status_code == 200
class TestDocsEndpoint:
"""Tests for documentation endpoints."""
def test_docs_accessible(self):
"""Swagger docs should be accessible."""
from app.main import app
client = TestClient(app)
response = client.get("/docs")
assert response.status_code == 200
def test_openapi_schema(self):
"""OpenAPI schema should be available."""
from app.main import app
client = TestClient(app)
response = client.get("/openapi.json")
assert response.status_code == 200
data = response.json()
assert "openapi" in data
assert "paths" in data
class TestRootRedirect:
"""Tests for root path handling."""
def test_root_redirects(self):
"""Root should redirect to docs."""
from app.main import app
client = TestClient(app, follow_redirects=False)
response = client.get("/")
# Should redirect (307 or 302)
assert response.status_code in [302, 307, 200]
class TestGenerateEndpoint:
"""Tests for generate API endpoint."""
def test_generate_requires_body(self):
"""Generate endpoint should require request body."""
from app.main import app
client = TestClient(app)
response = client.post("/api/generate")
# Should return 422 (validation error) without body
assert response.status_code == 422
def test_generate_validates_schema_correctly(self):
"""Generate endpoint should validate request schema."""
from app.main import app
client = TestClient(app)
# Wrong schema - should fail validation
wrong_request = {
"invalid_field": "value",
}
response = client.post("/api/generate", json=wrong_request)
assert response.status_code == 422 # Validation error
def test_generate_accepts_valid_request_format(self):
"""Generate endpoint should accept valid request format (schema check only)."""
from unittest.mock import AsyncMock, patch, MagicMock
from app.main import app
# Mock the orchestrator to avoid real API calls
mock_result = {
"product_owner": MagicMock(content="Mock PO output", metadata={}),
"business_analyst": MagicMock(content="Mock BA output", metadata={}),
}
with patch("app.routers.web.Orchestrator") as MockOrchestrator:
mock_instance = MagicMock()
mock_instance.run_pipeline = AsyncMock(return_value=mock_result)
MockOrchestrator.return_value = mock_instance
client = TestClient(app)
# Use the actual schema which requires 'description' field
request_data = {
"description": "Test project with user authentication",
}
response = client.post(
"/api/generate",
json=request_data,
timeout=5,
)
# Should not be a validation error - schema is correct
# With mocked orchestrator, should return 200
assert response.status_code == 200
class TestStreamingEndpoint:
"""Tests for streaming endpoint."""
def test_streaming_endpoint_exists(self):
"""Streaming endpoint should exist."""
from app.main import app
client = TestClient(app)
request_data = {
"description": "Test streaming",
}
response = client.post(
"/api/generate/stream",
json=request_data,
timeout=5,
)
# Should not be 404
assert response.status_code != 404
def test_streaming_query_param(self):
"""Stream query param should work."""
from app.main import app
client = TestClient(app)
request_data = {
"description": "Test streaming param",
}
response = client.post(
"/api/generate?stream=true",
json=request_data,
timeout=5,
)
# Should not be 404
assert response.status_code != 404
class TestCORSHeaders:
"""Tests for CORS configuration."""
def test_cors_headers_present(self):
"""CORS headers should be present in responses."""
from app.main import app
client = TestClient(app)
# Preflight request
response = client.options(
"/api/generate",
headers={
"Origin": "http://localhost:3000",
"Access-Control-Request-Method": "POST",
},
)
# CORS should be configured
assert response.status_code in [200, 204, 405]
|