"""Pipeline reliability: HTTP retries/backoff honouring Retry-After, country mapping, never publishing a shrunken snapshot,
source-health meta and build logs."""
from __future__ import annotations

import json

import duckdb
import httpx
import pytest
import respx

from countryatlas.config import settings
from countryatlas.connectors import base as base_mod
from countryatlas.connectors.base import Connector, retry_after_seconds
from countryatlas.pipeline import build as build_mod
from countryatlas.registry import lookup
from tests.test_build import _tiny_staging


class _Dummy(Connector):
    id = "worldbank"
    name = "dummy"
    rate_per_minute = 100_000

    def fetch(self, spec):  # pragma: no cover - not used
        raise NotImplementedError

    def normalize(self, raw, spec):  # pragma: no cover - not used
        return []


def test_retry_after_parsing() -> None:
    assert retry_after_seconds(None) is None and retry_after_seconds("abc") is None
    assert retry_after_seconds("7") == 7.0
    assert retry_after_seconds("999") == base_mod.RETRY_AFTER_CAP
    assert retry_after_seconds("Wed, 21 Oct 2015 07:28:00 GMT") == 0.0  # in the past → no wait


@respx.mock
def test_get_retries_on_transient_statuses_then_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
    sleeps: list[float] = []
    monkeypatch.setattr(base_mod.time, "sleep", lambda s: sleeps.append(s))  # Retry-After sleep
    # tenacity's wait: make it instant but record the computed deterministic delays
    waits: list[float] = []
    orig_wait = _Dummy.get.retry.wait

    def fake_wait(retry_state):
        w = orig_wait(retry_state)
        waits.append(w)
        return 0

    monkeypatch.setattr(_Dummy.get.retry, "wait", fake_wait)
    route = respx.get("https://example.test/x").mock(side_effect=[
        httpx.Response(429, headers={"Retry-After": "3"}),
        httpx.Response(503),
        httpx.Response(200, json={"ok": True}),
    ])
    c = _Dummy()
    r = c.get("https://example.test/x")
    assert r.status_code == 200 and route.call_count == 3
    assert [s for s in sleeps if s] == [3.0]  # Retry-After honoured once (zero-length tenacity naps filtered)
    assert waits == [2.0, 4.0]  # deterministic exponential backoff, no jitter
    c.close()


@respx.mock
def test_get_gives_up_after_five_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(_Dummy.get.retry, "wait", lambda rs: 0)
    route = respx.get("https://example.test/down").mock(return_value=httpx.Response(502))
    c = _Dummy()
    with pytest.raises(base_mod.TransientHTTPError):
        c.get("https://example.test/down")
    assert route.call_count == base_mod.RETRY_ATTEMPTS
    c.close()
    # a 404 is not retried
    route2 = respx.get("https://example.test/missing").mock(return_value=httpx.Response(404))
    c2 = _Dummy()
    with pytest.raises(httpx.HTTPStatusError):
        c2.get("https://example.test/missing")
    assert route2.call_count == 1
    c2.close()


def test_country_mapping_drops_aggregates_and_unknowns() -> None:
    lk = lookup()
    assert lk.any("CAN") == "CAN" and lk.any("ca") == "CAN" and lk.any("124") == "CAN"
    assert lk.any("Viet Nam") == "VNM" and lk.any("Türkiye") == "TUR" and lk.any("Kosovo") == "XKX"
    for agg in ("WLD", "OED", "EUU", "HIC", "EU27_2020", "OAVG"):
        assert lk.from_iso3(agg) is None
    assert lk.any("Atlantis") is None and lk.any("ZZZ") is None and lk.any("999") is None
    assert lk.from_iso2("EL") == "GRC" and lk.from_iso2("UK") == "GBR" and lk.from_iso2("XK") == "XKX"


@pytest.mark.usefixtures("data_dir")
def test_shrunken_snapshot_is_never_published_and_meta_has_health() -> None:
    _tiny_staging()
    first = build_mod.build(run_id="20260101T000000Z", strict=False)
    con = duckdb.connect(str(settings.db_path), read_only=True)
    meta = dict(con.execute("SELECT key, value FROM meta").fetchall())
    con.close()
    assert meta["previous_observation_count"] == ""  # no previous snapshot
    health = json.loads(meta["source_health"])
    assert health["worldbank"]["ok"] >= 5 and set(health["worldbank"]) == {"ok", "partial", "failed", "quarantined"}
    log_path = settings.logs_dir / f"build-{first.run_id}.json"
    assert log_path.exists()
    doc = json.loads(log_path.read_text())
    assert doc["published"] is True and doc["counts"]["observations"] > 0 and any(f.startswith("worldbank/") for f in doc["staging_files"])
    before = settings.db_path.stat()
    # second build sees the previous count
    second = build_mod.build(run_id="20260102T000000Z", strict=False)
    con = duckdb.connect(str(settings.db_path), read_only=True)
    meta2 = dict(con.execute("SELECT key, value FROM meta").fetchall())
    con.close()
    assert int(meta2["previous_observation_count"]) == first.counts["observations"] and second.counts["observations"] == first.counts["observations"]
    after = settings.db_path.stat()
    # drop most staging files → < 90 % retention → strict integrity failure, live DB untouched, build log says not published
    for f in sorted(settings.staging_dir.glob("*/*.parquet"))[:-1]:
        f.unlink()
    with pytest.raises(build_mod.IntegrityError, match="refusing to publish"):
        build_mod.build(run_id="20260103T000000Z", strict=True)
    assert settings.db_path.stat().st_ino == after.st_ino and settings.db_path.stat().st_mtime_ns == after.st_mtime_ns
    assert not list(settings.build_dir.glob("atlas-20260103*"))
    failed_log = json.loads((settings.logs_dir / "build-20260103T000000Z.json").read_text())
    assert failed_log["published"] is False and "refusing to publish" in (failed_log["error"] or "")
    assert before.st_ino != after.st_ino or before.st_mtime_ns != after.st_mtime_ns  # the second build did publish