web-fetcher/tests/test_web_fetcher.py

"""Offline tests (fixtures only). Usage: python3 -m unittest discover -s tests"""

import json
import unittest
from pathlib import Path

from web_fetcher.core import Fetched
from web_fetcher.registry import run_spec

FIX = Path(__file__).parent / "fixtures"


def fake(body: bytes, url: str = "https://example.test/x"):
    return lambda u: Fetched(url=u, body=body, retrieved_at="2026-10-12T00:00:00+00:00",
                             sha256="0" * 64, content_type="application/json")


class TestEurostat(unittest.TestCase):
    def test_parses_values_per_geo_and_period(self):
        body = (FIX / "eurostat_isoc_eb_ai.json").read_bytes()
        spec = {"id": "eu_ai_use", "adapter": "eurostat", "dataset": "isoc_eb_ai",
                "filters": {"geo": ["EU27_2020", "DE"], "unit": "PC_ENT", "size_emp": "GE10",
                            "indic_is": "E_AI_TANY"}, "split_by": ["geo"], "last": 3}
        obs = run_spec(spec, get=fake(body))
        got = {(o.labels["geo"], o.period): o.value for o in obs}
        self.assertEqual(got[("EU27_2020", "2025")], 19.95)
        self.assertEqual(got[("DE", "2023")], 11.55)
        self.assertTrue(all(o.status == "ok" and o.url and o.payload_sha256 for o in obs))
        self.assertEqual(obs[0].unit, "Percentage of enterprises")

    def test_url_contains_filters(self):
        from web_fetcher.adapters.eurostat import build_url
        url = build_url({"dataset": "isoc_eb_ai", "filters": {"geo": ["DE"], "unit": "PC_ENT"}, "last": 2})
        self.assertIn("isoc_eb_ai?", url)
        self.assertIn("geo=DE", url)
        self.assertIn("lastTimePeriod=2", url)


class TestXlsx(unittest.TestCase):
    def test_btos_ai_use_latest_period(self):
        body = (FIX / "btos_national.xlsx").read_bytes()
        obs = run_spec({"id": "us_ai_use", "adapter": "xlsx_wide", "url": "https://x.test/National.xlsx",
                        "sheet": "Response Estimates", "filter": {"Question ID": "7", "Answer ID": "1"},
                        "unit": "% of firms", "source": "Census BTOS"}, get=fake(body))
        self.assertEqual(obs[0].status, "ok")
        self.assertEqual(obs[0].period, "202620")
        self.assertAlmostEqual(obs[0].value, 23.8)


class TestHttp(unittest.TestCase):
    def test_json_path(self):
        body = json.dumps({"data": {"rows": [{"v": 7.5, "p": "2026-Q3"}]}}).encode()
        obs = run_spec({"id": "j", "adapter": "http_json", "url": "https://x.test", "path": "data.rows.0.v",
                        "period_path": "data.rows.0.p", "unit": "%", "source": "test"}, get=fake(body))
        self.assertEqual((obs[0].status, obs[0].value, obs[0].period), ("ok", 7.5, "2026-Q3"))

    def test_json_missing_path_is_not_found(self):
        obs = run_spec({"id": "j", "adapter": "http_json", "url": "https://x.test", "path": "nope"},
                       get=fake(b"{}"))
        self.assertEqual(obs[0].status, "not_found")

    def test_csv_latest_matching_row(self):
        body = b"geo,period,value\nUS,2026-08,8.1\nUS,2026-09,8.7\nDE,2026-09,9.9\n"
        obs = run_spec({"id": "c", "adapter": "http_csv", "url": "https://x.test", "filter": {"geo": "US"},
                        "value_column": "value", "period_column": "period", "source": "test"},
                      get=fake(body))
        self.assertEqual((obs[0].value, obs[0].period), (8.7, "2026-09"))

    def test_manual_and_unknown_adapters_never_raise(self):
        self.assertEqual(run_spec({"id": "m", "adapter": "manual"})[0].status, "not_found")
        self.assertEqual(run_spec({"id": "u", "adapter": "nope"})[0].status, "error")

    def test_fetch_error_becomes_error_observation(self):
        from web_fetcher.core import FetchError

        def boom(url):
            raise FetchError("offline")
        obs = run_spec({"id": "e", "adapter": "http_json", "url": "https://x.test", "path": "a"}, get=boom)
        self.assertEqual(obs[0].status, "error")


if __name__ == "__main__":
    unittest.main()


class FakeResponse:
    def __init__(self, body: bytes, headers: dict[str, str] | None = None):
        self.body, self.headers = body, headers or {}

    def read(self) -> bytes:
        return self.body

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False


