"""Email capture: the personal-data rules on the one public write route.

An in-memory SQLite holds the `subscribers` table so the rules are proven as behaviour:
the same answer for a new and an existing address, a filled honeypot stores nothing, the
throttle answers 429, no log line ever carries an address, and the export leaves out
withdrawn rows. Stream R's middleware will keep POST /api/subscribers on the public
allowlist; the route inventory fixture already lists it.

Then three pins that are not behaviour of this module but of everything around it, because
email capture only holds if they stay true: the sentence the form shows is the sentence it
sends, nothing outside the service and the owner's CLI reads the table, and the one public
write sits behind `SITE_ACCESS` like every storefront read.
"""

import logging
import pathlib
import re

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.db import get_db
from app.models.editorial import Subscriber, SubscribeIn
from app.routers import subscribers as route
from app.services import access, subscribers as svc
from app.services.throttle import Throttle

MAIN = pathlib.Path(__file__).resolve().parents[1]

CONSENT = "Yes, email me the Duty Free Professor newsletter. Unsubscribe any time."


def payload(**over):
    base = {
        "email": "Someone@Example.com ",
        "first_name": " Ada ",
        "last_name": "Lovelace",
        "home_airport": "lhr",
        "interests": ["Whisky", "whisky", "FRAGRANCE", "<b>x</b>"],
        "consent": True,
        "consent_text": CONSENT,
        "source": "home",
    }
    base.update(over)
    return base


@pytest.fixture
def db():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    Subscriber.__table__.create(engine)
    session = sessionmaker(bind=engine, expire_on_commit=False)()
    yield session
    session.close()


@pytest.fixture
def client(db, monkeypatch):
    monkeypatch.setattr(
        route, "throttle", Throttle(per_key=5, per_key_seconds=600, total=60, total_seconds=60)
    )
    app = FastAPI()
    app.include_router(route.router)

    def override():
        yield db

    app.dependency_overrides[get_db] = override
    return TestClient(app)


class TestService:
    def test_normalises_and_stores_the_minimum(self, db):
        row = svc.subscribe(db, SubscribeIn(**payload()))
        assert row.email == "someone@example.com"
        assert row.first_name == "Ada" and row.last_name == "Lovelace" and row.home_airport == "LHR"
        assert row.interests == ["whisky", "fragrance"]  # de-duplicated, lowercased, markup refused
        assert row.consent_text == CONSENT and row.consent_at is not None and row.source == "home"
        assert not hasattr(row, "ip") and not hasattr(row, "user_agent")

    def test_bad_shapes_are_refused_with_a_safe_message(self, db):
        for bad in ("nope", "a@b", "two@@x.com", "sp ace@x.com", "", "@x.com"):
            with pytest.raises(svc.InvalidSubscription):
                svc.subscribe(
                    db,
                    SubscribeIn(**payload(email=bad or "x"))
                    if bad
                    else SubscribeIn(**payload(email="   ")),
                )
        with pytest.raises(svc.InvalidSubscription):
            svc.subscribe(db, SubscribeIn(**payload(home_airport="LHRX")))
        with pytest.raises(svc.InvalidSubscription):
            svc.subscribe(db, SubscribeIn(**payload(consent=False)))
        assert db.scalar(select(Subscriber)) is None

    def test_second_submission_refreshes_and_resubscribes(self, db):
        first = svc.subscribe(db, SubscribeIn(**payload()))
        assert svc.unsubscribe(db, "SOMEONE@example.com") is True
        assert svc.unsubscribe(db, "someone@example.com") is False
        again = svc.subscribe(db, SubscribeIn(**payload(first_name="Augusta", source="article")))
        assert again.id == first.id and again.first_name == "Augusta" and again.source == "article"
        assert again.unsubscribed_at is None
        assert db.query(Subscriber).count() == 1

    def test_export_leaves_out_withdrawn_rows_unless_asked(self, db):
        svc.subscribe(db, SubscribeIn(**payload()))
        svc.subscribe(db, SubscribeIn(**payload(email="gone@example.com", interests=[])))
        svc.unsubscribe(db, "gone@example.com")
        csv_text = svc.export_csv(db)
        lines = csv_text.strip().split("\n")
        assert lines[0] == ",".join(svc.EXPORT_COLUMNS)
        assert len(lines) == 2 and lines[1].startswith(
            "someone@example.com,Ada,Lovelace,LHR,whisky; fragrance,"
        )
        assert "gone@example.com" in svc.export_csv(db, include_unsubscribed=True)
        counts = svc.stats(db)
        assert counts["total"] == 2 and counts["active"] == 1 and counts["source:home"] == 1


