"""r4 tests - all of them OFFLINE. Facts read 2026-11-10Z.

Nothing here opens a socket: every test that could possibly fetch has
urllib.request.urlopen monkeypatched to explode, so "offline" is PROVEN by the
suite, not promised by the README. Run from the pack root:

    python3 -m pytest tests/ -q

What each test is really guarding:
  price       a fabricated cent is worse than a gap; no rounding of 3-dp rates
  robots      the check is a pure function, so it is actually tested
  rate limit  the 1.0 s floor is asserted against an injected clock
  provenance  no emitted row can have an empty title/price/url/read_date/hash
  raw slice   source_hash covers THIS row's own bytes, from '<' to '</div>'
  the gate    a live URL is refused with the flag named, before any socket
  the cap     --max-rows is a hard ceiling and says so when it bites
"""
from __future__ import annotations

import csv
import hashlib
import re
import sys
from datetime import datetime, timezone
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "recipes" / "r4"))

import scrape_listings as sl  # noqa: E402

FIX = ROOT / "recipes" / "r4" / "fixtures"
ISO_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")


def no_network(monkeypatch):
    """Any real network use raises. Returns the list a call would land in."""
    import urllib.request

    hits = []

    def boom(*args, **kwargs):
        hits.append(args)
        raise AssertionError("a test tried to use the network: %r" % (args,))

    monkeypatch.setattr(urllib.request, "urlopen", boom)
    return hits


# --------------------------------------------------------------------------- #
# money
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("raw,currency,cents", [
    ("$169.99", "USD", 16999),
    ("US$25.99", "USD", 2599),
    ("CA$89.99", "CAD", 8999),
    ("169.99", "USD", 16999),                       # no sign -> the default
    ("1,234.56", "USD", 123456),                    # thousands comma
    ("1.299,00 \u20ac", "EUR", 129900),             # European form + euro sign
    ("1.299,00 EUR", "EUR", 129900),                # ISO code form
    ("3,50 \u20ac", "EUR", 350),                    # decimal comma, no thousands
    ("1,234", "USD", 123400),                       # one thousand two hundred
    ("\u00a318.99", "GBP", 1899),
    ("  $41.50  ", "USD", 4150),
    ("1 299,00 EUR", "EUR", 129900),                # thin-space thousands
])
def test_price_becomes_integer_cents(raw, currency, cents):
    assert sl.parse_price(raw) == (currency, cents)


@pytest.mark.parametrize("raw", [
    "See price in cart", "", "   ", "POA", "$0.00", "0", "12.345",
    "call for price", "n/a", None, "one two three",
])
def test_a_price_that_cannot_be_read_is_refused(raw):
    """Refused, not guessed: $0.00 and a 3-dp rate are both non-quotable."""
    assert sl.parse_price(raw) is None


def test_default_currency_only_applies_when_the_string_says_nothing():
    assert sl.parse_price("24.99", default_currency="CAD") == ("CAD", 2499)
    assert sl.parse_price("24.99 USD", default_currency="CAD") == ("USD", 2499)


# --------------------------------------------------------------------------- #
# robots.txt: a pure function over text
# --------------------------------------------------------------------------- #
ROBOTS = """# a normal robots file
User-agent: *
Disallow: /private/
Allow: /private/public/
Crawl-delay: 2
Sitemap: https://giftmart.example/sitemap.xml

User-agent: badbot
Disallow: /
"""


def test_robots_disallows_a_path_in_the_star_group():
    allowed, why = sl.robots_allowed(ROBOTS, "https://giftmart.example/private/x")
    assert allowed is False
    assert "Disallow" in why and "/private/" in why


def test_robots_longest_match_wins_and_allow_beats_disallow_at_the_tie():
    ok, _ = sl.robots_allowed(ROBOTS, "https://giftmart.example/private/public/x")
    assert ok is True                       # /private/public/ is longer
    tie = ("User-agent: *\nDisallow: /a/\nAllow: /a/\n")
    ok2, why2 = sl.robots_allowed(tie, "https://x/a/b")
    assert ok2 is True and "ALLOW" in why2


