"""Tests sans réseau : enveloppes, erreurs définitives/passagères, déduplication."""

import base64
import json

import pytest

from pipeline import Analyse, Event, PoisonMessage, Usage, handle, parse_envelope


def envelope(message_id="m1", event_type="OBJECT_FINALIZE", size=120, ctype="text/plain", attempt=None):
    meta = {"bucket": "b", "name": "entrant/f.txt", "generation": "17", "contentType": ctype, "size": str(size)}
    body = {"message": {"messageId": message_id, "publishTime": "2026-10-15T08:00:00Z",
                        "attributes": {"eventType": event_type},
                        "data": base64.b64encode(json.dumps(meta).encode()).decode()},
            "subscription": "projects/p/subscriptions/s"}
    if attempt:
        body["deliveryAttempt"] = attempt
    return body


class FakeStore:
    """Imite la contrainte du MERGE : une seule ligne par message_id."""

    def __init__(self):
        self.rows = {}

    def seen(self, message_id):
        return message_id in self.rows

    def record(self, event, status, reason=None, analyse=None, usage=None):
        if event.message_id in self.rows:
            return False
        self.rows[event.message_id] = (status, reason, analyse)
        return True


class FakeReader:
    def read(self, bucket, name, generation):
        return b"FACTURE 2026-118 Total TTC 1 440,00 EUR"


class FakeAnalyzer:
    def __init__(self):
        self.calls = 0

    def __call__(self, data, mime_type):
        self.calls += 1
        return (Analyse(type_document="facture", resume="Facture.", montant_ttc_eur=1440.0, urgence="normale", tiers="X"),
                Usage("fake", 10, 5, 1))


def test_parse_envelope_lit_les_metadonnees_storage():
    e = parse_envelope(envelope(attempt=3))
    assert (e.message_id, e.bucket, e.generation, e.size, e.delivery_attempt) == ("m1", "b", "17", 120, 3)
    assert e.uri == "gs://b/entrant/f.txt#17"


def test_evenement_autre_que_finalize_est_ignore():
    assert parse_envelope(envelope(event_type="OBJECT_DELETE")) is None


@pytest.mark.parametrize("body", [{}, {"message": {"messageId": "x", "data": "e30="}},
                                  {"message": {"messageId": "x", "attributes": {"eventType": "OBJECT_FINALIZE"}, "data": "%%%"}}])
def test_enveloppe_illisible_est_un_message_empoisonne(body):
    with pytest.raises(PoisonMessage):
        parse_envelope(body)


def test_document_valide_est_analyse_une_fois():
    store, analyzer = FakeStore(), FakeAnalyzer()
    e = parse_envelope(envelope())
    assert handle(e, store, FakeReader(), analyzer) == "ok"
    assert handle(e, store, FakeReader(), analyzer) == "duplicate"  # redélivrance du même message
    assert analyzer.calls == 1 and len(store.rows) == 1


def test_livraisons_simultanees_un_seul_enregistrement():
    """Les deux livraisons passent seen() ; record() tranche."""
    store, analyzer = FakeStore(), FakeAnalyzer()
    e = parse_envelope(envelope())
    store.seen = lambda _id: False  # simule la course : aucune ne voit l'autre
    assert sorted([handle(e, store, FakeReader(), analyzer), handle(e, store, FakeReader(), analyzer)]) == ["duplicate", "ok"]
    assert len(store.rows) == 1


@pytest.mark.parametrize("size,ctype,reason", [(0, "text/plain", "vide"), (10**8, "text/plain", "trop gros"),
                                               (50, "image/png", "non pris en charge")])
def test_rejet_definitif_sans_appel_gemini(size, ctype, reason):
    store, analyzer = FakeStore(), FakeAnalyzer()
    assert handle(parse_envelope(envelope(size=size, ctype=ctype)), store, FakeReader(), analyzer) == "rejected"
    assert analyzer.calls == 0 and reason in store.rows["m1"][1]


def test_erreur_passagere_remonte_pour_reessai():
    class Quota(FakeAnalyzer):
        def __call__(self, data, mime_type):
            raise RuntimeError("429 RESOURCE_EXHAUSTED")
    store = FakeStore()
    with pytest.raises(RuntimeError):
        handle(parse_envelope(envelope()), store, FakeReader(), Quota())
    assert store.rows == {}  # rien d'écrit : la prochaine livraison repartira de zéro


def test_sortie_gemini_corrompue_declenche_un_reessai():
    """Cas observé le 2026-10-15 : « RHÔNE » rendu « RH\\u0014NE » par la sortie structurée."""
    from types import SimpleNamespace

    from gcp import GeminiAnalyzer

    bad = '{"type_document":"facture","resume":"x","montant_ttc_eur":1.0,"urgence":"basse","tiers":"RH\\u0014NE"}'
    fake = SimpleNamespace(models=SimpleNamespace(generate_content=lambda **kw: SimpleNamespace(
        text=bad, usage_metadata=SimpleNamespace(prompt_token_count=1, candidates_token_count=1))))
    with pytest.raises(RuntimeError, match="contrôle"):
        GeminiAnalyzer(fake, "m")("RHÔNE".encode(), "text/plain")
    with pytest.raises(Exception, match="UTF-8"):
        GeminiAnalyzer(fake, "m")("RHÔNE".encode("latin-1"), "text/plain")