45 lines
1.9 KiB
Python
45 lines
1.9 KiB
Python
|
"""Live smoke tests against real endpoints, one spec per adapter.
|
||
|
|
|
||
|
|
Skipped unless WEB_FETCHER_LIVE=1. They detect upstream shape changes that the
|
||
|
|
recorded fixtures cannot; a failure here means a source changed, not
|
||
|
|
necessarily that web-fetcher is broken.
|
||
|
|
|
||
|
|
WEB_FETCHER_LIVE=1 python3 -B -m unittest tests.test_live -v
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
from web_fetcher.registry import run_spec
|
||
|
|
|
||
|
|
SPECS = [
|
||
|
|
{"id": "live-eurostat", "adapter": "eurostat", "dataset": "isoc_eb_ai",
|
||
|
|
"filters": {"geo": ["EU27_2020"], "unit": "PC_ENT", "size_emp": "GE10", "indic_is": "E_AI_TANY"},
|
||
|
|
"split_by": ["geo"]},
|
||
|
|
{"id": "live-json", "adapter": "http_json",
|
||
|
|
"url": "https://api.worldbank.org/v2/country/DEU/indicator/SE.XPD.TOTL.GD.ZS?format=json&mrnev=1",
|
||
|
|
"path": "1.0.value", "period_path": "1.0.date", "source": "World Bank WDI (SE.XPD.TOTL.GD.ZS)"},
|
||
|
|
{"id": "live-csv", "adapter": "http_csv", "url": "https://fred.stlouisfed.org/graph/fredgraph.csv?id=OPHNFB",
|
||
|
|
"value_column": "OPHNFB", "period_column": "observation_date", "source": "BLS via FRED (OPHNFB)"},
|
||
|
|
{"id": "live-xlsx", "adapter": "xlsx_wide", "url": "https://www.census.gov/hfp/btos/downloads/National.xlsx",
|
||
|
|
"sheet": "Response Estimates", "filter": {"Question ID": "7", "Answer ID": "1"},
|
||
|
|
"source": "U.S. Census Bureau BTOS"},
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
@unittest.skipUnless(os.environ.get("WEB_FETCHER_LIVE") == "1", "set WEB_FETCHER_LIVE=1 for live tests")
|
||
|
|
class TestLive(unittest.TestCase):
|
||
|
|
def test_each_adapter_returns_a_sourced_value(self):
|
||
|
|
for spec in SPECS:
|
||
|
|
with self.subTest(spec["id"]):
|
||
|
|
obs = run_spec(spec)
|
||
|
|
self.assertTrue(obs, "no observations")
|
||
|
|
o = obs[0]
|
||
|
|
self.assertEqual(o.status, "ok", o.note)
|
||
|
|
self.assertIsNotNone(o.value)
|
||
|
|
self.assertTrue(o.source and o.url and o.retrieved_at and o.payload_sha256)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|