Sibilla
f3fe13c60e
- Flask web app with Perplexity-style search + synthesis pipeline - Degoog → SearXNG → DuckDuckGo cascading search backends - Gemini LLM for query expansion + answer synthesis with citations - MCP server exposing search tools via stdio - 57 unit tests covering all modules
253 lines
9.4 KiB
Python
253 lines
9.4 KiB
Python
"""Tests for sibilla.app — Flask routes and search pipeline."""
|
|
|
|
import os
|
|
import pytest
|
|
from unittest import mock
|
|
from sibilla.app import create_app
|
|
from sibilla.search import SearchResult, SearchResponse
|
|
from sibilla.ai import SynthesisResult
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
"""Create a Flask app in testing mode with a fake Gemini API key."""
|
|
app = create_app(testing=True)
|
|
# Point templates to the project-level templates/ directory
|
|
app.template_folder = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "templates"
|
|
)
|
|
app.config.update({
|
|
"TESTING": True,
|
|
"GOOGLE_API_KEY": "fake-test-key",
|
|
"DEGOOG_URL": "http://degoog.test:4444",
|
|
"SEARXNG_URL": "http://searxng.test:8080",
|
|
"SIBILLA_DDG_FALLBACK": True,
|
|
})
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
return app.test_client()
|
|
|
|
|
|
class TestHealthEndpoint:
|
|
def test_health_returns_ok(self, client):
|
|
resp = client.get("/api/health")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["status"] == "ok"
|
|
assert "version" in data
|
|
assert data["gemini_configured"] is True
|
|
assert "degoog" in data["degoog_url"]
|
|
|
|
def test_health_gemini_not_configured(self, app):
|
|
app.config["GOOGLE_API_KEY"] = ""
|
|
client = app.test_client()
|
|
resp = client.get("/api/health")
|
|
data = resp.get_json()
|
|
assert data["gemini_configured"] is False
|
|
|
|
|
|
class TestIndexRoute:
|
|
def test_index_returns_html(self, client):
|
|
resp = client.get("/")
|
|
assert resp.status_code == 200
|
|
assert "text/html" in resp.content_type
|
|
html = resp.data.decode("utf-8")
|
|
assert "<title>Sibilla" in html
|
|
assert 'id="search-form"' in html
|
|
assert 'id="query"' in html
|
|
assert 'id="response-area"' in html
|
|
|
|
|
|
class TestAskEndpoint:
|
|
def test_empty_query_returns_400(self, client):
|
|
resp = client.post("/api/ask", json={"query": ""})
|
|
assert resp.status_code == 400
|
|
assert "non può essere vuota" in resp.get_json()["error"]
|
|
|
|
def test_ask_with_degoog_success(self, app):
|
|
"""Full pipeline test with mocked Degoog search and Gemini synthesis."""
|
|
client = app.test_client()
|
|
|
|
# Mock DegoogClient.search to return results
|
|
def mock_degoog_search(self, query, num_results=8, language="it"):
|
|
return SearchResponse(
|
|
query=query,
|
|
results=[
|
|
SearchResult(title="R1", url="https://a.com", snippet="Content A"),
|
|
SearchResult(title="R2", url="https://b.com", snippet="Content B"),
|
|
],
|
|
engine="degoog",
|
|
)
|
|
|
|
# Mock Gemini client to return a synthesized answer
|
|
def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096):
|
|
return SynthesisResult(
|
|
answer="Ecco la risposta sintetizzata [1][2].",
|
|
model="gemini-2.5-flash",
|
|
tokens_used=100,
|
|
)
|
|
|
|
def mock_rewrite_query(self, original_query, num_variants=3):
|
|
return [original_query, "rewritten variant"]
|
|
|
|
with mock.patch(
|
|
"sibilla.app.DegoogClient.search", mock_degoog_search
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.synthesize", mock_synthesize
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query
|
|
), mock.patch(
|
|
"sibilla.app.PageFetcher.fetch_results",
|
|
lambda self, results, max_results=5: results, # passthrough
|
|
):
|
|
resp = client.post("/api/ask", json={"query": "test query"})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "Ecco la risposta sintetizzata" in data["answer"]
|
|
assert data["search_engine"] == "degoog"
|
|
assert len(data["sources"]) == 2
|
|
assert data["sources"][0]["url"] == "https://a.com"
|
|
assert data["model"] == "gemini-2.5-flash"
|
|
assert data["tokens_used"] == 100
|
|
|
|
def test_ask_fallback_to_searxng(self, app):
|
|
"""When Degoog returns no results, fall back to SearXNG."""
|
|
client = app.test_client()
|
|
|
|
def mock_degoog_search_empty(self, query, num_results=8, language="it"):
|
|
return SearchResponse(query=query, results=[], engine="degoog")
|
|
|
|
def mock_searxng_search(self, query, num_results=8, engines=None, language="it"):
|
|
return SearchResponse(
|
|
query=query,
|
|
results=[
|
|
SearchResult(title="SR1", url="https://s.com", snippet="S content"),
|
|
],
|
|
engine="searxng",
|
|
)
|
|
|
|
def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096):
|
|
return SynthesisResult(answer="Answer from SearXNG sources [1].", model="g", tokens_used=50)
|
|
|
|
def mock_rewrite_query(self, original_query, num_variants=3):
|
|
return [original_query]
|
|
|
|
with mock.patch(
|
|
"sibilla.app.DegoogClient.search", mock_degoog_search_empty
|
|
), mock.patch(
|
|
"sibilla.app.SearXNGClient.search", mock_searxng_search
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.synthesize", mock_synthesize
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query
|
|
), mock.patch(
|
|
"sibilla.app.PageFetcher.fetch_results",
|
|
lambda self, results, max_results=5: results,
|
|
):
|
|
resp = client.post("/api/ask", json={"query": "fallback test"})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["search_engine"] == "searxng"
|
|
|
|
def test_ask_no_results_fallback_to_llm(self, app):
|
|
"""When all search backends fail, fall back to pure LLM."""
|
|
client = app.test_client()
|
|
|
|
def mock_empty_search(self, *args, **kwargs):
|
|
return SearchResponse(query="q", results=[], engine="none")
|
|
|
|
def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096):
|
|
return SynthesisResult(answer="Nessun risultato, ma posso aiutarti...", model="g", tokens_used=10)
|
|
|
|
def mock_rewrite_query(self, original_query, num_variants=3):
|
|
return [original_query]
|
|
|
|
with mock.patch(
|
|
"sibilla.app.DegoogClient.search", mock_empty_search
|
|
), mock.patch(
|
|
"sibilla.app.SearXNGClient.search", mock_empty_search
|
|
), mock.patch(
|
|
"sibilla.app.DuckDuckGoClient.search", mock_empty_search
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.synthesize", mock_synthesize
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query
|
|
):
|
|
resp = client.post("/api/ask", json={"query": "impossible query"})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["search_engine"] == "none"
|
|
assert data["sources"] == []
|
|
|
|
def test_ask_all_backends_error(self, app):
|
|
"""When all backends return errors, return 502."""
|
|
client = app.test_client()
|
|
|
|
def mock_error_search(self, *args, **kwargs):
|
|
return SearchResponse(query="q", results=[], error="Connection refused", engine="degoog")
|
|
|
|
def mock_rewrite_query(self, original_query, num_variants=3):
|
|
return [original_query]
|
|
|
|
with mock.patch(
|
|
"sibilla.app.DegoogClient.search", mock_error_search
|
|
), mock.patch(
|
|
"sibilla.app.SearXNGClient.search", mock_error_search
|
|
), mock.patch(
|
|
"sibilla.app.DuckDuckGoClient.search", mock_error_search
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query
|
|
):
|
|
resp = client.post("/api/ask", json={"query": "doomed query"})
|
|
|
|
assert resp.status_code == 502
|
|
|
|
def test_ask_respects_num_sources(self, app):
|
|
client = app.test_client()
|
|
|
|
call_count = 0
|
|
|
|
def mock_degoog_search(self, query, num_results=8, language="it"):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return SearchResponse(
|
|
query=query,
|
|
results=[
|
|
SearchResult(title=f"R{i}", url=f"https://{i}.com", snippet=f"S{i}")
|
|
for i in range(10)
|
|
],
|
|
engine="degoog",
|
|
)
|
|
|
|
def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096):
|
|
return SynthesisResult(answer="Ok.", model="g", tokens_used=1)
|
|
|
|
def mock_rewrite_query(self, original_query, num_variants=3):
|
|
return [original_query]
|
|
|
|
with mock.patch(
|
|
"sibilla.app.DegoogClient.search", mock_degoog_search
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.synthesize", mock_synthesize
|
|
), mock.patch(
|
|
"sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query
|
|
), mock.patch(
|
|
"sibilla.app.PageFetcher.fetch_results",
|
|
lambda self, results, max_results=5: results,
|
|
):
|
|
resp = client.post("/api/ask", json={"query": "test", "num_sources": 3})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert len(data["sources"]) <= 3
|
|
|
|
def test_ask_without_json_body(self, client):
|
|
resp = client.post("/api/ask", data="not json")
|
|
assert resp.status_code == 400
|