Initial commit: SearXNG CLI Toolkit v1.6.0

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
2026-08-01 17:02:34 +08:00
commit 0468dd4e9d
17 changed files with 4791 additions and 0 deletions
@@ -0,0 +1,133 @@
+"""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