Cover external web search with unit and route tests

tests/test_bot_search.py

@@ -0,0 +1,61 @@
+"""
+test_bot_search.py — сквозные тесты внешнего поиска в _handle_message_core.
+
+Что защищают: инцидент 11.10.2026 — явная просьба "погугли" при мёртвой квоте
+2.5 шла на модели без поиска и отвечала по памяти. Первый тест ловит подклейку
+сниппетов в промпт, второй — честную пометку, когда и внешний поиск пуст.
+Квота гасится точечно (только поисковые Gemini), fetch мокается на границе I/O.
+
+Запуск:
+    pytest tests/test_bot_search.py -v
+"""
+import asyncio
+
+import bot
+from tests.bot_test_helpers import _run_core_capturing_prompt
+
+_DEAD_SEARCH_QUOTA = {("gemini", "gemini-2.5-flash"), ("gemini", "gemini-2.5-flash-lite")}
+
+
+def _kill_search_quota(monkeypatch):
+    import lumen_router_config
+    monkeypatch.setattr(
+        lumen_router_config,
+        "_is_quota_exhausted",
+        lambda provider, model_id: (provider, model_id) in _DEAD_SEARCH_QUOTA,
+    )
+
+
+def test_explicit_search_request_gets_snippets_without_live_search(rate_guard_setup, monkeypatch):
+    message = rate_guard_setup()
+    message.text = "погугли что происходит с ДЦ Яндекса"
+    _kill_search_quota(monkeypatch)
+
+    async def fake_fetch(query, *, timeout_sec=None, max_results=None):
+        assert "ДЦ" in query
+        return [{"title": "Новость", "url": "https://example.com/n", "snippet": "Свежие данные"}]
+
+    monkeypatch.setattr("lumen_message_core.fetch_web_snippets", fake_fetch)
+    prompt = _run_core_capturing_prompt(message, monkeypatch)
+    assert "Web search snippets" in prompt
+    assert "https://example.com/n" in prompt
+    bot.chat_state.pop(123, None)
+
+
+def test_empty_fallback_search_marks_reply_without_live_search(rate_guard_setup, monkeypatch):
+    message = rate_guard_setup()
+    message.text = "погугли что происходит с ДЦ Яндекса"
+    _kill_search_quota(monkeypatch)
+
+    async def fake_fetch(query, *, timeout_sec=None, max_results=None):
+        return []
+
+    async def fake_run_route(chat_id, ai_prompt, route, message, **kwargs):
+        return "ok", False
+
+    monkeypatch.setattr("lumen_message_core.fetch_web_snippets", fake_fetch)
+    monkeypatch.setattr(bot, "_run_route", fake_run_route)
+    asyncio.run(bot._handle_message_core(message))
+    sent_text = bot._safe_reply.await_args[0][1]
+    assert "without live web search" in sent_text
+    bot.chat_state.pop(123, None)

tests/test_lumen_router_config.py

@@ -110,6 +110,25 @@ def test_looks_like_freshness_query_ignores_substrings():
     assert lumen_router_config._looks_like_freshness_query("groundbreaking discovery") is False
 
 
+def test_looks_like_search_request_detects_explicit_tool_asks():
+    # Прод 11.10.2026: "найди с веб поиском" без слов свежести уходило на модели
+    # без поиска — явная просьба искать теперь тоже ведёт на свежий маршрут.
+    assert lumen_router_config._looks_like_search_request("погугли что с ДЦ Яндекса") is True
+    assert lumen_router_config._looks_like_search_request("найди инфу с помощью веб-поиска") is True
+    assert lumen_router_config._looks_like_search_request("ищи через поиск, что там с дата-центрами") is True
+    assert lumen_router_config._looks_like_search_request("search the web for Yandex data centers") is True
+    assert lumen_router_config._looks_like_search_request("поищи в интернете свежие новости") is True
+
+
+def test_looks_like_search_request_ignores_code_searches():
+    # Голые "найди/ищи" без упоминания поиска — про код, а не про веб: иначе
+    # "найди баг" гнал бы тяжёлый разбор кода на поисковый маршрут.
+    assert lumen_router_config._looks_like_search_request("найди баг в скрипте") is False
+    assert lumen_router_config._looks_like_search_request("найди ошибку") is False
+    assert lumen_router_config._looks_like_search_request("привет") is False
+    assert lumen_router_config._looks_like_search_request("") is False
+
+
 def test_looks_like_heavy_query_detects_english():
     # Внешний аудит: "write a Python parser" пролетал в light-ветку.
     assert lumen_router_config._looks_like_heavy_query("write a Python parser for csv") is True

tests/test_lumen_search.py

