stack/tests/bls/test_ces.py

"""bls.ces — Current Employment Statistics health-care series via the BLS API (#695)."""

from __future__ import annotations

import json

import duckdb
import httpx
import pytest

from bib.store import Store
from bls.ces import (
    SERIES,
    CesRow,
    cite,
    fetch,
    read_series,
    to_rows,
    write,
)


def _payload(series_ids, years, *, message=None):
    return {
        "status": "REQUEST_SUCCEEDED",
        "responseTime": 1,
        "message": message or [],
        "Results": {
            "series": [
                {
                    "seriesID": sid,
                    "data": [
                        {
                            "year": str(y),
                            "period": f"M{m:02d}",
                            "periodName": "x",
                            "value": f"{y}.{m}",
                            "footnotes": [{}],
                        }
                        for y in years
                        for m in (12, 1)
                    ],
                }
                for sid in series_ids
            ]
        },
    }


class TestRegistry:
    def test_series_cover_health_care_employment_and_earnings(self):
        assert "CES6562000001" in SERIES and "CES6562110001" in SERIES
        assert SERIES["CES6562110001"].industry_name.startswith("Offices of physicians")
        assert SERIES["CES6562000003"].data_type == "avg_hourly_earnings"
        assert all(s.seasonal == "S" for s in SERIES.values())


class TestFetch:
    def test_chunks_years_and_series_and_posts_json(self):
        calls = []

        def handler(request):
            body = json.loads(request.content)
            calls.append(body)
            return httpx.Response(
                200,
                json=_payload(
                    body["seriesid"],
                    range(int(body["startyear"]), int(body["endyear"]) + 1),
                ),
            )

        client = httpx.Client(transport=httpx.MockTransport(handler))
        ids = [f"CES{i:010d}" for i in range(30)]
        out = fetch(client, ids, 2006, 2025)
        # unkeyed: 10 years and 25 series per request → 2 year windows × 2 series chunks
        assert len(calls) == 4
        assert {(c["startyear"], c["endyear"]) for c in calls} == {
            ("2006", "2015"),
            ("2016", "2025"),
        }
        assert max(len(c["seriesid"]) for c in calls) == 25
        assert "registrationkey" not in calls[0]
        assert len(out) == 30 and all(len(v) == 40 for v in out.values())

    def test_key_widens_the_windows(self):
        calls = []

        def handler(request):
            body = json.loads(request.content)
            calls.append(body)
            return httpx.Response(
                200,
                json=_payload(
                    body["seriesid"],
                    range(int(body["startyear"]), int(body["endyear"]) + 1),
                ),
            )

        client = httpx.Client(transport=httpx.MockTransport(handler))
        fetch(client, ["CES6562000001"], 2006, 2025, api_key="k")
        assert len(calls) == 1 and calls[0]["registrationkey"] == "k"

    def test_api_failure_status_raises(self):
        client = httpx.Client(
            transport=httpx.MockTransport(
                lambda r: httpx.Response(
                    200,
                    json={
                        "status": "REQUEST_NOT_PROCESSED",
                        "message": ["daily threshold"],
                        "Results": {},
                    },
                )
            )
        )
        with pytest.raises(RuntimeError, match="daily threshold"):
            fetch(client, ["CES6562000001"], 2016, 2025)

    def test_http_error_raises(self):
        client = httpx.Client(
            transport=httpx.MockTransport(lambda r: httpx.Response(500))
        )
        with pytest.raises(httpx.HTTPStatusError):
            fetch(client, ["CES6562000001"], 2016, 2025)


class TestToRows:
    def test_rows_carry_series_metadata(self):
        data = {
            "CES6562110001": [
                {
                    "year": "2024",
                    "period": "M03",
                    "periodName": "March",
                    "value": "3010.5",
                    "footnotes": [{}],
                }
            ]
        }
        rows = to_rows(data, item_keys={"CES6562110001": "K"})
        assert len(rows) == 1
        r = rows[0]
        assert isinstance(r, CesRow)
        assert (r.series_id, r.year, r.month, r.value) == (
            "CES6562110001",
            2024,
            3,
            3010.5,
        )
        assert (
            r.industry_code == "621100"
            and r.data_type == "employment"
            and r.item_key == "K"
        )

    def test_annual_average_and_blank_values(self):
        data = {
            "CES6562000001": [
                {"year": "2024", "period": "M13", "value": "-"},
                {"year": "2024", "period": "M01", "value": "1"},
            ]
        }
        rows = to_rows(data, item_keys={})
        assert [(r.month, r.value) for r in rows] == [(13, None), (1, 1.0)]

    def test_unknown_series_still_loads(self):
        rows = to_rows(
            {"CES9999999999": [{"year": "2024", "period": "M01", "value": "2"}]},
            item_keys={},
        )
        assert rows[0].industry_code == "" and rows[0].data_type == ""


@pytest.fixture
def con():
    c = duckdb.connect(":memory:")
    yield c
    c.close()


class TestWriteRead:
    def test_write_replaces_per_series(self, con):
        rows = to_rows(
            {
                "CES6562110001": [
                    {"year": "2024", "period": "M01", "value": "1"},
                    {"year": "2024", "period": "M02", "value": "2"},
                ],
                "CES6562000001": [{"year": "2024", "period": "M01", "value": "9"}],
            },
            item_keys={},
        )
        assert write(con, rows) == 3
        assert write(con, rows) == 3
        assert con.execute("SELECT count(*) FROM bls.ces").fetchone()[0] == 3
        out = read_series(con, ["CES6562110001"])
        assert [(r.year, r.month, r.value) for r in out] == [
            (2024, 1, 1.0),
            (2024, 2, 2.0),
        ]
        assert read_series(con, ["CES6562110001"], annual=True)[0].value == 1.5


class TestCite:
    def test_one_source_per_series(self, tmp_path):
        s = Store(":memory:", storage_dir=tmp_path / "st")
        k1, k2 = cite(s, "CES6562110001"), cite(s, "CES6562110001")
        assert k1 == k2
        item = s.get(k1)
        assert "CES6562110001" in item.title and item.url.endswith("CES6562110001")
        assert {"module:bls", "table:bls.ces", "source:bls-website"} <= set(item.tags)
        s.close()