thzxx
471818074d
真实环境问题: 抓取被墙/反爬站点 (如 baike.baidu.com) 返回 403 时, classify_error 统一归为 E_AUTH, AI Agent 会误判为凭证问题而做无效的 认证重试。被封锁不是认证失败。 改动: - common.py: 新增 E_BLOCKED 错误码 + recovery_hint (提示换 URL/镜像/ 用 Wayback 兜底/--exclude-domain); 新增 classify_fetch_error() 与 _extract_status_code(): fetch 场景 403→E_BLOCKED, 401→E_AUTH, 其余委托 classify_error (搜索场景 403 仍为 E_AUTH, 不变) - fetch.py main(): 错误路径改用 classify_fetch_error (替代 classify_error) - search.py: fetch_page 与 fetch_top_results 线程错误路径同样切换 - 新增 11 个测试 (test_v240_blocked_code.py), 全量 572 测试通过 - 真实环境验证: baike 403 → E_BLOCKED, 正常站点不受影响 - 文档同步 (SKILL.md/README.md 错误码表, 版本号 2.4.0)
113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
"""Tests for v2.4.0 E_BLOCKED error classification (fetch scenario).
|
|
|
|
Real-world motivation: fetching baike.baidu.com etc. returns 403 because
|
|
the site blocks automated access (WAF/anti-bot), not because credentials
|
|
are wrong. classify_error maps 401/403 -> E_AUTH (correct for SearXNG
|
|
instance auth), but for page fetches that misleads agents into checking
|
|
credentials. classify_fetch_error narrows fetch-scenario 403 -> E_BLOCKED.
|
|
|
|
Covers:
|
|
* classify_fetch_error 403/401/5xx/4xx + message-only fallback
|
|
* classify_error keeps 403 -> E_AUTH (search scenario untouched)
|
|
* fetch_page error_code uses the fetch-scenario classification
|
|
"""
|
|
import urllib.error
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from common import (
|
|
E_AUTH,
|
|
E_BLOCKED,
|
|
E_INPUT,
|
|
E_NETWORK,
|
|
classify_error,
|
|
classify_fetch_error,
|
|
)
|
|
import search as search_mod
|
|
from search import fetch_page
|
|
|
|
|
|
def _http_error(code):
|
|
try:
|
|
raise urllib.error.HTTPError(
|
|
"https://x.example.com", code, "Error", {}, None)
|
|
except urllib.error.HTTPError as e:
|
|
return e
|
|
|
|
|
|
# ===== classify_fetch_error =====
|
|
|
|
def test_fetch_403_is_blocked():
|
|
"""Fetch 403 = anti-bot block, NOT auth."""
|
|
assert classify_fetch_error(_http_error(403)) == E_BLOCKED
|
|
|
|
|
|
def test_fetch_401_is_auth():
|
|
"""Fetch 401 stays E_AUTH (real credentials problem)."""
|
|
assert classify_fetch_error(_http_error(401)) == E_AUTH
|
|
|
|
|
|
def test_fetch_5xx_delegates_to_network():
|
|
assert classify_fetch_error(_http_error(503)) == E_NETWORK
|
|
|
|
|
|
def test_fetch_404_delegates_to_input():
|
|
"""4xx other than 403/401 keeps classify_error's E_INPUT."""
|
|
assert classify_fetch_error(_http_error(404)) == E_INPUT
|
|
|
|
|
|
def test_fetch_runtime_error_message_fallback():
|
|
"""No __cause__ but message contains 'HTTP 403' -> E_BLOCKED."""
|
|
err = RuntimeError("HTTP 403 for https://x.example.com")
|
|
assert classify_fetch_error(err) == E_BLOCKED
|
|
|
|
|
|
def test_fetch_explicit_status_code_shortcut():
|
|
"""Caller-provided status_code avoids re-parsing the exception."""
|
|
assert classify_fetch_error(RuntimeError("no code in message"),
|
|
status_code=403) == E_BLOCKED
|
|
|
|
|
|
def test_fetch_generic_error_delegates():
|
|
"""Connection-level exceptions delegate to classify_error -> E_NETWORK."""
|
|
err = urllib.error.URLError("connection refused")
|
|
assert classify_fetch_error(err) == E_NETWORK
|
|
|
|
|
|
# ===== classify_error unchanged (search scenario) =====
|
|
|
|
def test_classify_error_403_still_auth():
|
|
"""SearXNG instance 403 remains E_AUTH — search scenario untouched."""
|
|
assert classify_error(_http_error(403)) == E_AUTH
|
|
|
|
|
|
# ===== fetch_page integration =====
|
|
|
|
def test_fetch_page_403_error_code_blocked():
|
|
"""fetch_page reports 403 as E_BLOCKED (not E_AUTH)."""
|
|
with patch.object(search_mod, "fetch_url",
|
|
side_effect=RuntimeError("HTTP 403 for url")):
|
|
r = fetch_page("https://blocked.example.com", fallback_enabled=False)
|
|
assert r["status"] == "error"
|
|
assert r["error_code"] == E_BLOCKED
|
|
assert "403" in r["error"]
|
|
|
|
|
|
def test_fetch_page_404_error_code_input():
|
|
"""fetch_page 404 stays E_INPUT (not blocked/auth)."""
|
|
with patch.object(search_mod, "fetch_url",
|
|
side_effect=RuntimeError("HTTP 404 for url")):
|
|
r = fetch_page("https://missing.example.com", fallback_enabled=False)
|
|
assert r["status"] == "error"
|
|
assert r["error_code"] == E_INPUT
|
|
|
|
|
|
def test_fetch_page_401_error_code_auth():
|
|
"""fetch_page 401 stays E_AUTH (genuine credentials problem)."""
|
|
with patch.object(search_mod, "fetch_url",
|
|
side_effect=RuntimeError("HTTP 401 for url")):
|
|
r = fetch_page("https://private.example.com", fallback_enabled=False)
|
|
assert r["status"] == "error"
|
|
assert r["error_code"] == E_AUTH
|