"""A page that is not there is not a refusal.

The record. On staging at 2026-09-20 02:49:59 UTC, collection run 186 (source `shopify-yul`,
Montreal, mode recheck, limit 30, started from the page) read its held listings back through
`read_one`. The ninth request,
`https://www.montrealdutyfree.ca/products/rouge-clair-by-michel-chapoutier-750ml.json`,
answered HTTP 404 with an empty body: the shop no longer sells that wine. `fetch()` raised
SourceBlocked for every HTTPError, and an empty body satisfies the block test anyway, so the
run ended `status=blocked` after 8 prices with "... refused with HTTP 404", the log said
`collector_blocked`, and the source read `refused` on /collectors#live with Start disabled.

What it cost: a refusal is FINAL -- only a decision rian records lifts it -- so a public
airport's source was closed to every later run, and `sweep plan` would have skipped it. A
product that vanished is what `existing_missing` counts, not a host turning us away.
"""

from __future__ import annotations

import email.message
import io
import logging
import urllib.error
import urllib.request
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app import cli
from app.models import Account, Base, CollectionRun, Source
from app.services import collector_view as view
from app.services import ingest, verify
from app.services.collectors import control, fetch as fetch_module
from app.services.collectors.base import RawListing, gone
from app.services.collectors.fetch import (
    FetchError,
    PageGone,
    RenderRefused,
    SourceBlocked,
    fetch,
    render,
    render_budget,
)
from app.services.collectors.robots import parse as parse_robots
from tests.test_collector_control import RATES, SLUG, TABLES, FakeClock, FakeCollector, _raw

#: The ninth read of run 186, verbatim.
GONE_URL = "https://www.montrealdutyfree.ca/products/rouge-clair-by-michel-chapoutier-750ml.json"


def _http_error(url: str, code: int, body: bytes = b""):
    """What urllib raises for a non-2xx: the 404 that ended run 186 carried no body at all."""
    return urllib.error.HTTPError(url, code, "", email.message.Message(), io.BytesIO(body))


@pytest.fixture
def db():
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine, tables=TABLES)
    ingest._table_cache.clear()
    factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False, future=True)
    with factory() as session:
        session.add(Account(id=1, username="rian", display_name="rian"))
        session.add(Source(id=1, slug=SLUG, name="Fake Retailer", delay_seconds=1.0))
        session.commit()
        yield session


# --------------------------------------------------------------------------- the fetch port

class TestTheFetchPortTellsGoneFromRefused:
    @pytest.mark.parametrize("code", [404, 410])
    def test_a_missing_page_is_gone_and_never_a_refusal(self, monkeypatch, code):
        """Run 186's ninth request, with the empty body it actually had."""
        monkeypatch.setattr(urllib.request, "urlopen",
                            lambda *a, **k: (_ for _ in ()).throw(_http_error(GONE_URL, code)))
        with pytest.raises(PageGone) as raised:
            fetch(GONE_URL, delay=0.0)
        assert not isinstance(raised.value, SourceBlocked), "a refusal is final; a gone page must never be one"
        assert isinstance(raised.value, FetchError), "every walk that skips an unreadable page skips this one"
        assert GONE_URL in str(raised.value) and str(code) in str(raised.value)

    @pytest.mark.parametrize("code", [401, 403, 406, 429])
    def test_the_refusal_codes_are_still_refusals(self, monkeypatch, code):
        monkeypatch.setattr(urllib.request, "urlopen",
                            lambda *a, **k: (_ for _ in ()).throw(_http_error("https://h/p", code)))
        with pytest.raises(SourceBlocked) as raised:
            fetch("https://h/p", delay=0.0)
        assert not isinstance(raised.value, PageGone)

    def test_a_challenge_body_behind_a_200_is_still_a_refusal(self, monkeypatch):
        class Answer:
            status = 200
            headers = email.message.Message()

            def read(self):
                return b"Attention Required! | Cloudflare"

            def __enter__(self):
                return self

            def __exit__(self, *exc):
                return False

        monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **k: Answer())
        with pytest.raises(SourceBlocked):
            fetch("https://h/p", delay=0.0)

    def test_a_server_error_is_still_worth_retrying(self, monkeypatch):
        monkeypatch.setattr(urllib.request, "urlopen",
                            lambda *a, **k: (_ for _ in ()).throw(_http_error("https://h/p", 500, b"broke")))
        with pytest.raises(FetchError) as raised:
            fetch("https://h/p", delay=0.0)
        assert not isinstance(raised.value, PageGone) and not isinstance(raised.value, SourceBlocked)

    def test_the_rendered_door_classifies_the_same_way(self, monkeypatch):
        """A 404 draws nothing, so the refusal test would have called it an empty render."""
        monkeypatch.setattr(fetch_module, "_last_render_at", {})
        monkeypatch.setattr(fetch_module, "RENDER_FLOOR_DELAY", 0.0)
        robots = parse_robots("", host="https://h")

        monkeypatch.setattr(fetch_module, "_sidecar_render", lambda payload, base_url: {
            "final_url": payload["url"], "status": 404, "html": "", "text_length": 0, "title": "",
        })
        with render_budget(2), pytest.raises(PageGone) as raised:
            render("https://h/p/1", robots=robots, delay=0, sidecar_url="http://x")
        assert not isinstance(raised.value, SourceBlocked) and "https://h/p/1" in str(raised.value)

        monkeypatch.setattr(fetch_module, "_sidecar_render", lambda payload, base_url: {
            "final_url": payload["url"], "status": 403, "html": "", "text_length": 0, "title": "",
            "headers": {"cf-mitigated": "challenge"},
        })
        with render_budget(2), pytest.raises(RenderRefused):
            render("https://h/p/2", robots=robots, delay=0, sidecar_url="http://x")

    def test_the_collectors_shared_reader_knows_a_gone_page(self):
        assert gone(PageGone(f"{GONE_URL} is gone (HTTP 404)"))
        assert not gone(SourceBlocked("https://h/p refused with HTTP 403"))
        assert not gone(FetchError("https://h/p failed: timed out"))


