spb/countryatlas

1"""Eurostat JSON-stat connector — recorded fixtures (une_rt_a with EL + EU27_2020; rd_e_gerdtot with `p` flags)."""2from __future__ import annotations34import json5from datetime import date67import pytest89from countryatlas.connectors._util import ConnectorError10from countryatlas.connectors.eurostat import EurostatConnector, decode_jsonstat11from countryatlas.models import IndicatorSourceSpec, RawPayload121314def _raw(fixtures_dir, name: str, dataset: str, code: str, retrieved_at) -> RawPayload:15    body = (fixtures_dir / "eurostat" / name).read_bytes()16    doc = json.loads(body)17    return RawPayload(connector="eurostat", dataset=dataset, code=code, url="fixture://eurostat", retrieved_at=retrieved_at,18                      status_code=200, content_type="application/json", body=body, meta={"updated": doc.get("updated")})192021@pytest.fixture22def es() -> EurostatConnector:23    return EurostatConnector()242526def test_decoder_sparse_row_major():27    doc = {28        "id": ["freq", "geo", "time"], "size": [1, 2, 3],29        "dimension": {30            "freq": {"category": {"index": {"A": 0}}},31            "geo": {"category": {"index": {"DE": 0, "EL": 1}}},32            "time": {"category": {"index": {"2022": 0, "2023": 1, "2024": 2}}},33        },34        "value": {"0": 1.0, "2": 3.0, "4": 5.0},35        "status": {"4": "p"},36    }37    cells = decode_jsonstat(doc)38    assert [(c["dims"]["geo"], c["dims"]["time"], c["value"], c["status"]) for c in cells] == [39        ("DE", "2022", 1.0, None), ("DE", "2024", 3.0, None), ("EL", "2023", 5.0, "p"),40    ]414243def test_unemployment_maps_el_and_drops_eu_aggregate(es, fixtures_dir, retrieved_at):44    spec = IndicatorSourceSpec(indicator_id="unemployment-rate", connector="eurostat", dataset="une_rt_a", code="Y15-74.PC_ACT.T",45                               params={"age": "Y15-74", "unit": "PC_ACT", "sex": "T"}, priority=3)46    rows = es.normalize(_raw(fixtures_dir, "une_rt_a_sample.json", "une_rt_a", spec.code, retrieved_at), spec)47    assert {r.country_id for r in rows} == {"DEU", "GRC"}, "EL → GRC, EU27_2020 dropped"48    grc = sorted((r for r in rows if r.country_id == "GRC"), key=lambda r: r.year)49    assert [r.year for r in grc] == [2022, 2023, 2024, 2025]50    assert grc[0].period == date(2022, 1, 1) and grc[0].frequency == "A"51    assert grc[0].value == pytest.approx(12.5)52    assert grc[0].unit == "% of labour force"53    assert grc[0].source_updated_at is not None and grc[0].source_updated_at.year == 202654    assert grc[0].metadata["filters"] == {"age": "Y15-74", "unit": "PC_ACT", "sex": "T"}55    assert es.validate(rows).errors == 0565758def test_status_flags_mark_estimates(es, fixtures_dir, retrieved_at):59    spec = IndicatorSourceSpec(indicator_id="rd-expenditure-pct-gdp", connector="eurostat", dataset="rd_e_gerdtot",60                               code="TOTAL.PC_GDP", params={"sectperf": "TOTAL", "unit": "PC_GDP"}, priority=3)61    rows = es.normalize(_raw(fixtures_dir, "rd_e_gerdtot_sample.json", "rd_e_gerdtot", spec.code, retrieved_at), spec)62    assert {r.country_id for r in rows} == {"DEU", "FRA"}63    flagged = [r for r in rows if r.is_estimate]64    assert len(flagged) == 2 and all(r.year == 2024 for r in flagged) and all(r.metadata["flags"] == "p" for r in flagged)65    assert not any(r.is_forecast for r in rows)66    deu = {r.year: r.value for r in rows if r.country_id == "DEU"}67    assert deu == {2021: pytest.approx(3.07), 2022: pytest.approx(3.04), 2023: pytest.approx(3.13), 2024: pytest.approx(3.13)}686970def test_forecast_flag_and_quarterly_time(es, retrieved_at):71    doc = {72        "id": ["freq", "unit", "geo", "time"], "size": [1, 1, 2, 2], "updated": "2026-07-02T11:00:00+0200",73        "dimension": {74            "freq": {"category": {"index": {"Q": 0}}},75            "unit": {"category": {"index": {"PC": 0}}},76            "geo": {"category": {"index": {"UK": 0, "FR": 1}}},77            "time": {"category": {"index": {"2025-Q4": 0, "2026-Q1": 1}}},78        },79        "value": {"0": 1.5, "1": 1.7, "2": 2.0, "3": 2.2},80        "status": {"3": "f", "1": "bp"},81    }82    raw = RawPayload(connector="eurostat", dataset="x_q", code="PC", url="fixture://", retrieved_at=retrieved_at,83                     status_code=200, body=json.dumps(doc).encode())84    spec = IndicatorSourceSpec(indicator_id="inflation", connector="eurostat", dataset="x_q", code="PC", params={"unit": "PC"})85    rows = es.normalize(raw, spec)86    by = {(r.country_id, r.period): r for r in rows}87    assert set(by) == {("GBR", date(2025, 10, 1)), ("GBR", date(2026, 1, 1)), ("FRA", date(2025, 10, 1)), ("FRA", date(2026, 1, 1))}88    assert all(r.frequency == "Q" for r in rows)89    assert by[("FRA", date(2026, 1, 1))].is_forecast is True90    assert by[("GBR", date(2026, 1, 1))].is_estimate is True and by[("GBR", date(2026, 1, 1))].metadata["flags"] == "bp"919293def test_ambiguous_dimension_raises(es, fixtures_dir, retrieved_at):94    body = json.loads((fixtures_dir / "eurostat" / "une_rt_a_sample.json").read_text())95    # pretend the sex dimension was not filtered: 2 categories with the same cell count96    body["id"] = ["freq", "age", "unit", "sex", "geo", "time"]97    body["size"] = [1, 1, 1, 2, 3, len(body["dimension"]["time"]["category"]["index"])]98    body["dimension"]["sex"]["category"]["index"] = {"T": 0, "M": 1}99    raw = RawPayload(connector="eurostat", dataset="une_rt_a", code="x", url="fixture://", retrieved_at=retrieved_at,100                     status_code=200, body=json.dumps(body).encode())101    spec = IndicatorSourceSpec(indicator_id="unemployment-rate", connector="eurostat", dataset="une_rt_a", code="x")102    with pytest.raises(ConnectorError, match="sex="):103        es.normalize(raw, spec)104105106def test_registry_specs_for_eurostat():107    from countryatlas import registry108109    specs = registry.source_specs("eurostat")110    by_ind = {s.indicator_id: s for s in specs}111    assert {"employment-rate", "median-household-income", "at-risk-of-poverty-rate", "homeownership-rate",112            "housing-cost-overburden-rate", "inflation"} <= set(by_ind)113    for s in specs:114        assert s.dataset and s.params, s.indicator_id115        assert "geo" not in s.params and "time" not in s.params116117118@pytest.mark.live119def test_live_small_query():120    es = EurostatConnector()121    spec = IndicatorSourceSpec(indicator_id="unemployment-rate", connector="eurostat", dataset="une_rt_a", code="Y15-74.PC_ACT.T",122                               params={"age": "Y15-74", "unit": "PC_ACT", "sex": "T", "geo": ["DE", "EL"], "sinceTimePeriod": 2023})123    raw = es.fetch(spec)124    rows = es.normalize(raw, spec)125    assert {r.country_id for r in rows} == {"DEU", "GRC"}126    es.close()127