"""Briques transverses : rate limit persistant, IP derrière un proxy, config de production,
élection du leader, alertes, organizer (codes QR), e-mails."""
import asyncio
import types

import pytest
from fastapi import HTTPException

from core import config, security
from core.rate_limit import enforce, is_blocked, record_hit


# ── Rate limit ──────────────────────────────────────────────────────────────────

def test_rate_limit_blocks_after_max_and_is_shared_through_the_database():
    for _ in range(3):
        enforce("k1", max_attempts=3, window_seconds=60)
    with pytest.raises(HTTPException) as e:
        enforce("k1", max_attempts=3, window_seconds=60)
    assert e.value.status_code == 429
    # une clé différente n'est pas affectée
    enforce("k2", max_attempts=3, window_seconds=60)


def test_rate_limit_state_survives_a_new_connection_like_a_restart():
    for _ in range(2):
        enforce("persist", max_attempts=2, window_seconds=60)
    # simule un autre worker / redémarrage : tout l'état est en base
    from core.db import get_db
    with get_db() as conn:
        assert conn.execute("SELECT COUNT(*) AS n FROM rate_hits WHERE key='persist'").fetchone()["n"] == 2
    assert is_blocked("persist", 2, 60)


def test_rate_limit_window_expires():
    import time
    from core.db import get_db
    record_hit("old")
    with get_db() as conn:
        conn.execute("UPDATE rate_hits SET ts=? WHERE key='old'", (time.time() - 120,))
        conn.commit()
    assert not is_blocked("old", 1, 60)


# ── IP client ───────────────────────────────────────────────────────────────────

def _req(peer, xff=None):
    headers = {"x-forwarded-for": xff} if xff else {}
    return types.SimpleNamespace(client=types.SimpleNamespace(host=peer), headers=headers)


def test_client_ip_ignores_forwarded_header_from_untrusted_peer():
    assert security.client_ip(_req("203.0.113.9", "1.2.3.4")) == "203.0.113.9"


def test_client_ip_uses_forwarded_header_behind_a_trusted_proxy():
    assert security.client_ip(_req("127.0.0.1", "198.51.100.7")) == "198.51.100.7"


def test_client_ip_cannot_be_spoofed_through_a_trusted_proxy():
    # le client envoie « 9.9.9.9 », le proxy ajoute la vraie IP à droite
    assert security.client_ip(_req("127.0.0.1", "9.9.9.9, 198.51.100.7")) == "198.51.100.7"


def test_client_ip_without_forwarded_header():
    assert security.client_ip(_req("127.0.0.1")) == "127.0.0.1"


# ── Mots de passe / signatures ─────────────────────────────────────────────────

def test_password_hash_roundtrip():
    h = security.hash_password("un-mot-de-passe")
    assert h.startswith("scrypt$") and "un-mot-de-passe" not in h
    assert security.verify_password("un-mot-de-passe", h)
    assert not security.verify_password("autre", h)
    assert not security.verify_password("x", None)
    assert not security.verify_password("x", "garbage")


def test_signed_urls_expire_and_are_payload_bound():
    exp, sig = security.sign("payload", 60)
    assert security.verify_signature("payload", exp, sig)
    assert not security.verify_signature("autre", exp, sig)
    assert not security.verify_signature("payload", exp, sig[:-1] + ("1" if sig[-1] == "0" else "0"))
    exp2, sig2 = security.sign("payload", -5)
    assert not security.verify_signature("payload", exp2, sig2)


# ── Config de production ───────────────────────────────────────────────────────

def test_production_validation_flags_every_weak_setting(monkeypatch):
    monkeypatch.setattr(config, "ADMIN_PASSWORD", "meeconic2026")
    monkeypatch.setattr(config, "ADMIN_JWT_SECRET", "change-me-generate-a-random-secret")
    monkeypatch.setattr(config, "STRIPE_SECRET_KEY", "")
    monkeypatch.setattr(config, "STRIPE_WEBHOOK_SECRET", "")
    assert len(security.validate_production_config()) == 4


def test_production_validation_passes_with_a_strong_config(monkeypatch):
    monkeypatch.setattr(config, "ADMIN_PASSWORD", "un-vrai-mot-de-passe-long")
    monkeypatch.setattr(config, "ADMIN_JWT_SECRET", "x" * 40)
    monkeypatch.setattr(config, "STRIPE_SECRET_KEY", "sk_live_x")
    monkeypatch.setattr(config, "STRIPE_WEBHOOK_SECRET", "whsec_x")
    assert security.validate_production_config() == []


async def test_production_refuses_to_start_with_a_bad_config(monkeypatch):
    import main
    monkeypatch.setattr(main, "IS_PRODUCTION", True)
    monkeypatch.setattr(config, "ADMIN_PASSWORD", "admin")
    with pytest.raises(RuntimeError, match="Configuration de production invalide"):
        async with main.lifespan(main.app):
            pass


async def test_lifespan_starts_and_stops_background_tasks_cleanly(monkeypatch):
    import main
    monkeypatch.setattr(config, "BACKGROUND_TASKS_ENABLED", True)
    monkeypatch.setattr(config, "ORGANIZER_ENABLED", False)
    async with main.lifespan(main.app):
        await asyncio.sleep(0.3)
        names = {t.get_name() for t in asyncio.all_tasks()}
        assert {"leader-cleanup", "leader-backup", "leader-monitor"} <= names
    await asyncio.sleep(0.1)
    assert not any(t.get_name().startswith("leader-") for t in asyncio.all_tasks())