# --------------------------------------------------------------------------- the recheck loop

class TestTheRecheckLoopCountsItMissing:
    def test_run_186_ends_ok_with_eight_prices_and_one_missing(self, db):
        """Nine held listings, the ninth gone: what run 186 met. It ended `blocked` after its
        eight prices and closed the source; it must end `ok` and count the ninth missing."""
        clock = FakeClock()
        held = [_raw(n) for n in range(9)]
        first = ingest.run_collector(db, FakeCollector(held), rates=RATES, sleeper=clock.sleep, clock=clock.clock)
        assert first.status == "ok" and first.prices_written == 9

        results = {f"sku{n}": _raw(n) for n in range(8)}
        results["sku8"] = PageGone(f"{GONE_URL} is gone (HTTP 404)")
        run = ingest.run_collector(db, FakeCollector([], read_results=results), rates=RATES, mode="recheck",
                                   sleeper=clock.sleep, clock=clock.clock)
        assert run.status == "ok", "a product that vanished is not the shop refusing us"
        assert (run.prices_written, run.existing_checked, run.existing_missing) == (8, 8, 1)
        assert run.skip_counts == {}, "missing is counted as missing, never also as a skip"
        assert run.error is None

    def test_a_gone_page_reads_the_same_as_a_read_one_that_answers_none(self, db):
        clock = FakeClock()
        ingest.run_collector(db, FakeCollector([_raw(0), _raw(1)]), rates=RATES, sleeper=clock.sleep, clock=clock.clock)
        none = ingest.run_collector(db, FakeCollector([], read_results={"sku0": _raw(0)}), rates=RATES,
                                    mode="recheck", sleeper=clock.sleep, clock=clock.clock)
        gone_page = ingest.run_collector(
            db, FakeCollector([], read_results={"sku0": _raw(0), "sku1": PageGone(f"{GONE_URL} is gone (HTTP 404)")}),
            rates=RATES, mode="recheck", sleeper=clock.sleep, clock=clock.clock)
        assert (none.status, none.existing_missing, none.skip_counts) == ("ok", 1, {})
        assert (gone_page.status, gone_page.existing_missing, gone_page.skip_counts) == ("ok", 1, {})

    def test_a_real_refusal_still_ends_the_run_blocked(self, db):
        clock = FakeClock()
        ingest.run_collector(db, FakeCollector([_raw(0)]), rates=RATES, sleeper=clock.sleep, clock=clock.clock)
        run = ingest.run_collector(db, FakeCollector([], read_results={"sku0": SourceBlocked("https://h refused with HTTP 403")}),
                                   rates=RATES, mode="recheck", sleeper=clock.sleep, clock=clock.clock)
        assert run.status == "blocked" and "403" in run.error


# --------------------------------------------------------------------------- what escapes a walk

class _GoneWalker(FakeCollector):
    """A discovery walk that could not skip: the page it needed is not there."""

    def collect(self, *, limit=None, delay=1.0):
        control.wait(delay)
        raise PageGone(f"{GONE_URL} is gone (HTTP 404)")


