[eric] web: bing + brave join the keyless race behind ddg; bing went 50/50 on the burst that tripped both ddg and brave

This commit is contained in:
ciregenz
2026-08-10 14:02:42 -07:00
parent 321b1ef549
commit 784b33c15a
10 changed files with 480 additions and 37 deletions
@@ -0,0 +1,16 @@
"""What a keyless search engine actually said: results, an honest nothing, or a refusal.
"Closed" and "genuinely no hits" look identical on the wire (both are a 200
with no result anchors) and the caller has to act on them differently: a
refusal is a failed tier that should count against the engine's breaker, an
empty result set is the honest answer to a nonsense query."""
from pydantic import BaseModel, ConfigDict
class EngineAnswer(BaseModel):
model_config = ConfigDict(validate_assignment=True)
results: str = ""
# Challenge, HTTP error, or markup we no longer recognise: the engine did not answer the question.
refused: bool = False
@@ -0,0 +1,88 @@
"""Bing search: the third keyless engine, straight from Microsoft's own index.
DuckDuckGo largely serves Bing's index through DuckDuckGo's frontend, so when
DDG's anti-bot wall is up the index itself is usually still reachable here.
Probed live 2026-08-10 through the Chrome-impersonating client: 200 with
server-rendered results on 3/3 queries at 0.2-0.3s, the fastest of every
engine probed.
Organic results are `<li class="b_algo">` blocks (ads live in `b_ad`, so this
match skips them by construction): title inside an `<h2><a>`, snippet in the
`b_caption` paragraph. The href is a click-tracking redirect whose `u=a1<b64>`
parameter carries the real URL base64url-encoded; a result pointing at
bing.com/ck/a would be junk to a model, so decoding it is load-bearing.
"There are no results for" is Bing's honest empty page; zero parsed blocks
without that marker means a challenge or markup drift and reads as a refusal."""
import base64
import html
import re
from typing import List
from typeguard import typechecked
from backend.apps.agents.tools.browser_http import browser_request
from backend.apps.agents.tools.search.engine_answer import EngineAnswer
from backend.apps.agents.tools.search.strip_tags import strip_tags
P_SEARCH_URL = "https://www.bing.com/search"
P_TIMEOUT = 12.0
P_BLOCK_RE = re.compile(r'<li class="b_algo.*?</li>', flags=re.DOTALL)
P_H2_RE = re.compile(r"<h2[^>]*>(.*?)</h2>", flags=re.DOTALL)
P_HREF_RE = re.compile(r'<a[^>]*href="([^"]+)"')
P_SNIPPET_RE = re.compile(r'class="b_caption[^"]*".*?<p[^>]*>(.*?)</p>', flags=re.DOTALL)
P_REDIRECT_RE = re.compile(r"[?&]u=a1([A-Za-z0-9_\-]+)")
P_NO_RESULTS_MARKER = "There are no results for"
@typechecked
def p_real_url(raw: str) -> str:
raw = html.unescape(raw)
m = P_REDIRECT_RE.search(raw)
if not m:
return raw
token = m.group(1)
try:
return base64.urlsafe_b64decode(token + "=" * (-len(token) % 4)).decode("utf-8", "replace")
except Exception:
return raw
@typechecked
def parse_bing_results(body: str, num_results: int) -> str:
"""Format Bing's b_algo rows with their redirect hrefs decoded to real URLs."""
entries: List[str] = []
for block in P_BLOCK_RE.findall(body):
if len(entries) >= num_results:
break
h2 = P_H2_RE.search(block)
if not h2:
continue
href = P_HREF_RE.search(h2.group(1))
title = strip_tags(h2.group(1))
if not href or not title:
continue
entry = f"[{len(entries) + 1}] {title}\n {p_real_url(href.group(1))}"
snippet_match = P_SNIPPET_RE.search(block)
if snippet_match:
snippet = strip_tags(snippet_match.group(1))
if snippet:
entry += f"\n {snippet}"
entries.append(entry)
return "\n\n".join(entries)
@typechecked
async def search_bing(query: str, num_results: int) -> EngineAnswer:
"""Bing's answer: results, an honest nothing, or a refusal."""
reply = await browser_request(P_SEARCH_URL, params={"q": query}, timeout=P_TIMEOUT)
if reply.status != 200:
return EngineAnswer(refused=True)
results = parse_bing_results(reply.text, num_results)
if results:
return EngineAnswer(results=results)
if P_NO_RESULTS_MARKER in reply.text:
return EngineAnswer()
return EngineAnswer(refused=True)
@@ -0,0 +1,78 @@
"""Brave search: a keyless engine with its OWN index behind DuckDuckGo.
Brave runs its own crawler, so it fails independently of the Bing-fed engines
(DuckDuckGo, Bing itself) and of Google (Startpage). Probed live 2026-08-10
through the Chrome-impersonating client: 200 with server-rendered results on
3/3 queries at 0.7-0.9s, while Mojeek, Ecosia and Yep 403'd the same client
on the same machine and Qwant served a JS shell.
The SERP is server-rendered Svelte: each organic result is a
`<div class="snippet ..." data-type="web">` block whose first anchor carries
the REAL target URL (no redirect wrapper), whose title div carries the clean
text in its `title` attribute, and whose description sits in a
`<div class="content ...">`. Ads carry a different data-type, so matching
`data-type="web"` skips them by construction.
A gibberish query still returns fuzzy matches (measured: 19 blocks plus a
"Not many great matches" banner), so zero parsed blocks on a 200 is markup
drift or a challenge page, never an honest no-hits; both read as a refusal.
Hammered with 50 back-to-back queries it answered the first 11 then throttled,
and recovered within about a minute, so it belongs BEHIND an unthrottled rung
in the race (Bing went 50/50 on the same burst) where it only sees the
occasional rescue query, not the firehose."""
import html
import re
from typing import List
from typeguard import typechecked
from backend.apps.agents.tools.browser_http import browser_request
from backend.apps.agents.tools.search.engine_answer import EngineAnswer
from backend.apps.agents.tools.search.strip_tags import strip_tags
P_SEARCH_URL = "https://search.brave.com/search"
P_TIMEOUT = 12.0
P_BLOCK_SPLIT_RE = re.compile(r'data-type="web"')
P_HREF_RE = re.compile(r'<a href="(https?://[^"]+)"')
P_TITLE_RE = re.compile(r'class="title[^"]*"[^>]*\btitle="([^"]*)"')
P_DESC_RE = re.compile(r'<div class="content[^"]*"[^>]*>(.*?)</div>', flags=re.DOTALL)
@typechecked
def parse_brave_results(body: str, num_results: int) -> str:
"""Format Brave's organic rows; each split chunk starts with one result's own markup."""
chunks = P_BLOCK_SPLIT_RE.split(body)[1:]
entries: List[str] = []
for chunk in chunks:
if len(entries) >= num_results:
break
href = P_HREF_RE.search(chunk)
title = P_TITLE_RE.search(chunk)
if not href or not title:
continue
title_text = html.unescape(title.group(1)).strip()
if not title_text:
continue
entry = f"[{len(entries) + 1}] {title_text}\n {html.unescape(href.group(1))}"
desc = P_DESC_RE.search(chunk)
if desc:
snippet = strip_tags(desc.group(1))
if snippet:
entry += f"\n {snippet}"
entries.append(entry)
return "\n\n".join(entries)
@typechecked
async def search_brave(query: str, num_results: int) -> EngineAnswer:
"""Brave's answer: results, or a refusal (it fuzzy-matches, so empty means blocked or drifted)."""
reply = await browser_request(P_SEARCH_URL, params={"q": query}, timeout=P_TIMEOUT)
if reply.status != 200:
return EngineAnswer(refused=True)
results = parse_brave_results(reply.text, num_results)
if results:
return EngineAnswer(results=results)
return EngineAnswer(refused=True)
@@ -11,21 +11,20 @@ returns the real result page. Parsing is anchored on `result-link` /
`gl-title-link` and the `description` paragraph, never on the emotion CSS
hashes in the same class attributes, which change build to build.
Answers a `StartpageAnswer` rather than a string because "closed" and
"genuinely no hits" look identical on the wire (both are a 200 with no result
anchors) and the caller has to act on them differently: a refusal is a failed
tier that should count against the engine, an empty result set is the honest
answer to a nonsense query. Startpage names the second case itself, in an
Answers an `EngineAnswer` rather than a string because "closed" and
"genuinely no hits" look identical on the wire and the caller has to act on
them differently. Startpage names the honest-empty case itself, in an
"Uh-oh, there are no results for this search" page."""
import html
import re
from typing import List
from pydantic import BaseModel, ConfigDict
from typeguard import typechecked
from backend.apps.agents.tools.browser_http import browser_request
from backend.apps.agents.tools.search.engine_answer import EngineAnswer
from backend.apps.agents.tools.search.strip_tags import strip_tags
P_SEARCH_URL = "https://www.startpage.com/sp/search"
P_TIMEOUT = 12.0
@@ -38,25 +37,9 @@ P_TITLE_RE = re.compile(r"<h2[^>]*>(.*?)</h2>", flags=re.DOTALL)
P_DESC_RE = re.compile(
r'<p[^>]*class="[^"]*\bdescription\b[^"]*"[^>]*>(.*?)</p>', flags=re.DOTALL,
)
# Startpage inlines a <style> block inside each result anchor, so tag-stripping alone would emit CSS as the title.
P_NOISE_RE = re.compile(r"<(script|style)\b[^>]*>.*?</\1>", flags=re.DOTALL | re.IGNORECASE)
P_TAG_RE = re.compile(r"<[^>]+>")
P_NO_RESULTS_MARKER = "there are no results for this search"
class StartpageAnswer(BaseModel):
model_config = ConfigDict(validate_assignment=True)
results: str = ""
# Challenge, error, or markup we no longer recognise: Startpage did not answer the question.
refused: bool = False
@typechecked
def p_strip(raw: str) -> str:
return re.sub(r"\s+", " ", html.unescape(P_TAG_RE.sub("", P_NOISE_RE.sub("", raw)))).strip()
@typechecked
def parse_startpage_results(body: str, num_results: int) -> str:
"""Format Startpage's result rows; a snippet is only paired when it sits INSIDE its own result block."""
@@ -67,14 +50,14 @@ def parse_startpage_results(body: str, num_results: int) -> str:
if not href:
continue
title_match = P_TITLE_RE.search(match.group(2))
title = p_strip(title_match.group(1)) if title_match else p_strip(match.group(2))
title = strip_tags(title_match.group(1)) if title_match else strip_tags(match.group(2))
if not title:
continue
entry = f"[{len(entries) + 1}] {title}\n {html.unescape(href.group(1))}"
block_end = anchors[i + 1].start() if i + 1 < len(anchors) else len(body)
desc = P_DESC_RE.search(body, match.end(), block_end)
if desc:
snippet = p_strip(desc.group(1))
snippet = strip_tags(desc.group(1))
if snippet:
entry += f"\n {snippet}"
entries.append(entry)
@@ -82,20 +65,20 @@ def parse_startpage_results(body: str, num_results: int) -> str:
@typechecked
async def search_startpage(query: str, num_results: int) -> StartpageAnswer:
async def search_startpage(query: str, num_results: int) -> EngineAnswer:
"""Startpage's answer: results, an honest nothing, or a refusal."""
reply = await browser_request(
P_SEARCH_URL, method="POST", params={"query": query, "cat": "web"}, timeout=P_TIMEOUT,
)
if reply.status != 200:
return StartpageAnswer(refused=True)
return EngineAnswer(refused=True)
results = parse_startpage_results(reply.text, num_results)
if results:
return StartpageAnswer(results=results)
return EngineAnswer(results=results)
# Since ~2026-08 the POST gets the same ~10KB proof-of-work interstitial as GET (measured: 10,349 bytes, zero result anchors, 'challenge' markers); name it a refusal so the breaker benches the engine instead of treating the wall as a mystery empty page.
if "challenge" in reply.text and len(reply.text) < 40_000:
return StartpageAnswer(refused=True)
return EngineAnswer(refused=True)
# Measured once in 14 tight-loop requests: Startpage serves its own no-results page for a query that answered 10 results a second later, so an empty page is only trustworthy when it says so.
if P_NO_RESULTS_MARKER in reply.text:
return StartpageAnswer()
return StartpageAnswer(refused=True)
return EngineAnswer()
return EngineAnswer(refused=True)
@@ -0,0 +1,15 @@
"""Collapse a fragment of SERP markup to its visible one-line text."""
import html
import re
from typeguard import typechecked
# Some engines inline <style> blocks INSIDE result anchors, so tag-stripping alone would emit CSS as the title.
P_NOISE_RE = re.compile(r"<(script|style)\b[^>]*>.*?</\1>", flags=re.DOTALL | re.IGNORECASE)
P_TAG_RE = re.compile(r"<[^>]+>")
@typechecked
def strip_tags(raw: str) -> str:
return re.sub(r"\s+", " ", html.unescape(P_TAG_RE.sub(" ", P_NOISE_RE.sub("", raw)))).strip()
+23 -1
View File
@@ -138,7 +138,7 @@ async def search(body: SearchBody) -> Dict:
return {"query": body.query, "results": text, "backend": "ddg"}
async def try_startpage() -> Optional[Dict]:
# Second independent engine (Google's index), so DuckDuckGo's bot challenge is no longer a single point of failure for keyless users.
# Google's index, when its proof-of-work wall is down; benched by the breaker while it isn't.
from backend.apps.agents.tools.search.search_startpage import search_startpage
answer = await search_startpage(body.query, body.num_results)
# Raise rather than return None: a refusal is the engine failing, and the breaker must count it so a closed Startpage stops costing its budget too.
@@ -148,6 +148,26 @@ async def search(body: SearchBody) -> Dict:
return None
return {"query": body.query, "results": answer.results, "backend": "startpage"}
async def try_bing() -> Optional[Dict]:
# The index DuckDuckGo mostly serves, reachable directly even while DDG's challenge wall is up; went 50/50 on a zero-delay burst that tripped both DDG and Brave.
from backend.apps.agents.tools.search.search_bing import search_bing
answer = await search_bing(body.query, body.num_results)
if answer.refused:
raise RuntimeError("Bing answered with a challenge instead of results")
if not answer.results:
return None
return {"query": body.query, "results": answer.results, "backend": "bing"}
async def try_brave() -> Optional[Dict]:
# Brave's own crawler: coverage independent of both Bing-fed engines and Google, but it throttles bursts, so it sits behind Bing and only sees rescue traffic.
from backend.apps.agents.tools.search.search_brave import search_brave
answer = await search_brave(body.query, body.num_results)
if answer.refused:
raise RuntimeError("Brave answered with a challenge instead of results")
if not answer.results:
return None
return {"query": body.query, "results": answer.results, "backend": "brave"}
async def try_browser_search() -> Optional[Dict]:
# Packaged-app tier: a real Chromium's fingerprint isn't subject to the headless-client challenge, and it can scrape Google/Bing directly. Skipped (None) when no Electron main bridge is connected.
res = await p_browser_bridge("browser_search", {"query": body.query, "num_results": body.num_results})
@@ -196,6 +216,8 @@ async def search(body: SearchBody) -> Dict:
async def try_keyless_engines() -> Optional[Dict]:
outcome = await race_keyless(
[KeylessEngine(name="ddg", run=try_keyless),
KeylessEngine(name="bing", run=try_bing),
KeylessEngine(name="brave", run=try_brave),
KeylessEngine(name="startpage", run=try_startpage)],
KEYLESS_TIER_SECONDS,
)
+97
View File
@@ -0,0 +1,97 @@
"""Bing rung: Microsoft's index direct, with the click-tracker redirect decoded.
The fixture mirrors the real markup captured live 2026-08-10: b_algo list
items, the title inside an h2 anchor whose href is a bing.com/ck/a redirect
carrying the real URL base64url-encoded in `u=a1...`, the snippet in the
b_caption paragraph, and ads living in b_ad blocks."""
import base64
import pytest
import backend.apps.agents.tools.search.search_bing as BI
from backend.apps.agents.tools.browser_http import HttpReply
from backend.apps.agents.tools.search.search_bing import (
parse_bing_results,
search_bing,
)
def p_b64u(url: str) -> str:
return base64.urlsafe_b64encode(url.encode()).decode().rstrip("=")
P_BODY = (
'<ol id="b_results">'
'<li class="b_ad"><h2><a href="https://www.bing.com/aclick?ad=1">Sponsored Thing</a></h2></li>'
'<li class="b_algo" data-id iid=SERP.1><h2><a target="_blank" '
f'href="https://www.bing.com/ck/a?!&amp;&amp;p=xx&amp;u=a1{p_b64u("https://docs.python-requests.org/en/latest/")}&amp;ntb=1" '
'h="ID=SERP,1.1">Requests: <strong>HTTP for Humans</strong></a></h2>'
'<div class="b_caption hasdl"><p class="b_lineclamp2">Requests is an elegant and simple '
"<strong>HTTP</strong> library for Python.</p></div></li>"
'<li class="b_algo"><h2><a href="https://pypi.org/project/requests/">requests · PyPI</a></h2></li>'
"</ol>"
)
def p_reply(status: int, text: str) -> HttpReply:
return HttpReply(status=status, text=text, content=text.encode(),
content_type="text/html", url="https://www.bing.com/search")
def p_answer(monkeypatch, reply: HttpReply):
async def p_req(url, **kw):
return reply
monkeypatch.setattr(BI, "browser_request", p_req)
def test_redirect_href_is_decoded_to_the_real_url():
out = parse_bing_results(P_BODY, 5)
assert "https://docs.python-requests.org/en/latest/" in out
assert "bing.com/ck/a" not in out
def test_parses_title_snippet_and_direct_hrefs():
out = parse_bing_results(P_BODY, 5)
assert "[1] Requests: HTTP for Humans" in out
assert "Requests is an elegant and simple HTTP library for Python." in out
assert "[2] requests · PyPI" in out
assert "https://pypi.org/project/requests/" in out
def test_ads_are_skipped_by_block_class():
out = parse_bing_results(P_BODY, 5)
assert "Sponsored Thing" not in out
def test_respects_num_results():
out = parse_bing_results(P_BODY, 1)
assert "[1]" in out and "[2]" not in out
@pytest.mark.asyncio
async def test_http_error_reads_as_refused(monkeypatch):
p_answer(monkeypatch, p_reply(403, "nope"))
assert (await search_bing("q", 5)).refused
@pytest.mark.asyncio
async def test_bings_own_no_results_page_is_an_answer_not_a_refusal(monkeypatch):
p_answer(monkeypatch, p_reply(200, "<html><body>There are no results for <b>zxqv</b>.</body></html>"))
answer = await search_bing("zxqv", 5)
assert not answer.refused
assert answer.results == ""
@pytest.mark.asyncio
async def test_unrecognised_markup_is_a_refusal_not_zero_hits(monkeypatch):
p_answer(monkeypatch, p_reply(200, "<html><body><div>redesigned</div></body></html>"))
assert (await search_bing("python", 5)).refused
@pytest.mark.asyncio
async def test_real_markup_answers_results(monkeypatch):
p_answer(monkeypatch, p_reply(200, P_BODY))
answer = await search_bing("requests", 5)
assert not answer.refused
assert "docs.python-requests.org" in answer.results
+85
View File
@@ -0,0 +1,85 @@
"""Brave rung: independent-index engine behind DuckDuckGo and Bing.
The fixture is the real markup shape captured live 2026-08-10: server-rendered
Svelte blocks keyed on data-type="web", the clean title in the title div's
`title` attribute, the real target URL on the first anchor (no redirect
wrapper), and ads carrying a different data-type."""
import pytest
import backend.apps.agents.tools.search.search_brave as BR
from backend.apps.agents.tools.browser_http import HttpReply
from backend.apps.agents.tools.search.search_brave import (
parse_brave_results,
search_brave,
)
P_BODY = (
'<div id="results">'
'<div class="snippet svelte-x" data-pos="0" data-type="ad">'
'<a href="https://ad.example/click"><div class="title x" title="Buy Things Now">Buy Things Now</div></a></div>'
'<div class="snippet svelte-x" data-pos="1" data-type="web" data-keynav="true">'
'<div class="result-wrapper"><a href="https://pypi.org/project/requests/" target="_self" class="l1">'
'<div class="site-name-content"><cite class="snippet-url">pypi.org</cite></div>'
'<div class="title search-snippet-title line-clamp-1 svelte-y" title="requests &#183; PyPI">requests · PyPI</div></a>'
'<div class="generic-snippet"><div class="content desktop-default-regular t-primary">'
'<span class="t-secondary">May 14, 2026 -</span> Python HTTP for Humans. Requests is '
"<strong>a simple, yet elegant, HTTP library</strong>.</div></div></div></div>"
'<div class="snippet svelte-x" data-pos="2" data-type="web">'
'<a href="https://en.wikipedia.org/wiki/Requests_(software)">'
'<div class="title x" title="Requests (software) - Wikipedia">clipped visible text</div></a>'
'<div class="generic-snippet"><div class="content x">Requests is an HTTP client library.</div></div></div>'
"</div>"
)
def p_reply(status: int, text: str) -> HttpReply:
return HttpReply(status=status, text=text, content=text.encode(),
content_type="text/html", url="https://search.brave.com/search")
def p_answer(monkeypatch, reply: HttpReply):
async def p_req(url, **kw):
return reply
monkeypatch.setattr(BR, "browser_request", p_req)
def test_parses_title_from_attribute_url_and_snippet():
out = parse_brave_results(P_BODY, 5)
assert "[1] requests · PyPI" in out
assert "https://pypi.org/project/requests/" in out
assert "Python HTTP for Humans" in out
assert "[2] Requests (software) - Wikipedia" in out
assert "https://en.wikipedia.org/wiki/Requests_(software)" in out
def test_ads_are_skipped_by_data_type():
out = parse_brave_results(P_BODY, 5)
assert "Buy Things Now" not in out
assert "ad.example" not in out
def test_respects_num_results():
out = parse_brave_results(P_BODY, 1)
assert "[1]" in out and "[2]" not in out
@pytest.mark.asyncio
async def test_http_error_reads_as_refused(monkeypatch):
p_answer(monkeypatch, p_reply(429, "slow down"))
assert (await search_brave("q", 5)).refused
@pytest.mark.asyncio
async def test_unrecognised_markup_is_a_refusal_not_zero_hits(monkeypatch):
"""Brave fuzzy-matches even gibberish, so an empty parse is never an honest no-hits."""
p_answer(monkeypatch, p_reply(200, "<html><body><div>redesigned</div></body></html>"))
assert (await search_brave("python", 5)).refused
@pytest.mark.asyncio
async def test_real_markup_answers_results(monkeypatch):
p_answer(monkeypatch, p_reply(200, P_BODY))
answer = await search_brave("requests", 5)
assert not answer.refused
assert "pypi.org" in answer.results
+31
View File
@@ -18,6 +18,9 @@ import backend.apps.web.web as W
from backend.apps.web.web import search, SearchBody
from backend.tests.web_cascade_fixtures import ( # noqa: F401
allow_urls,
bing_refuses,
bing_returns,
brave_returns,
fresh_breaker,
patch_browser_bridge,
ddg_returns,
@@ -190,6 +193,34 @@ async def test_browser_search_skipped_when_no_bridge(monkeypatch):
assert res["backend"] == "openai_native"
@pytest.mark.asyncio
async def test_bing_rescues_a_ddg_challenge_before_any_paid_backend(monkeypatch):
"""Bing is the unthrottled second rung: it went 50/50 on the burst that tripped DDG."""
ddg_throttled(monkeypatch)
bing_returns(monkeypatch, "[1] Rescued\n https://bing-hit.example")
async def p_boom(*a, **k):
raise AssertionError("a paid backend must not run while a free engine still answers")
monkeypatch.setattr(W, "gemini_grounded_call", p_boom)
res = await search(SearchBody(query="x"))
assert res["backend"] == "bing"
assert "bing-hit.example" in res["results"]
assert any("ddg" in e for e in res["cascade_errors"])
@pytest.mark.asyncio
async def test_brave_rescues_when_ddg_and_bing_both_refuse(monkeypatch):
ddg_throttled(monkeypatch)
bing_refuses(monkeypatch)
brave_returns(monkeypatch, "[1] Independent index\n https://brave-hit.example")
res = await search(SearchBody(query="x"))
assert res["backend"] == "brave"
assert "brave-hit.example" in res["results"]
assert any("Bing" in e for e in res["cascade_errors"])
@pytest.mark.asyncio
async def test_a_refusing_startpage_counts_against_it(monkeypatch):
"""A challenge and a genuinely empty web look identical on the wire, and treating a
+34 -6
View File
@@ -3,8 +3,10 @@
import pytest
import backend.apps.agents.tools.fetch.wayback as WB
import backend.apps.agents.tools.search.search_bing as SBI
import backend.apps.agents.tools.search.search_brave as SBR
import backend.apps.agents.tools.search.search_startpage as SP
from backend.apps.agents.tools.search.search_startpage import StartpageAnswer
from backend.apps.agents.tools.search.engine_answer import EngineAnswer
import backend.apps.web.web as W
from backend.apps.agents.tools.web import DDGRateLimited, WebSearchTool
import backend.apps.agents.tools.ssrf_guard as p_ssrf
@@ -35,9 +37,11 @@ def no_network(monkeypatch):
monkeypatch.setattr(W, "gemini_grounded_via_9router", p_empty)
monkeypatch.setattr(W, "openai_websearch_via_9router", p_empty)
async def p_startpage_closed(query, num):
return StartpageAnswer()
monkeypatch.setattr(SP, "search_startpage", p_startpage_closed)
async def p_engine_closed(query, num):
return EngineAnswer()
monkeypatch.setattr(SP, "search_startpage", p_engine_closed)
monkeypatch.setattr(SBR, "search_brave", p_engine_closed)
monkeypatch.setattr(SBI, "search_bing", p_engine_closed)
async def p_no_snapshot(url):
return None
@@ -65,16 +69,40 @@ def ddg_throttled(monkeypatch):
def startpage_returns(monkeypatch, text):
async def p_f(query, num):
return StartpageAnswer(results=text)
return EngineAnswer(results=text)
monkeypatch.setattr(SP, "search_startpage", p_f)
def startpage_refuses(monkeypatch):
async def p_f(query, num):
return StartpageAnswer(refused=True)
return EngineAnswer(refused=True)
monkeypatch.setattr(SP, "search_startpage", p_f)
def bing_returns(monkeypatch, text):
async def p_f(query, num):
return EngineAnswer(results=text)
monkeypatch.setattr(SBI, "search_bing", p_f)
def bing_refuses(monkeypatch):
async def p_f(query, num):
return EngineAnswer(refused=True)
monkeypatch.setattr(SBI, "search_bing", p_f)
def brave_returns(monkeypatch, text):
async def p_f(query, num):
return EngineAnswer(results=text)
monkeypatch.setattr(SBR, "search_brave", p_f)
def brave_refuses(monkeypatch):
async def p_f(query, num):
return EngineAnswer(refused=True)
monkeypatch.setattr(SBR, "search_brave", p_f)
def patch_browser_bridge(monkeypatch, result):
"""Patch the offscreen-browser bridge; result=None simulates 'no Electron main bridge connected'."""
async def p_f(action, params):