"""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, 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 import fetch as fetch_mod from fetch import fetch_url def _mock_urlopen(data: bytes, content_type="application/json"): """Build a MagicMock that quacks like an urlopen context manager.""" resp = MagicMock() resp.read.return_value = data resp.headers = {"Content-Type": content_type} resp.__enter__.return_value = resp resp.__exit__.return_value = None return resp # ----- search_json ----- def test_search_json_success(): payload = json.dumps({"results": [{"title": "hi", "url": "https://x.com"}]}) with patch("urllib.request.urlopen", return_value=_mock_urlopen(payload.encode())): r = search_json("https://s.example.com", {"q": "test", "format": "json"}) assert r is not None assert r["results"][0]["title"] == "hi" def test_search_json_html_returns_none(): """HTML response (not JSON) → return None (JSON unsupported).""" with patch("urllib.request.urlopen", return_value=_mock_urlopen(b"not json")): r = search_json("https://s.example.com", {"q": "test", "format": "json"}) assert r is None def test_search_json_404_returns_none(): """404 = JSON endpoint absent → None (triggers HTML fallback upstream).""" err = urllib.error.HTTPError("url", 404, "Not Found", {}, None) with patch("urllib.request.urlopen", side_effect=err): r = search_json("https://s.example.com", {"q": "test", "format": "json"}) assert r is None def test_search_json_403_raises(): """403 = auth/IP issue → must raise, not silently fall back to HTML.""" err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None) with patch("urllib.request.urlopen", side_effect=err): try: search_json("https://s.example.com", {"q": "test", "format": "json"}) assert False, "should have raised" except urllib.error.HTTPError: pass # expected # ----- search_multi (serial failover) ----- def test_search_multi_serial_failover(): """First instance fails, second succeeds → return second's results.""" payload = json.dumps({"results": [{"title": "from-b", "url": "https://b.com"}]}) err = urllib.error.URLError("connection refused") with patch("urllib.request.urlopen", side_effect=[err, _mock_urlopen(payload.encode())]): r = search_multi( ["https://a.example.com", "https://b.example.com"], {"q": "test", "format": "json"}, parallel=False, retry_per=0, ) assert r["results"][0]["title"] == "from-b" def test_search_multi_all_fail_raises(): """All instances fail → RuntimeError.""" err = urllib.error.URLError("connection refused") with patch("urllib.request.urlopen", side_effect=err): try: search_multi( ["https://a.example.com", "https://b.example.com"], {"q": "test", "format": "json"}, parallel=False, retry_per=0, ) assert False, "should have raised" except RuntimeError as e: assert "All 2 instances failed" in str(e) def test_search_multi_single_instance_success(): """Single instance, serial mode, success → return results.""" payload = json.dumps({"results": [{"title": "ok", "url": "https://a.com"}]}) with patch("urllib.request.urlopen", return_value=_mock_urlopen(payload.encode())): r = search_multi( ["https://a.example.com"], {"q": "test", "format": "json"}, parallel=False, retry_per=0, ) assert r["results"][0]["title"] == "ok" # ----- fetch_url (stdlib path) ----- def test_fetch_url_stdlib_success(): """fetch_url with stdlib path returns content + content_type.""" html = b"

Hello

" resp = _mock_urlopen(html, content_type="text/html; charset=utf-8") 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 "Hello" in result.content assert "text/html" in result.content_type def test_fetch_url_stdlib_max_size_truncates(): """max_size sets truncated=True when response exceeds the cap. Note: mock's read() ignores the size arg, so content length is not accurately capped here — we only verify the truncated flag is set. """ html = b"" + b"x" * 200 + b"" resp = _mock_urlopen(html, content_type="text/html") with patch("urllib.request.urlopen", return_value=resp), \ 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"

Gzip content

" 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("") def test_fetch_url_stdlib_deflate_decompress(): """stdlib 路径正确解压 deflate 压缩的响应（v2.1.1 修复）。""" html = b"

Deflate content

" # 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("") def test_fetch_url_stdlib_brotli_decompress_when_brotli_available(): """stdlib 路径在 brotli 可用时正确解压 br 压缩的响应（v2.2.1 修复）。使用 mock brotli 模块避免依赖真实 brotli 包。 """ html = b"

Brotli content

" # 用 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("") # 验证 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"

content

" 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"

BR via requests

" 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)