"""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="OK", 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="Archived page", 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="Archived") 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 = "Just a moment... cf-ray: 123" 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 = "Protected by DataDome" 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 = "

Normal article content.

" 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（_should_try_fallback 防御性检查）。当前实现：主抓取成功返回反爬页内容 → result 非 None → 不触发兜底。此测试记录此行为：反爬页被当作"成功抓取"返回，在 fetch_page 内部检测。 """ cf_html = "Just a moment... cf-ray: 123" with patch("search.fetch_url", return_value=_make_result(content=cf_html)): result = fetch_page("https://example.com", fallback_enabled=True) # 反爬页被检测到，标记为 error + anti_bot_detected assert result["status"] == "error" assert result["anti_bot_detected"] is True # 但不会触发 Wayback（因为主抓取"成功"了，只是内容是反爬页） assert result["fallback_used"] is None 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}"