import json import urllib.error import pytest import web class FakeResp: """Minimal stand-in for an http.client.HTTPResponse context manager.""" def __init__(self, payload): self._b = json.dumps(payload).encode("utf-8") def read(self): return self._b def __enter__(self): return self def __exit__(self, *exc): return False def test_endpoints_defaults(monkeypatch): for var in ("WEBTOOLS_HOST", "SEARXNG_PORT", "FIRECRAWL_PORT"): monkeypatch.delenv(var, raising=False) searx, fire = web.endpoints() assert searx == "http://10.10.20.37:8080" assert fire == "http://10.10.20.37:3002" def test_endpoints_env_override(monkeypatch): monkeypatch.setenv("WEBTOOLS_HOST", "192.168.1.5") monkeypatch.setenv("SEARXNG_PORT", "9999") monkeypatch.setenv("FIRECRAWL_PORT", "4000") searx, fire = web.endpoints() assert searx == "http://192.168.1.5:9999" assert fire == "http://192.168.1.5:4000" def test_request_parses_json(monkeypatch): monkeypatch.setattr(web.urllib.request, "urlopen", lambda req, timeout=None: FakeResp({"ok": True})) assert web._request("http://x/y") == {"ok": True} def test_request_retries_then_succeeds(monkeypatch): calls = {"n": 0} def flaky(req, timeout=None): calls["n"] += 1 if calls["n"] < 3: raise urllib.error.URLError("boom") return FakeResp({"ok": True}) monkeypatch.setattr(web.urllib.request, "urlopen", flaky) monkeypatch.setattr(web.time, "sleep", lambda *_: None) assert web._request("http://x/y", retries=2) == {"ok": True} assert calls["n"] == 3 def test_request_raises_after_exhausting_retries(monkeypatch): def always_fail(req, timeout=None): raise urllib.error.URLError("down") monkeypatch.setattr(web.urllib.request, "urlopen", always_fail) monkeypatch.setattr(web.time, "sleep", lambda *_: None) with pytest.raises(web.WebToolsError): web._request("http://x/y", retries=2) def test_search_parses_and_caps(monkeypatch): payload = {"results": [ {"title": "A", "url": "http://a", "content": "snip a", "engine": "ddg"}, {"title": "B", "url": "http://b", "content": "snip b", "engine": "bing"}, {"title": "no-url", "content": "dropped"}, {"title": "C", "url": "http://c", "content": "snip c", "engine": "ddg"}, ]} monkeypatch.setattr(web, "_request", lambda url, **kw: payload) hits = web.search("demons", max_results=2) assert hits == [ {"title": "A", "url": "http://a", "snippet": "snip a", "engine": "ddg"}, {"title": "B", "url": "http://b", "snippet": "snip b", "engine": "bing"}, ] def test_search_builds_json_query_url(monkeypatch): captured = {} def fake_request(url, **kw): captured["url"] = url return {"results": []} monkeypatch.setattr(web, "_request", fake_request) monkeypatch.setenv("WEBTOOLS_HOST", "10.10.20.37") monkeypatch.delenv("SEARXNG_PORT", raising=False) web.search("demon hierarchy") assert captured["url"].startswith("http://10.10.20.37:8080/search?") assert "q=demon+hierarchy" in captured["url"] assert "format=json" in captured["url"]