f6638d5406
## Added
- conftest.py: Shared fixtures and mocks
- test_core/test_config.py: 35 tests for Settings
- test_core/test_logging.py: 15 tests for logging
- test_api/test_chat.py: 27 tests for chat endpoints
- test_api/test_health.py: 27 tests for health endpoints
- test_services/test_document_service.py: 38 tests
- test_services/test_rag_service.py: 66 tests
- test_services/test_vector_store.py: 32 tests
## Coverage
- auth.py: 100%
- config.py: 100%
- logging.py: 100%
- chat.py: 100%
- health.py: 100%
- document_service.py: 96%
- rag_service.py: 100%
- vector_store.py: 100%
Total: 240 tests passing, 64% coverage
🧪 Core functionality fully tested
286 lines
8.6 KiB
Python
286 lines
8.6 KiB
Python
"""Tests for health check endpoints."""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import Mock, patch
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestHealthCheck:
|
|
"""Tests for health check endpoint."""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
"""Create test client for health routes."""
|
|
from fastapi import FastAPI
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
def test_health_check_returns_200(self, client):
|
|
"""Test health check returns HTTP 200."""
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_health_check_returns_json(self, client):
|
|
"""Test health check returns JSON response."""
|
|
response = client.get("/health")
|
|
|
|
assert response.headers["content-type"] == "application/json"
|
|
|
|
def test_health_check_status_healthy(self, client):
|
|
"""Test health check status is healthy."""
|
|
response = client.get("/health")
|
|
data = response.json()
|
|
|
|
assert data["status"] == "healthy"
|
|
|
|
def test_health_check_service_name(self, client):
|
|
"""Test health check includes service name."""
|
|
response = client.get("/health")
|
|
data = response.json()
|
|
|
|
assert data["service"] == "agentic-rag"
|
|
|
|
def test_health_check_version(self, client):
|
|
"""Test health check includes version."""
|
|
response = client.get("/health")
|
|
data = response.json()
|
|
|
|
assert data["version"] == "2.0.0"
|
|
|
|
def test_health_check_all_fields(self, client):
|
|
"""Test health check contains all expected fields."""
|
|
response = client.get("/health")
|
|
data = response.json()
|
|
|
|
assert "status" in data
|
|
assert "service" in data
|
|
assert "version" in data
|
|
assert len(data) == 3
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestReadinessCheck:
|
|
"""Tests for readiness probe endpoint."""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
"""Create test client for health routes."""
|
|
from fastapi import FastAPI
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
def test_readiness_check_returns_200(self, client):
|
|
"""Test readiness check returns HTTP 200."""
|
|
response = client.get("/health/ready")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_readiness_check_returns_json(self, client):
|
|
"""Test readiness check returns JSON response."""
|
|
response = client.get("/health/ready")
|
|
|
|
assert response.headers["content-type"] == "application/json"
|
|
|
|
def test_readiness_check_status_ready(self, client):
|
|
"""Test readiness check status is ready."""
|
|
response = client.get("/health/ready")
|
|
data = response.json()
|
|
|
|
assert data["status"] == "ready"
|
|
|
|
def test_readiness_check_single_field(self, client):
|
|
"""Test readiness check only contains status field."""
|
|
response = client.get("/health/ready")
|
|
data = response.json()
|
|
|
|
assert len(data) == 1
|
|
assert "status" in data
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLivenessCheck:
|
|
"""Tests for liveness probe endpoint."""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
"""Create test client for health routes."""
|
|
from fastapi import FastAPI
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
def test_liveness_check_returns_200(self, client):
|
|
"""Test liveness check returns HTTP 200."""
|
|
response = client.get("/health/live")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_liveness_check_returns_json(self, client):
|
|
"""Test liveness check returns JSON response."""
|
|
response = client.get("/health/live")
|
|
|
|
assert response.headers["content-type"] == "application/json"
|
|
|
|
def test_liveness_check_status_alive(self, client):
|
|
"""Test liveness check status is alive."""
|
|
response = client.get("/health/live")
|
|
data = response.json()
|
|
|
|
assert data["status"] == "alive"
|
|
|
|
def test_liveness_check_single_field(self, client):
|
|
"""Test liveness check only contains status field."""
|
|
response = client.get("/health/live")
|
|
data = response.json()
|
|
|
|
assert len(data) == 1
|
|
assert "status" in data
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestHealthEndpointsAsync:
|
|
"""Tests to verify endpoints are async."""
|
|
|
|
def test_health_check_is_async(self):
|
|
"""Test health check endpoint is async function."""
|
|
from agentic_rag.api.routes.health import health_check
|
|
|
|
import asyncio
|
|
|
|
assert asyncio.iscoroutinefunction(health_check)
|
|
|
|
def test_readiness_check_is_async(self):
|
|
"""Test readiness check endpoint is async function."""
|
|
from agentic_rag.api.routes.health import readiness_check
|
|
|
|
import asyncio
|
|
|
|
assert asyncio.iscoroutinefunction(readiness_check)
|
|
|
|
def test_liveness_check_is_async(self):
|
|
"""Test liveness check endpoint is async function."""
|
|
from agentic_rag.api.routes.health import liveness_check
|
|
|
|
import asyncio
|
|
|
|
assert asyncio.iscoroutinefunction(liveness_check)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestHealthRouterConfiguration:
|
|
"""Tests for router configuration."""
|
|
|
|
def test_router_exists(self):
|
|
"""Test router module exports router."""
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
assert router is not None
|
|
|
|
def test_router_is_api_router(self):
|
|
"""Test router is FastAPI APIRouter."""
|
|
from agentic_rag.api.routes.health import router
|
|
from fastapi import APIRouter
|
|
|
|
assert isinstance(router, APIRouter)
|
|
|
|
def test_health_endpoint_path(self):
|
|
"""Test health endpoint has correct path."""
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
routes = [route for route in router.routes if hasattr(route, "path")]
|
|
paths = [route.path for route in routes]
|
|
|
|
assert "/health" in paths
|
|
|
|
def test_readiness_endpoint_path(self):
|
|
"""Test readiness endpoint has correct path."""
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
routes = [route for route in router.routes if hasattr(route, "path")]
|
|
paths = [route.path for route in routes]
|
|
|
|
assert "/health/ready" in paths
|
|
|
|
def test_liveness_endpoint_path(self):
|
|
"""Test liveness endpoint has correct path."""
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
routes = [route for route in router.routes if hasattr(route, "path")]
|
|
paths = [route.path for route in routes]
|
|
|
|
assert "/health/live" in paths
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestHealthEndpointsMethods:
|
|
"""Tests for HTTP methods on health endpoints."""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
"""Create test client for health routes."""
|
|
from fastapi import FastAPI
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
def test_health_check_only_get(self, client):
|
|
"""Test health check only accepts GET."""
|
|
# POST should not be allowed
|
|
response = client.post("/health")
|
|
assert response.status_code == 405
|
|
|
|
def test_readiness_check_only_get(self, client):
|
|
"""Test readiness check only accepts GET."""
|
|
response = client.post("/health/ready")
|
|
assert response.status_code == 405
|
|
|
|
def test_liveness_check_only_get(self, client):
|
|
"""Test liveness check only accepts GET."""
|
|
response = client.post("/health/live")
|
|
assert response.status_code == 405
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestHealthEndpointPerformance:
|
|
"""Tests for health endpoint performance characteristics."""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
"""Create test client for health routes."""
|
|
from fastapi import FastAPI
|
|
from agentic_rag.api.routes.health import router
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return TestClient(app)
|
|
|
|
def test_health_check_response_time(self, client):
|
|
"""Test health check responds quickly."""
|
|
import time
|
|
|
|
start = time.time()
|
|
response = client.get("/health")
|
|
elapsed = time.time() - start
|
|
|
|
assert response.status_code == 200
|
|
assert elapsed < 1.0 # Should respond in less than 1 second
|
|
|
|
def test_health_check_small_response(self, client):
|
|
"""Test health check returns small response."""
|
|
response = client.get("/health")
|
|
|
|
# Response should be small (less than 1KB)
|
|
assert len(response.content) < 1024
|