thzxx
fb9b2af45f
稳定性修复: - 修复 cache.py SQLite 连接泄漏（contextlib.closing 包装） - 修复 fetch.py requests stream=True 连接泄漏（try/finally resp.close()） - RETRYABLE_STATUS 新增 403，激活 UA fallback 切换逻辑 - --cache-stats 移至实例解析前，无需实例即可查询 - classify_error 从错误消息提取 HTTP 状态码，正确分类 E_AUTH/E_RATE_LIMIT - --stream 与 --queries-file 互斥检查，违规报 E_INPUT - batch 退出码语义统一（0=有结果 / 1=全部错误 / 2=全部空结果） AI Agent 体验增强: - 错误码体系完善：E_CONFIG/E_AUTH/E_NETWORK/E_RATE_LIMIT/E_PARSE/E_EMPTY/E_INPUT/E_INTERNAL - recovery_hint 恢复提示字段，AI Agent 可程序化决策恢复策略 - stream 模式新增 error 事件类型（含 error_code + recovery_hint） - 进度事件扩展：instance_try/instance_ok/instance_fail - batch 模式统一 schema（status 字段区分 success/failed） - JSON 输出含 schema_version 字段确保版本兼容 测试与文档: - 测试覆盖：330 -> 352 - SKILL.md / README.md 同步更新
192 lines
5.6 KiB
Python
192 lines
5.6 KiB
Python
"""Tests for scripts/common.py — auth, charset, retry policy, proxy.
|
|
|
|
Covers: build_auth_headers (Bearer/Basic/precedence), resolve_auth_* priority
|
|
(CLI > file > env, file parsing, empty-file errors), detect_charset
|
|
(header/meta/fallback), is_retryable_error (429/5xx yes, 404/403 no,
|
|
connection errors yes), and apply_proxy (env vars + NO_PROXY preservation).
|
|
"""
|
|
import base64
|
|
import os
|
|
import urllib.error
|
|
|
|
import pytest
|
|
from common import (
|
|
RETRYABLE_STATUS,
|
|
apply_proxy,
|
|
build_auth_headers,
|
|
detect_charset,
|
|
is_retryable_error,
|
|
resolve_auth_basic,
|
|
resolve_auth_bearer,
|
|
)
|
|
|
|
|
|
# ----- build_auth_headers -----
|
|
|
|
def test_build_auth_bearer():
|
|
assert build_auth_headers(bearer_token="tok123") == \
|
|
{"Authorization": "Bearer tok123"}
|
|
|
|
|
|
def test_build_auth_basic():
|
|
h = build_auth_headers(basic_auth="user:pass")
|
|
assert h["Authorization"].startswith("Basic ")
|
|
decoded = base64.b64decode(h["Authorization"].split(" ", 1)[1]).decode()
|
|
assert decoded == "user:pass"
|
|
|
|
|
|
def test_build_auth_bearer_wins_over_basic():
|
|
"""Bearer takes precedence when both are provided."""
|
|
h = build_auth_headers(bearer_token="tok", basic_auth="u:p")
|
|
assert h == {"Authorization": "Bearer tok"}
|
|
|
|
|
|
def test_build_auth_none_returns_empty():
|
|
assert build_auth_headers() == {}
|
|
|
|
|
|
# ----- resolve_auth_basic -----
|
|
|
|
def test_resolve_auth_basic_cli_wins():
|
|
assert resolve_auth_basic(cli_value="cliu:clip") == "cliu:clip"
|
|
|
|
|
|
def test_resolve_auth_basic_file(tmp_path):
|
|
f = tmp_path / "auth.txt"
|
|
f.write_text("# header\nu:secret\n", encoding="utf-8")
|
|
assert resolve_auth_basic(file_path=str(f)) == "u:secret"
|
|
|
|
|
|
def test_resolve_auth_basic_env(monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_BASIC_AUTH", "envu:envp")
|
|
assert resolve_auth_basic() == "envu:envp"
|
|
|
|
|
|
def test_resolve_auth_basic_priority_cli_over_file(tmp_path):
|
|
f = tmp_path / "auth.txt"
|
|
f.write_text("fileu:filep", encoding="utf-8")
|
|
assert resolve_auth_basic(cli_value="cliu:clip", file_path=str(f)) == "cliu:clip"
|
|
|
|
|
|
def test_resolve_auth_basic_empty_file_raises(tmp_path):
|
|
f = tmp_path / "empty.txt"
|
|
f.write_text("# only comment\n", encoding="utf-8")
|
|
with pytest.raises(RuntimeError):
|
|
resolve_auth_basic(file_path=str(f))
|
|
|
|
|
|
def test_resolve_auth_basic_missing_file_raises():
|
|
with pytest.raises(RuntimeError):
|
|
resolve_auth_basic(file_path="nonexistent_file.txt")
|
|
|
|
|
|
# ----- resolve_auth_bearer -----
|
|
|
|
def test_resolve_auth_bearer_cli():
|
|
assert resolve_auth_bearer(cli_value="tok") == "tok"
|
|
|
|
|
|
def test_resolve_auth_bearer_env(monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_BEARER_TOKEN", "envtok")
|
|
assert resolve_auth_bearer() == "envtok"
|
|
|
|
|
|
def test_resolve_auth_bearer_file(tmp_path):
|
|
f = tmp_path / "tok.txt"
|
|
f.write_text("# header\ntoken123\n", encoding="utf-8")
|
|
assert resolve_auth_bearer(file_path=str(f)) == "token123"
|
|
|
|
|
|
# ----- detect_charset -----
|
|
|
|
def test_detect_charset_from_header():
|
|
assert detect_charset(b"hello", "text/html; charset=utf-8") == "utf-8"
|
|
|
|
|
|
def test_detect_charset_from_meta_tag():
|
|
html = b'<html><head><meta charset="gbk"></head><body>hello</body></html>'
|
|
assert detect_charset(html, "text/html") == "gbk"
|
|
|
|
|
|
def test_detect_charset_fallback_utf8():
|
|
assert detect_charset(b"plain", "text/html") == "utf-8"
|
|
|
|
|
|
def test_detect_charset_invalid_header_falls_back():
|
|
"""An invalid charset in the header should not raise; falls back to utf-8."""
|
|
assert detect_charset(b"hello", "text/html; charset=nonexistent_encoding") == "utf-8"
|
|
|
|
|
|
# ----- is_retryable_error -----
|
|
|
|
def _http_error(code):
|
|
return urllib.error.HTTPError("http://x", code, "msg", {}, None)
|
|
|
|
|
|
def test_retryable_429():
|
|
assert is_retryable_error(_http_error(429)) is True
|
|
|
|
|
|
def test_retryable_502():
|
|
assert is_retryable_error(_http_error(502)) is True
|
|
|
|
|
|
def test_retryable_503():
|
|
assert is_retryable_error(_http_error(503)) is True
|
|
|
|
|
|
def test_retryable_504():
|
|
assert is_retryable_error(_http_error(504)) is True
|
|
|
|
|
|
def test_non_retryable_404():
|
|
assert is_retryable_error(_http_error(404)) is False
|
|
|
|
|
|
def test_retryable_403():
|
|
# 403 是可重试的：让 fetch_url 的 UA-fallback 循环有机会切换到浏览器 UA。
|
|
# 真正的认证错误会在重试耗尽后由 classify_error 归为 E_AUTH。
|
|
assert is_retryable_error(_http_error(403)) is True
|
|
|
|
|
|
def test_non_retryable_200():
|
|
assert is_retryable_error(_http_error(200)) is False
|
|
|
|
|
|
def test_retryable_url_error():
|
|
assert is_retryable_error(urllib.error.URLError("refused")) is True
|
|
|
|
|
|
def test_retryable_os_error():
|
|
assert is_retryable_error(OSError("timeout")) is True
|
|
|
|
|
|
def test_retryable_status_set_contents():
|
|
# 403 加入可重试集合，让 UA-fallback 在 UA 被屏蔽时有机会切换浏览器 UA
|
|
assert RETRYABLE_STATUS == frozenset({403, 429, 502, 503, 504})
|
|
|
|
|
|
# ----- apply_proxy -----
|
|
|
|
def test_apply_proxy_sets_env_vars(monkeypatch):
|
|
monkeypatch.delenv("HTTP_PROXY", raising=False)
|
|
monkeypatch.delenv("HTTPS_PROXY", raising=False)
|
|
monkeypatch.delenv("NO_PROXY", raising=False)
|
|
apply_proxy("http://corp:8080")
|
|
assert os.environ["HTTP_PROXY"] == "http://corp:8080"
|
|
assert os.environ["HTTPS_PROXY"] == "http://corp:8080"
|
|
assert os.environ["NO_PROXY"] == "localhost,127.0.0.1,::1"
|
|
|
|
|
|
def test_apply_proxy_empty_is_noop(monkeypatch):
|
|
monkeypatch.delenv("HTTP_PROXY", raising=False)
|
|
apply_proxy("")
|
|
assert "HTTP_PROXY" not in os.environ
|
|
|
|
|
|
def test_apply_proxy_preserves_existing_no_proxy(monkeypatch):
|
|
"""setdefault must NOT overwrite a user-set NO_PROXY."""
|
|
monkeypatch.setenv("NO_PROXY", "custom.host")
|
|
apply_proxy("http://corp:8080")
|
|
assert os.environ["NO_PROXY"] == "custom.host"
|