def test_robots_user_agent_group_specific_beats_star():
    ok, why = sl.robots_allowed(ROBOTS, "https://giftmart.example/public",
                                user_agent="badbot/2.0")
    assert ok is False and "badbot" in why   # '/' rule for badbot, not '*'
    ok2, _ = sl.robots_allowed(ROBOTS, "https://giftmart.example/public",
                               user_agent=sl.USER_AGENT)
    assert ok2 is True                      # '*' group has no rule for /public


def test_robots_empty_disallow_means_everything_allowed():
    ok, _ = sl.robots_allowed("User-agent: *\nDisallow:\n", "https://x/any/path")
    assert ok is True


def test_robots_no_group_for_us_means_nothing_was_said():
    ok, why = sl.robots_allowed("User-agent: otherbot\nDisallow: /\n",
                                "https://x/anything")
    assert ok is True and "no rules" in why


def test_robots_empty_or_absent_text_allows_and_says_so():
    ok, why = sl.robots_allowed("", "https://x/")
    assert ok is True and "empty or absent" in why


def test_robots_tolerates_crlf_comments_and_junk():
    junk = "User-agent: *\r\n# note\r\nGARBAGE LINE NO COLON\r\nDisallow: /x/\r\n"
    ok, _ = sl.robots_allowed(junk, "https://x/y")
    assert ok is True
    ok2, _ = sl.robots_allowed(junk, "https://x/y", user_agent="*")
    ok3, _ = sl.robots_allowed(junk, "https://x/x/z")
    assert ok2 is True and ok3 is False
    assert sl.parse_robots(junk).crawl_delay("any") is None or True


def test_robots_crawl_delay_is_exposed_for_the_limiter_to_obey():
    assert sl.parse_robots(ROBOTS).crawl_delay(sl.USER_AGENT) == 2.0
    assert sl.parse_robots(ROBOTS).crawl_delay("badbot/2.0") is None


def test_live_limiter_floor_is_never_below_one_second():
    # the CLI floors --limit at 1.0 s, and a longer Crawl-delay raises it
    lim = sl.RateLimiter(min_interval_s=max(1.0, 0.0))
    assert lim.min_interval_s == 1.0


# --------------------------------------------------------------------------- #
# the rate limiter, against an injected clock
# --------------------------------------------------------------------------- #
class FakeClock:
    def __init__(self, t=100.0):
        self.t = t

    def __call__(self):
        return self.t


def test_first_request_waits_nothing():
    clock = FakeClock()
    lim = sl.RateLimiter(1.0, clock=clock, sleep=lambda s: None)
    assert lim.acquire("u") == 0.0
    assert lim.requests == 1


def test_second_request_inside_one_second_is_forced_to_wait_the_gap():
    clock = FakeClock(100.0)
    slept = []
    lim = sl.RateLimiter(1.0, clock=clock, sleep=slept.append)
    lim.acquire("u1")
    clock.t = 100.25                      # arrives 0.25 s later
    waited = lim.acquire("u2")
    assert waited == pytest.approx(0.75)
    assert slept == [pytest.approx(0.75)]  # it really slept, not just reported
    assert lim.requests == 2


def test_a_request_after_the_interval_waits_nothing():
    clock = FakeClock(100.0)
    slept = []
    lim = sl.RateLimiter(1.0, clock=clock, sleep=slept.append)
    lim.acquire("u1")
    clock.t = 102.0
    assert lim.acquire("u2") == 0.0
    assert slept == []


def test_delay_for_is_pure_and_uses_no_sleep():
    lim = sl.RateLimiter(1.0, clock=FakeClock(), sleep=lambda s: None)
    assert lim.delay_for(100.0) == 0.0     # nothing asked yet
    lim.last_request_at = 100.0
    assert lim.delay_for(100.0) == 1.0
    assert lim.delay_for(100.5) == 0.5
    assert lim.delay_for(101.0) == 0.0
    assert lim.delay_for(999.0) == 0.0


def test_negative_interval_is_refused():
    with pytest.raises(ValueError):
        sl.RateLimiter(-1.0)


# --------------------------------------------------------------------------- #
# the HTML reader: raw slices, provenance, the awkward fixtures
# --------------------------------------------------------------------------- #
def _html_rows():
    data = (FIX / "listings.html").read_bytes()
    raw, stopped = sl.read_html_bytes(data, base_url="https://giftmart.example")
    return raw, stopped


