Sibilla
76bc667adf
✨ Nuove funzionalità: - Nuovo endpoint API REST /api/v1/query (GET+POST) per consumo programmatico - Campo speakable_text ottimizzato per sintesi vocale (Alexa, Google Home) - Funzione strip_markdown() per conversione Markdown → plain text TTS - Supporto CORS per accesso cross-origin - UI responsive mobile-first con micro-interazioni e stati - Nuovo pulsante 'Nuova ricerca' e metadata bar 📚 Documentazione completa: - docs/ARCHITECTURE.md — architettura di sistema, stack, componenti - docs/FLOW.md — diagrammi Mermaid della pipeline e flussi - docs/API.md — documentazione endpoint REST con esempi curl - docs/WIKI.md — FAQ, glossario, roadmap - docs/DEPLOYMENT.md — guida deployment (Gunicorn, Nginx, Docker, systemd) - docs/CONTRIBUTING.md — guida alla contribuzione 🔧 Migliorie tecniche: - Pipeline di ricerca condivisa in _run_search_pipeline() - Gestione errori Gemini migliorata con _clean_error() - Sanitizzazione HTML con allowlist per prevenzione XSS - .env.example completo con commenti 🧪 Test: 109/109 passati
454 lines
17 KiB
Python
454 lines
17 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
|
|
|
|
|
|
class TestApiV1QueryEndpoint:
|
|
"""Tests for the /api/v1/query endpoint (voice-assistant / curl-friendly API)."""
|
|
|
|
def test_get_missing_query_returns_400(self, client):
|
|
resp = client.get("/api/v1/query")
|
|
assert resp.status_code == 400
|
|
assert "Missing query" in resp.get_json()["error"]
|
|
|
|
def test_get_empty_query_returns_400(self, client):
|
|
resp = client.get("/api/v1/query?q=")
|
|
assert resp.status_code == 400
|
|
|
|
def test_post_empty_query_returns_400(self, client):
|
|
resp = client.post("/api/v1/query", json={"query": ""})
|
|
assert resp.status_code == 400
|
|
|
|
def test_cors_headers_present(self, client):
|
|
resp = client.get("/api/v1/query?q=test")
|
|
assert resp.headers.get("Access-Control-Allow-Origin") == "*"
|
|
assert "GET" in resp.headers.get("Access-Control-Allow-Methods", "")
|
|
|
|
def test_get_query_full_pipeline(self, app):
|
|
"""Full pipeline via GET /api/v1/query with mocked services."""
|
|
client = app.test_client()
|
|
|
|
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"),
|
|
],
|
|
engine="degoog",
|
|
)
|
|
|
|
def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096):
|
|
return SynthesisResult(
|
|
answer="Ecco la **risposta** con [link](https://x.com) e `code`.",
|
|
model="gemini-2.5-flash",
|
|
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
|
|
), 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.get("/api/v1/query?q=test+query")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["query"] == "test query"
|
|
assert "risposta" in data["answer"]
|
|
assert "speakable_text" in data
|
|
# speakable_text must be clean of markdown
|
|
assert "**" not in data["speakable_text"]
|
|
assert "[" not in data["speakable_text"]
|
|
assert "`" not in data["speakable_text"]
|
|
assert "Ecco la risposta con link e code." in data["speakable_text"]
|
|
assert data["model"] == "gemini-2.5-flash"
|
|
assert len(data["sources"]) == 1
|
|
|
|
def test_post_query(self, app):
|
|
"""POST /api/v1/query with JSON body."""
|
|
client = app.test_client()
|
|
|
|
def mock_degoog_search(self, query, num_results=8, language="it"):
|
|
return SearchResponse(
|
|
query=query,
|
|
results=[
|
|
SearchResult(title="R1", url="https://a.com", snippet="S"),
|
|
],
|
|
engine="degoog",
|
|
)
|
|
|
|
def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096):
|
|
return SynthesisResult(answer="Risposta semplice.", 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_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/v1/query", json={"query": "hello"})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["query"] == "hello"
|
|
assert data["answer"] == "Risposta semplice."
|
|
assert data["speakable_text"] == "Risposta semplice."
|
|
|
|
def test_no_results_fallback(self, app):
|
|
"""When no search results, falls 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="Risposta senza fonti.", model="g", tokens_used=5)
|
|
|
|
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.get("/api/v1/query?q=test")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["speakable_text"] == "Risposta senza fonti."
|
|
assert data["sources"] == []
|
|
|
|
def test_search_error_returns_502(self, app):
|
|
"""When all backends error, return 502."""
|
|
client = app.test_client()
|
|
|
|
def mock_error_search(self, *args, **kwargs):
|
|
return SearchResponse(query="q", results=[], error="Down", 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.get("/api/v1/query?q=test")
|
|
|
|
assert resp.status_code == 502
|
|
|
|
def test_speakable_text_in_ask_endpoint(self, app):
|
|
"""Verify that /api/ask also includes speakable_text after our changes."""
|
|
client = app.test_client()
|
|
|
|
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"),
|
|
],
|
|
engine="degoog",
|
|
)
|
|
|
|
def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096):
|
|
return SynthesisResult(
|
|
answer="Test **bold** answer.",
|
|
model="g",
|
|
tokens_used=5,
|
|
)
|
|
|
|
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": "t"})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert "speakable_text" in data
|
|
assert data["speakable_text"] == "Test bold answer."
|