class TestRoute:
    def test_new_and_existing_addresses_get_the_same_answer(self, client, db):
        a = client.post("/api/subscribers", json=payload())
        b = client.post("/api/subscribers", json=payload())
        assert a.status_code == 200 and a.json() == b.json() and a.json()["ok"] is True
        assert db.query(Subscriber).count() == 1

    def test_honeypot_answers_success_and_stores_nothing(self, client, db):
        r = client.post("/api/subscribers", json=payload(website="https://spam.example"))
        assert r.status_code == 200 and r.json()["ok"] is True
        assert db.query(Subscriber).count() == 0

    def test_shape_errors_are_422_and_name_no_address(self, client, db):
        r = client.post("/api/subscribers", json=payload(email="not-an-address"))
        assert r.status_code == 422 and "not-an-address" not in r.text
        r = client.post("/api/subscribers", json=payload(consent=False))
        assert r.status_code == 422
        r = client.post("/api/subscribers", json={"email": "a@b.co"})  # no consent text at all
        assert r.status_code == 422
        assert db.query(Subscriber).count() == 0

    def test_throttle_per_client_then_429(self, client, db):
        for n in range(5):
            assert (
                client.post("/api/subscribers", json=payload(email=f"p{n}@example.com")).status_code
                == 200
            )
        r = client.post("/api/subscribers", json=payload(email="p6@example.com"))
        assert r.status_code == 429
        # Another client (as Caddy reports it) still gets through.
        r = client.post(
            "/api/subscribers",
            json=payload(email="p7@example.com"),
            headers={"X-Forwarded-For": "203.0.113.9"},
        )
        assert r.status_code == 200
        assert db.query(Subscriber).count() == 6

    def test_no_log_line_carries_the_address(self, client, caplog):
        with caplog.at_level(logging.DEBUG):
            client.post("/api/subscribers", json=payload(email="secret.person@example.com"))
        assert "secret.person" not in caplog.text
        assert "subscriber" in caplog.text and "source=home" in caplog.text

    def test_there_is_no_read_route(self, client):
        assert client.get("/api/subscribers").status_code in (404, 405)


class TestThrottle:
    def test_windows_slide_and_global_budget_holds(self):
        t = Throttle(per_key=2, per_key_seconds=10, total=3, total_seconds=10)
        assert t.allow("a", now=0) and t.allow("a", now=1) and not t.allow("a", now=2)
        assert t.allow("b", now=2) and not t.allow("c", now=3)  # global budget of 3 spent
        assert t.allow("a", now=11)  # a's first attempt has slid out; global too


def _web(name: str) -> str:
    return (MAIN / "web" / "src" / name).read_text()


def _consent_text() -> str:
    """The one constant, read out of the SPA source: there is no frontend test runner here."""
    found = re.search(r'export const CONSENT_TEXT\s*=\s*\n?\s*"([^"]+)"', _web("api/editorial.ts"))
    assert found, "CONSENT_TEXT is no longer a single double-quoted string in api/editorial.ts"
    return found.group(1)