def http_error(code: int, headers: dict[str, str] | None = None):
    import email.message
    import urllib.error
    msg = email.message.Message()
    for k, v in (headers or {}).items():
        msg[k] = v
    return urllib.error.HTTPError("https://x.test", code, "err", msg, None)


class TestFetchBytes(unittest.TestCase):
    def run_with(self, responses, **kw):
        from unittest import mock

        from web_fetcher import core
        calls, sleeps = [], []

        def opener(req, timeout):
            calls.append(req.full_url)
            r = responses.pop(0)
            if isinstance(r, Exception):
                raise r
            return r
        with mock.patch.object(core, "_urlopen", opener), mock.patch.object(core, "_sleep", sleeps.append):
            try:
                return core.fetch_bytes("https://x.test/a", **kw), calls, sleeps
            except core.FetchError as exc:
                return exc, calls, sleeps

    def test_404_fails_fast(self):
        res, calls, sleeps = self.run_with([http_error(404)])
        self.assertIn("HTTP 404", str(res))
        self.assertEqual((len(calls), sleeps), (1, []))

    def test_503_honours_retry_after(self):
        res, calls, sleeps = self.run_with([http_error(503, {"Retry-After": "7"}), FakeResponse(b"ok")])
        self.assertEqual((res.body, len(calls), sleeps), (b"ok", 2, [7.0]))

    def test_connection_errors_back_off_exponentially(self):
        import urllib.error
        res, calls, sleeps = self.run_with([urllib.error.URLError("down")] * 3)
        self.assertEqual((len(calls), sleeps), (3, [1.5, 3.0]))
        self.assertIn("down", str(res))

    def test_last_modified_becomes_published(self):
        res, _, _ = self.run_with([FakeResponse(b"{}", {"Last-Modified": "Wed, 01 Oct 2026 08:00:00 GMT"})])
        self.assertEqual(res.published, "2026-10-01T08:00:00+00:00")


class TestReplay(unittest.TestCase):
    def test_replay_reproduces_cached_run(self):
        import tempfile
        from unittest import mock

        from web_fetcher import core
        body = json.dumps({"v": 3.2}).encode()
        spec = {"id": "r", "adapter": "http_json", "url": "https://x.test/r", "path": "v", "source": "test"}
        with tempfile.TemporaryDirectory() as cache:
            with mock.patch.object(core, "_urlopen", lambda req, timeout: FakeResponse(body)):
                first = run_spec(spec, cache_dir=cache)

            def no_network(req, timeout):
                raise AssertionError("network used during replay")
            with mock.patch.object(core, "_urlopen", no_network):
                again = run_spec(spec, replay_dir=cache)
                missing = run_spec({**spec, "url": "https://x.test/other"}, replay_dir=cache)
        self.assertEqual([o.to_dict() for o in again], [o.to_dict() for o in first])
        self.assertEqual(first[0].value, 3.2)
        self.assertEqual(missing[0].status, "error")
        self.assertIn("not in replay cache", missing[0].note)


class TestProvenanceRules(unittest.TestCase):
    def test_value_without_source_is_error(self):
        obs = run_spec({"id": "s", "adapter": "http_json", "url": "https://x.test", "path": "v"},
                       get=fake(b'{"v": 1}'))
        self.assertEqual(obs[0].status, "error")
        self.assertIn("source", obs[0].note)

    def csv(self, body: bytes, **extra):
        spec = {"id": "c", "adapter": "http_csv", "url": "https://x.test", "value_column": "value",
                "period_column": "period", "source": "test", **extra}
        return run_spec(spec, get=fake(body))[0]

    def test_quarters_ordered_by_value_not_string(self):
        o = self.csv(b"period,value\n2025-Q4,1\n2026Q1,2\n")
        self.assertEqual((o.status, o.period, o.value), ("ok", "2026Q1", 2.0))
        self.assertEqual(o.note, "latest of 2 periods by quarter order")

    def test_mixed_period_formats_are_error(self):
        o = self.csv(b"period,value\n2026-Q1,1\n2026-02,2\n")
        self.assertEqual(o.status, "error")
        self.assertIn("mixed period formats", o.note)

    def test_explicit_string_order_is_allowed(self):
        o = self.csv(b"period,value\nb,1\na,2\n", period_format="string")
        self.assertEqual((o.status, o.period), ("ok", "b"))

    def test_non_numeric_value_is_not_found_unless_string(self):
        body = b"period,value\n2026-01,(D)\n"
        self.assertEqual(self.csv(body).status, "not_found")
        self.assertEqual(self.csv(body, value_type="string").value, "(D)")