82 lines
3.8 KiB
Python
82 lines
3.8 KiB
Python
"""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"}, 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": "%"}, 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"}, 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()
|