Metona Team
0468dd4e9d
Multi-instance failover, exponential-backoff retry, SQLite cache, batch mode, domain filter, cross-engine dedup, result sorting, CSV export, structured logging, enhanced Markdown conversion, 155 pytest tests, Gitea Actions CI
134 lines
5.1 KiB
Python
134 lines
5.1 KiB
Python
"""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.
|
|
|
|
No real network calls are made; ``urllib.request.urlopen`` is patched.
|
|
"""
|
|
import json
|
|
import urllib.error
|
|
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"<html>not json</html>")):
|
|
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"<html><body><p>Hello</p></body></html>"
|
|
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"<html>" + b"x" * 200 + b"</html>"
|
|
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
|