@@ -0,0 +1,138 @@
+"""
+test_lumen_search.py — юнит-тесты на lumen_search.py: парсер lite-выдачи DuckDuckGo,
+блок сниппетов в промпт, детект живого поиска в маршруте, килл-свитч и фолбэк
+fetch на [] без сети.
+
+Что защищают: инцидент 11.10.2026 — свежий запрос при мёртвой квоте 2.5 отвечал
+по памяти, будто искал. Парсер тестируется статичной фикстурой (без сети), fetch —
+стабом сессии (граница I/O, не перереализация поведения).
+
+Запуск:
+    pytest tests/test_lumen_search.py -v
+"""
+import asyncio
+from types import SimpleNamespace
+
+import aiohttp
+import lumen_search
+
+
+_DDG_FIXTURE = """
+<html><body><table>
+<tr><td><a rel="nofollow" href="//duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com%2Fnews&amp;rut=abc">Example news</a></td></tr>
+<tr><td class='result-snippet'>Fresh snippet text here</td></tr>
+<tr><td><a rel="nofollow" href="https://direct.example.org/page">Direct <b>title</b></a></td></tr>
+<tr><td class='result-snippet'>Second snippet</td></tr>
+<tr><td><a rel="nofollow" href="/internal/path">Internal link</a></td></tr>
+<tr><td><a rel="nofollow" href="//duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com%2Fnews">Example news again</a></td></tr>
+<tr><td class='result-snippet'>Duplicate url snippet</td></tr>
+</table></body></html>
+"""
+
+
+def test_parse_ddg_lite_html_unwraps_redirect_and_pairs_snippets():
+    out = lumen_search.parse_ddg_lite_html(_DDG_FIXTURE)
+    assert out[0]["url"] == "https://example.com/news"
+    assert out[0]["title"] == "Example news"
+    assert out[0]["snippet"] == "Fresh snippet text here"
+    assert out[1]["url"] == "https://direct.example.org/page"
+    assert out[1]["title"] == "Direct title"
+    # Внутренняя ссылка и дубль целевого URL отсечены.
+    assert all("duckduckgo.com" not in item["url"] for item in out)
+    assert len({item["url"] for item in out}) == len(out) == 2
+
+
+def test_parse_ddg_lite_html_empty_page_gives_empty_list():
+    assert lumen_search.parse_ddg_lite_html("") == []
+    assert lumen_search.parse_ddg_lite_html("<html></html>") == []
+
+
+def test_format_snippets_for_prompt_marks_data_and_skips_empties():
+    block = lumen_search.format_snippets_for_prompt([
+        {"title": "T", "url": "https://example.com", "snippet": "S"},
+        {"title": "", "url": "", "snippet": ""},
+    ])
+    assert "untrusted" in block and "not instructions" in block
+    assert "https://example.com" in block
+    assert lumen_search.format_snippets_for_prompt([]) == ""
+    assert lumen_search.format_snippets_for_prompt([{"title": "", "url": "", "snippet": ""}]) == ""
+
+
+def test_route_has_live_search_follows_grounding_flags():
+    assert lumen_search.route_has_live_search([("groq", "qwen/qwen3.8-27b")]) is False
+    assert lumen_search.route_has_live_search([("gemini", "gemini-3.8-flash")]) is False
+    assert lumen_search.route_has_live_search([("gemini", "gemini-2.5-flash")]) is True
+    assert lumen_search.route_has_live_search([
+        ("groq", "qwen/qwen3.8-27b"), ("gemini", "gemini-2.5-flash-lite"),
+    ]) is True
+
+
+def test_web_search_kill_switch_and_limits(monkeypatch):
+    monkeypatch.setenv("WEB_SEARCH_ENABLED", "0")
+    assert lumen_search.web_search_is_enabled() is False
+    monkeypatch.setenv("WEB_SEARCH_ENABLED", "1")
+    assert lumen_search.web_search_is_enabled() is True
+    monkeypatch.setenv("WEB_SEARCH_TIMEOUT_SEC", "мусор")
+    assert lumen_search.web_search_timeout_sec() == 8.0
+    monkeypatch.setenv("WEB_SEARCH_MAX_RESULTS", "3")
+    assert lumen_search.web_search_max_results() == 3
+
+
+class _StubResp:
+    def __init__(self, status=200, text=""):
+        self.status = status
+        self._text = text
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+    async def text(self):
+        return self._text
+
+    def raise_for_status(self):
+        raise RuntimeError(f"http {self.status}")
+
+
+class _StubSession:
+    def __init__(self, resp=None, exc=None):
+        self._resp = resp
+        self._exc = exc
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+    def post(self, *args, **kwargs):
+        if self._exc is not None:
+            raise self._exc
+        return self._resp
+
+    def get(self, *args, **kwargs):
+        if self._exc is not None:
+            raise self._exc
+        return self._resp
+
+
+def _stub_aiohttp(resp=None, exc=None):
+    return SimpleNamespace(
+        ClientSession=lambda *args, **kwargs: _StubSession(resp, exc),
+        ClientTimeout=aiohttp.ClientTimeout,
+    )
+
+
+def test_fetch_web_snippets_parses_stubbed_page(monkeypatch):
+    monkeypatch.setattr(lumen_search, "aiohttp", _stub_aiohttp(_StubResp(200, _DDG_FIXTURE)))
+    out = asyncio.run(lumen_search.fetch_web_snippets("что с ДЦ Яндекса"))
+    assert [item["url"] for item in out] == ["https://example.com/news", "https://direct.example.org/page"]
+
+
+def test_fetch_web_snippets_fails_open_on_network_error(monkeypatch):
+    # Любая сетевая ошибка — пустой список, вызывающий код ставит честную пометку, а не падает.
+    monkeypatch.setattr(lumen_search, "aiohttp", _stub_aiohttp(exc=ConnectionError("down")))
+    assert asyncio.run(lumen_search.fetch_web_snippets("что с ДЦ Яндекса")) == []
+    assert asyncio.run(lumen_search.fetch_web_snippets("   ")) == []