Spaces:
Running
Running
File size: 6,793 Bytes
31a2688 db45c50 31a2688 9612292 31a2688 3f19c23 31a2688 9612292 | 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 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 | """Tests for the FastAPI API routes."""
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
from src.api.routes import router, set_dependencies
from src.models import DocumentChunk, GenerationResponse, IntentType, QueryResult
@pytest.fixture()
def mock_deps() -> dict[str, MagicMock]:
"""Create mock dependencies and inject them into the routes module."""
query_router = MagicMock()
ingestion_pipeline = MagicMock()
embedder = MagicMock()
vector_store = MagicMock()
bm25_search = MagicMock()
settings = MagicMock()
settings.llm_provider = "ollama"
settings.embedding_provider = "local"
settings.embedding_model = "paraphrase-multilingual-MiniLM-L12-v2"
settings.ollama_model = "llama3"
settings.generation_model = "llama3"
set_dependencies(
query_router=query_router,
ingestion_pipeline=ingestion_pipeline,
embedder=embedder,
vector_store=vector_store,
bm25_search=bm25_search,
settings=settings,
)
return {
"query_router": query_router,
"ingestion_pipeline": ingestion_pipeline,
"embedder": embedder,
"vector_store": vector_store,
"bm25_search": bm25_search,
"settings": settings,
}
@pytest.fixture()
def client(mock_deps: dict[str, MagicMock]) -> TestClient:
"""Create a TestClient with mocked dependencies."""
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
return TestClient(app)
class TestHealthCheck:
"""Tests for the /health endpoint."""
def test_health_check_returns_ok(self, client: TestClient) -> None:
response = client.get("/health")
assert response.status_code == 200
body = response.json()
assert body["status"] == "ok"
assert body["version"] == "0.1.0"
class TestLivenessProbe:
"""Tests for the /health/live endpoint."""
def test_liveness_returns_ok(self, client: TestClient) -> None:
response = client.get("/health/live")
assert response.status_code == 200
body = response.json()
assert body["status"] == "ok"
class TestReadinessProbe:
"""Tests for the /health/ready endpoint."""
def test_readiness_returns_ready_when_all_deps_available(
self, client: TestClient, mock_deps: dict[str, MagicMock]
) -> None:
mock_deps["vector_store"].get_all_chunks.return_value = []
mock_deps["bm25_search"].is_indexed = True
response = client.get("/health/ready")
assert response.status_code == 200
body = response.json()
assert body["status"] == "ready"
assert body["checks"]["vector_store"] is True
assert body["checks"]["bm25_index"] is True
assert body["checks"]["router"] is True
def test_readiness_returns_503_when_bm25_not_indexed(
self, client: TestClient, mock_deps: dict[str, MagicMock]
) -> None:
mock_deps["vector_store"].get_all_chunks.return_value = []
mock_deps["bm25_search"].is_indexed = False
response = client.get("/health/ready")
assert response.status_code == 503
class TestQueryEndpoint:
"""Tests for the /query endpoint."""
def test_query_returns_structured_response(
self, client: TestClient, mock_deps: dict[str, MagicMock]
) -> None:
chunk = DocumentChunk(
chunk_id="c1",
document_id="doc1",
text="Some policy text.",
)
generation_response = GenerationResponse(
answer="The policy states that...",
sources=[QueryResult(chunk=chunk, score=0.95, source="hybrid")],
intent=IntentType.FACTUAL,
confidence=0.88,
)
mock_deps["query_router"].route.return_value = generation_response
response = client.post("/query", json={"question": "What is the policy?"})
assert response.status_code == 200
body = response.json()
assert body["answer"] == "The policy states that..."
assert body["intent"] == "factual"
assert body["confidence"] == 0.88
assert len(body["sources"]) == 1
assert body["sources"][0]["chunk_id"] == "c1"
assert body["sources"][0]["document_id"] == "doc1"
assert body["sources"][0]["score"] == 0.95
def test_empty_question_returns_422(self, client: TestClient) -> None:
"""FastAPI returns 422 for missing required fields; empty string passes validation."""
response = client.post("/query", json={})
assert response.status_code == 422
def test_missing_body_returns_422(self, client: TestClient) -> None:
response = client.post("/query")
assert response.status_code == 422
def test_query_uses_default_top_k(
self, client: TestClient, mock_deps: dict[str, MagicMock]
) -> None:
mock_deps["query_router"].route.return_value = GenerationResponse(
answer="answer",
sources=[],
intent=IntentType.UNKNOWN,
confidence=0.5,
)
client.post("/query", json={"question": "test"})
mock_deps["query_router"].route.assert_called_once_with(query="test", top_k=5)
class TestIngestEndpoint:
"""Tests for the /ingest endpoint."""
@patch("src.api.routes.os.path.isfile", return_value=False)
def test_ingest_missing_file_returns_404(
self, _mock_isfile: MagicMock, client: TestClient
) -> None:
response = client.post("/ingest", json={"file_path": "/nonexistent.pdf"})
assert response.status_code == 404
assert "File not found" in response.json()["detail"]
@patch("src.api.routes.os.path.isfile", return_value=True)
def test_ingest_success(
self,
_mock_isfile: MagicMock,
client: TestClient,
mock_deps: dict[str, MagicMock],
) -> None:
chunks = [
DocumentChunk(chunk_id="c1", document_id="doc.pdf", text="chunk1"),
DocumentChunk(chunk_id="c2", document_id="doc.pdf", text="chunk2"),
]
mock_deps["ingestion_pipeline"].ingest_pdf.return_value = chunks
mock_deps["embedder"].embed_batch.return_value = [[0.1] * 1536, [0.2] * 1536]
response = client.post("/ingest", json={"file_path": "/tmp/doc.pdf"})
assert response.status_code == 200
body = response.json()
assert body["document_id"] == "doc.pdf"
assert body["chunks_created"] == 2
mock_deps["vector_store"].add_chunks.assert_called_once()
mock_deps["vector_store"].get_all_chunks.assert_called_once()
mock_deps["bm25_search"].index.assert_called_once_with(
mock_deps["vector_store"].get_all_chunks.return_value
)
|