class TestTheConsentSentence:
    """The sentence shown beside the checkbox and the sentence stored as consent are one constant.

    Consent is only provable if the words kept are the words shown: `consent_text` is written
    to the row verbatim from what the form sent, and nothing re-derives it later. A form that
    printed one sentence and posted another would leave every signup after that edit with a
    record of something nobody was shown, and the rows would be indistinguishable from the
    honest ones. `SubscribeForm.tsx` reads `CONSENT_TEXT` in both places; this is what keeps
    a literal from creeping into either.
    """

    def test_the_form_shows_and_sends_the_same_constant(self):
        form = _web("components/SubscribeForm.tsx")
        assert "consent_text: CONSENT_TEXT," in form, "the form posts something other than CONSENT_TEXT"
        label = form.split('className="subscribe-form__consent"', 1)[1].split("</label>", 1)[0]
        assert "{CONSENT_TEXT}" in label, "the checkbox no longer prints CONSENT_TEXT"
        assert 'consent_text: "' not in form, "a literal consent sentence is being posted"

    def test_the_sentence_the_form_sends_is_stored_word_for_word(self, db):
        sentence = _consent_text()
        assert 10 <= len(sentence) <= 400  # the SubscribeIn bounds, so the form can never be refused on it
        row = svc.subscribe(db, SubscribeIn(**payload(consent_text=sentence)))
        assert row.consent_text == sentence
        assert sentence in svc.export_csv(db)  # and it is what the export hands over as the proof


class TestNothingElseTouchesTheList:
    """The table is declared in the model, reached only through the service, and left only
    by the owner's CLI.

    The list is personal data with no route in front of it and no mailer behind it: there is
    no GET, no count, no lookup, and nothing that could send to it by accident. Those are
    properties of the whole app, not of one module, so they are checked as such rather than
    remembered.
    """

    def test_only_the_service_and_the_cli_read_the_table(self):
        readers = sorted(
            str(p.relative_to(MAIN))
            for p in (MAIN / "app").rglob("*.py")
            if re.search(r"\bSubscriber\b|\bexport_csv\b", p.read_text())
        )
        # The model that declares it, the service that is its one door, and the owner's CLI.
        # The route never names the table: it hands the form's payload to the service.
        assert readers == [
            "app/cli_editorial.py",
            "app/models/editorial.py",
            "app/services/subscribers.py",
        ], readers

    def test_no_mailer_stands_behind_the_form(self):
        for name in ("app/services/subscribers.py", "app/routers/subscribers.py", "app/cli_editorial.py"):
            text = (MAIN / name).read_text()
            assert "services.mail" not in text and "import mail" not in text, name

    def test_the_route_table_offers_no_way_to_read_it(self):
        keys = [k for k in access.PERMISSION if "subscriber" in k]
        assert keys == [], keys
        assert "POST /api/subscribers" in access.PUBLIC_WHEN_OPEN
        assert "POST /api/subscribers" in access.PUBLIC_WRITES_WHEN_OPEN


class TestTheGateInFrontOfTheForm:
    """Capture follows `SITE_ACCESS`, like every storefront read.

    While the site is members-only the form is not a way in: an anonymous post is refused
    401 before the route is reached, so nothing is captured until the go-live flip. Worth
    pinning because it reads like a broken form (rehearsed on a restored copy, 19 Sep: the
    same post answered 200 with `public` and 401 with `members`), and the wrong repair would
    be to move the route into `public_always`, which would open the one write that stores a
    person's details to anyone, in any mode, before launch.
    """

    def _decide(self, **kw):
        base = dict(method="POST", key="POST /api/subscribers", page=None, site_open=False,
                    signed_in=False, active=False, must_change=False, holds=lambda p: False,
                    read_only=False, origin_ok=True, wants_html=False)
        base.update(kw)
        return access.decide(**base)

    def test_open_for_anyone_when_the_site_is_open_and_for_members_before_that(self):
        assert self._decide(site_open=True) is None
        assert self._decide().code == "NOT_SIGNED_IN"
        assert self._decide().status == 401
        assert self._decide(signed_in=True, active=True) is None