71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
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)
|