thzxx
d9a08716bc
Bug 1: Wayback 兜底未触发 (search.py fetch_page) - 根因: Cloudflare JS 质询页常返回 HTTP 200 (非 403), 主抓取 result 非 None - _should_try_fallback 在 result 非 None 时直接返回 False, 跳过兜底 - 反爬检测在兜底判断之后执行, 错过兜底入口 - 修复: 反爬检测提前到兜底判断之前, anti_bot_detected=True 也触发 Wayback - Wayback 结果重新做反爬检测 (防御性) Bug 2: WAF 指纹库误判正常内容 (search.py WAF_FINGERPRINTS) - 根因: 裸公司名 (cloudflare/akamai) 和宽泛词 (captcha/challenge/dd-) 做全文匹配 - DataCamp 文章引用 cloudflare.com 文档链接 -> 误判为 cloudflare WAF - 'coding challenges' 正常内容 -> 误判为 generic 反爬 - Wayback 归档正文被误判, 兜底返回的有效内容被丢弃 - 修复: 移除裸公司名和宽泛词, 改用技术标识符 (cf-ray/incap_ses/bm_sz 等) - 通用文案用完整短语 (please complete the captcha) 替代单词 - 增加 <title> 标签精准检测 (反爬页 title 是特征文案, 误判率极低) - 新增 Anubis 反爬系统检测 (anubis_challenge/miserere) 真实测试验证 (search.metona.cn 实例): - v2.0.0: --fetch 3 全部失败 (3 ERR: cloudflare/generic, Wayback 未触发) - v2.0.1: --fetch 3 全部成功 (3 OK: 38100/93442/1945 chars, UA 轮换绕过 Cloudflare) 测试: 458 个全部通过 (新增 7 个测试覆盖修复行为)
301 lines
12 KiB
Python
301 lines
12 KiB
Python
"""Tests for v2.0.0 Wayback Machine fallback and fetch_page enhancements.
|
||
|
||
Covers: _should_try_fallback (trigger conditions), _try_wayback_fallback
|
||
(mocked), fetch_page with fallback_enabled flag, new result fields
|
||
(anti_bot_detected, waf_type, fallback_used), fallback disabled path.
|
||
"""
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
|
||
|
||
from unittest.mock import patch, MagicMock
|
||
from fetch import FetchResult
|
||
import search as search_mod
|
||
from search import (
|
||
_should_try_fallback,
|
||
_try_wayback_fallback,
|
||
fetch_page,
|
||
)
|
||
|
||
|
||
def _make_result(content="<html><body>OK</body></html>",
|
||
content_type="text/html", final_url="https://example.com",
|
||
truncated=False, ua="TestUA/1.0"):
|
||
return FetchResult(content, content_type, final_url, truncated, ua)
|
||
|
||
|
||
# ----- _should_try_fallback -----
|
||
|
||
def test_should_try_fallback_on_404():
|
||
assert _should_try_fallback(None, "HTTP 404 for https://example.com") is True
|
||
|
||
|
||
def test_should_try_fallback_on_403():
|
||
assert _should_try_fallback(None, "HTTP 403 for https://example.com") is True
|
||
|
||
|
||
def test_should_try_fallback_on_timeout():
|
||
assert _should_try_fallback(None, "Request failed: timed out") is True
|
||
|
||
|
||
def test_should_try_fallback_on_connection_reset():
|
||
assert _should_try_fallback(None, "Connection reset by peer") is True
|
||
|
||
|
||
def test_should_try_fallback_on_max_retries():
|
||
assert _should_try_fallback(None, "Max retries exceeded") is True
|
||
|
||
|
||
def test_should_try_fallback_not_on_success():
|
||
"""主抓取成功时不触发兜底。"""
|
||
result = _make_result()
|
||
assert _should_try_fallback(result, None) is False
|
||
|
||
|
||
def test_should_try_fallback_not_on_500():
|
||
"""500 错误不触发 Wayback（服务器内部错误，Wayback 也未必有）。"""
|
||
assert _should_try_fallback(None, "HTTP 500 for https://example.com") is False
|
||
|
||
|
||
def test_should_try_fallback_not_on_empty_error():
|
||
assert _should_try_fallback(None, "") is False
|
||
assert _should_try_fallback(None, None) is False
|
||
|
||
|
||
def test_should_try_fallback_not_on_dns():
|
||
"""DNS 失败不触发（Wayback 也访问不到）。"""
|
||
assert _should_try_fallback(None, "Name or service not known") is False
|
||
|
||
|
||
# ----- _try_wayback_fallback (mocked) -----
|
||
|
||
def test_try_wayback_success():
|
||
"""Wayback 返回成功时，返回 FetchResult。"""
|
||
wb_result = _make_result(content="<html>Archived page</html>",
|
||
final_url="https://web.archive.org/web/2024/https://example.com")
|
||
with patch("search.fetch_url", return_value=wb_result) as mock_fetch:
|
||
result = _try_wayback_fallback("https://example.com")
|
||
assert result is not None
|
||
assert "Archived page" in result.content
|
||
# 确认调用了 Wayback URL
|
||
call_args = mock_fetch.call_args
|
||
assert "web.archive.org/web/2/" in call_args[0][0]
|
||
|
||
|
||
def test_try_wayback_failure_returns_none():
|
||
"""Wayback 也失败时返回 None。"""
|
||
with patch("search.fetch_url", side_effect=RuntimeError("timeout")):
|
||
result = _try_wayback_fallback("https://example.com")
|
||
assert result is None
|
||
|
||
|
||
def test_try_wayback_uses_reduced_timeout():
|
||
"""Wayback 使用 min(timeout, 10) 避免长时间阻塞。"""
|
||
wb_result = _make_result()
|
||
with patch("search.fetch_url", return_value=wb_result) as mock_fetch:
|
||
_try_wayback_fallback("https://example.com", timeout=30)
|
||
call_kwargs = mock_fetch.call_args[1]
|
||
assert call_kwargs["timeout"] == 10
|
||
|
||
|
||
def test_try_wayback_no_auth_headers():
|
||
"""Wayback 是公共服务，不传 auth_headers。"""
|
||
wb_result = _make_result()
|
||
with patch("search.fetch_url", return_value=wb_result) as mock_fetch:
|
||
_try_wayback_fallback("https://example.com",
|
||
auth_headers={"Authorization": "Bearer x"})
|
||
call_kwargs = mock_fetch.call_args[1]
|
||
# auth_headers 应为 None（Wayback 不需要认证）
|
||
assert call_kwargs.get("auth_headers") is None
|
||
|
||
|
||
# ----- fetch_page with fallback -----
|
||
|
||
def test_fetch_page_success_no_fallback():
|
||
"""主抓取成功，不触发兜底。"""
|
||
with patch("search.fetch_url", return_value=_make_result()):
|
||
result = fetch_page("https://example.com", fallback_enabled=True)
|
||
assert result["status"] == "ok"
|
||
assert result["fallback_used"] is None
|
||
assert result["anti_bot_detected"] is False
|
||
assert result["waf_type"] is None
|
||
|
||
|
||
def test_fetch_page_404_triggers_wayback_success():
|
||
"""404 触发 Wayback，Wayback 成功。"""
|
||
wb_result = _make_result(content="<html>Archived</html>")
|
||
with patch("search.fetch_url",
|
||
side_effect=[RuntimeError("HTTP 404"), wb_result]):
|
||
result = fetch_page("https://example.com", fallback_enabled=True)
|
||
assert result["status"] == "ok"
|
||
assert result["fallback_used"] == "wayback"
|
||
|
||
|
||
def test_fetch_page_403_triggers_wayback_success():
|
||
"""403 触发 Wayback，Wayback 成功。"""
|
||
wb_result = _make_result()
|
||
with patch("search.fetch_url",
|
||
side_effect=[RuntimeError("HTTP 403"), wb_result]):
|
||
result = fetch_page("https://example.com", fallback_enabled=True)
|
||
assert result["status"] == "ok"
|
||
assert result["fallback_used"] == "wayback"
|
||
|
||
|
||
def test_fetch_page_timeout_triggers_wayback():
|
||
"""超时触发 Wayback。"""
|
||
wb_result = _make_result()
|
||
with patch("search.fetch_url",
|
||
side_effect=[RuntimeError("timed out"), wb_result]):
|
||
result = fetch_page("https://example.com", fallback_enabled=True)
|
||
assert result["status"] == "ok"
|
||
assert result["fallback_used"] == "wayback"
|
||
|
||
|
||
def test_fetch_page_fallback_disabled():
|
||
"""fallback_enabled=False 时不触发 Wayback。"""
|
||
with patch("search.fetch_url",
|
||
side_effect=RuntimeError("HTTP 404")) as mock_fetch:
|
||
result = fetch_page("https://example.com", fallback_enabled=False)
|
||
assert result["status"] == "error"
|
||
assert result["fallback_used"] is None
|
||
# 只调用一次（主抓取），不调用 Wayback
|
||
assert mock_fetch.call_count == 1
|
||
|
||
|
||
def test_fetch_page_both_fail():
|
||
"""主抓取和 Wayback 都失败。"""
|
||
with patch("search.fetch_url",
|
||
side_effect=[RuntimeError("HTTP 404"), RuntimeError("timeout")]):
|
||
result = fetch_page("https://example.com", fallback_enabled=True)
|
||
assert result["status"] == "error"
|
||
assert result["fallback_used"] is None
|
||
|
||
|
||
# ----- fetch_page anti-bot fields -----
|
||
|
||
def test_fetch_page_detects_cloudflare():
|
||
"""抓到 Cloudflare 拦截页，标记 anti_bot_detected + waf_type。"""
|
||
cf_html = "<html><body>Just a moment... cf-ray: 123</body></html>"
|
||
with patch("search.fetch_url",
|
||
return_value=_make_result(content=cf_html)):
|
||
result = fetch_page("https://example.com", fallback_enabled=False)
|
||
assert result["status"] == "error"
|
||
assert result["anti_bot_detected"] is True
|
||
assert result["waf_type"] == "cloudflare"
|
||
assert "cloudflare" in result["error"]
|
||
|
||
|
||
def test_fetch_page_detects_datadome():
|
||
"""抓到 DataDome 拦截页。"""
|
||
dd_html = "<html><body>Protected by DataDome</body></html>"
|
||
with patch("search.fetch_url",
|
||
return_value=_make_result(content=dd_html)):
|
||
result = fetch_page("https://example.com", fallback_enabled=False)
|
||
assert result["status"] == "error"
|
||
assert result["anti_bot_detected"] is True
|
||
assert result["waf_type"] == "datadome"
|
||
|
||
|
||
def test_fetch_page_normal_page_no_anti_bot():
|
||
"""正常页面 anti_bot_detected=False。"""
|
||
normal_html = "<html><body><p>Normal article content.</p></body></html>"
|
||
with patch("search.fetch_url",
|
||
return_value=_make_result(content=normal_html)):
|
||
result = fetch_page("https://example.com", fallback_enabled=False)
|
||
assert result["status"] == "ok"
|
||
assert result["anti_bot_detected"] is False
|
||
assert result["waf_type"] is None
|
||
|
||
|
||
def test_fetch_page_anti_bot_triggers_wayback():
|
||
"""被反爬拦截后应尝试 Wayback 兜底（v2.0.1 修复）。
|
||
|
||
v2.0.0 bug：主抓取"成功"返回反爬页（HTTP 200 + Cloudflare 质询）时，
|
||
result 非 None 导致 _should_try_fallback 返回 False，Wayback 永不触发。
|
||
v2.0.1 修复：反爬检测提前到兜底判断之前，反爬阳性也触发 Wayback。
|
||
|
||
本测试 mock fetch_url 两次调用：
|
||
1. 主抓取 → 返回 Cloudflare 质询页
|
||
2. Wayback 兜底 → 返回正常归档页
|
||
"""
|
||
cf_html = "<html><body>Just a moment... cf-ray: 123</body></html>"
|
||
wb_html = "<html><body><p>Archived article content.</p></body></html>"
|
||
cf_result = _make_result(content=cf_html, final_url="https://example.com")
|
||
wb_result = _make_result(content=wb_html,
|
||
final_url="https://web.archive.org/web/2024/https://example.com")
|
||
with patch("search.fetch_url",
|
||
side_effect=[cf_result, wb_result]) as mock_fetch:
|
||
result = fetch_page("https://example.com", fallback_enabled=True)
|
||
# Wayback 兜底成功，反爬标记清除，status=ok
|
||
assert result["status"] == "ok"
|
||
assert result["anti_bot_detected"] is False
|
||
assert result["waf_type"] is None
|
||
assert result["fallback_used"] == "wayback"
|
||
assert "Archived article content." in result["text"]
|
||
# 确认 fetch_url 被调用两次：主抓取 + Wayback
|
||
assert mock_fetch.call_count == 2
|
||
# 第二次调用应该是 Wayback URL
|
||
assert "web.archive.org/web/2/" in mock_fetch.call_args_list[1][0][0]
|
||
|
||
|
||
def test_fetch_page_anti_bot_wayback_also_blocked():
|
||
"""主抓取反爬 + Wayback 也反爬/失败 → 最终返回 error。
|
||
|
||
Wayback 兜底返回 None（失败）时，保留主抓取的反爬检测结果。
|
||
"""
|
||
cf_html = "<html><body>Just a moment... cf-ray: 123</body></html>"
|
||
cf_result = _make_result(content=cf_html, final_url="https://example.com")
|
||
with patch("search.fetch_url",
|
||
side_effect=[cf_result, RuntimeError("wayback timeout")]):
|
||
result = fetch_page("https://example.com", fallback_enabled=True)
|
||
# Wayback 失败，保留反爬 error
|
||
assert result["status"] == "error"
|
||
assert result["anti_bot_detected"] is True
|
||
assert result["waf_type"] == "cloudflare"
|
||
assert result["fallback_used"] is None
|
||
|
||
|
||
def test_fetch_page_anti_bot_fallback_disabled():
|
||
"""--no-fallback 时反爬页直接返回 error，不尝试 Wayback。"""
|
||
cf_html = "<html><body>Just a moment... cf-ray: 123</body></html>"
|
||
with patch("search.fetch_url",
|
||
return_value=_make_result(content=cf_html)) as mock_fetch:
|
||
result = fetch_page("https://example.com", fallback_enabled=False)
|
||
assert result["status"] == "error"
|
||
assert result["anti_bot_detected"] is True
|
||
assert result["waf_type"] == "cloudflare"
|
||
assert result["fallback_used"] is None
|
||
# 只调用一次（主抓取），没有 Wayback
|
||
assert mock_fetch.call_count == 1
|
||
|
||
|
||
def test_fetch_page_passes_referer():
|
||
"""referer 透传给 fetch_url。"""
|
||
with patch("search.fetch_url", return_value=_make_result()) as mock_fetch:
|
||
fetch_page("https://example.com", referer="https://google.com/",
|
||
fallback_enabled=False)
|
||
call_kwargs = mock_fetch.call_args[1]
|
||
assert call_kwargs.get("referer") == "https://google.com/"
|
||
|
||
|
||
# ----- fetch_page result structure -----
|
||
|
||
def test_fetch_page_result_has_all_v2_fields():
|
||
"""结果 dict 包含所有 v2.0.0 新字段。"""
|
||
with patch("search.fetch_url", return_value=_make_result()):
|
||
result = fetch_page("https://example.com", fallback_enabled=False)
|
||
required_fields = ["anti_bot_detected", "waf_type", "fallback_used",
|
||
"status", "url", "text", "text_length", "truncated"]
|
||
for field in required_fields:
|
||
assert field in result, f"missing field: {field}"
|
||
|
||
|
||
def test_fetch_page_error_result_has_all_v2_fields():
|
||
"""错误结果也包含所有 v2.0.0 新字段。"""
|
||
with patch("search.fetch_url", side_effect=RuntimeError("HTTP 500")):
|
||
result = fetch_page("https://example.com", fallback_enabled=False)
|
||
required_fields = ["anti_bot_detected", "waf_type", "fallback_used",
|
||
"status", "url", "error", "text", "text_length"]
|
||
for field in required_fields:
|
||
assert field in result, f"missing field: {field}"
|