zonza/tests/test_gemini.py

"""Moteur Gemini : la transcription part chez Google, avec repli local.

Aucun test ne touche le réseau : l'envoi HTTP et la lecture de la clé sont
injectés. Ce qu'on éprouve ici, c'est la forme de la requête, la lecture de
la réponse, et surtout le repli — une dictée ne doit JAMAIS être perdue parce
que le réseau ou l'API tombe.
"""
import base64
import io
import wave

import numpy as np
import pytest

from moteur import choisir_transcripteur, detecter_moteur
from moteur.gemini import (
    ErreurGemini, TranscripteurAvecRepli, TranscripteurGemini, audio_en_wav,
    lire_cle,
)


def _reponse(texte):
    return {"candidates": [{"content": {"parts": [{"text": texte}]}}]}


def _transcripteur(envoyer, cle="cle-test"):
    return TranscripteurGemini(
        "gemini-x", "fr", "Vocabulaire : Gitea, iCloud.",
        lire_cle=lambda: cle, envoyer=envoyer,
    )


# --- encodage audio ---------------------------------------------------------

def test_l_audio_devient_un_wav_16_bits_mono_a_16_khz():
    audio = np.array([0.0, 0.5, -0.5, 1.0], dtype="float32")
    with wave.open(io.BytesIO(audio_en_wav(audio, 16000))) as w:
        assert (w.getnchannels(), w.getsampwidth(), w.getframerate()) == (1, 2, 16000)
        echantillons = np.frombuffer(w.readframes(4), dtype="<i2")
    assert list(echantillons) == [0, 16383, -16383, 32767]


def test_un_audio_sature_est_borne_au_lieu_de_deborder():
    with wave.open(io.BytesIO(audio_en_wav(np.array([3.0, -3.0], dtype="float32"), 16000))) as w:
        assert list(np.frombuffer(w.readframes(2), dtype="<i2")) == [32767, -32767]


# --- requête et réponse -----------------------------------------------------

def test_la_requete_porte_l_audio_le_modele_et_le_vocabulaire():
    vu = {}

    def envoyer(url, cle, corps, delai_s):
        vu.update(url=url, cle=cle, corps=corps)
        return _reponse("Bonjour.")

    _transcripteur(envoyer).transcrire(np.zeros(1600, dtype="float32"))

    assert vu["url"].endswith("/models/gemini-x:generateContent")
    assert vu["cle"] == "cle-test"
    parts = vu["corps"]["contents"][0]["parts"]
    audio = next(p["inline_data"] for p in parts if "inline_data" in p)
    assert audio["mime_type"] == "audio/wav"
    assert base64.b64decode(audio["data"])[:4] == b"RIFF"
    consigne = vu["corps"]["system_instruction"]["parts"][0]["text"]
    assert "Gitea" in consigne and "iCloud" in consigne


def test_la_cle_ne_part_jamais_dans_l_url():
    vu = {}
    _transcripteur(lambda url, cle, corps, delai_s: vu.setdefault("url", url) and _reponse("x")) \
        .transcrire(np.zeros(1600, dtype="float32"))
    assert "cle-test" not in vu["url"]


def test_le_texte_rendu_est_nettoye_et_les_morceaux_joints():
    reponse = {"candidates": [{"content": {"parts": [{"text": "Je pousse "}, {"text": "sur Gitea.  "}]}}]}
    t = _transcripteur(lambda *a: reponse)
    assert t.transcrire(np.zeros(1600, dtype="float32")) == "Je pousse sur Gitea."


def test_une_reponse_sans_candidat_est_une_erreur_explicite():
    t = _transcripteur(lambda *a: {"promptFeedback": {"blockReason": "OTHER"}})
    with pytest.raises(ErreurGemini) as e:
        t.transcrire(np.zeros(1600, dtype="float32"))
    assert "OTHER" in str(e.value)


def test_une_cle_absente_est_signalee_des_le_warmup():
    t = _transcripteur(lambda *a: _reponse("x"), cle=None)
    with pytest.raises(ErreurGemini) as e:
        t.warmup()
    assert "zonza-gemini" in str(e.value)


def test_le_warmup_ne_consomme_aucun_appel_payant():
    appels = []
    _transcripteur(lambda *a: appels.append(a) or _reponse("x")).warmup()
    assert appels == []


# --- lecture de la clé ------------------------------------------------------

def test_la_cle_du_trousseau_prime_sur_l_environnement():
    assert lire_cle(trousseau=lambda: "du-trousseau", environ={"GEMINI_API_KEY": "env"}) == "du-trousseau"


def test_sans_trousseau_on_se_rabat_sur_l_environnement():
    assert lire_cle(trousseau=lambda: None, environ={"GEMINI_API_KEY": "env"}) == "env"


def test_sans_aucune_cle_on_rend_none():
    assert lire_cle(trousseau=lambda: None, environ={}) is None


# --- repli local ------------------------------------------------------------

class _Local:
    def __init__(self):
        self.appels = 0

    def warmup(self):
        pass

    def transcrire(self, audio):
        self.appels += 1
        return "texte local"


def test_si_gemini_tombe_la_dictee_passe_par_le_moteur_local():
    def en_panne(audio):
        raise ErreurGemini("réseau coupé")

    principal = type("P", (), {"warmup": lambda s: None, "transcrire": lambda s, a: en_panne(a)})()
    local = _Local()
    t = TranscripteurAvecRepli(principal, fabriquer_secours=lambda: local)
    assert t.transcrire(np.zeros(10, dtype="float32")) == "texte local"


def test_le_moteur_local_n_est_charge_qu_au_premier_repli():
    fabrications = []
    principal = type("P", (), {"warmup": lambda s: None, "transcrire": lambda s, a: "de gemini"})()
    t = TranscripteurAvecRepli(principal, fabriquer_secours=lambda: fabrications.append(1) or _Local())
    t.warmup()
    assert t.transcrire(np.zeros(10, dtype="float32")) == "de gemini"
    assert fabrications == []


def test_une_cle_absente_au_warmup_ne_bloque_pas_le_demarrage():
    def sans_cle(self):
        raise ErreurGemini("aucune clé")

    principal = type("P", (), {"warmup": sans_cle, "transcrire": lambda s, a: "x"})()
    local = _Local()
    t = TranscripteurAvecRepli(principal, fabriquer_secours=lambda: local)
    t.warmup()  # ne lève pas : le moteur local prendra le relais
    assert t.secours_charge


# --- sélection du moteur ----------------------------------------------------

def test_le_forcage_de_config_prime_sur_la_detection_materielle():
    assert detecter_moteur(forcage="gemini") == "gemini"


def test_la_fabrique_rend_gemini_enveloppe_d_un_repli():
    t = choisir_transcripteur(nom="gemini", modele="gemini-x", langue="fr", amorce="")
    assert isinstance(t, TranscripteurAvecRepli)
    assert isinstance(t.principal, TranscripteurGemini)
    assert callable(t.warmup) and callable(t.transcrire)