host
3a56f83114
- 默认 auto 只走免费链 searxng -> duckduckgo；即使环境里有 付费 key 也不调用 Perplexity/Exa/Tavily/Brave，SearXNG 失败也不回落付费，默认使用零付费花费。 - 付费后端代码/测试/文档全部保留；显式 pin（tavily/exa/ perplexity/brave）仍照常调用，是有意覆盖不是默认。 - 解冻开关 WEBSEARCH_ENABLE_PAID，仅精确值 "1" 才恢复旧链 （perplexity/exa/tavily/brave/searxng/duckduckgo），默认关； 开关经 worker search_env 透传进 MCP config env。 - HttpClient 超时/DNS 失败改报显式错误（timeout after Ns / URL error），绝不把超时写成“无结果”。 - SearXNG 默认参数核实：代码无 SEARXNG_URL 硬编码默认值； SEARXNG_ENGINES 默认 bing,360search 已在基线；WEBSEARCH_TIMEOUT 默认 20 秒。主机 env 实测仅 SEARXNG_URL=http://searx.diante.me 一行，故默认 pin 生效；文档 https 漂移改回 http（主机 nginx 仅听 80、无 443 vhost，https 在 TLS 层即失败）。 - 旧测试更新原因：test_websearch_mcp 原断言默认 auto 为 perplexity 优先，与新冻结语义直接冲突，故改为断言冻结后 的 searxng -> duckduckgo 回退 + 不泄漏 key；旧 perplexity -> exa 回退断言移至解冻开关测试（WEBSEARCH_ENABLE_PAID=1 下）。 未删断言、未加 skip、未放宽门禁。 - 文档另删 WEBSEARCH_PROVIDERS 说法：代码从未实现该变量。 - 未推、未建 PR、未合并、未部署、未改主机 env/nginx/filter/ 证书、未做付费后端 live 验证、未创建生产任务。
275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""Web search MCP v1 tests: chain fallback, fetch, MCP framing, config.
|
|
|
|
No network, no keys: every HTTP call goes through a FakeHttp, and
|
|
the fake key values are deliberately distinctive so a leak into an
|
|
error string or a tool result would be caught by assertion, not by
|
|
eyeballing.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.websearch import (
|
|
WebSearchError, ensure_codex_home_websearch, fetch_url, search)
|
|
from cloud.websearch_mcp import handle_request
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.websearch import (
|
|
WebSearchError, ensure_codex_home_websearch, fetch_url, search)
|
|
from cloud.websearch_mcp import handle_request
|
|
|
|
_FAKE_KEYS = {
|
|
"PERPLEXITY_API_KEY": "pplx-SECRET-111",
|
|
"EXA_API_KEY": "exa-SECRET-222",
|
|
"TAVILY_API_KEY": "tvly-SECRET-333",
|
|
"BRAVE_API_KEY": "brave-SECRET-444",
|
|
}
|
|
|
|
|
|
class FakeHttp:
|
|
"""post_json/get keyed by URL substring; failures raise like HTTP."""
|
|
|
|
def __init__(self, routes):
|
|
# routes: list of (url_substring, body_dict_or_text, status)
|
|
self.routes = routes
|
|
self.seen = []
|
|
|
|
def post_json(self, url, headers, payload):
|
|
self.seen.append(("POST", url, payload))
|
|
return self._route(url)
|
|
|
|
def get(self, url, headers=None):
|
|
self.seen.append(("GET", url, headers))
|
|
return self._route(url)
|
|
|
|
def _route(self, url):
|
|
for needle, body, status in self.routes:
|
|
if needle in url:
|
|
if status != 200:
|
|
# Mimic HttpClient._read: raise with status only.
|
|
raise WebSearchError("HTTP %s" % status)
|
|
if isinstance(body, dict):
|
|
return 200, json.dumps(body).encode(), "application/json"
|
|
return 200, body.encode(), "text/html"
|
|
raise WebSearchError("HTTP 404")
|
|
|
|
|
|
def _exa_body():
|
|
return {"results": [
|
|
{"title": "Exa Hit", "url": "https://example.com/exa",
|
|
"text": "exa snippet", "score": 0.9,
|
|
"highlights": ["exa highlight"]}]}
|
|
|
|
|
|
class ChainTestCase(unittest.TestCase):
|
|
def test_default_auto_frozen_falls_back_searxng_to_duckduckgo_and_key_not_leaked(self):
|
|
# Paid-freeze update of the old perplexity -> exa fallback
|
|
# test: default auto no longer contains any paid backend,
|
|
# even when all four paid keys are set, so the fallback
|
|
# that still exists by default is searxng -> duckduckgo
|
|
# and no paid URL is ever touched. The old perplexity ->
|
|
# exa fallback is preserved, but only under
|
|
# WEBSEARCH_ENABLE_PAID=1; see test_websearch_paid_freeze.py.
|
|
html = (
|
|
'<a class="result__a" href="//duckduckgo.com/l/?uddg='
|
|
'https%3A%2F%2Fexample.com%2Fddg">DDG Title</a>'
|
|
'<a class="result__snippet">ddg snippet</a>')
|
|
http = FakeHttp([
|
|
("searx", {}, 500),
|
|
("duckduckgo", html, 200),
|
|
])
|
|
env = dict(_FAKE_KEYS)
|
|
env["SEARXNG_URL"] = "http://searx.test"
|
|
out = search("test query", provider="auto",
|
|
env=env, http=http)
|
|
self.assertEqual(out["provider"], "duckduckgo")
|
|
self.assertEqual(out["results"][0]["url"],
|
|
"https://example.com/ddg")
|
|
self.assertEqual(out["attempted"][0]["provider"], "searxng")
|
|
self.assertFalse(out["attempted"][0]["ok"])
|
|
self.assertEqual(
|
|
[a["provider"] for a in out["attempted"]],
|
|
["searxng", "duckduckgo"])
|
|
for _method, url, _payload in http.seen:
|
|
for paid in ("perplexity", "exa.ai", "tavily",
|
|
"search.brave.com"):
|
|
self.assertNotIn(paid, url)
|
|
blob = json.dumps(out)
|
|
for secret in _FAKE_KEYS.values():
|
|
self.assertNotIn(secret, blob)
|
|
|
|
def test_all_fail_error_lists_attempts_without_keys(self):
|
|
http = FakeHttp([("perplexity", {}, 401),
|
|
("api.exa.ai", {}, 429),
|
|
("api.tavily.com", {}, 500),
|
|
("api.search.brave.com", {}, 403)])
|
|
with self.assertRaises(WebSearchError) as ctx:
|
|
search("q", provider="perplexity,exa,tavily,brave",
|
|
env=dict(_FAKE_KEYS), http=http)
|
|
for secret in _FAKE_KEYS.values():
|
|
self.assertNotIn(secret, str(ctx.exception))
|
|
self.assertEqual(
|
|
[a["provider"] for a in ctx.exception.attempted],
|
|
["perplexity", "exa", "tavily", "brave"])
|
|
|
|
def test_pinned_provider_does_not_fall_back(self):
|
|
http = FakeHttp([("perplexity", {}, 500),
|
|
("api.exa.ai", _exa_body(), 200)])
|
|
with self.assertRaises(WebSearchError):
|
|
search("q", provider="perplexity",
|
|
env=dict(_FAKE_KEYS), http=http)
|
|
# Exa was never even attempted.
|
|
self.assertFalse(any("exa" in url for _m, url, _p in http.seen))
|
|
|
|
def test_brave_mapping_and_tavily_mapping(self):
|
|
http = FakeHttp([
|
|
("api.search.brave.com",
|
|
{"web": {"results": [
|
|
{"title": "Brave Hit", "url": "https://example.com/b",
|
|
"description": "desc",
|
|
"extra_snippets": ["extra"]}]}}, 200),
|
|
])
|
|
out = search("q", provider="brave",
|
|
env=dict(_FAKE_KEYS), http=http)
|
|
self.assertIn("desc", out["results"][0]["snippet"])
|
|
self.assertIn("extra", out["results"][0]["snippet"])
|
|
|
|
http2 = FakeHttp([
|
|
("api.tavily.com",
|
|
{"results": [{"title": "T", "url": "https://example.com/t",
|
|
"content": "tavily content", "score": 0.5}]},
|
|
200),
|
|
])
|
|
out2 = search("q", provider="tavily",
|
|
env=dict(_FAKE_KEYS), http=http2)
|
|
self.assertEqual(out2["results"][0]["snippet"], "tavily content")
|
|
# Tavily request must not ask for a synthesised answer: the
|
|
# tool returns results, the model synthesises.
|
|
self.assertFalse(http2.seen[0][2]["include_answer"])
|
|
|
|
def test_perplexity_answer_and_citations(self):
|
|
http = FakeHttp([
|
|
("api.perplexity.ai",
|
|
{"choices": [{"message": {"content": "synthesised answer"}}],
|
|
"search_results": [
|
|
{"title": "P", "url": "https://example.com/p",
|
|
"snippet": "p snippet"}]}, 200),
|
|
])
|
|
out = search("q", provider="perplexity",
|
|
env=dict(_FAKE_KEYS), http=http)
|
|
self.assertEqual(out["answer"], "synthesised answer")
|
|
self.assertEqual(out["results"][0]["url"], "https://example.com/p")
|
|
|
|
def test_empty_results_continue_chain(self):
|
|
http = FakeHttp([
|
|
("api.exa.ai", {"results": []}, 200),
|
|
("api.tavily.com",
|
|
{"results": [{"title": "T", "url": "https://example.com/t",
|
|
"content": "c"}]}, 200),
|
|
])
|
|
out = search("q", provider="exa,tavily",
|
|
env=dict(_FAKE_KEYS), http=http)
|
|
self.assertEqual(out["provider"], "tavily")
|
|
|
|
def test_duckduckgo_html_parser(self):
|
|
html = (
|
|
'<a class="result__a" href="//duckduckgo.com/l/?uddg='
|
|
'https%3A%2F%2Fexample.com%2Fddg">DDG Title</a>'
|
|
'<a class="result__snippet">ddg snippet</a>')
|
|
http = FakeHttp([("duckduckgo", html, 200)])
|
|
out = search("q", provider="duckduckgo", env={}, http=http)
|
|
self.assertEqual(out["results"][0]["url"], "https://example.com/ddg")
|
|
self.assertEqual(out["results"][0]["title"], "DDG Title")
|
|
|
|
|
|
class FetchTestCase(unittest.TestCase):
|
|
def test_strips_script_and_truncates_head_tail(self):
|
|
body = ("<html><head><title>My Page</title></head><body>"
|
|
"<script>var secret = 1;</script>"
|
|
"<p>" + "A" * 10000 + "</p>"
|
|
"<p>" + "B" * 10000 + "</p></body></html>")
|
|
http = FakeHttp([("example.com", body, 200)])
|
|
out = fetch_url("https://example.com/page", char_limit=2000,
|
|
http=http)
|
|
self.assertEqual(out["title"], "My Page")
|
|
self.assertNotIn("secret", out["text"])
|
|
self.assertTrue(out["truncated"])
|
|
self.assertIn("[TRUNCATED]", out["text"])
|
|
self.assertIn("B" * 10, out["text"])
|
|
|
|
def test_rejects_non_http_and_localhost(self):
|
|
http = FakeHttp([])
|
|
with self.assertRaises(WebSearchError):
|
|
fetch_url("file:///etc/passwd", http=http)
|
|
with self.assertRaises(WebSearchError):
|
|
fetch_url("http://127.0.0.1/admin", http=http)
|
|
with self.assertRaises(WebSearchError):
|
|
fetch_url("http://10.0.0.5/admin", http=http)
|
|
|
|
|
|
class McpFramingTestCase(unittest.TestCase):
|
|
def test_initialize_tools_list_and_call(self):
|
|
resp = handle_request(
|
|
{"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
|
"params": {"protocolVersion": "2024-11-05"}})
|
|
self.assertEqual(resp["result"]["serverInfo"]["name"], "websearch")
|
|
resp = handle_request(
|
|
{"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
|
|
names = [t["name"] for t in resp["result"]["tools"]]
|
|
self.assertEqual(names, ["web_search", "web_fetch"])
|
|
|
|
def fake_search(query, limit=5, provider="auto",
|
|
scope="general"):
|
|
return {"provider": "fake", "results": [], "answer": None,
|
|
"attempted": [], "scope": scope}
|
|
|
|
resp = handle_request(
|
|
{"jsonrpc": "2.0", "id": 3, "method": "tools/call",
|
|
"params": {"name": "web_search",
|
|
"arguments": {"query": "hi"}}},
|
|
search_fn=fake_search)
|
|
self.assertFalse(resp["result"]["isError"])
|
|
self.assertIn("fake", resp["result"]["content"][0]["text"])
|
|
|
|
def test_notification_gets_no_response_and_unknown_tool_errors(self):
|
|
self.assertIsNone(handle_request(
|
|
{"jsonrpc": "2.0", "method": "notifications/initialized"}))
|
|
resp = handle_request(
|
|
{"jsonrpc": "2.0", "id": 9, "method": "tools/call",
|
|
"params": {"name": "nope", "arguments": {}}})
|
|
self.assertEqual(resp["error"]["code"], -32602)
|
|
|
|
|
|
class CodexHomeTestCase(unittest.TestCase):
|
|
def test_merge_is_idempotent_and_preserves_existing(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
home = os.path.join(tmp, "codex-home")
|
|
os.makedirs(home)
|
|
existing = ("# tenant config\n"
|
|
"[model_providers.relay]\n"
|
|
"api_key = \"keep-me\"\n")
|
|
with open(os.path.join(home, "config.toml"), "w") as fh:
|
|
fh.write(existing)
|
|
out = ensure_codex_home_websearch(
|
|
home, command_args=["python3", "/tmp/ws/websearch_mcp.py"])
|
|
self.assertTrue(out["config_changed"])
|
|
with open(out["config_path"]) as fh:
|
|
merged = fh.read()
|
|
self.assertIn("keep-me", merged)
|
|
self.assertIn("[mcp_servers.websearch]", merged)
|
|
out2 = ensure_codex_home_websearch(
|
|
home, command_args=["python3", "/tmp/ws/websearch_mcp.py"])
|
|
self.assertFalse(out2["config_changed"])
|
|
with open(out2["config_path"]) as fh:
|
|
self.assertEqual(fh.read().count("[mcp_servers.websearch]"),
|
|
1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|