Dave Boyd
9e2c55c568
Item 1 — QCEW county/industry detail: - new bls_client/qcew.py: thin client over the QCEW Open Data CSV service (area/industry/size slices; no API key, no quota), with filter_rows helper - covers the county/industry detail the timeseries API can't reach Item 2 — ECEC benefit breakdown: - new series.ecec() builder from the authoritative cm.estimate decode table - real helpers: wages, total benefits, paid leave, supplemental pay, health insurance, retirement & savings, legally required + ecec_dashboard() - FIX: ecec_total_benefits was CMU1036... (education/health industries only); correct all-civilian total benefits is CMU1030000000000D Item 3 — caching + typed errors: - bls_client/cache.py FileCache (on-disk, TTL); BLSClient(cache=True), queries_used counter (cache hits don't spend quota) - bls_client/errors.py: BLSError / BLSQuotaError / BLSRequestError; quota exhaustion now raises a clear BLSQuotaError instead of a generic RuntimeError Tests: 119 offline (+24) incl. ECEC goldens, QCEW fixture parse, cache/quota behavior; live smoke extended to ECEC + QCEW. Helper sweep: 96/96 live. README: caching/QCEW usage, updated coverage + limitations. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
"""Tests for the QCEW Open Data CSV module.
|
|
|
|
Offline tests parse a committed fixture (no network). The live test (opt-in via
|
|
`-m live`) hits the real CSV service, which needs no API key and no quota.
|
|
"""
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from bls_client import qcew
|
|
|
|
FIXTURE = os.path.join(os.path.dirname(__file__), "fixtures", "qcew_dc_sample.csv")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def dc_rows():
|
|
with open(FIXTURE) as f:
|
|
return qcew.parse_csv(f.read())
|
|
|
|
|
|
def test_parse_returns_dict_rows(dc_rows):
|
|
assert dc_rows and isinstance(dc_rows[0], dict)
|
|
# core QCEW columns present
|
|
for col in ("area_fips", "own_code", "industry_code", "annual_avg_emplvl", "annual_avg_wkly_wage"):
|
|
assert col in dc_rows[0]
|
|
|
|
|
|
def test_all_rows_are_dc_all_industries(dc_rows):
|
|
assert all(r["area_fips"] == "11000" for r in dc_rows)
|
|
assert all(r["industry_code"] == "10" for r in dc_rows)
|
|
|
|
|
|
def test_filter_by_ownership(dc_rows):
|
|
# ownership name resolves to its code, and total (own_code 0) exists
|
|
total = qcew.filter_rows(dc_rows, own_code="total")
|
|
assert len(total) == 1
|
|
assert int(total[0]["annual_avg_emplvl"]) > 0
|
|
|
|
|
|
def test_ownership_name_mapping():
|
|
assert qcew.OWNERSHIP["private"] == "5"
|
|
assert qcew.OWNERSHIP["federal"] == "1"
|
|
|
|
|
|
@pytest.mark.live
|
|
def test_live_area_fetch():
|
|
rows = qcew.area("11000", 2024, "a") # DC annual, no key / no quota
|
|
assert len(rows) > 100
|
|
private_total = qcew.filter_rows(rows, own_code="private", industry_code="10", agglvl_code="51")
|
|
assert len(private_total) == 1
|
|
assert int(private_total[0]["annual_avg_emplvl"]) > 100_000 # DC private ~525k
|