feat(v2.2.1): 修复 Brotli 乱码 + 缓存治理 + 多页聚合 + 研究模式增强
核心修复(v2.2.1): - 修复 Brotli 乱码 bug: build_browser_headers 智能声明 Accept-Encoding, 仅在 brotli 可用时才声明 br; fetch.py 双路径 br 解压(requests + stdlib) 此前 Chrome/Edge UA 抓取 example.com 等返回 br 的站点输出乱码 v2.2.0 新功能: - main() 拆分为 _handle_verify/_handle_research/_handle_batch/_handle_single - --cache-max-size MB: 缓存大小上限 + LRU 淘汰(默认 100MB) - --pages N: 多页聚合 + 跨页去重 - --research 跨角度合并: 新增 merged_results 字段 - --stream / --progress: JSON Lines 流式输出 + request_id 贯穿 - --dry-run / --save-config / --log-format json - --similarity-dedup / --throttle-* 参数化 - 15-UA 池 + PDF/docx 解析 + error_code 字段 文档与测试: - SKILL.md: 版本号唯一(元数据),删除版本标记干扰 - README.md: 测试数量 539 -> 544 - 544 passed (新增 5 个 Content-Encoding 解压测试)
This commit is contained in:
@@ -150,9 +150,22 @@ def test_build_headers_includes_accept_language():
|
||||
|
||||
|
||||
def test_build_headers_includes_accept_encoding():
|
||||
"""Accept-Encoding 必须存在;Firefox 不发 br。"""
|
||||
"""Accept-Encoding 必须存在;Firefox 不发 br;Chrome/Edge 仅在
|
||||
本机安装了 brotli 解压器时才声明 br(v2.2.1 智能声明)。
|
||||
|
||||
未安装 brotli 时声明 br 会导致服务器返回 br 压缩字节而 requests
|
||||
无法解压 → 全页乱码。这是 fetch.py 真实环境的 bug 修复。
|
||||
"""
|
||||
# 导入 common 的 brotli 检测状态
|
||||
from common import _HAS_BROTLI
|
||||
|
||||
chrome_headers = build_browser_headers(FALLBACK_UAS[0])
|
||||
assert "br" in chrome_headers["Accept-Encoding"]
|
||||
# Chrome/Edge:根据 brotli 可用性决定是否声明 br
|
||||
if _HAS_BROTLI:
|
||||
assert "br" in chrome_headers["Accept-Encoding"]
|
||||
else:
|
||||
assert "br" not in chrome_headers["Accept-Encoding"]
|
||||
assert "gzip" in chrome_headers["Accept-Encoding"]
|
||||
|
||||
firefox_ua = next(ua for ua in FALLBACK_UAS if "Firefox/" in ua)
|
||||
firefox_headers = build_browser_headers(firefox_ua)
|
||||
|
||||
+119
-1
@@ -1,12 +1,15 @@
|
||||
"""Integration tests — mock urllib to test search_multi / fetch_url end-to-end.
|
||||
|
||||
Covers: search_json success/HTML-fallback/404/403, search_multi serial
|
||||
failover + all-fail, fetch_url stdlib-path success.
|
||||
failover + all-fail, fetch_url stdlib-path success, Content-Encoding
|
||||
decompression (gzip/deflate/br).
|
||||
|
||||
No real network calls are made; ``urllib.request.urlopen`` is patched.
|
||||
"""
|
||||
import gzip
|
||||
import json
|
||||
import urllib.error
|
||||
import zlib
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from search import search_json, search_multi
|
||||
@@ -131,3 +134,118 @@ def test_fetch_url_stdlib_max_size_truncates():
|
||||
patch.object(fetch_mod, "_HAS_REQUESTS", False):
|
||||
result = fetch_url("https://example.com", max_retries=0, max_size=50)
|
||||
assert result.truncated is True
|
||||
|
||||
|
||||
# ----- fetch_url Content-Encoding 解压(v2.1.1 + v2.2.1)-----
|
||||
# 这些测试防止真实环境的乱码 bug 回归:
|
||||
# - v2.1.1: stdlib urllib 不自动解压 gzip/deflate → 乱码
|
||||
# - v2.2.1: requests 不自动解压 br(未装 brotli 时)→ 乱码
|
||||
|
||||
def test_fetch_url_stdlib_gzip_decompress():
|
||||
"""stdlib 路径正确解压 gzip 压缩的响应(v2.1.1 修复)。"""
|
||||
html = b"<html><body><p>Gzip content</p></body></html>"
|
||||
compressed = gzip.compress(html)
|
||||
resp = _mock_urlopen(compressed, content_type="text/html")
|
||||
resp.headers = {"Content-Type": "text/html", "Content-Encoding": "gzip"}
|
||||
with patch("urllib.request.urlopen", return_value=resp), \
|
||||
patch.object(fetch_mod, "_HAS_REQUESTS", False):
|
||||
result = fetch_url("https://example.com", max_retries=0)
|
||||
assert "Gzip content" in result.content
|
||||
assert result.content.startswith("<html>")
|
||||
|
||||
|
||||
def test_fetch_url_stdlib_deflate_decompress():
|
||||
"""stdlib 路径正确解压 deflate 压缩的响应(v2.1.1 修复)。"""
|
||||
html = b"<html><body><p>Deflate content</p></body></html>"
|
||||
# zlib.compress 产生带 zlib 头的 deflate 流
|
||||
compressed = zlib.compress(html)
|
||||
resp = _mock_urlopen(compressed, content_type="text/html")
|
||||
resp.headers = {"Content-Type": "text/html", "Content-Encoding": "deflate"}
|
||||
with patch("urllib.request.urlopen", return_value=resp), \
|
||||
patch.object(fetch_mod, "_HAS_REQUESTS", False):
|
||||
result = fetch_url("https://example.com", max_retries=0)
|
||||
assert "Deflate content" in result.content
|
||||
assert result.content.startswith("<html>")
|
||||
|
||||
|
||||
def test_fetch_url_stdlib_brotli_decompress_when_brotli_available():
|
||||
"""stdlib 路径在 brotli 可用时正确解压 br 压缩的响应(v2.2.1 修复)。
|
||||
|
||||
使用 mock brotli 模块避免依赖真实 brotli 包。
|
||||
"""
|
||||
html = b"<html><body><p>Brotli content</p></body></html>"
|
||||
# 用 gzip 模拟 br 压缩字节(仅用于测试解压逻辑被正确调用)
|
||||
compressed = gzip.compress(html)
|
||||
|
||||
# 构造 mock brotli 模块
|
||||
fake_brotli = MagicMock()
|
||||
fake_brotli.decompress = MagicMock(return_value=html)
|
||||
|
||||
resp = _mock_urlopen(compressed, content_type="text/html")
|
||||
resp.headers = {"Content-Type": "text/html", "Content-Encoding": "br"}
|
||||
with patch("urllib.request.urlopen", return_value=resp), \
|
||||
patch.object(fetch_mod, "_HAS_REQUESTS", False), \
|
||||
patch.object(fetch_mod, "_HAS_BROTLI", True), \
|
||||
patch.object(fetch_mod, "_brotli", fake_brotli):
|
||||
result = fetch_url("https://example.com", max_retries=0)
|
||||
assert "Brotli content" in result.content
|
||||
assert result.content.startswith("<html>")
|
||||
# 验证 brotli.decompress 确实被调用
|
||||
fake_brotli.decompress.assert_called_once_with(compressed)
|
||||
|
||||
|
||||
def test_fetch_url_stdlib_brotli_skipped_when_unavailable():
|
||||
"""stdlib 路径在 brotli 不可用时跳过 br 解压(保留原 raw,由 decode 兜底)。
|
||||
|
||||
这对应真实环境中未安装 brotli 包的情况:build_browser_headers 不会
|
||||
声明 br,所以正常情况下不会收到 br 响应。此测试验证即使收到 br
|
||||
响应也不会崩溃(虽然内容会是乱码)。
|
||||
"""
|
||||
html = b"<html><body><p>content</p></body></html>"
|
||||
compressed = gzip.compress(html) # 假装是 br 压缩
|
||||
resp = _mock_urlopen(compressed, content_type="text/html")
|
||||
resp.headers = {"Content-Type": "text/html", "Content-Encoding": "br"}
|
||||
with patch("urllib.request.urlopen", return_value=resp), \
|
||||
patch.object(fetch_mod, "_HAS_REQUESTS", False), \
|
||||
patch.object(fetch_mod, "_HAS_BROTLI", False):
|
||||
result = fetch_url("https://example.com", max_retries=0)
|
||||
# 未解压时 content 是乱码但不会崩溃(errors="replace" 兜底)
|
||||
assert result.content # 有内容(虽然是乱码)
|
||||
|
||||
|
||||
def test_fetch_url_requests_brotli_decompress_when_available():
|
||||
"""requests 路径在 brotli 可用时手动解压 br(v2.2.1 修复)。
|
||||
|
||||
requests 不自动解压 br(除非安装 brotli 包)。此测试验证双保险逻辑:
|
||||
即使 requests 路径,也会在 _HAS_BROTLI=True 时手动解压 br。
|
||||
|
||||
使用 mock requests Session 避免 HTTP 请求。
|
||||
"""
|
||||
html = b"<html><body><p>BR via requests</p></body></html>"
|
||||
compressed = gzip.compress(html) # 假装是 br 压缩
|
||||
|
||||
fake_brotli = MagicMock()
|
||||
fake_brotli.decompress = MagicMock(return_value=html)
|
||||
|
||||
# 构造 mock requests Response
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.headers = {
|
||||
"Content-Type": "text/html",
|
||||
"Content-Encoding": "br",
|
||||
}
|
||||
mock_resp.content = compressed
|
||||
mock_resp.url = "https://example.com"
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_resp.close = MagicMock()
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.get = MagicMock(return_value=mock_resp)
|
||||
|
||||
with patch.object(fetch_mod, "_HAS_REQUESTS", True), \
|
||||
patch.object(fetch_mod, "_HAS_BROTLI", True), \
|
||||
patch.object(fetch_mod, "_brotli", fake_brotli), \
|
||||
patch.object(fetch_mod, "_get_session", return_value=mock_session):
|
||||
result = fetch_url("https://example.com", max_retries=0)
|
||||
assert "BR via requests" in result.content
|
||||
fake_brotli.decompress.assert_called_once_with(compressed)
|
||||
|
||||
Reference in New Issue
Block a user