"""Tests for the keyless DuckDuckGo ``web_search`` backend. Covers the parser (title/url/snippet extraction, redirect decoding, ad skipping, whitespace, the result cap), the HTTP path (success + error handling, mocked via ``respx``), and the selector wiring (no ``search_provider`` defaults to DuckDuckGo). """ from __future__ import annotations import pathlib import httpx import pytest import respx from omnigent.tools.builtins.web_search_duckduckgo import ( _DDG_HTML_URL, _MAX_RESULTS, _decode_result_href, _format_results, _parse_results, _search_duckduckgo, ) # A real captured html.duckduckgo.com/html/ body (query "wikipedia", # captured 2026-06). Re-capture with tests/tools/fixtures/refresh_ddg_fixture.py # when the live drift canary (tests/e2e_live/) goes red. _DDG_GOLDEN = (pathlib.Path(__file__).parent / "fixtures" / "ddg_html_2026-06.html").read_text( encoding="utf-8" ) # A realistic slice of html.duckduckgo.com/html/: two organic results # (redirect-wrapped hrefs, with ``&`` entities) plus one ad whose ``y.js`` # href has no ``uddg`` target and must be skipped. _FIXTURE_HTML = """
Sponsored thing An ad snippet

Example Title

A useful snippet about example.
Other Docs Docs snippet.
""" def test_decode_uddg_redirect() -> None: """A scheme-relative DDG redirect resolves to its decoded ``uddg`` target.""" href = "//duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com%2Fa+b&rut=x" assert _decode_result_href(href) == "https://example.com/a b" def test_decode_direct_http_href_passthrough() -> None: """A direct ``http(s)`` href is returned unchanged.""" assert _decode_result_href("https://direct.example/x") == "https://direct.example/x" def test_decode_ad_link_returns_none() -> None: """An ad / JS ``y.js`` link (no ``uddg`` target) is skipped.""" assert _decode_result_href("//duckduckgo.com/y.js?ad_provider=foo") is None def test_parse_results_extracts_and_skips_ads() -> None: """Parsing yields organic results (title/url/snippet), skipping the ad, and normalizes whitespace in titles and snippets.""" results = _parse_results(_FIXTURE_HTML) assert len(results) == 2, results # the ad result is skipped first = results[0] assert first["title"] == "Example Title" # newline/indentation collapsed assert first["url"] == "https://example.com/page" assert first["snippet"] == "A useful snippet about example." second = results[1] assert second["title"] == "Other Docs" assert second["url"] == "https://other.org/d" def test_parse_results_caps_at_max_results() -> None: """No more than ``_MAX_RESULTS`` results are returned.""" block = ( 'Title {n}' 'snip {n}' ) html = "".join(block.format(n=i) for i in range(_MAX_RESULTS + 5)) assert len(_parse_results(html)) == _MAX_RESULTS def test_format_results_numbered_blocks() -> None: """Results format as numbered ``title / url / snippet`` blocks.""" out = _format_results([{"title": "T", "url": "https://x.example", "snippet": "S"}]) assert out == "1. T\n https://x.example\n S" def test_format_results_empty() -> None: """No results → a clear message, not an empty string.""" assert _format_results([]) == "No results found." @respx.mock def test_search_duckduckgo_success() -> None: """A 200 from the HTML endpoint is parsed and formatted.""" route = respx.post(_DDG_HTML_URL).mock(return_value=httpx.Response(200, text=_FIXTURE_HTML)) out = _search_duckduckgo("example query", {}) assert route.called # Sent as a form POST with the query. assert b"q=example" in route.calls.last.request.content assert out.startswith("1. Example Title") assert "https://example.com/page" in out @respx.mock def test_search_duckduckgo_http_error() -> None: """A non-2xx response surfaces as a readable error, not an exception.""" respx.post(_DDG_HTML_URL).mock(return_value=httpx.Response(503)) assert _search_duckduckgo("q", {}) == "DuckDuckGo search error: HTTP 503" @respx.mock def test_search_duckduckgo_timeout() -> None: """A network timeout surfaces as a readable error.""" respx.post(_DDG_HTML_URL).mock(side_effect=httpx.TimeoutException("slow")) out = _search_duckduckgo("q", {}) assert out.startswith("DuckDuckGo search error") def test_web_search_no_provider_fails_loudly( monkeypatch: pytest.MonkeyPatch, ) -> None: """With no ``search_provider``, the selector returns a loud, helpful error naming the engines — it does NOT silently pick one (per maintainer review), so it's always explicit which engine ran. The DDG backend is not invoked.""" import omnigent.tools.builtins.web_search_duckduckgo as ddg from omnigent.tools.builtins.web_search import _search monkeypatch.setattr( ddg, "_search_duckduckgo", lambda q, c: pytest.fail("must not auto-run DDG") ) out = _search("hello world", {}) assert out.startswith("web_search error: no search_provider") assert "duckduckgo" in out def test_web_search_explicit_duckduckgo_provider( monkeypatch: pytest.MonkeyPatch, ) -> None: """``search_provider: duckduckgo`` selects the DDG backend explicitly.""" import omnigent.tools.builtins.web_search_duckduckgo as ddg from omnigent.tools.builtins.web_search import _search monkeypatch.setattr(ddg, "_search_duckduckgo", lambda q, c: "DDG-OK") assert _search("hi", {"search_provider": "duckduckgo"}) == "DDG-OK" # ── robustness: HTML scraping is best-effort, so it must degrade, never crash ── @respx.mock @pytest.mark.parametrize( "exc", [ httpx.RemoteProtocolError("peer closed connection"), httpx.ReadError("connection reset"), httpx.DecodingError("bad gzip"), ], ) def test_search_duckduckgo_transport_errors_no_raise(exc: httpx.HTTPError) -> None: """A flaky/rate-limiting DDG endpoint raises RemoteProtocol/Read/Decoding errors — none of which are ``TransportError`` — and they must surface as a readable string, not crash the tool. (Regression: the old narrow catch let these escape.)""" respx.post(_DDG_HTML_URL).mock(side_effect=exc) assert _search_duckduckgo("q", {}).startswith("DuckDuckGo search error") @respx.mock def test_search_duckduckgo_connect_error() -> None: """A connect failure still surfaces as a readable error (broadened catch).""" respx.post(_DDG_HTML_URL).mock(side_effect=httpx.ConnectError("no route")) assert _search_duckduckgo("q", {}).startswith("DuckDuckGo search error") @respx.mock def test_search_duckduckgo_rate_limit_429() -> None: """The 429 throttle still surfaces its status number (HTTPStatusError stays distinct from the broadened transport catch).""" respx.post(_DDG_HTML_URL).mock(return_value=httpx.Response(429)) assert _search_duckduckgo("q", {}) == "DuckDuckGo search error: HTTP 429" @respx.mock def test_search_duckduckgo_blocked_200_is_empty() -> None: """A 200 block/anomaly page (no ``result__a``) yields "No results found." — we intentionally do NOT phrase-match block pages, so it's indistinguishable from a genuine zero-result query (best-effort contract).""" blocked = ( "
" "If this persists, please let us know. Performed automatically." "
" ) respx.post(_DDG_HTML_URL).mock(return_value=httpx.Response(200, text=blocked)) assert _search_duckduckgo("q", {}) == "No results found." @respx.mock def test_search_duckduckgo_empty_body() -> None: """An empty 200 body degrades to "No results found." over the full path.""" respx.post(_DDG_HTML_URL).mock(return_value=httpx.Response(200, text="")) assert _search_duckduckgo("q", {}) == "No results found." @respx.mock def test_search_duckduckgo_garbage_body() -> None: """Garbage HTML degrades gracefully (never raises) over the full path.""" respx.post(_DDG_HTML_URL).mock( return_value=httpx.Response(200, text="<> not html") ) assert _search_duckduckgo("q", {}) == "No results found." @respx.mock def test_search_duckduckgo_caps_rendered_output_at_max_results() -> None: """More than _MAX_RESULTS results render only the first _MAX_RESULTS — the cap is tied to the rendered string, not just the parsed list.""" block = ( '' "Title {n}" 'snip {n}' ) html = "".join(block.format(n=i) for i in range(_MAX_RESULTS + 5)) respx.post(_DDG_HTML_URL).mock(return_value=httpx.Response(200, text=html)) out = _search_duckduckgo("q", {}) assert f"{_MAX_RESULTS}. " in out assert f"{_MAX_RESULTS + 1}. " not in out @respx.mock def test_search_duckduckgo_sends_browser_user_agent() -> None: """The request carries a browser-like UA — the endpoint blocks/empties requests without one, so this is a load-bearing contract.""" route = respx.post(_DDG_HTML_URL).mock(return_value=httpx.Response(200, text=_FIXTURE_HTML)) _search_duckduckgo("q", {}) assert route.calls.last.request.headers["user-agent"].startswith("Mozilla/5.0") def test_parse_results_empty_and_garbage() -> None: """The parser returns [] (never raises) for empty or non-HTML input.""" assert _parse_results("") == [] assert _parse_results("<<< not really html & broken >>>") == [] def test_parse_results_truncated_html_no_raise() -> None: """Truncated markup (stream cut mid-result) degrades gracefully — no raise, at most a partial result; it never crashes search.""" html = ( '
Titl' ) results = _parse_results(html) # must not raise assert len(results) <= 1 def test_decode_result_href_preserves_encoded_ampersand_in_target() -> None: """A redirect whose target has an encoded ``&`` (``%26``) plus a literal ``&`` separator decodes to the correct URL — the query inside the target survives.""" href = "//duckduckgo.com/l/?uddg=https%3A%2F%2Fa.com%2Fx%26y%3D1&rut=z" assert _decode_result_href(href) == "https://a.com/x&y=1" # Verbatim slice of a LIVE html.duckduckgo.com/html/ response (captured # 2026-06). Note the current markup drift the parser must tolerate: # * href is now a DIRECT url, NOT the //duckduckgo.com/l/?uddg= redirect # * result__a carries rel="nofollow" # * snippets embed ... highlight tags _LIVE_DIRECT_HTML = ( '
' '' "Sign in - Claude" '' "Claude is an AI assistant by Anthropic, " "designed to assist with creative tasks." "
" ) def test_parse_results_live_direct_url_markup() -> None: """Parse the shape DDG actually serves today (verified live): a direct ``href`` (no ``/l/?uddg=`` redirect), ``rel="nofollow"`` on ``result__a``, and ```` highlight tags inside the snippet. The parser passes the direct URL through, ignores ``rel``, and strips the bold tags to plain text. This is the regression guard for the exact markup drift the reviewer flagged as fragile — it already happened, and the backend handles it.""" results = _parse_results(_LIVE_DIRECT_HTML) assert len(results) == 1 assert results[0]["url"] == "https://claude.ai/" assert results[0]["title"] == "Sign in - Claude" assert "" not in results[0]["snippet"] assert results[0]["snippet"] == ( "Claude is an AI assistant by Anthropic, designed to assist with creative tasks." ) def test_parse_results_returns_partial_when_parser_raises( monkeypatch: pytest.MonkeyPatch, ) -> None: """If the underlying HTMLParser ever raises mid-stream, ``_parse_results`` must not propagate it — it returns whatever was parsed before the fault. (``html.parser`` is tolerant enough that no real input triggers this, so we inject the failure to prove the H2 guard.)""" import omnigent.tools.builtins.web_search_duckduckgo as ddg def _boom(self: ddg._ResultParser, data: str) -> None: self.results.append({"title": "Partial", "url": "https://a.com", "snippet": "s"}) raise RuntimeError("parser blew up mid-stream") monkeypatch.setattr(ddg._ResultParser, "feed", _boom) out = ddg._parse_results("") # must not raise assert out == [{"title": "Partial", "url": "https://a.com", "snippet": "s"}] # ── recorded real-response fixture: parse a genuine captured DDG page offline ── def test_parse_real_captured_ddg_page_structure() -> None: """Parse a REAL captured html.duckduckgo.com/html/ body (captured 2026-06). Asserts structure, not upstream content, so it can't flake on result churn — but it WILL catch a markup change that the inline fixtures don't model.""" results = _parse_results(_DDG_GOLDEN) assert len(results) >= 5, f"expected several organic results, got {len(results)}" for r in results: assert r["title"], r # non-empty title assert r["url"].startswith(("http://", "https://")), r["url"] assert "duckduckgo.com/l/" not in r["url"], r["url"] # redirect fully decoded assert "/y.js" not in r["url"], r["url"] # no ad / JS link leaked assert "" not in r["snippet"], r["snippet"] # highlight tags stripped @respx.mock def test_search_duckduckgo_over_real_captured_page() -> None: """The full HTTP path renders the real captured body to numbered blocks.""" respx.post(_DDG_HTML_URL).mock(return_value=httpx.Response(200, text=_DDG_GOLDEN)) out = _search_duckduckgo("wikipedia", {}) assert out.startswith("1. ") assert out != "No results found." def test_parse_results_snippet_in_non_anchor_tag() -> None: """Drift guard (parser hardening): the snippet is captured by CSS class even when DDG moves it out of an ```` into a ``
`` (with nested children), and capture ends only when that ``
`` closes — not an inner one.""" html = ( 'Title' '
A snippet ' "
with a nested wrapper
and a tail.
" ) results = _parse_results(html) assert len(results) == 1 assert results[0]["url"] == "https://example.com/p" assert results[0]["title"] == "Title" assert results[0]["snippet"] == "A snippet with a nested wrapper and a tail."