spb/countryatlas

1"""Pipeline reliability: HTTP retries/backoff honouring Retry-After, country mapping, never publishing a shrunken snapshot,2source-health meta and build logs."""3from __future__ import annotations45import json67import duckdb8import httpx9import pytest10import respx1112from countryatlas.config import settings13from countryatlas.connectors import base as base_mod14from countryatlas.connectors.base import Connector, retry_after_seconds15from countryatlas.pipeline import build as build_mod16from countryatlas.registry import lookup17from tests.test_build import _tiny_staging181920class _Dummy(Connector):21    id = "worldbank"22    name = "dummy"23    rate_per_minute = 100_0002425    def fetch(self, spec):  # pragma: no cover - not used26        raise NotImplementedError2728    def normalize(self, raw, spec):  # pragma: no cover - not used29        return []303132def test_retry_after_parsing() -> None:33    assert retry_after_seconds(None) is None and retry_after_seconds("abc") is None34    assert retry_after_seconds("7") == 7.035    assert retry_after_seconds("999") == base_mod.RETRY_AFTER_CAP36    assert retry_after_seconds("Wed, 21 Oct 2015 07:28:00 GMT") == 0.0  # in the past → no wait373839@respx.mock40def test_get_retries_on_transient_statuses_then_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:41    sleeps: list[float] = []42    monkeypatch.setattr(base_mod.time, "sleep", lambda s: sleeps.append(s))  # Retry-After sleep43    # tenacity's wait: make it instant but record the computed deterministic delays44    waits: list[float] = []45    orig_wait = _Dummy.get.retry.wait4647    def fake_wait(retry_state):48        w = orig_wait(retry_state)49        waits.append(w)50        return 05152    monkeypatch.setattr(_Dummy.get.retry, "wait", fake_wait)53    route = respx.get("https://example.test/x").mock(side_effect=[54        httpx.Response(429, headers={"Retry-After": "3"}),55        httpx.Response(503),56        httpx.Response(200, json={"ok": True}),57    ])58    c = _Dummy()59    r = c.get("https://example.test/x")60    assert r.status_code == 200 and route.call_count == 361    assert [s for s in sleeps if s] == [3.0]  # Retry-After honoured once (zero-length tenacity naps filtered)62    assert waits == [2.0, 4.0]  # deterministic exponential backoff, no jitter63    c.close()646566@respx.mock67def test_get_gives_up_after_five_attempts(monkeypatch: pytest.MonkeyPatch) -> None:68    monkeypatch.setattr(_Dummy.get.retry, "wait", lambda rs: 0)69    route = respx.get("https://example.test/down").mock(return_value=httpx.Response(502))70    c = _Dummy()71    with pytest.raises(base_mod.TransientHTTPError):72        c.get("https://example.test/down")73    assert route.call_count == base_mod.RETRY_ATTEMPTS74    c.close()75    # a 404 is not retried76    route2 = respx.get("https://example.test/missing").mock(return_value=httpx.Response(404))77    c2 = _Dummy()78    with pytest.raises(httpx.HTTPStatusError):79        c2.get("https://example.test/missing")80    assert route2.call_count == 181    c2.close()828384def test_country_mapping_drops_aggregates_and_unknowns() -> None:85    lk = lookup()86    assert lk.any("CAN") == "CAN" and lk.any("ca") == "CAN" and lk.any("124") == "CAN"87    assert lk.any("Viet Nam") == "VNM" and lk.any("Türkiye") == "TUR" and lk.any("Kosovo") == "XKX"88    for agg in ("WLD", "OED", "EUU", "HIC", "EU27_2020", "OAVG"):89        assert lk.from_iso3(agg) is None90    assert lk.any("Atlantis") is None and lk.any("ZZZ") is None and lk.any("999") is None91    assert lk.from_iso2("EL") == "GRC" and lk.from_iso2("UK") == "GBR" and lk.from_iso2("XK") == "XKX"929394@pytest.mark.usefixtures("data_dir")95def test_shrunken_snapshot_is_never_published_and_meta_has_health() -> None:96    _tiny_staging()97    first = build_mod.build(run_id="20260101T000000Z", strict=False)98    con = duckdb.connect(str(settings.db_path), read_only=True)99    meta = dict(con.execute("SELECT key, value FROM meta").fetchall())100    con.close()101    assert meta["previous_observation_count"] == ""  # no previous snapshot102    health = json.loads(meta["source_health"])103    assert health["worldbank"]["ok"] >= 5 and set(health["worldbank"]) == {"ok", "partial", "failed", "quarantined"}104    log_path = settings.logs_dir / f"build-{first.run_id}.json"105    assert log_path.exists()106    doc = json.loads(log_path.read_text())107    assert doc["published"] is True and doc["counts"]["observations"] > 0 and any(f.startswith("worldbank/") for f in doc["staging_files"])108    before = settings.db_path.stat()109    # second build sees the previous count110    second = build_mod.build(run_id="20260102T000000Z", strict=False)111    con = duckdb.connect(str(settings.db_path), read_only=True)112    meta2 = dict(con.execute("SELECT key, value FROM meta").fetchall())113    con.close()114    assert int(meta2["previous_observation_count"]) == first.counts["observations"] and second.counts["observations"] == first.counts["observations"]115    after = settings.db_path.stat()116    # drop most staging files → < 90 % retention → strict integrity failure, live DB untouched, build log says not published117    for f in sorted(settings.staging_dir.glob("*/*.parquet"))[:-1]:118        f.unlink()119    with pytest.raises(build_mod.IntegrityError, match="refusing to publish"):120        build_mod.build(run_id="20260103T000000Z", strict=True)121    assert settings.db_path.stat().st_ino == after.st_ino and settings.db_path.stat().st_mtime_ns == after.st_mtime_ns122    assert not list(settings.build_dir.glob("atlas-20260103*"))123    failed_log = json.loads((settings.logs_dir / "build-20260103T000000Z.json").read_text())124    assert failed_log["published"] is False and "refusing to publish" in (failed_log["error"] or "")125    assert before.st_ino != after.st_ino or before.st_mtime_ns != after.st_mtime_ns  # the second build did publish126