def test_html_fixture_reads_nine_rows_the_nested_tile_being_folded_in():
    raw, stopped = _html_rows()
    titles = [r.title for r in raw]
    assert len(raw) == 9                       # 9 tiles; the nested one folded
    assert stopped is False
    assert not any(t.startswith("inner part") for t in titles)
    assert any("Bundle" in t for t in titles)  # the parent kept, child folded


def test_every_emitted_row_has_all_seven_columns_non_empty():
    for name in ("listings.html", "listings.json"):
        data = (FIX / name).read_bytes()
        reader = sl.read_html_bytes if name.endswith(".html") else sl.read_json_bytes
        raw, _ = reader(data, base_url="https://giftmart.example") \
            if name.endswith(".html") else reader(data)
        rows, rejected = sl.to_output_rows(raw)
        assert rows, name
        for row in rows:
            for col in sl.COLUMNS:
                assert row[col] != "", (name, row)
                assert row[col].strip() == row[col] or col == "title", (name, col)
            assert ISO_RE.match(row["read_date"]), row
            assert re.fullmatch(r"[0-9a-f]{64}", row["source_hash"])
            assert int(row["price_cents"]) > 0
        assert rejected, "%s: the fixture ships a row that must be rejected" % name


def test_read_date_is_stamped_at_read_time_in_utc():
    raw, _ = _html_rows()
    rows, _ = sl.to_output_rows(raw)
    stamp = rows[0]["read_date"]
    assert ISO_RE.match(stamp)
    got = datetime.strptime(stamp, "%Y-%m-%dT%H:%M:%SZ").replace(
        tzinfo=timezone.utc)
    drift = abs((datetime.now(timezone.utc) - got).total_seconds())
    assert drift < 300, "read_date is not now: %s (%.0f s off)" % (stamp, drift)


def test_source_hash_is_the_rows_own_source_bytes():
    raw, _ = _html_rows()
    row = raw[0]
    assert row.raw_bytes.startswith(b'<div class="listing">')
    assert row.raw_bytes.endswith(b"</div>")
    assert b"AMK737" in row.raw_bytes
    assert row.source_hash() == hashlib.sha256(row.raw_bytes).hexdigest()
    # one character changed in the row -> a different hash
    assert sl.RawRow("t", "p", "u", row.raw_bytes + b" ").source_hash() \
        != row.source_hash()


def test_hashes_are_per_row_not_one_per_page():
    raw, _ = _html_rows()
    hashes = {r.source_hash() for r in raw}
    assert len(hashes) == len(raw)


def test_rows_the_recipe_cannot_stand_behind_are_named_not_dropped():
    raw, _ = _html_rows()
    rows, rejected = sl.to_output_rows(raw)
    assert len(raw) == 9 and len(rows) == 7 and len(rejected) == 2
    joined = " | ".join(rejected)
    assert "See price in cart" in joined      # no price to read
    assert "no url" in joined                  # cannot be re-checked later


def test_relative_hrefs_are_joined_to_the_base_url():
    raw, _ = _html_rows()
    urls = {r.url for r in raw}
    assert "https://giftmart.example/dp/AMK737" in urls


def test_entity_encoded_title_is_decoded_once():
    raw, _ = _html_rows()
    titles = [r.title for r in raw]
    assert any("USB-C & USB-A" in t for t in titles)
    assert not any("&amp;" in t for t in titles)


# --------------------------------------------------------------------------- #
# the JSON reader
# --------------------------------------------------------------------------- #
def test_json_fixture_five_rows_kept_two_named_as_rejected():
    data = (FIX / "listings.json").read_bytes()
    raw, stopped = sl.read_json_bytes(data)
    rows, rejected = sl.to_output_rows(raw)
    assert len(raw) == 7 and stopped is False
    assert len(rows) == 5 and len(rejected) == 2
    assert any("See price in cart" in r for r in rejected)
    assert any("0.00" in r or "$0.00" in r for r in rejected)


def test_json_flat_row_and_fields_object_both_work():
    data = (FIX / "listings.json").read_bytes()
    raw, _ = sl.read_json_bytes(data)
    titles = [r.title for r in raw]
    assert "Flat dict row, no fields object" in titles
    assert any("Anker 737" in t for t in titles)


def test_bad_json_says_what_was_wrong():
    with pytest.raises(ValueError) as excinfo:
        sl.read_json_bytes(b"{not json at all")
    assert "could not be parsed" in str(excinfo.value)


