spb/countryatlas

1"""BIS SDMX connector — offline tests on trimmed WS_CBPOL / WS_SPP CSVs, plus live smoke tests."""2from __future__ import annotations34from datetime import date56import pytest78from countryatlas.connectors._series import parse_sdmx_period9from countryatlas.connectors.bis import BISConnector10from countryatlas.models import IndicatorSourceSpec11from tests.connectors._helpers import install_fake_get, raw_from_file121314@pytest.fixture15def fake_get(monkeypatch):16    return install_fake_get(monkeypatch)171819CBPOL =IndicatorSourceSpec(indicator_id="policy-rate", connector="bis", dataset="WS_CBPOL", code="M.", frequency="M", priority=1)20SPP_R = IndicatorSourceSpec(indicator_id="real-house-price-index", connector="bis", dataset="WS_SPP", code="Q..R.628",21                            frequency="Q", priority=1, transform="rebase:2015")22SPP_G = IndicatorSourceSpec(indicator_id="house-price-growth", connector="bis", dataset="WS_SPP", code="Q..R.771",23                            frequency="Q", priority=1)242526def test_sdmx_period_parsing():27    assert parse_sdmx_period("2026-08") == (date(2026, 8, 1), 2026, "M")28    assert parse_sdmx_period("2026-Q3") == (date(2026, 7, 1), 2026, "Q")29    assert parse_sdmx_period("2024") == (date(2024, 1, 1), 2024, "A")30    assert parse_sdmx_period("2026-13") is None313233def test_policy_rate_iso2_mapping_and_nan(fixtures_dir):34    conn = BISConnector()35    raw = raw_from_file("bis", "WS_CBPOL", "M.", fixtures_dir / "bis" / "WS_CBPOL_M.csv", "text/csv")36    rows = conn.normalize(raw, CBPOL)37    # CA/US → CAN/USA; XM (euro area) dropped; AR rows are NaN with OBS_STATUS=M → dropped38    assert {r.country_id for r in rows} == {"CAN", "USA"}39    assert all(r.frequency == "M" and r.period.day == 1 and r.unit == "% per annum" for r in rows)40    us = {r.period: r.value for r in rows if r.country_id == "USA"}41    assert us[date(2026, 8, 1)] == pytest.approx(3.625)42    assert conn.validate(rows).quarantine_dataset is False434445def test_real_index_rebased_to_2015(fixtures_dir):46    conn = BISConnector()47    raw = raw_from_file("bis", "WS_SPP", "Q..R.628", fixtures_dir / "bis" / "WS_SPP_R_628.csv", "text/csv")48    rows = conn.normalize(raw, SPP_R)49    assert {r.country_id for r in rows} == {"CAN", "USA"}50    for iso in ("CAN", "USA"):51        base = [r.value for r in rows if r.country_id == iso and r.year == 2015]52        assert len(base) == 4 and sum(base) / 4 == pytest.approx(100.0)53    q = [r for r in rows if r.country_id == "CAN"]54    assert q[0].period == date(2014, 1, 1) and q[0].frequency == "Q" and q[0].unit == "index (2015 = 100)"55    assert len(q) == 12565758def test_growth_series_taken_as_published(fixtures_dir):59    conn = BISConnector()60    raw = raw_from_file("bis", "WS_SPP", "Q..R.771", fixtures_dir / "bis" / "WS_SPP_R_771.csv", "text/csv")61    rows = conn.normalize(raw, SPP_G)62    us = {r.period: r.value for r in rows if r.country_id == "USA"}63    assert us[date(2025, 1, 1)] == pytest.approx(-0.0324)64    assert all(r.unit == "annual %" for r in rows)656667def test_rebase_drops_country_without_base_year(fixtures_dir):68    conn = BISConnector()69    text = (fixtures_dir / "bis" / "WS_SPP_R_628.csv").read_text()70    lines = [ln for ln in text.splitlines() if not (",CA," in ln and ",2015-Q" in ln)]71    raw = raw_from_file("bis", "WS_SPP", "Q..R.628", fixtures_dir / "bis" / "WS_SPP_R_628.csv", "text/csv")72    raw.body = "\n".join(lines).encode()73    rows = conn.normalize(raw, SPP_R)74    assert {r.country_id for r in rows} == {"USA"}757677def test_fetch_builds_verified_url(fixtures_dir, fake_get):78    conn = BISConnector()79    body = (fixtures_dir / "bis" / "WS_CBPOL_M.csv").read_bytes()80    calls = fake_get(conn, lambda url, params: (body, "text/csv"))81    p = conn.fetch(CBPOL)82    assert calls[0]["url"] == "https://stats.bis.org/api/v2/data/dataflow/BIS/WS_CBPOL/1.0/M."83    assert calls[0]["params"] == {"format": "csv"}84    assert p.content_type == "text/csv" and p.dataset == "WS_CBPOL" and p.code == "M."858687@pytest.mark.live88def test_live_policy_rates_all_areas():89    conn = BISConnector()90    spec = CBPOL.model_copy(update={"params": {"startPeriod": "2026-01"}})91    rows = conn.normalize(conn.fetch(spec), spec)92    isos = {r.country_id for r in rows}93    assert len(isos) >= 30 and {"CAN", "USA", "GBR", "JPN"} <= isos94    assert conn.validate(rows).quarantine_dataset is False959697@pytest.mark.live98def test_live_property_prices_real_index():99    conn = BISConnector()100    spec = SPP_R.model_copy(update={"params": {"startPeriod": "2015-Q1"}})101    rows = conn.normalize(conn.fetch(spec), spec)102    assert len({r.country_id for r in rows}) >= 50103