"""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.
"""

import logging

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 subscribers as svc
from app.services.throttle import Throttle

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