class TestAnEscapedGonePageIsOurError:
    def test_the_run_is_error_and_names_the_url_never_blocked(self, db):
        clock = FakeClock()
        run = ingest.run_collector(db, _GoneWalker([]), rates=RATES, sleeper=clock.sleep, clock=clock.clock)
        assert run.status == "error", "our failure to read, not the host refusing us"
        assert GONE_URL in run.error

    def test_an_error_run_refuses_no_start_and_tells_no_refusal_story(self, db):
        now = datetime.now(UTC)
        db.add(CollectionRun(source_id=1, started_at=now - timedelta(hours=1), finished_at=now - timedelta(hours=1),
                             status="error", error=f"a page the walk could not skip is gone: {GONE_URL} is gone (HTTP 404)"))
        db.commit()
        last = db.scalar(select(CollectionRun).order_by(CollectionRun.id.desc()))
        source = db.get(Source, 1)
        assert view.state_of(source, None, last, None, now, None, boot=None)[0] == "idle"
        assert control.start_refusal(db, source, FakeCollector([]), "discover", None, None, now,
                                     last=last, running=None, refused_since=None, prefetched=True) is None
        told = view._story([SimpleNamespace(status="error", started_at=now)])
        assert told["refused"] == 0 and told["failed"] == 1

        last.status, last.error = "blocked", "https://h refused with HTTP 403"
        db.commit()
        assert view.state_of(source, None, last, None, now, None, boot=None)[0] == "refused"
        assert control.start_refusal(db, source, FakeCollector([]), "discover", None, None, now,
                                     last=last, running=None, refused_since=None,
                                     prefetched=True).code == "SOURCE_REFUSED"


# --------------------------------------------------------------------------- a discovery walker

class _Robots:
    host = "h"
    crawl_delay = None
    disallows: list[str] = []

    def allows(self, url):  # noqa: ARG002
        return True

    def delay_for(self, ours):  # noqa: ARG002
        return 0.0


class TestAWalkerSkipsAMissingPage:
    def test_a_category_that_is_gone_is_logged_and_the_next_one_is_read(self, monkeypatch, caplog):
        import dataclasses

        from app.services.collectors import heinemann_platform

        site = dataclasses.replace(heinemann_platform.SITES[0], category_ids=("cat_gone", "cat_here"))
        collector = heinemann_platform.HeinemannPlatformCollector(site)
        monkeypatch.setattr(heinemann_platform, "check_allowed", lambda *a, **k: _Robots())

        def fake_fetch_json(url, delay=0.0):  # noqa: ARG001
            if "cat_gone" in url:
                raise PageGone(f"{url} is gone (HTTP 404)")
            return {"pagination": {"numberOfPages": 1}, "results": [
                {"code": "1", "name": "Fake Whisky 1L", "brand": "Fake", "price": {"value": 10.0, "currencyIso": "EUR"}},
            ]}

        monkeypatch.setattr(heinemann_platform, "fetch_json", fake_fetch_json)
        with caplog.at_level(logging.WARNING):
            rows = list(collector.collect())
        assert [row.source_sku for row in rows] == ["1"], "the walk carried on past the gone category"
        gone_lines = [r for r in caplog.records if r.getMessage().startswith("heinemann_page_gone")]
        assert len(gone_lines) == 1 and "cat_gone" in gone_lines[0].getMessage()

    def test_a_walk_that_reads_nothing_at_all_raises_rather_than_reporting_a_clean_run(self, monkeypatch):
        from app.services.collectors import shopify

        collector = shopify.shopify_collectors()[0]
        monkeypatch.setattr(shopify, "check_allowed", lambda *a, **k: _Robots())
        monkeypatch.setattr(shopify, "fetch_json",
                            lambda url, delay=0.0: (_ for _ in ()).throw(PageGone(f"{url} is gone (HTTP 404)")))
        with pytest.raises(PageGone):
            list(collector.collect())


# --------------------------------------------------------------------------- verification

class _FakeDb:
    """Enough session for `verify_source`: it selects Listings and Shops and adds check rows."""

    def __init__(self, rows: dict[str, list]):
        self.rows = rows
        self.added: list = []

    def scalars(self, statement):
        return list(self.rows.get(statement.column_descriptions[0]["entity"].__name__, []))

    def scalar(self, statement):  # noqa: ARG002
        return None

    def add(self, obj):
        self.added.append(obj)

    def commit(self):
        pass


def _latest(listing_id: int):
    from app.services.audit import LatestRow

    return LatestRow(listing_id=listing_id, variant_id=listing_id, shop_id=1, shop_code="FAKE1",
                     shop_currency="USD", visible=True, source_slug=SLUG, price=10.0, currency="USD",
                     price_usd=10.0, was_price=None, in_stock=True, observed_at=datetime.now(UTC),
                     fx_source="live", url=GONE_URL, source_sku=f"sku{listing_id}",
                     name="Fake Whisky 1L", brand="Fake", gtin=None, quantity_ml=1000,
                     vertical="liquor", category=None)


