"""Eurostat JSON-stat connector — recorded fixtures (une_rt_a with EL + EU27_2020; rd_e_gerdtot with `p` flags).""" from __future__ import annotations import json from datetime import date import pytest from countryatlas.connectors._util import ConnectorError from countryatlas.connectors.eurostat import EurostatConnector, decode_jsonstat from countryatlas.models import IndicatorSourceSpec, RawPayload def _raw(fixtures_dir, name: str, dataset: str, code: str, retrieved_at) -> RawPayload: body = (fixtures_dir / "eurostat" / name).read_bytes() doc = json.loads(body) return RawPayload(connector="eurostat", dataset=dataset, code=code, url="fixture://eurostat", retrieved_at=retrieved_at, status_code=200, content_type="application/json", body=body, meta={"updated": doc.get("updated")}) @pytest.fixture def es() -> EurostatConnector: return EurostatConnector() def test_decoder_sparse_row_major(): doc = { "id": ["freq", "geo", "time"], "size": [1, 2, 3], "dimension": { "freq": {"category": {"index": {"A": 0}}}, "geo": {"category": {"index": {"DE": 0, "EL": 1}}}, "time": {"category": {"index": {"2022": 0, "2023": 1, "2024": 2}}}, }, "value": {"0": 1.0, "2": 3.0, "4": 5.0}, "status": {"4": "p"}, } cells = decode_jsonstat(doc) assert [(c["dims"]["geo"], c["dims"]["time"], c["value"], c["status"]) for c in cells] == [ ("DE", "2022", 1.0, None), ("DE", "2024", 3.0, None), ("EL", "2023", 5.0, "p"), ] def test_unemployment_maps_el_and_drops_eu_aggregate(es, fixtures_dir, retrieved_at): spec = IndicatorSourceSpec(indicator_id="unemployment-rate", connector="eurostat", dataset="une_rt_a", code="Y15-74.PC_ACT.T", params={"age": "Y15-74", "unit": "PC_ACT", "sex": "T"}, priority=3) rows = es.normalize(_raw(fixtures_dir, "une_rt_a_sample.json", "une_rt_a", spec.code, retrieved_at), spec) assert {r.country_id for r in rows} == {"DEU", "GRC"}, "EL → GRC, EU27_2020 dropped" grc = sorted((r for r in rows if r.country_id == "GRC"), key=lambda r: r.year) assert [r.year for r in grc] == [2022, 2023, 2024, 2025] assert grc[0].period == date(2022, 1, 1) and grc[0].frequency == "A" assert grc[0].value == pytest.approx(12.5) assert grc[0].unit == "% of labour force" assert grc[0].source_updated_at is not None and grc[0].source_updated_at.year == 2026 assert grc[0].metadata["filters"] == {"age": "Y15-74", "unit": "PC_ACT", "sex": "T"} assert es.validate(rows).errors == 0 def test_status_flags_mark_estimates(es, fixtures_dir, retrieved_at): spec = IndicatorSourceSpec(indicator_id="rd-expenditure-pct-gdp", connector="eurostat", dataset="rd_e_gerdtot", code="TOTAL.PC_GDP", params={"sectperf": "TOTAL", "unit": "PC_GDP"}, priority=3) rows = es.normalize(_raw(fixtures_dir, "rd_e_gerdtot_sample.json", "rd_e_gerdtot", spec.code, retrieved_at), spec) assert {r.country_id for r in rows} == {"DEU", "FRA"} flagged = [r for r in rows if r.is_estimate] assert len(flagged) == 2 and all(r.year == 2024 for r in flagged) and all(r.metadata["flags"] == "p" for r in flagged) assert not any(r.is_forecast for r in rows) deu = {r.year: r.value for r in rows if r.country_id == "DEU"} assert deu == {2021: pytest.approx(3.07), 2022: pytest.approx(3.04), 2023: pytest.approx(3.13), 2024: pytest.approx(3.13)} def test_forecast_flag_and_quarterly_time(es, retrieved_at): doc = { "id": ["freq", "unit", "geo", "time"], "size": [1, 1, 2, 2], "updated": "2026-07-02T11:00:00+0200", "dimension": { "freq": {"category": {"index": {"Q": 0}}}, "unit": {"category": {"index": {"PC": 0}}}, "geo": {"category": {"index": {"UK": 0, "FR": 1}}}, "time": {"category": {"index": {"2025-Q4": 0, "2026-Q1": 1}}}, }, "value": {"0": 1.5, "1": 1.7, "2": 2.0, "3": 2.2}, "status": {"3": "f", "1": "bp"}, } raw = RawPayload(connector="eurostat", dataset="x_q", code="PC", url="fixture://", retrieved_at=retrieved_at, status_code=200, body=json.dumps(doc).encode()) spec = IndicatorSourceSpec(indicator_id="inflation", connector="eurostat", dataset="x_q", code="PC", params={"unit": "PC"}) rows = es.normalize(raw, spec) by = {(r.country_id, r.period): r for r in rows} assert set(by) == {("GBR", date(2025, 10, 1)), ("GBR", date(2026, 1, 1)), ("FRA", date(2025, 10, 1)), ("FRA", date(2026, 1, 1))} assert all(r.frequency == "Q" for r in rows) assert by[("FRA", date(2026, 1, 1))].is_forecast is True assert by[("GBR", date(2026, 1, 1))].is_estimate is True and by[("GBR", date(2026, 1, 1))].metadata["flags"] == "bp" def test_ambiguous_dimension_raises(es, fixtures_dir, retrieved_at): body = json.loads((fixtures_dir / "eurostat" / "une_rt_a_sample.json").read_text()) # pretend the sex dimension was not filtered: 2 categories with the same cell count body["id"] = ["freq", "age", "unit", "sex", "geo", "time"] body["size"] = [1, 1, 1, 2, 3, len(body["dimension"]["time"]["category"]["index"])] body["dimension"]["sex"]["category"]["index"] = {"T": 0, "M": 1} raw = RawPayload(connector="eurostat", dataset="une_rt_a", code="x", url="fixture://", retrieved_at=retrieved_at, status_code=200, body=json.dumps(body).encode()) spec = IndicatorSourceSpec(indicator_id="unemployment-rate", connector="eurostat", dataset="une_rt_a", code="x") with pytest.raises(ConnectorError, match="sex="): es.normalize(raw, spec) def test_registry_specs_for_eurostat(): from countryatlas import registry specs = registry.source_specs("eurostat") by_ind = {s.indicator_id: s for s in specs} assert {"employment-rate", "median-household-income", "at-risk-of-poverty-rate", "homeownership-rate", "housing-cost-overburden-rate", "inflation"} <= set(by_ind) for s in specs: assert s.dataset and s.params, s.indicator_id assert "geo" not in s.params and "time" not in s.params @pytest.mark.live def test_live_small_query(): es = EurostatConnector() spec = IndicatorSourceSpec(indicator_id="unemployment-rate", connector="eurostat", dataset="une_rt_a", code="Y15-74.PC_ACT.T", params={"age": "Y15-74", "unit": "PC_ACT", "sex": "T", "geo": ["DE", "EL"], "sinceTimePeriod": 2023}) raw = es.fetch(spec) rows = es.normalize(raw, spec) assert {r.country_id for r in rows} == {"DEU", "GRC"} es.close()