261 lines
10 KiB
Python
261 lines
10 KiB
Python
"""Unit tests for API router."""
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.api.app import create_app
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
app = create_app()
|
|
return TestClient(app)
|
|
|
|
|
|
class TestHealthEndpoint:
|
|
"""Test cases for health endpoint."""
|
|
|
|
def test_health_searxng_healthy(self, client):
|
|
"""Test health check when SearXNG is up."""
|
|
with patch("src.api.router.SearXNGClient") as mock_class:
|
|
mock_instance = AsyncMock()
|
|
mock_instance.health_check = AsyncMock(return_value=True)
|
|
mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
|
|
mock_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_class.return_value = mock_instance
|
|
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "degraded" # No Kimi key configured in test
|
|
assert data["searxng_connected"] is True
|
|
assert data["kimi_coding_available"] is False
|
|
|
|
def test_health_searxng_down(self, client):
|
|
"""Test health check when SearXNG is down."""
|
|
with patch("src.api.router.SearXNGClient") as mock_class:
|
|
mock_instance = AsyncMock()
|
|
mock_instance.health_check = AsyncMock(return_value=False)
|
|
mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
|
|
mock_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_class.return_value = mock_instance
|
|
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "degraded"
|
|
assert data["searxng_connected"] is False
|
|
|
|
|
|
class TestSearchEndpoint:
|
|
"""Test cases for search endpoint."""
|
|
|
|
def test_search_get_success(self, client):
|
|
"""Test GET search with successful response."""
|
|
mock_response = Mock()
|
|
mock_response.query = "python"
|
|
mock_response.results = []
|
|
mock_response.total = 0
|
|
mock_response.page = 1
|
|
mock_response.metadata = {}
|
|
|
|
with patch("src.api.router.SearXNGClient") as mock_class:
|
|
mock_instance = AsyncMock()
|
|
mock_instance.search = AsyncMock(return_value=mock_response)
|
|
mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
|
|
mock_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_class.return_value = mock_instance
|
|
|
|
response = client.get("/search?q=python")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["query"] == "python"
|
|
assert "results" in data
|
|
|
|
def test_search_post_success(self, client):
|
|
"""Test POST search with successful response."""
|
|
mock_response = Mock()
|
|
mock_response.query = "asyncio"
|
|
mock_response.results = []
|
|
mock_response.total = 0
|
|
mock_response.page = 1
|
|
mock_response.metadata = {}
|
|
|
|
with patch("src.api.router.SearXNGClient") as mock_class:
|
|
mock_instance = AsyncMock()
|
|
mock_instance.search = AsyncMock(return_value=mock_response)
|
|
mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
|
|
mock_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_class.return_value = mock_instance
|
|
|
|
response = client.post("/search", json={
|
|
"q": "asyncio",
|
|
"engines": ["google"],
|
|
"page": 1
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_search_validation_error(self, client):
|
|
"""Test search with invalid parameters."""
|
|
response = client.get("/search?q=a")
|
|
# Just test that it accepts the request
|
|
assert response.status_code in [200, 502] # 502 if no SearXNG
|
|
|
|
|
|
class TestResearchEndpoint:
|
|
"""Test cases for research endpoint (Phase 2 - with synthesis)."""
|
|
|
|
def test_research_phase2_with_synthesis(self, client):
|
|
"""Test research endpoint returns synthesis (Phase 2)."""
|
|
from src.models.schemas import SearchResult
|
|
import src.api.router as router_module
|
|
|
|
mock_search_response = Mock()
|
|
mock_search_response.results = [
|
|
SearchResult(title="Test", url="https://example.com", source="google")
|
|
]
|
|
mock_search_response.total = 1
|
|
mock_search_response.page = 1
|
|
|
|
mock_synthesis = Mock()
|
|
mock_synthesis.content = "This is a synthesized answer."
|
|
mock_synthesis.sources = [{"index": 1, "title": "Test", "url": "https://example.com"}]
|
|
mock_synthesis.tokens_used = 100
|
|
|
|
# Temporarily set API key in router module
|
|
original_key = router_module.KIMI_API_KEY
|
|
router_module.KIMI_API_KEY = "sk-test"
|
|
|
|
try:
|
|
with patch("src.api.router.SearXNGClient") as mock_search_class, \
|
|
patch("src.api.router.Synthesizer") as mock_synth_class:
|
|
|
|
# Mock SearXNG
|
|
mock_search_instance = AsyncMock()
|
|
mock_search_instance.search = AsyncMock(return_value=mock_search_response)
|
|
mock_search_instance.__aenter__ = AsyncMock(return_value=mock_search_instance)
|
|
mock_search_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_search_class.return_value = mock_search_instance
|
|
|
|
# Mock Synthesizer
|
|
mock_synth_instance = AsyncMock()
|
|
mock_synth_instance.synthesize = AsyncMock(return_value=mock_synthesis)
|
|
mock_synth_instance.__aenter__ = AsyncMock(return_value=mock_synth_instance)
|
|
mock_synth_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_synth_class.return_value = mock_synth_instance
|
|
|
|
response = client.post("/research", json={
|
|
"query": "python asyncio",
|
|
"depth": "shallow",
|
|
"sources": ["web"]
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["query"] == "python asyncio"
|
|
assert data["depth"] == "shallow"
|
|
assert data["synthesis"] == "This is a synthesized answer."
|
|
assert data["metadata"]["phase"] == "2"
|
|
assert len(data["sources"]) == 1
|
|
finally:
|
|
router_module.KIMI_API_KEY = original_key
|
|
|
|
def test_research_no_api_key_returns_message(self, client):
|
|
"""Test research endpoint without API key returns appropriate message."""
|
|
from src.models.schemas import SearchResult
|
|
|
|
mock_search_response = Mock()
|
|
mock_search_response.results = [
|
|
SearchResult(title="Test", url="https://example.com", source="google")
|
|
]
|
|
|
|
with patch("src.api.router.SearXNGClient") as mock_class:
|
|
mock_instance = AsyncMock()
|
|
mock_instance.search = AsyncMock(return_value=mock_search_response)
|
|
mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
|
|
mock_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_class.return_value = mock_instance
|
|
|
|
# Ensure no API key
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
with patch("src.api.router.KIMI_API_KEY", None):
|
|
response = client.post("/research", json={
|
|
"query": "test",
|
|
"sources": ["web"]
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "not configured" in data["synthesis"].lower() or "API key" in data["synthesis"]
|
|
|
|
def test_research_no_results(self, client):
|
|
"""Test research endpoint with no search results."""
|
|
mock_search_response = Mock()
|
|
mock_search_response.results = []
|
|
|
|
with patch("src.api.router.SearXNGClient") as mock_class:
|
|
mock_instance = AsyncMock()
|
|
mock_instance.search = AsyncMock(return_value=mock_search_response)
|
|
mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
|
|
mock_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_class.return_value = mock_instance
|
|
|
|
response = client.post("/research", json={
|
|
"query": "xyzabc123nonexistent",
|
|
"sources": ["web"]
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "no results" in data["synthesis"].lower()
|
|
|
|
def test_research_with_omit_raw(self, client):
|
|
"""Test research endpoint with omit_raw=true."""
|
|
from src.models.schemas import SearchResult
|
|
import src.api.router as router_module
|
|
|
|
mock_search_response = Mock()
|
|
mock_search_response.results = [
|
|
SearchResult(title="Test", url="https://example.com", source="google")
|
|
]
|
|
|
|
mock_synthesis = Mock()
|
|
mock_synthesis.content = "Answer"
|
|
mock_synthesis.sources = []
|
|
mock_synthesis.tokens_used = 50
|
|
|
|
original_key = router_module.KIMI_API_KEY
|
|
router_module.KIMI_API_KEY = "sk-test"
|
|
|
|
try:
|
|
with patch("src.api.router.SearXNGClient") as mock_search_class, \
|
|
patch("src.api.router.Synthesizer") as mock_synth_class:
|
|
|
|
mock_search_instance = AsyncMock()
|
|
mock_search_instance.search = AsyncMock(return_value=mock_search_response)
|
|
mock_search_instance.__aenter__ = AsyncMock(return_value=mock_search_instance)
|
|
mock_search_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_search_class.return_value = mock_search_instance
|
|
|
|
mock_synth_instance = AsyncMock()
|
|
mock_synth_instance.synthesize = AsyncMock(return_value=mock_synthesis)
|
|
mock_synth_instance.__aenter__ = AsyncMock(return_value=mock_synth_instance)
|
|
mock_synth_instance.__aexit__ = AsyncMock(return_value=None)
|
|
mock_synth_class.return_value = mock_synth_instance
|
|
|
|
response = client.post("/research", json={
|
|
"query": "test",
|
|
"omit_raw": True
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["raw_results"] is None
|
|
finally:
|
|
router_module.KIMI_API_KEY = original_key
|