"""web_search must prefer SearXNG, fall back in order, and never run unbounded.
The server is a separate stdio process: its environment is fixed at connect time,
so the SearXNG URL reaches it only through the ``${SEARXNG_URL:-}`` placeholder in
``mcp_servers.d/navi-web.json``. What these tests pin down is the chain itself —
order, per-backend budgets, the retry rule, the batch rules — because until
2026-11 those waits added up to half a minute and a broken backend was
indistinguishable from a slow one.
These tests live in the root suite rather than in ``mcp-servers/navi-web/tests/``
because ``pyproject.toml`` collects ``testpaths = ["tests"]`` only; the loader is
in ``conftest.py`` next to this file.
"""
import asyncio
import time
def _hit(title: str = "t") -> dict:
return {"title": title, "url": "https://example.com", "content": ""}
def _ok(source: str = "SearXNG") -> dict:
return {"success": True, "output": f"[{source}]\n[1] t", "error": None,
"metadata": {"source": source, "results": [_hit()]}}
class TestOrder:
async def test_searxng_answers_first(self, navi_web_search, monkeypatch, navi_web_settings):
calls: list[str] = []
async def searxng(s, q, n):
calls.append("searxng")
return [_hit()]
def ddg(s, q, n):
calls.append("ddg")
return [_hit()]
monkeypatch.setattr(navi_web_search, "_searxng", searxng)
monkeypatch.setattr(navi_web_search, "_ddg", ddg)
navi_web_settings.searxng_url = "http://searx"
result = await navi_web_search.web_search(navi_web_settings, "q")
assert result["success"] is True
assert result["metadata"]["source"] == "SearXNG"
assert calls == ["searxng"]
async def test_empty_searxng_falls_back_to_ddg(
self, navi_web_search, monkeypatch, navi_web_settings
):
async def searxng(s, q, n):
return []
monkeypatch.setattr(navi_web_search, "_searxng", searxng)
monkeypatch.setattr(navi_web_search, "_ddg", lambda s, q, n: [_hit()])
navi_web_settings.searxng_url = "http://searx"
result = await navi_web_search.web_search(navi_web_settings, "q")
assert result["metadata"]["source"] == "DDG"
async def test_brave_is_last_and_needs_a_key(
self, navi_web_search, monkeypatch, navi_web_settings
):
calls: list[str] = []
async def searxng(s, q, n):
calls.append("searxng")
return []
def ddg(s, q, n):
calls.append("ddg")
return []
async def brave(s, q, n):
calls.append("brave")
return [_hit()]
monkeypatch.setattr(navi_web_search, "_searxng", searxng)
monkeypatch.setattr(navi_web_search, "_ddg", ddg)
monkeypatch.setattr(navi_web_search, "_brave", brave)
navi_web_settings.searxng_url = "http://searx"
without_key = await navi_web_search.web_search(navi_web_settings, "q")
assert without_key["success"] is False
assert "brave" not in calls
navi_web_settings.brave_api_key = "key"
with_key = await navi_web_search.web_search(navi_web_settings, "q")
assert with_key["metadata"]["source"] == "Brave"
assert calls[-1] == "brave"
async def test_all_backends_fail_names_the_last_error(
self, navi_web_search, monkeypatch, navi_web_settings
):
async def searxng(s, q, n):
return []
monkeypatch.setattr(navi_web_search, "_searxng", searxng)
monkeypatch.setattr(navi_web_search, "_ddg", lambda s, q, n: [])
navi_web_settings.searxng_url = "http://searx"
result = await navi_web_search.web_search(navi_web_settings, "q")
assert result["success"] is False
assert result["error"] == "DDG: empty response"
assert "All search backends failed" in result["output"]
class TestBudgets:
async def test_slow_searxng_falls_through_quickly(
self, navi_web_search, monkeypatch, navi_web_settings
):
async def slow(s, q, n):
await asyncio.sleep(10)
monkeypatch.setattr(navi_web_search, "_searxng", slow)
monkeypatch.setattr(navi_web_search, "_ddg", lambda s, q, n: [_hit()])
monkeypatch.setattr(navi_web_search, "_SEARXNG_TIMEOUT", 0.05)
navi_web_settings.searxng_url = "http://searx"
started = time.monotonic()
result = await navi_web_search.web_search(navi_web_settings, "q")
assert result["metadata"]["source"] == "DDG"
assert time.monotonic() - started < 1.0
async def test_total_budget_is_the_hard_cap(
self, navi_web_search, monkeypatch, navi_web_settings
):
async def slow(s, q, n):
await asyncio.sleep(10)
monkeypatch.setattr(navi_web_search, "_searxng", slow)
monkeypatch.setattr(navi_web_search, "_TOTAL_TIMEOUT", 0.05)
navi_web_settings.searxng_url = "http://searx"
started = time.monotonic()
result = await navi_web_search.web_search(navi_web_settings, "q")
assert result["success"] is False
assert result["error"] == "timeout"
assert time.monotonic() - started < 1.0
class TestDdgRetry:
async def test_retries_once_after_an_error(
self, navi_web_search, monkeypatch, navi_web_settings
):
attempts: list[int] = []
def flaky(s, q, n):
attempts.append(len(attempts) + 1)
if len(attempts) == 1:
raise RuntimeError("blocked")
return [_hit()]
monkeypatch.setattr(navi_web_search, "_ddg", flaky)
monkeypatch.setattr(navi_web_search, "_DDG_RETRY_DELAY", 0.0)
result = await navi_web_search.web_search(navi_web_settings, "q")
assert result["metadata"]["source"] == "DDG"
assert len(attempts) == 2
async def test_empty_result_is_not_retried(
self, navi_web_search, monkeypatch, navi_web_settings
):
attempts: list[int] = []
def empty(s, q, n):
attempts.append(1)
return []
monkeypatch.setattr(navi_web_search, "_ddg", empty)
result = await navi_web_search.web_search(navi_web_settings, "q")
assert result["success"] is False
assert len(attempts) == 1
class TestDdgClient:
def test_budget_and_arguments_reach_ddgs(
self, navi_web_search, monkeypatch, navi_web_settings
):
seen: dict = {}
class FakeDDGS:
def __init__(self, timeout=None):
seen["timeout"] = timeout
def text(self, query, **kwargs):
seen["query"] = query
seen.update(kwargs)
return [_hit()]
monkeypatch.setattr(navi_web_search, "DDGS", FakeDDGS)
assert navi_web_search._ddg(navi_web_settings, "q", 3)[0]["title"] == "t"
assert seen["timeout"] == int(navi_web_search._DDG_TIMEOUT)
assert seen["query"] == "q"
assert seen["backend"] == navi_web_search._DDG_BACKENDS
assert seen["max_results"] == 3
assert seen["region"] == "ru-ru"
def test_the_pinned_engines_are_real_ddgs_names(self, navi_web_search):
"""An unknown name is not an error in ddgs — it silently means "all".
``backend="html"`` did exactly that, quietly adding yandex and every other
engine to the merge. Pin the names against the library's own engine map.
"""
from ddgs.engines import ENGINES
known = set(ENGINES["text"].keys())
pinned = set(navi_web_search._DDG_BACKENDS.split(","))
assert pinned <= known, f"unknown ddgs backends: {sorted(pinned - known)}"
assert "yandex" not in pinned
class TestSearxngRequest:
async def test_engines_and_language_are_pinned(
self, navi_web_search, monkeypatch, navi_web_settings
):
"""Bing is deliberately absent and the language is not fixed.
Bing answered with trending/homepage noise on this host and, being
non-empty, won the chain; a pinned language made the instance's DDG engine
return CAPTCHA on every request.
"""
seen: dict = {}
class FakeResponse:
def raise_for_status(self):
pass
def json(self):
return {"results": [{"title": "t", "url": "https://e", "content": "c"}]}
class FakeClient:
def __init__(self, timeout=None):
seen["timeout"] = timeout
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
async def get(self, url, params=None):
seen["url"], seen["params"] = url, params
return FakeResponse()
monkeypatch.setattr(navi_web_search.httpx, "AsyncClient", FakeClient)
navi_web_settings.searxng_url = "http://searx"
results = await navi_web_search._searxng(navi_web_settings, "q", 5)
assert results == [{"title": "t", "href": "https://e", "body": "c"}]
assert seen["url"] == "http://searx/search"
engines = seen["params"]["engines"].split(",")
assert "bing" not in engines
assert "yandex" not in engines
assert seen["params"]["language"] == "all"
assert seen["params"]["format"] == "json"
class TestBatch:
"""One call may carry up to three independent queries.
The point of the batch is wall time — three queries cost the slowest, not the
sum — and the point of the rules around it is that a batch never hides which
query answered and which did not.
"""
async def test_a_string_keeps_the_single_query_shape(
self, navi_web_search, monkeypatch, navi_web_settings
):
async def chain(s, q, n):
return _ok()
monkeypatch.setattr(navi_web_search, "_search_chain", chain)
result = await navi_web_search.web_search(navi_web_settings, "one")
assert result["metadata"]["source"] == "SearXNG"
assert "###" not in result["output"]
async def test_queries_run_concurrently(
self, navi_web_search, monkeypatch, navi_web_settings
):
started: list[str] = []
async def chain(s, q, n):
started.append(q)
await asyncio.sleep(0.1)
return _ok()
monkeypatch.setattr(navi_web_search, "_search_chain", chain)
began = time.monotonic()
result = await navi_web_search.web_search(navi_web_settings, ["a", "b", "c"])
elapsed = time.monotonic() - began
assert result["success"] is True
assert started == ["a", "b", "c"]
assert elapsed < 0.25, f"three 0.1s queries took {elapsed:.2f}s — not concurrent"
assert result["output"].count("### ") == 3
async def test_blocks_follow_the_request_order(
self, navi_web_search, monkeypatch, navi_web_settings
):
async def chain(s, q, n):
await asyncio.sleep(0.1 if q == "a" else 0)
return _ok()
monkeypatch.setattr(navi_web_search, "_search_chain", chain)
result = await navi_web_search.web_search(navi_web_settings, ["a", "b"])
assert result["output"].index('"a"') < result["output"].index('"b"')
async def test_more_than_three_are_capped_with_a_note(
self, navi_web_search, monkeypatch, navi_web_settings
):
started: list[str] = []
async def chain(s, q, n):
started.append(q)
return _ok()
monkeypatch.setattr(navi_web_search, "_search_chain", chain)
result = await navi_web_search.web_search(
navi_web_settings, ["a", "b", "c", "d", "e"]
)
assert started == ["a", "b", "c"]
assert "Only the first 3 of 5 queries ran" in result["output"]
async def test_blanks_and_duplicates_collapse(
self, navi_web_search, monkeypatch, navi_web_settings
):
started: list[str] = []
async def chain(s, q, n):
started.append(q)
return _ok()
monkeypatch.setattr(navi_web_search, "_search_chain", chain)
await navi_web_search.web_search(navi_web_settings, ["a", " ", "a", " a "])
assert started == ["a"]
async def test_one_failed_query_does_not_erase_the_others(
self, navi_web_search, monkeypatch, navi_web_settings
):
async def chain(s, q, n):
if q == "b":
return {"success": False, "output": "x", "error": "DDG: blocked",
"metadata": {}}
return _ok()
monkeypatch.setattr(navi_web_search, "_search_chain", chain)
result = await navi_web_search.web_search(navi_web_settings, ["a", "b", "c"])
assert result["success"] is True
assert '### "b" -> failed: DDG: blocked' in result["output"]
assert [item["success"] for item in result["metadata"]["queries"]] == [
True, False, True,
]
async def test_a_query_timing_out_does_not_erase_the_others(
self, navi_web_search, monkeypatch, navi_web_settings
):
async def chain(s, q, n):
if q == "b":
await asyncio.sleep(10)
return _ok()
monkeypatch.setattr(navi_web_search, "_search_chain", chain)
monkeypatch.setattr(navi_web_search, "_TOTAL_TIMEOUT", 0.05)
result = await navi_web_search.web_search(navi_web_settings, ["a", "b"])
assert result["success"] is True
assert '### "b" -> failed: timeout' in result["output"]
class TestEmptyBatch:
async def test_no_query_is_an_error(self, navi_web_search, navi_web_settings):
result = await navi_web_search.web_search(navi_web_settings, [" ", ""])
assert result["success"] is False
assert result["error"] == "empty_query"
async def test_a_batch_with_no_answers_fails(
self, navi_web_search, monkeypatch, navi_web_settings
):
"""The tool raises on this, which is what makes a dead search visible in
the journal — so a batch that answered nothing must stay a failure."""
async def chain(s, q, n):
return {"success": False, "output": "x", "error": "DDG: blocked",
"metadata": {}}
monkeypatch.setattr(navi_web_search, "_search_chain", chain)
result = await navi_web_search.web_search(navi_web_settings, ["a", "b"])
assert result["success"] is False
assert result["error"] == "all_queries_failed"
assert "a: DDG: blocked" in result["output"]
assert "b: DDG: blocked" in result["output"]
