|
|
""" |
|
|
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("/") |
|
|
|
|
|
|
|
|
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") |
|
|
|
|
|
|
|
|
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_request = { |
|
|
"invalid_field": "value", |
|
|
} |
|
|
response = client.post("/api/generate", json=wrong_request) |
|
|
assert response.status_code == 422 |
|
|
|
|
|
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_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) |
|
|
|
|
|
|
|
|
request_data = { |
|
|
"description": "Test project with user authentication", |
|
|
} |
|
|
|
|
|
response = client.post( |
|
|
"/api/generate", |
|
|
json=request_data, |
|
|
timeout=5, |
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
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, |
|
|
) |
|
|
|
|
|
|
|
|
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, |
|
|
) |
|
|
|
|
|
|
|
|
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) |
|
|
|
|
|
|
|
|
response = client.options( |
|
|
"/api/generate", |
|
|
headers={ |
|
|
"Origin": "http://localhost:3000", |
|
|
"Access-Control-Request-Method": "POST", |
|
|
}, |
|
|
) |
|
|
|
|
|
|
|
|
assert response.status_code in [200, 204, 405] |
|
|
|