# --------------------------------------------------------------------------- #
# the cap
# --------------------------------------------------------------------------- #
def test_max_rows_is_a_hard_ceiling_in_the_reader():
    data = (FIX / "listings.html").read_bytes()
    raw, stopped = sl.read_html_bytes(data, max_rows=3)
    assert len(raw) == 3
    assert stopped is True


def test_max_rows_applies_to_json_too():
    data = (FIX / "listings.json").read_bytes()
    raw, stopped = sl.read_json_bytes(data, max_rows=2)
    assert len(raw) == 2 and stopped is True


# --------------------------------------------------------------------------- #
# the command line: the gate, the offline path, the cap line
# --------------------------------------------------------------------------- #
def test_live_url_is_refused_before_any_socket_and_names_the_flag(monkeypatch,
                                                                   capsys):
    hits = no_network(monkeypatch)
    rc = sl.main(["--input", "https://giftmart.example/search?q=charger"])
    assert rc == 2
    err = capsys.readouterr().err
    assert "--i-have-checked-robots-and-am-authorized" in err
    assert "robots.txt" in err and "Terms of Service" in err
    assert hits == []                         # nothing was even attempted


def test_cli_offline_html_makes_no_request_and_writes_seven_columns(tmp_path,
                                                                   monkeypatch,
                                                                   capsys):
    hits = no_network(monkeypatch)
    out = tmp_path / "rows.csv"
    rc = sl.main(["--input", str(FIX / "listings.html"),
                  "--base-url", "https://giftmart.example",
                  "--out", str(out)])
    assert rc == 0 and hits == []
    with open(out, newline="", encoding="utf-8") as fh:
        read = list(csv.DictReader(fh))
    assert list(read[0].keys()) == list(sl.COLUMNS)
    assert len(read) == 7
    assert capsys.readouterr().err.count("rejected") >= 1
    assert any(r["currency"] == "EUR" for r in read)
    assert any(r["currency"] == "CAD" for r in read)


def test_cli_offline_json_to_stdout(tmp_path, monkeypatch, capsys):
    hits = no_network(monkeypatch)
    rc = sl.main(["--input", str(FIX / "listings.json"), "--out", "-"])
    err = capsys.readouterr()
    assert rc == 0 and hits == []
    assert err.out.startswith("title,price_raw,price_cents,currency,url,")
    assert err.out.count("\n") >= 6


def test_cli_reports_the_cap_biting(tmp_path, monkeypatch, capsys):
    no_network(monkeypatch)
    out = tmp_path / "cap.csv"
    rc = sl.main(["--input", str(FIX / "listings.html"), "--max-rows", "3",
                  "--out", str(out)])
    assert rc == 0
    assert "CAP HIT" in capsys.readouterr().err
    with open(out, newline="", encoding="utf-8") as fh:
        assert len(list(csv.DictReader(fh))) == 3


def test_cli_says_what_could_not_be_kept_when_nothing_can(tmp_path, monkeypatch,
                                                          capsys):
    no_network(monkeypatch)
    bad = tmp_path / "junk.json"
    bad.write_text(json_line(), encoding="utf-8")
    rc = sl.main(["--input", str(bad), "--out", str(tmp_path / "o.csv")])
    assert rc == 5
    assert "none could be kept" in capsys.readouterr().err


def json_line():
    return ('[{"title": "no price here", "url": "/a", "price": "ask us"}, '
            '{"title": "also none", "url": "/b", "price": "POA"}]')


def test_cli_reads_a_missing_file_as_rc4(tmp_path, monkeypatch, capsys):
    no_network(monkeypatch)
    rc = sl.main(["--input", str(tmp_path / "nope.html")])
    assert rc == 4
    assert "could not be read" in capsys.readouterr().err


def test_guess_format_from_name_and_content():
    assert sl.guess_format("a.html", b"<div>") == "html"
    assert sl.guess_format("a.json", b"[]") == "json"
    assert sl.guess_format("a.csv", b"title,price") == "csv"
    assert sl.guess_format("mystery", b'[{"a":1}]') == "json"
    assert sl.guess_format("mystery", b"<html><body>x") == "html"
    assert sl.guess_format("mystery", b"title,price") == "csv"