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
399 lines
14 KiB
Python
399 lines
14 KiB
Python
"""Tests for sibilla.search — search clients and page fetcher."""
|
|
|
|
from unittest import mock
|
|
from sibilla.search import (
|
|
DegoogClient,
|
|
SearXNGClient,
|
|
DuckDuckGoClient,
|
|
PageFetcher,
|
|
SearchResult,
|
|
SearchResponse,
|
|
_clean_html,
|
|
)
|
|
|
|
|
|
class TestCleanHtml:
|
|
def test_removes_tags(self):
|
|
assert _clean_html("<p>Hello <b>world</b></p>") == "Hello world"
|
|
|
|
def test_plain_text_unchanged(self):
|
|
assert _clean_html("Plain text") == "Plain text"
|
|
|
|
def test_empty_string(self):
|
|
assert _clean_html("") == ""
|
|
|
|
def test_nested_tags(self):
|
|
assert _clean_html('<div><span class="x">Text</span></div>') == "Text"
|
|
|
|
def test_strips_html_entities(self):
|
|
# HTML entities stay as-is (not entity-decoded), just tags stripped
|
|
assert _clean_html("<b>Bold & Brave</b>") == "Bold & Brave"
|
|
|
|
|
|
class TestSearchDataclasses:
|
|
def test_search_result_defaults(self):
|
|
sr = SearchResult(title="T", url="https://u.com")
|
|
assert sr.snippet == ""
|
|
assert sr.full_content == ""
|
|
assert sr.fetched is False
|
|
|
|
def test_search_response_defaults(self):
|
|
sr = SearchResponse(query="q")
|
|
assert sr.results == []
|
|
assert sr.total_results == 0
|
|
assert sr.engine == "searxng"
|
|
assert sr.error is None
|
|
|
|
|
|
# ── DegoogClient tests ────────────────────────────────────────────
|
|
|
|
class TestDegoogClient:
|
|
def test_search_success(self):
|
|
client = DegoogClient(base_url="http://degoog.test:4444")
|
|
|
|
mock_data = {
|
|
"results": [
|
|
{"title": "R1", "url": "https://a.com", "snippet": "S1"},
|
|
{"title": "R2", "url": "https://b.com", "snippet": "S2"},
|
|
]
|
|
}
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
mock_get.return_value.json.return_value = mock_data
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
resp = client.search("test query", num_results=5)
|
|
|
|
assert len(resp.results) == 2
|
|
assert resp.results[0].title == "R1"
|
|
assert resp.results[0].url == "https://a.com"
|
|
assert resp.results[0].snippet == "S1"
|
|
assert resp.engine == "degoog"
|
|
assert resp.error is None
|
|
|
|
def test_search_respects_num_results(self):
|
|
client = DegoogClient(base_url="http://degoog.test:4444")
|
|
|
|
mock_data = {
|
|
"results": [
|
|
{"title": f"R{i}", "url": f"https://{i}.com", "snippet": f"S{i}"}
|
|
for i in range(10)
|
|
]
|
|
}
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
mock_get.return_value.json.return_value = mock_data
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
resp = client.search("q", num_results=3)
|
|
|
|
assert len(resp.results) == 3
|
|
|
|
def test_search_empty_results(self):
|
|
client = DegoogClient(base_url="http://degoog.test:4444")
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
mock_get.return_value.json.return_value = {"results": []}
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
resp = client.search("no results query")
|
|
|
|
assert resp.results == []
|
|
assert resp.error is None
|
|
|
|
def test_search_handles_request_exception(self):
|
|
client = DegoogClient(base_url="http://degoog.test:4444")
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
import requests
|
|
mock_get.side_effect = requests.RequestException("Connection refused")
|
|
|
|
resp = client.search("q")
|
|
|
|
assert resp.error is not None
|
|
assert resp.results == []
|
|
|
|
def test_search_cleans_html_in_results(self):
|
|
"""Verify that _clean_html is applied to title and snippet."""
|
|
client = DegoogClient(base_url="http://degoog.test:4444")
|
|
|
|
mock_data = {
|
|
"results": [
|
|
{
|
|
"title": "<b>Bold Title</b>",
|
|
"url": "https://x.com",
|
|
"snippet": "<p>Some <em>text</em></p>",
|
|
}
|
|
]
|
|
}
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
mock_get.return_value.json.return_value = mock_data
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
resp = client.search("q")
|
|
|
|
assert resp.results[0].title == "Bold Title"
|
|
assert resp.results[0].snippet == "Some text"
|
|
|
|
def test_search_uses_content_fallback_for_snippet(self):
|
|
"""Degoog may return 'content' instead of 'snippet' (OpenWebUI compat)."""
|
|
client = DegoogClient(base_url="http://degoog.test:4444")
|
|
|
|
mock_data = {
|
|
"results": [
|
|
{"title": "R", "url": "https://x.com", "content": "Content field"}
|
|
]
|
|
}
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
mock_get.return_value.json.return_value = mock_data
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
resp = client.search("q")
|
|
|
|
assert resp.results[0].snippet == "Content field"
|
|
|
|
def test_search_query_params(self):
|
|
client = DegoogClient(base_url="http://degoog.test:4444")
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
mock_get.return_value.json.return_value = {"results": []}
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
client.search("test query", language="en")
|
|
|
|
call_args = mock_get.call_args
|
|
assert call_args[1]["params"]["q"] == "test query"
|
|
assert call_args[1]["params"]["lang"] == "en"
|
|
|
|
|
|
# ── SearXNGClient tests ───────────────────────────────────────────
|
|
|
|
class TestSearXNGClient:
|
|
def test_search_success(self):
|
|
client = SearXNGClient(base_url="http://searxng.test:8080")
|
|
|
|
mock_data = {
|
|
"results": [
|
|
{"title": "R1", "url": "https://a.com", "content": "S1 content"},
|
|
],
|
|
"number_of_results": 1,
|
|
}
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
mock_get.return_value.json.return_value = mock_data
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
resp = client.search("test", num_results=5)
|
|
|
|
assert len(resp.results) == 1
|
|
assert resp.results[0].snippet == "S1 content"
|
|
assert resp.total_results == 1
|
|
|
|
def test_search_handles_error(self):
|
|
client = SearXNGClient(base_url="http://searxng.test:8080")
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
import requests
|
|
mock_get.side_effect = requests.RequestException("Timeout")
|
|
|
|
resp = client.search("q")
|
|
|
|
assert resp.error is not None
|
|
assert "Timeout" in resp.error
|
|
|
|
def test_search_with_engines_param(self):
|
|
client = SearXNGClient(base_url="http://searxng.test:8080")
|
|
|
|
with mock.patch.object(client._session, "get") as mock_get:
|
|
mock_get.return_value.json.return_value = {"results": []}
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
client.search("q", engines="google,bing")
|
|
|
|
call_args = mock_get.call_args
|
|
assert call_args[1]["params"]["engines"] == "google,bing"
|
|
|
|
|
|
# ── DuckDuckGoClient tests ────────────────────────────────────────
|
|
|
|
DUCKDUCKGO_HTML = """
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<body>
|
|
<div class="result">
|
|
<div class="result__title">
|
|
<a href="https://duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com&rut=abc">Example Page</a>
|
|
</div>
|
|
<div class="result__snippet">This is the snippet text for the first result.</div>
|
|
</div>
|
|
<div class="result">
|
|
<div class="result__title">
|
|
<a href="https://another-site.org/page">Another Site</a>
|
|
</div>
|
|
<div class="result__snippet">Second snippet over here.</div>
|
|
</div>
|
|
<div class="result">
|
|
<div class="result__title">
|
|
<a href="/l/?uddg=https%3A%2F%2Fthird.com">Third Result</a>
|
|
</div>
|
|
<!-- No snippet -->
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
class TestDuckDuckGoClient:
|
|
def test_extract_redirect(self):
|
|
client = DuckDuckGoClient()
|
|
url = "https://duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com&rut=abc"
|
|
result = client._extract_redirect(url)
|
|
assert result == "https://example.com"
|
|
|
|
def test_extract_redirect_no_uddg(self):
|
|
client = DuckDuckGoClient()
|
|
result = client._extract_redirect("https://example.com/page")
|
|
assert result == "https://example.com/page"
|
|
|
|
def test_search_parses_html_results(self):
|
|
"""Test that DuckDuckGo HTML results are parsed correctly."""
|
|
client = DuckDuckGoClient()
|
|
|
|
mock_resp = mock.MagicMock()
|
|
mock_resp.text = DUCKDUCKGO_HTML
|
|
mock_resp.raise_for_status.return_value = None
|
|
|
|
with mock.patch.object(client._session, "post") as mock_post:
|
|
mock_post.return_value = mock_resp
|
|
|
|
resp = client.search("test query", num_results=5)
|
|
|
|
assert len(resp.results) == 3
|
|
assert resp.results[0].title == "Example Page"
|
|
assert resp.results[0].url == "https://example.com" # redirect extracted
|
|
assert resp.results[0].snippet == "This is the snippet text for the first result."
|
|
assert resp.results[1].url == "https://another-site.org/page"
|
|
assert resp.results[2].snippet == "" # no snippet element
|
|
assert resp.engine == "duckduckgo"
|
|
assert resp.error is None
|
|
|
|
def test_search_handles_error(self):
|
|
client = DuckDuckGoClient()
|
|
|
|
with mock.patch.object(client._session, "post") as mock_post:
|
|
import requests
|
|
mock_post.side_effect = requests.RequestException("Network error")
|
|
|
|
resp = client.search("q")
|
|
|
|
assert resp.error is not None
|
|
assert "Network error" in resp.error
|
|
|
|
|
|
# ── PageFetcher tests ─────────────────────────────────────────────
|
|
|
|
class TestPageFetcher:
|
|
def test_fetch_extracts_text(self):
|
|
fetcher = PageFetcher()
|
|
|
|
html = """<html><body>
|
|
<script>alert('xss')</script>
|
|
<nav>Menu</nav>
|
|
<p>Hello world</p>
|
|
<footer>Footer</footer>
|
|
</body></html>"""
|
|
|
|
with mock.patch.object(fetcher._session, "get") as mock_get:
|
|
mock_get.return_value.text = html
|
|
mock_get.return_value.headers = {"Content-Type": "text/html; charset=utf-8"}
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
content = fetcher.fetch("https://example.com")
|
|
|
|
assert "Hello world" in content
|
|
assert "alert" not in content # script removed
|
|
assert "Footer" not in content # footer removed
|
|
|
|
def test_fetch_non_html_returns_empty(self):
|
|
fetcher = PageFetcher()
|
|
|
|
with mock.patch.object(fetcher._session, "get") as mock_get:
|
|
mock_get.return_value.headers = {"Content-Type": "application/pdf"}
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
content = fetcher.fetch("https://example.com/doc.pdf")
|
|
|
|
assert content == ""
|
|
|
|
def test_fetch_handles_error(self):
|
|
fetcher = PageFetcher()
|
|
|
|
with mock.patch.object(fetcher._session, "get") as mock_get:
|
|
import requests
|
|
mock_get.side_effect = requests.RequestException("Connection refused")
|
|
|
|
content = fetcher.fetch("https://down.example.com")
|
|
|
|
assert content == ""
|
|
|
|
def test_fetch_truncates_to_max_chars(self):
|
|
fetcher = PageFetcher(max_chars=50)
|
|
|
|
long_text = "A" * 200
|
|
html = f"<html><body><p>{long_text}</p></body></html>"
|
|
|
|
with mock.patch.object(fetcher._session, "get") as mock_get:
|
|
mock_get.return_value.text = html
|
|
mock_get.return_value.headers = {"Content-Type": "text/html"}
|
|
mock_get.return_value.raise_for_status.return_value = None
|
|
|
|
content = fetcher.fetch("https://example.com")
|
|
|
|
assert len(content) <= 53 # 50 + "..." truncation marker
|
|
|
|
def test_fetch_results_enriches_objects(self):
|
|
fetcher = PageFetcher()
|
|
results = [
|
|
SearchResult(title="A", url="https://a.com"),
|
|
SearchResult(title="B", url="https://b.com"),
|
|
]
|
|
|
|
with mock.patch.object(fetcher, "fetch") as mock_fetch:
|
|
mock_fetch.side_effect = ["Content A", "Content B"]
|
|
|
|
enriched = fetcher.fetch_results(results, max_results=2)
|
|
|
|
assert enriched[0].full_content == "Content A"
|
|
assert enriched[0].fetched is True
|
|
assert enriched[1].full_content == "Content B"
|
|
assert enriched[1].fetched is True
|
|
|
|
def test_fetch_results_respects_max(self):
|
|
fetcher = PageFetcher()
|
|
results = [SearchResult(title=f"R{i}", url=f"https://{i}.com") for i in range(5)]
|
|
|
|
with mock.patch.object(fetcher, "fetch") as mock_fetch:
|
|
mock_fetch.return_value = "Content"
|
|
|
|
fetcher.fetch_results(results, max_results=2)
|
|
|
|
assert mock_fetch.call_count == 2
|
|
|
|
def test_fetch_results_skips_already_fetched(self):
|
|
fetcher = PageFetcher()
|
|
results = [
|
|
SearchResult(title="A", url="https://a.com", fetched=True, full_content="Old"),
|
|
SearchResult(title="B", url="https://b.com"),
|
|
]
|
|
|
|
with mock.patch.object(fetcher, "fetch") as mock_fetch:
|
|
mock_fetch.return_value = "New"
|
|
|
|
fetcher.fetch_results(results, max_results=2)
|
|
|
|
# Only the non-fetched result should be fetched
|
|
assert mock_fetch.call_count == 1
|
|
assert results[0].full_content == "Old" # unchanged
|