research/tests/test_web.py

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


def test_scrape_returns_markdown(monkeypatch):
    monkeypatch.setattr(web, "_request",
                        lambda url, **kw: {"success": True, "data": {"markdown": "# Hi"}})
    assert web.scrape("http://a") == "# Hi"


def test_scrape_posts_url_and_format(monkeypatch):
    captured = {}

    def fake_request(url, **kw):
        captured["url"] = url
        captured["data"] = kw.get("data")
        return {"success": True, "data": {"markdown": "ok"}}

    monkeypatch.setattr(web, "_request", fake_request)
    web.scrape("http://example.com/page")
    assert captured["url"].endswith(":3002/v1/scrape")
    assert captured["data"] == {"url": "http://example.com/page", "formats": ["markdown"]}


def test_scrape_raises_on_failure_flag(monkeypatch):
    monkeypatch.setattr(web, "_request",
                        lambda url, **kw: {"success": False, "error": "blocked"})
    with pytest.raises(web.WebToolsError):
        web.scrape("http://a")


def test_scrape_raises_on_empty_markdown(monkeypatch):
    monkeypatch.setattr(web, "_request",
                        lambda url, **kw: {"success": True, "data": {"markdown": ""}})
    with pytest.raises(web.WebToolsError):
        web.scrape("http://a")


def test_main_search_json(monkeypatch, capsys):
    monkeypatch.setattr(web, "search",
                        lambda q, max_results=12: [{"title": "T", "url": "http://u",
                                                    "snippet": "s", "engine": "e"}])
    rc = web.main(["search", "demons", "--max", "5", "--json"])
    assert rc == 0
    out = capsys.readouterr().out
    assert json.loads(out) == [{"title": "T", "url": "http://u",
                                "snippet": "s", "engine": "e"}]


def test_main_scrape_to_stdout(monkeypatch, capsys):
    monkeypatch.setattr(web, "scrape", lambda url: "# Page")
    rc = web.main(["scrape", "http://a"])
    assert rc == 0
    assert "# Page" in capsys.readouterr().out


def test_main_scrape_to_file(monkeypatch, tmp_path):
    monkeypatch.setattr(web, "scrape", lambda url: "# Saved")
    out = tmp_path / "p.md"
    rc = web.main(["scrape", "http://a", "--out", str(out)])
    assert rc == 0
    assert out.read_text(encoding="utf-8") == "# Saved"


def test_main_returns_1_on_error(monkeypatch, capsys):
    def boom(*a, **k):
        raise web.WebToolsError("nope")

    monkeypatch.setattr(web, "search", boom)
    rc = web.main(["search", "x"])
    assert rc == 1
    assert "error:" in capsys.readouterr().err


def test_request_does_not_retry_on_bad_json(monkeypatch):
    calls = {"n": 0}

    class BadResp:
        def read(self):
            return b"<html>not json</html>"

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

    def once(req, timeout=None):
        calls["n"] += 1
        return BadResp()

    monkeypatch.setattr(web.urllib.request, "urlopen", once)
    monkeypatch.setattr(web.time, "sleep", lambda *_: None)
    with pytest.raises(web.WebToolsError):
        web._request("http://x/y", retries=2)
    assert calls["n"] == 1  # deterministic failure, not retried


def test_scrape_raises_on_null_data(monkeypatch):
    monkeypatch.setattr(web, "_request",
                        lambda url, **kw: {"success": True, "data": None})
    with pytest.raises(web.WebToolsError):
        web.scrape("http://a")


def test_main_search_human_format(monkeypatch, capsys):
    monkeypatch.setattr(web, "search",
                        lambda q, max_results=12: [{"title": "T", "url": "http://u",
                                                    "snippet": "s" * 200, "engine": "e"}])
    rc = web.main(["search", "x"])
    assert rc == 0
    out = capsys.readouterr().out
    assert "T" in out and "http://u" in out


def test_main_scrape_out_prints_status(monkeypatch, capsys, tmp_path):
    monkeypatch.setattr(web, "scrape", lambda url: "# X")
    out = tmp_path / "p.md"
    rc = web.main(["scrape", "http://a", "--out", str(out)])
    assert rc == 0
    assert f"wrote {out}" in capsys.readouterr().out