tegwick
96fe5b022b
- --replay DIR / replay_dir: offline replay from the cache, hash-verified - ok values without source become error; Last-Modified recorded as published - latest period by parsed order (web_fetcher/periods.py); mixed formats refused - non-numeric values not_found unless value_type: string - retry only connection errors, timeouts, 429/5xx; honour Retry-After - opt-in live smoke test; 18 offline tests - CHANGELOG, README install docs, SCOPE update, capability D2/A3/C2/R2 - WEBFETCH-WP-0003 (repo hygiene) added Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 153021@bnt-lap001 Assistant-Session: 46917be4-6400-4106-b8b7-61953a0caf24
44 lines
1.9 KiB
Python
44 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()
|