thzxx
f983a9377e
核心新增（面向 AI Agent 程序化使用）： - 结构化错误码体系：E_CONFIG/E_AUTH/E_NETWORK/E_RATE_LIMIT/E_PARSE/E_EMPTY/E_INPUT/E_INTERNAL classify_error() 自动分类异常，JSON 错误输出含 error_code 字段 - JSON Lines 流式输出 (--stream)：每条结果独立一行，AI 可增量处理 - 进度事件 (--progress)：JSON Lines 事件流到 stderr（start/cache_hit/fetch_ok/done 等） 测试补全（+154 例，覆盖全部高风险盲区）： - HTML 回退搜索路径 (19) - --fetch 自动抓取 (21) - --verify 健康检查 (15) - 输出格式化 (15) - 实例解析链 (20) - 并行多实例搜索 (10) - CLI 入口与端到端 (17) - 错误码分类 (27) - 流式输出与进度事件 (10) 源码改进： - search.py: h3 内 a 标签 href 作为 url fallback，提升 SearXNG 主题兼容性 - common.py: 新增 classify_error/emit_progress/set_progress_enabled 文档同步：SKILL.md 新增 AI Agent Integration Guide 章节，README.md 更新参数与错误码表
215 lines
8.2 KiB
Python
215 lines
8.2 KiB
Python
"""Tests for the --verify instance health-check feature.
|
|
|
|
Covers: _probe_config_endpoint (reachable/disabled/error), verify_instances
|
|
(reachability, JSON support, POST probe, latency, auth status, error
|
|
classification), _print_verify_report (JSON and table output).
|
|
"""
|
|
import json
|
|
import urllib.error
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from search import _probe_config_endpoint, verify_instances, _print_verify_report
|
|
|
|
|
|
def _mock_urlopen(data: bytes, content_type="application/json"):
|
|
"""Build a MagicMock that quacks like an urlopen context manager."""
|
|
resp = MagicMock()
|
|
resp.read.return_value = data
|
|
resp.headers = {"Content-Type": content_type}
|
|
resp.__enter__.return_value = resp
|
|
resp.__exit__.return_value = None
|
|
return resp
|
|
|
|
|
|
# ----- _probe_config_endpoint -----
|
|
|
|
def test_probe_config_reachable():
|
|
"""A working /config endpoint returns engines and categories."""
|
|
payload = json.dumps({
|
|
"engines": [{"name": "google"}, {"name": "bing"}, {"name": "wikipedia"}],
|
|
"categories": {"general": [], "news": [], "images": []},
|
|
})
|
|
with patch("urllib.request.urlopen",
|
|
return_value=_mock_urlopen(payload.encode())):
|
|
r = _probe_config_endpoint("https://s.example.com", timeout=10)
|
|
assert r["reachable"] is True
|
|
assert r["engines"] == ["google", "bing", "wikipedia"]
|
|
assert set(r["categories"]) == {"general", "news", "images"}
|
|
assert r["error"] is None
|
|
|
|
|
|
def test_probe_config_categories_as_list():
|
|
"""Some instances return categories as a list instead of a dict."""
|
|
payload = json.dumps({
|
|
"engines": [{"name": "google"}],
|
|
"categories": ["general", "news"],
|
|
})
|
|
with patch("urllib.request.urlopen",
|
|
return_value=_mock_urlopen(payload.encode())):
|
|
r = _probe_config_endpoint("https://s.example.com", timeout=10)
|
|
assert r["reachable"] is True
|
|
assert r["categories"] == ["general", "news"]
|
|
|
|
|
|
def test_probe_config_http_error():
|
|
"""/config returns 403 → reachable=False with HTTP code in error."""
|
|
err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None)
|
|
with patch("urllib.request.urlopen", side_effect=err):
|
|
r = _probe_config_endpoint("https://s.example.com", timeout=10)
|
|
assert r["reachable"] is False
|
|
assert "403" in r["error"]
|
|
assert r["engines"] == []
|
|
|
|
|
|
def test_probe_config_connection_error():
|
|
"""Connection error → reachable=False with error message."""
|
|
err = urllib.error.URLError("connection refused")
|
|
with patch("urllib.request.urlopen", side_effect=err):
|
|
r = _probe_config_endpoint("https://s.example.com", timeout=10)
|
|
assert r["reachable"] is False
|
|
assert r["engines"] == []
|
|
assert "connection refused" in r["error"]
|
|
|
|
|
|
# ----- verify_instances -----
|
|
|
|
def _mock_json_search_response(results=None):
|
|
"""Mock a successful JSON search response."""
|
|
payload = json.dumps({"results": results or [{"title": "t", "url": "https://x.com"}]})
|
|
return _mock_urlopen(payload.encode(), content_type="application/json")
|
|
|
|
|
|
def _mock_html_search_response():
|
|
"""Mock an HTML search response (JSON disabled)."""
|
|
return _mock_urlopen(b"<html>not json</html>", content_type="text/html")
|
|
|
|
|
|
def test_verify_reachable_json_supported():
|
|
"""Instance returns JSON → reachable + json_supported + post probe."""
|
|
with patch("urllib.request.urlopen",
|
|
return_value=_mock_json_search_response()):
|
|
# /config also needs to respond
|
|
config_payload = json.dumps({"engines": [{"name": "google"}], "categories": {}})
|
|
config_resp = _mock_urlopen(config_payload.encode())
|
|
with patch("urllib.request.urlopen",
|
|
return_value=_mock_json_search_response()):
|
|
report = verify_instances(["https://s.example.com"])
|
|
r = report[0]
|
|
assert r["url"] == "https://s.example.com"
|
|
assert r["reachable"] is True
|
|
assert r["json_supported"] is True
|
|
|
|
|
|
def test_verify_html_only_instance():
|
|
"""Instance returns HTML (no JSON) → reachable but json_supported=False."""
|
|
with patch("urllib.request.urlopen",
|
|
return_value=_mock_html_search_response()):
|
|
report = verify_instances(["https://s.example.com"])
|
|
r = report[0]
|
|
assert r["reachable"] is True
|
|
assert r["json_supported"] is False
|
|
|
|
|
|
def test_verify_auth_rejected():
|
|
"""401/403 with auth → auth_status='rejected'."""
|
|
err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None)
|
|
auth_headers = {"Authorization": "Bearer token123"}
|
|
with patch("urllib.request.urlopen", side_effect=err):
|
|
report = verify_instances(["https://s.example.com"],
|
|
auth_headers=auth_headers)
|
|
r = report[0]
|
|
assert r["auth_status"] == "rejected"
|
|
|
|
|
|
def test_verify_no_auth_returns_na():
|
|
"""Without auth headers, auth_status is 'n/a'."""
|
|
err = urllib.error.URLError("timeout")
|
|
with patch("urllib.request.urlopen", side_effect=err):
|
|
report = verify_instances(["https://s.example.com"])
|
|
r = report[0]
|
|
assert r["auth_status"] == "n/a"
|
|
|
|
|
|
def test_verify_connection_error():
|
|
"""Connection error → reachable=False, latency=None (no response received)."""
|
|
err = urllib.error.URLError("connection refused")
|
|
with patch("urllib.request.urlopen", side_effect=err):
|
|
report = verify_instances(["https://s.example.com"])
|
|
r = report[0]
|
|
assert r["reachable"] is False
|
|
assert r["json_supported"] is False
|
|
assert r["latency"] is None # no response → no latency
|
|
assert "connection refused" in r["error"]
|
|
|
|
|
|
def test_verify_preserves_input_order():
|
|
"""Multiple instances: report preserves the original input order."""
|
|
urls = ["https://a.example.com", "https://b.example.com", "https://c.example.com"]
|
|
with patch("urllib.request.urlopen",
|
|
return_value=_mock_json_search_response()):
|
|
report = verify_instances(urls)
|
|
assert [r["url"] for r in report] == urls
|
|
|
|
|
|
def test_verify_has_latency():
|
|
"""Latency is a positive float for reachable instances."""
|
|
with patch("urllib.request.urlopen",
|
|
return_value=_mock_json_search_response()):
|
|
report = verify_instances(["https://s.example.com"])
|
|
assert report[0]["latency"] is not None
|
|
assert report[0]["latency"] >= 0
|
|
|
|
|
|
def test_verify_has_result_count():
|
|
"""Result count from the test query is recorded."""
|
|
with patch("urllib.request.urlopen",
|
|
return_value=_mock_json_search_response(
|
|
[{"title": "a"}, {"title": "b"}, {"title": "c"}])):
|
|
report = verify_instances(["https://s.example.com"])
|
|
assert report[0]["result_count"] == 3
|
|
|
|
|
|
# ----- _print_verify_report -----
|
|
|
|
def test_print_report_json(capsys):
|
|
"""JSON output is valid JSON with the full report."""
|
|
report = [
|
|
{"url": "https://a.com", "reachable": True, "json_supported": True,
|
|
"post_supported": True, "latency": 0.5, "result_count": 10,
|
|
"engines": ["google"], "auth_status": "n/a", "error": None,
|
|
"config_endpoint": True},
|
|
]
|
|
_print_verify_report(report, as_json=True)
|
|
out, _ = capsys.readouterr()
|
|
data = json.loads(out)
|
|
assert data[0]["url"] == "https://a.com"
|
|
assert data[0]["reachable"] is True
|
|
|
|
|
|
def test_print_report_table(capsys):
|
|
"""Table output includes the header and a summary line."""
|
|
report = [
|
|
{"url": "https://a.com", "reachable": True, "json_supported": True,
|
|
"post_supported": True, "latency": 0.5, "result_count": 10,
|
|
"engines": ["google", "bing"], "auth_status": "n/a", "error": None,
|
|
"config_endpoint": True},
|
|
{"url": "https://b.com", "reachable": False, "json_supported": False,
|
|
"post_supported": None, "latency": None, "result_count": None,
|
|
"engines": [], "auth_status": "n/a", "error": "timeout",
|
|
"config_endpoint": None},
|
|
]
|
|
_print_verify_report(report, as_json=False)
|
|
out, _ = capsys.readouterr()
|
|
assert "URL" in out
|
|
assert "REACH" in out
|
|
assert "https://a.com" in out
|
|
assert "https://b.com" in out
|
|
assert "1/2 instances reachable" in out
|
|
|
|
|
|
def test_print_report_empty(capsys):
|
|
"""Empty report produces a table with 0/0 summary."""
|
|
_print_verify_report([], as_json=False)
|
|
out, _ = capsys.readouterr()
|
|
assert "0/0 instances reachable" in out
|