"""Egress-block detection: 403 storms must rotate the IP, then fail the run.

Regression cover for 2026-07-15, when Rightmove's edge 403'd every typeahead
call, fetch_with_retry returned None, resolve_outcode_id read that as "no such
outcode" for all 366 outcodes, and the run published a Rightmove-less parquet
with errors: [] and exit 0.
"""

import httpx
import pytest

import http_client
import scraper
from http_client import BlockedError, _EgressBlockTracker


class _StubResponse:
    def __init__(self, status_code: int, payload: dict | None = None):
        self.status_code = status_code
        self._payload = payload or {}

    def json(self) -> dict:
        return self._payload


class _StubClient:
    """Replays a fixed status sequence, recording the URLs it was asked for."""

    def __init__(self, statuses: list[int], payload: dict | None = None):
        self._statuses = list(statuses)
        self._payload = payload or {"matches": []}
        self.calls: list[str] = []

    def get(self, url, params=None, headers=None):
        self.calls.append(url)
        status = self._statuses.pop(0) if self._statuses else 200
        return _StubResponse(status, self._payload)


@pytest.fixture(autouse=True)
def _no_sleeping(monkeypatch):
    # Backoff delays would otherwise make this suite take minutes.
    monkeypatch.setattr(http_client.shutdown, "sleep", lambda _s: None)
    monkeypatch.setattr(http_client.RATE_LIMITER, "acquire", lambda: None)


@pytest.fixture(autouse=True)
def _fresh_tracker(monkeypatch):
    tracker = _EgressBlockTracker(threshold=3, max_rotations=1)
    monkeypatch.setattr(http_client, "BLOCK_TRACKER", tracker)
    return tracker


def test_403_is_retried_not_swallowed(monkeypatch):
    """A transient 403 must not end the call: this is what broke last time."""
    rotations = []
    monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: rotations.append(1))

    client = _StubClient([403, 200], payload={"matches": ["ok"]})
    result = http_client.fetch_with_retry(client, "https://los.example.com/typeahead")

    assert result == {"matches": ["ok"]}
    assert len(client.calls) == 2, "the 403 should have been retried"
    assert rotations == [], "one 403 is not a block; rotating would be disruptive"


def test_sustained_403s_rotate_the_egress_ip(monkeypatch):
    """Threshold consecutive 403s from one host trigger exactly one rotation."""
    rotations = []

    def _rotate():
        rotations.append(1)
        return True

    monkeypatch.setattr(http_client.gluetun, "rotate_ip", _rotate)

    # 3 x 403 hits the threshold and rotates; the retry budget resets and the
    # next call succeeds on the "new IP".
    client = _StubClient([403, 403, 403, 200], payload={"matches": ["ok"]})
    result = http_client.fetch_with_retry(client, "https://los.example.com/typeahead")

    assert result == {"matches": ["ok"]}
    assert rotations == [1], "expected exactly one rotation"


def test_403s_surviving_rotation_raise_blocked_error(monkeypatch):
    """Once the rotation budget is spent, a block must raise, never return None.

    Returning None is what let resolve_outcode_id read a site-wide block as
    "this outcode does not exist" and silently produce zero listings.
    """
    monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: True)

    client = _StubClient([403] * 20)
    with pytest.raises(BlockedError, match="refusing this egress IP"):
        http_client.fetch_with_retry(client, "https://los.example.com/typeahead")


def test_failed_rotation_is_treated_as_blocked(monkeypatch):
    """If we cannot rotate away from a blocked IP, we are blocked. Say so."""
    monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: False)

    client = _StubClient([403] * 10)
    with pytest.raises(BlockedError):
        http_client.fetch_with_retry(client, "https://los.example.com/typeahead")


def test_success_resets_the_consecutive_count(monkeypatch):
    """Interleaved successes mean the IP is fine, so 403s must not accumulate."""
    rotations = []
    monkeypatch.setattr(
        http_client.gluetun, "rotate_ip", lambda: rotations.append(1) or True
    )

    # 403,200 repeated: never 3 consecutive, so never a rotation.
    for _ in range(5):
        client = _StubClient([403, 200], payload={"matches": ["ok"]})
        http_client.fetch_with_retry(client, "https://los.example.com/typeahead")

    assert rotations == []


def test_on_403_false_opts_out_of_block_detection(monkeypatch):
    monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: True)
    client = _StubClient([403])
    assert (
        http_client.fetch_with_retry(client, "https://x.example.com/y", on_403=False)
        is None
    )


def test_403_tracking_is_per_host(monkeypatch):
    """A block on one host must not be charged against another."""
    rotations = []
    monkeypatch.setattr(
        http_client.gluetun, "rotate_ip", lambda: rotations.append(1) or True
    )

    for host in ("a.example.com", "b.example.com", "c.example.com"):
        client = _StubClient([403, 200], payload={"matches": []})
        http_client.fetch_with_retry(client, f"https://{host}/typeahead")

    assert rotations == [], "one 403 each across three hosts is not a block"