def _verify_with(reader):
    from app.models import Listing, Shop
    from app.models.quality import VerificationRun
    from app.services.collectors.registry import COLLECTORS

    rows = [_latest(1), _latest(2)]
    db = _FakeDb({
        "Listing": [Listing(id=n, shop_id=1, variant_id=n, source_sku=f"sku{n}", url=GONE_URL) for n in (1, 2)],
        "Shop": [Shop(id=1, retailer_id=1, code="FAKE1", name="Fake shop", currency="USD")],
    })
    source = Source(id=1, slug=next(iter(COLLECTORS)), name="Fake", delay_seconds=0)
    run = VerificationRun(id=1, started_at=datetime.now(UTC), seed=1, n=2, per_source={}, mode="on_demand")
    from app.services.fx import FxRates

    summary = verify.verify_source(db, run, source, rows, targeted={}, medians={}, n=2,
                                   rates=FxRates({"USD": 1.0}, datetime.now(UTC), False), reader=reader)
    return summary, [check.verdict for check in db.added]


class TestVerificationDoesNotAbandonASourceOverAGonePage:
    def test_a_gone_page_is_one_gone_check_and_the_sample_carries_on(self):
        def reader(collector, ref):  # noqa: ARG001
            if ref.source_sku == "sku1":
                raise PageGone(f"{GONE_URL} is gone (HTTP 404)")
            return RawListing(source_sku="sku2", name="Fake Whisky 1L", price=10.0, currency="USD",
                              shop_code="FAKE1", brand="Fake", quantity_ml=1000, vertical="liquor")

        summary, verdicts = _verify_with(reader)
        assert verdicts == ["GONE", "PASS"], "the second listing was still read"
        assert summary["checked"] == 2 and not summary["blocked"]

    def test_a_real_refusal_still_abandons_the_source_at_the_first_check(self):
        def reader(collector, ref):  # noqa: ARG001
            raise SourceBlocked("https://h refused with HTTP 403")

        summary, verdicts = _verify_with(reader)
        assert verdicts == ["BLOCKED"] and summary["blocked"] and summary["checked"] == 1


# --------------------------------------------------------------------------- the data correction

BLOCKED_186 = f"{GONE_URL} refused with HTTP 404"


def _seed_runs(db):
    now = datetime.now(UTC)
    db.add_all([
        CollectionRun(id=186, source_id=1, started_at=now, finished_at=now, status="blocked",
                      error=BLOCKED_186, prices_written=8),
        CollectionRun(id=187, source_id=1, started_at=now, finished_at=now, status="blocked",
                      error="https://h/p/1 refused with HTTP 410"),
        CollectionRun(id=188, source_id=1, started_at=now, finished_at=now, status="blocked",
                      error="shop.example.com robots.txt answered HTTP 403"),
        CollectionRun(id=189, source_id=1, started_at=now, finished_at=now, status="ok", prices_written=412),
    ])
    db.commit()


class TestTheBackfillCorrectsTheRowsAlreadyWritten:
    def test_it_moves_only_the_false_refusals_and_keeps_what_the_run_recorded(self, db):
        _seed_runs(db)
        message = cli.backfill_false_refusals(db)
        assert "2 run(s)" in message and "186, 187" in message
        by_id = {run.id: run for run in db.scalars(select(CollectionRun))}
        assert by_id[186].status == "error" and by_id[187].status == "error"
        assert by_id[186].error.startswith(BLOCKED_186), "the text the run recorded is kept"
        assert "reclassified from blocked to error" in by_id[186].error
        assert by_id[188].status == "blocked" and by_id[188].error == "shop.example.com robots.txt answered HTTP 403", \
            "a 403 is a real refusal and is never touched"
        assert by_id[189].status == "ok"
        assert by_id[186].prices_written == 8, "the eight prices run 186 did collect stay"

    def test_a_second_run_changes_nothing(self, db):
        _seed_runs(db)
        cli.backfill_false_refusals(db)
        assert cli.backfill_false_refusals(db).startswith("false_refusals: 0 run(s)")
        assert db.get(CollectionRun, 186).error.count("reclassified") == 1

    def test_check_writes_nothing(self, db):
        """`--check` is `cmd_backfill`'s generic mechanism: every commit becomes a flush and the
        transaction is rolled back at the end. Pinned here so this backfill honours it."""
        _seed_runs(db)
        db.commit = db.flush  # type: ignore[method-assign]
        message = cli.backfill_false_refusals(db)
        db.rollback()
        assert "2 run(s)" in message
        assert db.get(CollectionRun, 186).status == "blocked"

    def test_it_is_registered_under_its_name(self):
        assert cli.BACKFILLS["false_refusals"] is cli.backfill_false_refusals