spb/countryatlas

1"""IMF WEO connector — recorded SDMX-CSV fixtures (CAN, USA, KOS, G001 for NGDP_RPCH; CAN, USA for NGDPD)."""2from __future__ import annotations34from datetime import date56import pytest78from countryatlas.connectors.imf import IMFConnector9from countryatlas.models import IndicatorSourceSpec, RawPayload101112def _raw(fixtures_dir, name: str, code: str, retrieved_at) -> RawPayload:13    body = (fixtures_dir / "imf" / name).read_bytes()14    return RawPayload(15        connector="imf", dataset="WEO", code=code, url="fixture://imf", retrieved_at=retrieved_at, status_code=200,16        content_type="text/csv", body=body, meta={"publication_date": "2026-04-14T13:00:00Z"},17    )181920@pytest.fixture21def imf() -> IMFConnector:22    return IMFConnector()232425def test_normalize_growth_maps_iso3_and_flags_forecasts(imf, fixtures_dir, retrieved_at):26    spec = IndicatorSourceSpec(indicator_id="gdp-growth", connector="imf", dataset="WEO", code="NGDP_RPCH", priority=2)27    rows = imf.normalize(_raw(fixtures_dir, "weo_NGDP_RPCH_sample.csv", "NGDP_RPCH", retrieved_at), spec)28    countries = {r.country_id for r in rows}29    assert countries == {"CAN", "USA", "XKX"}, "G001 (World aggregate) dropped, KOS mapped to XKX"30    can = sorted((r for r in rows if r.country_id == "CAN"), key=lambda r: r.year)31    assert [r.year for r in can] == list(range(2022, 2029))32    assert can[0].period == date(2022, 1, 1) and can[0].frequency == "A"33    # LATEST_ACTUAL_ANNUAL_DATA = 2025 → 2026+ are projections34    assert {r.year: r.is_forecast for r in can} == {2022: False, 2023: False, 2024: False, 2025: False, 2026: True,35                                                    2027: True, 2028: True}36    assert all(r.is_estimate is False for r in rows)37    assert can[0].value == pytest.approx(4.69542)38    assert can[0].unit == "annual %"39    assert can[0].source_updated_at is not None and can[0].source_updated_at.date() == date(2026, 4, 14)40    assert can[0].metadata["latest_actual_annual_data"] == 202541    assert can[0].source_dataset == "WEO" and can[0].source_series_code == "NGDP_RPCH"424344def test_normalize_gdp_values_are_base_units_without_transform(imf, fixtures_dir, retrieved_at):45    spec = IndicatorSourceSpec(indicator_id="gdp", connector="imf", dataset="WEO", code="NGDPD", priority=2)46    rows = imf.normalize(_raw(fixtures_dir, "weo_NGDPD_sample.csv", "NGDPD", retrieved_at), spec)47    can2023 = next(r for r in rows if r.country_id == "CAN" and r.year == 2023)48    assert can2023.value == pytest.approx(2.196593836e12, rel=1e-6), "OBS_VALUE already in US$, SCALE=9 is display-only"49    assert can2023.metadata.get("scale") == "9"50    assert can2023.unit == "current US$"515253def test_transform_is_applied_when_present(imf, fixtures_dir, retrieved_at):54    spec = IndicatorSourceSpec(indicator_id="gdp", connector="imf", dataset="WEO", code="NGDPD", priority=2, transform="x/1e9")55    rows = imf.normalize(_raw(fixtures_dir, "weo_NGDPD_sample.csv", "NGDPD", retrieved_at), spec)56    can2023 = next(r for r in rows if r.country_id == "CAN" and r.year == 2023)57    assert can2023.value == pytest.approx(2196.593836, rel=1e-6)585960def test_validate_no_duplicates(imf, fixtures_dir, retrieved_at):61    spec = IndicatorSourceSpec(indicator_id="gdp-growth", connector="imf", dataset="WEO", code="NGDP_RPCH", priority=2)62    rows = imf.normalize(_raw(fixtures_dir, "weo_NGDP_RPCH_sample.csv", "NGDP_RPCH", retrieved_at), spec)63    report = imf.validate(rows)64    assert report.errors == 0 and not report.quarantine_dataset656667def test_registry_specs_for_imf_include_inline_and_extra():68    from countryatlas import registry6970    specs = registry.source_specs("imf")71    codes = {s.code for s in specs}72    assert {"NGDP_RPCH", "PCPIPCH", "LUR", "GGXWDG_NGDP", "NGSD_NGDP", "NID_NGDP", "LP"} <= codes73    for s in specs:74        assert s.dataset == "WEO"75        assert s.transform is None or "1e9" not in s.transform, f"{s.code}: OBS_VALUE is already in base units"767778@pytest.mark.live79def test_live_fetch_small_key():80    imf = IMFConnector()81    spec = IndicatorSourceSpec(indicator_id="unemployment-rate", connector="imf", dataset="WEO", code="LUR", priority=2,82                               params={"countries": ["CAN", "USA"], "startPeriod": 2023, "endPeriod": 2027})83    raw = imf.fetch(spec)84    assert raw.status_code == 200 and raw.body.startswith(b"DATAFLOW")85    rows = imf.normalize(raw, spec)86    assert {r.country_id for r in rows} == {"CAN", "USA"}87    assert any(r.is_forecast for r in rows) and any(not r.is_forecast for r in rows)88    imf.close()89