def test_connection_errors_still_retry(monkeypatch):
    """The rewritten loop must not regress non-403 retry behaviour."""

    class _FlakyClient:
        def __init__(self):
            self.calls = 0

        def get(self, url, params=None, headers=None):
            self.calls += 1
            if self.calls == 1:
                raise httpx.ConnectError("boom")
            return _StubResponse(200, {"matches": ["ok"]})

    client = _FlakyClient()
    assert http_client.fetch_with_retry(client, "https://x.example.com/y") == {
        "matches": ["ok"]
    }
    assert client.calls == 2


# ---------------------------------------------------------------------------
# Run-level sanity gates
# ---------------------------------------------------------------------------


def test_zero_yield_from_a_selected_source_is_a_failure(tmp_path):
    results = {"rightmove": [], "onthemarket": [{"a": 1}], "zoopla": []}
    failures = scraper._sanity_failures(
        results,
        ["rightmove", "onthemarket"],
        merged=[{"a": 1}],
        output_path=tmp_path / "missing.parquet",
        abandoned=set(),
    )
    assert len(failures) == 1
    assert "rightmove yielded 0 listings" in failures[0]


def test_healthy_run_has_no_failures(tmp_path):
    results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
    assert (
        scraper._sanity_failures(
            results,
            ["rightmove", "onthemarket"],
            merged=[{"a": 1}, {"b": 2}],
            output_path=tmp_path / "missing.parquet",
            abandoned=set(),
        )
        == []
    )


def test_unselected_source_yielding_zero_is_fine(tmp_path):
    """zoopla is not scheduled in production; its 0 must not fail the run."""
    results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
    assert (
        scraper._sanity_failures(
            results,
            ["rightmove", "onthemarket"],
            merged=[{"a": 1}],
            output_path=tmp_path / "missing.parquet",
            abandoned=set(),
        )
        == []
    )


def test_source_abandoned_mid_run_is_a_failure_even_with_listings(tmp_path):
    """A block that starts mid-run must fail even though the source is non-empty.

    Without this gate a single banked listing defeats the zero-yield check: the
    source stops at outcode 300 of 366, keeps its 60k listings, OnTheMarket
    backfills the merged total to within the drop limit, and the partial
    dataset publishes with exit 0. That is the original incident with one
    outcode's head start.
    """
    import polars as pl

    path = tmp_path / "online_listings_buy.parquet"
    pl.DataFrame({"x": range(103008)}).write_parquet(path)

    results = {
        "rightmove": [{"a": 1}] * 60000,
        "onthemarket": [{"b": 2}] * 83785,
        "zoopla": [],
    }
    failures = scraper._sanity_failures(
        results,
        ["rightmove", "onthemarket"],
        merged=[{"a": 1}] * 99000,  # only a 4% dip: both other gates pass
        output_path=path,
        abandoned={"rightmove"},
    )
    assert len(failures) == 1
    assert "abandoned mid-run" in failures[0]
    assert "60000 listings cover only part" in failures[0]


def test_the_real_incident_drop_now_trips_the_collapse_gate(tmp_path):
    """103,008 -> 83,785 is an 18.7% drop, which the original 25% limit missed."""
    import polars as pl

    path = tmp_path / "online_listings_buy.parquet"
    pl.DataFrame({"x": range(103008)}).write_parquet(path)

    # Pretend rightmove banked listings so only the collapse gate can fire.
    results = {
        "rightmove": [{"a": 1}] * 100,
        "onthemarket": [{"b": 2}] * 83785,
        "zoopla": [],
    }
    failures = scraper._sanity_failures(
        results,
        ["rightmove", "onthemarket"],
        merged=[{"b": 2}] * 83785,
        output_path=path,
        abandoned=set(),
    )
    assert len(failures) == 1
    assert "collapsed from 103008 to 83785" in failures[0]


def test_row_count_collapse_is_a_failure(tmp_path):
    """Backstop for degradation that is neither a clean zero nor a raised block.

    Dedup masks a lost source (OnTheMarket backfilled the keys Rightmove used to
    win), so a partial block can still look plausible per-source.
    """
    import polars as pl

    path = tmp_path / "online_listings_buy.parquet"
    pl.DataFrame({"x": range(1000)}).write_parquet(path)

    results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
    failures = scraper._sanity_failures(
        results,
        ["rightmove", "onthemarket"],
        merged=[{"a": 1}] * 700,  # 30% drop, over the 10% limit
        output_path=path,
        abandoned=set(),
    )
    assert len(failures) == 1
    assert "collapsed from 1000 to 700" in failures[0]


def test_normal_market_movement_is_not_a_failure(tmp_path):
    import polars as pl

    path = tmp_path / "online_listings_buy.parquet"
    pl.DataFrame({"x": range(1000)}).write_parquet(path)

    results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
    assert (
        scraper._sanity_failures(
            results,
            ["rightmove", "onthemarket"],
            merged=[{"a": 1}] * 970,  # 3% drop, within normal movement
            output_path=path,
            abandoned=set(),
        )
        == []
    )


def test_first_ever_run_has_no_baseline_to_compare(tmp_path):
    results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
    assert (
        scraper._sanity_failures(
            results,
            ["rightmove", "onthemarket"],
            merged=[{"a": 1}],
            output_path=tmp_path / "does-not-exist.parquet",
            abandoned=set(),
        )
        == []
    )