# ── Leader ─────────────────────────────────────────────────────────────────────

def test_only_one_worker_can_hold_a_leader_lock(monkeypatch):
    from core import leader
    assert leader._try_acquire("t-lock") is True
    assert leader.is_leader("t-lock")
    monkeypatch.setattr(leader, "_OWNER", "autre-worker")
    assert leader._try_acquire("t-lock") is False       # déjà détenu
    assert not leader.is_leader("t-lock")
    from core.db import get_db
    with get_db() as conn:                              # le bail du premier expire → reprise
        conn.execute("UPDATE leader_locks SET expires_at=0 WHERE name='t-lock'")
        conn.commit()
    assert leader._try_acquire("t-lock") is True


# ── Alertes ────────────────────────────────────────────────────────────────────

def test_alerts_are_deduplicated_by_cooldown(monkeypatch):
    from services import alert_service
    sent = []
    monkeypatch.setattr(config, "APP_ENV", "production")
    monkeypatch.setattr(config, "ALERT_EMAIL", "ops@example.com")
    monkeypatch.setattr(alert_service, "_smtp_send", lambda subject, body: sent.append(subject))
    import threading
    monkeypatch.setattr(threading, "Thread", lambda target, args, daemon: types.SimpleNamespace(start=lambda: target(*args)))
    alert_service.notify("same-key-xyz", "Sujet 1", "corps")
    alert_service.notify("same-key-xyz", "Sujet 2", "corps")
    alert_service.notify("autre-cle-xyz", "Sujet 3", "corps")
    assert sent == ["Sujet 1", "Sujet 3"]


def test_alert_never_raises_even_if_smtp_blows_up(monkeypatch):
    from services import alert_service
    monkeypatch.setattr(config, "APP_ENV", "production")
    monkeypatch.setattr(config, "ALERT_EMAIL", "ops@example.com")
    monkeypatch.setattr(alert_service, "_cooldown_ok", lambda key: (_ for _ in ()).throw(RuntimeError("boom")))
    alert_service.notify("k", "s", "b")   # ne doit pas lever


async def test_unhandled_error_returns_500_and_raises_an_alert(client, monkeypatch):
    import main
    seen = []
    monkeypatch.setattr(main.alert_service, "notify", lambda key, subject, body="": seen.append(key))

    @main.app.get("/api/_boom")
    async def boom():
        raise RuntimeError("kaboom")

    from httpx import ASGITransport, AsyncClient
    async with AsyncClient(transport=ASGITransport(app=main.app, raise_app_exceptions=False), base_url="http://t") as c:
        r = await c.get("/api/_boom")
    assert r.status_code == 500
    assert r.json() == {"detail": "Erreur interne du serveur."}
    assert any(k.startswith("http-500") for k in seen)


# ── E-mails ────────────────────────────────────────────────────────────────────

def test_order_emails_escape_user_supplied_html():
    from services import email_service
    html_out = email_service._confirmation_html(
        '<script>alert(1)</script>', "abc12345", "TST-1001",
        [{"format_label": '<img src=x onerror=alert(1)>', "quantity": 1, "unit_price": 9, "is_magic": False}], 9.0)
    assert "<script>" not in html_out and "<img src=x" not in html_out
    assert "&lt;script&gt;" in html_out


def test_confirmation_email_mentions_downloads_only_for_digital_orders():
    from services import email_service
    digital = email_service._confirmation_html("A", "abc12345", "TST-1001", [], 7.0, True, "https://p.example")
    paper = email_service._confirmation_html("A", "abc12345", "TST-1001", [], 9.0, False, "https://p.example")
    assert "téléchargement" in digital and "https://p.example/?code=TST-1001" in digital
    assert "téléchargement" not in paper


def test_order_email_attaches_originals_within_size_limit(monkeypatch, tmp_path):
    import main
    from core import config
    from services import media_service

    big, small = tmp_path / "big.jpg", tmp_path / "small.jpg"
    big.write_bytes(b"x" * 600_000)
    small.write_bytes(b"y" * 1_000)
    paths = {"big.jpg": big, "small.jpg": small}
    monkeypatch.setattr(media_service, "safe_photo_path", lambda sid, code, name: paths.get(name))
    entitled = [{"photo_filename": "small.jpg"}, {"photo_filename": "gone.jpg"}, {"photo_filename": "big.jpg"}]

    # A purged original is skipped; the rest are attached.
    monkeypatch.setattr(config, "EMAIL_ATTACH_MAX_MB", 1)
    got = main._collect_attachments("s", "TST-1", entitled)
    assert [n for n, _ in got] == ["small.jpg", "big.jpg"]

    # Over the limit nothing is attached (all-or-nothing, the family uses the portal).
    monkeypatch.setattr(config, "EMAIL_ATTACH_MAX_MB", 0)
    assert main._collect_attachments("s", "TST-1", entitled) == []


def test_confirmation_email_mentions_attached_photos():
    from services import email_service
    out = email_service._confirmation_html("A", "abc12345", "TST-1001", [], 7.0, True, "https://p.example", attached=2)
    assert "jointes à cet e-mail (2)" in out