Dave Boyd
881d62cf32
- pyproject.toml: setuptools build, deps (requests), dev extra (pytest), pytest config with opt-in `live` marker - tests/: 95 offline tests locking every series-ID encoding against known-good IDs (regression guard for the JOLTS/OES/ECI/productivity/QCEW fixes), plus a live API smoke test behind `-m live` - README: install-as-package + test instructions; drop the now-resolved "no packaging"/"no tests" limitations - gitignore build/test artifacts Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
"""Live smoke test — hits the real BLS API.
|
|
|
|
Opt-in only: deselected by default (see addopts in pyproject.toml). Run with:
|
|
pytest -m live
|
|
Requires network and a BLS API key (BLS_API_KEY env var, or config.py).
|
|
|
|
Guards against silent series-ID drift: BLS occasionally retires codes (e.g. the
|
|
2010->2018 SOC change), which unit tests alone can't catch.
|
|
"""
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from bls_client import BLSClient
|
|
from bls_client.queries import employment as E, prices as P, wages as W, productivity as PR
|
|
|
|
|
|
def _api_key():
|
|
key = os.environ.get("BLS_API_KEY")
|
|
if key:
|
|
return key
|
|
try:
|
|
from config import BLS_API_KEY # local, gitignored
|
|
if BLS_API_KEY and BLS_API_KEY != "YOUR_KEY_HERE":
|
|
return BLS_API_KEY
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
pytestmark = pytest.mark.live
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
key = _api_key()
|
|
if not key:
|
|
pytest.skip("no BLS_API_KEY available")
|
|
return BLSClient(key)
|
|
|
|
|
|
# One representative, previously-broken helper per repaired survey.
|
|
REPAIRED = {
|
|
"JOLTS": E.jolts_dashboard(), # dict of 5
|
|
"OES wage": W.occupation_annual_median_wage("151252"),
|
|
"OES employ": W.all_occupations_employment(),
|
|
"ECI": W.eci_total_compensation(),
|
|
"Productivity": PR.business_output_per_hour(),
|
|
"QCEW": E.qcew_national_private(),
|
|
"PPI final": P.ppi_final_demand(),
|
|
"ECEC": W.ecec_total_benefits(),
|
|
}
|
|
|
|
|
|
def _flatten(v):
|
|
return list(v.values()) if isinstance(v, dict) else [v]
|
|
|
|
|
|
def test_repaired_surveys_return_data(client):
|
|
"""Every repaired survey must return at least one real observation."""
|
|
all_ids = []
|
|
for ids in REPAIRED.values():
|
|
all_ids.extend(_flatten(ids))
|
|
res = client.fetch(all_ids, 2024, 2025)
|
|
missing = [sid for sid in all_ids if not res.get(sid, {}).get("data")]
|
|
assert not missing, f"no data returned for: {missing}"
|
|
|
|
|
|
def test_headline_payrolls_plausible(client):
|
|
res = client.fetch_latest(E.nonfarm_payrolls(), years=1)
|
|
obs = client.latest_obs(next(iter(res.values())))
|
|
assert obs is not None
|
|
# US nonfarm payrolls are ~150,000 (thousands); sanity bound, not exact.
|
|
assert 100_000 < float(obs["value"]) < 200_000
|