327 lines
11 KiB
Python
327 lines
11 KiB
Python
"""`zonza --diagnostic` : enchaînement des épreuves et mise en forme.
|
||
|
||
Chaque épreuve qui toucherait normalement du matériel réel (micro, moteur de
|
||
transcription, presse-papier, verrou) est ici exercée avec une doublure
|
||
injectée — ces tests éprouvent l'ORCHESTRATION (ordre, propagation d'un échec,
|
||
non-invention d'un résultat), pas le matériel lui-même.
|
||
"""
|
||
import numpy as np
|
||
import pytest
|
||
|
||
from diagnostic import (
|
||
ResultatEpreuve,
|
||
diagnostiquer,
|
||
epreuve_micro,
|
||
epreuve_moteur,
|
||
epreuve_plateforme,
|
||
epreuve_presse_papier,
|
||
epreuve_raccourci,
|
||
epreuve_transcription,
|
||
epreuve_verrou,
|
||
formater_resultat,
|
||
mettre_en_forme,
|
||
)
|
||
|
||
|
||
# --- epreuve_plateforme ------------------------------------------------------
|
||
|
||
def test_epreuve_plateforme_rapporte_cuda_present():
|
||
r = epreuve_plateforme(plateforme="win32", machine="AMD64", cuda_disponible=lambda: True)
|
||
assert r.reussite is True
|
||
assert r.detail == "win32 / AMD64 / CUDA présent"
|
||
|
||
|
||
def test_epreuve_plateforme_rapporte_cuda_absent():
|
||
r = epreuve_plateforme(plateforme="darwin", machine="arm64", cuda_disponible=lambda: False)
|
||
assert "CUDA absent" in r.detail
|
||
|
||
|
||
def test_epreuve_plateforme_non_concluante_si_la_sonde_cuda_leve():
|
||
def leve():
|
||
raise RuntimeError("ctypes en échec")
|
||
r = epreuve_plateforme(plateforme="win32", machine="AMD64", cuda_disponible=leve)
|
||
assert r.reussite is None
|
||
|
||
|
||
# --- epreuve_moteur -----------------------------------------------------------
|
||
|
||
def test_epreuve_moteur_rapporte_le_nom_et_le_modele():
|
||
r, transcripteur = epreuve_moteur(
|
||
detecter_moteur=lambda: "faster-cpu",
|
||
resoudre_modele=lambda nom, forcage: "Systran/faster-whisper-small",
|
||
creer_transcripteur=lambda **kw: "un-faux-transcripteur",
|
||
)
|
||
assert r.reussite is True
|
||
assert r.detail == "faster-cpu + Systran/faster-whisper-small"
|
||
assert transcripteur == "un-faux-transcripteur"
|
||
|
||
|
||
def test_epreuve_moteur_non_concluante_si_la_detection_echoue():
|
||
def leve():
|
||
raise RuntimeError("aucun moteur compatible")
|
||
r, transcripteur = epreuve_moteur(detecter_moteur=leve)
|
||
assert r.reussite is None
|
||
assert transcripteur is None
|
||
|
||
|
||
# --- epreuve_micro -------------------------------------------------------------
|
||
|
||
class _FauxFlux:
|
||
def __init__(self, callback, n_blocs):
|
||
self._callback = callback
|
||
self._n_blocs = n_blocs
|
||
|
||
def start(self):
|
||
for _ in range(self._n_blocs):
|
||
self._callback(np.zeros(10, dtype="float32"), 10, None, None)
|
||
|
||
def stop(self):
|
||
pass
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
|
||
def test_epreuve_micro_compte_les_blocs_recus():
|
||
def ouvrir_flux(**kw):
|
||
return _FauxFlux(kw["callback"], n_blocs=7)
|
||
r = epreuve_micro(ouvrir_flux=ouvrir_flux, dormir=lambda s: None)
|
||
assert r.reussite is True
|
||
assert r.detail == "7 blocs reçus en 1 s"
|
||
|
||
|
||
def test_epreuve_micro_echoue_si_aucun_bloc_recu():
|
||
def ouvrir_flux(**kw):
|
||
return _FauxFlux(kw["callback"], n_blocs=0)
|
||
r = epreuve_micro(ouvrir_flux=ouvrir_flux, dormir=lambda s: None)
|
||
assert r.reussite is False
|
||
|
||
|
||
def test_epreuve_micro_non_concluante_si_louverture_echoue():
|
||
def ouvrir_flux(**kw):
|
||
raise OSError("aucun périphérique")
|
||
r = epreuve_micro(ouvrir_flux=ouvrir_flux, dormir=lambda s: None)
|
||
assert r.reussite is None
|
||
|
||
|
||
def test_epreuve_micro_utilise_bien_la_duree_demandee():
|
||
appels = []
|
||
|
||
def ouvrir_flux(**kw):
|
||
return _FauxFlux(kw["callback"], n_blocs=1)
|
||
|
||
def dormir(s):
|
||
appels.append(s)
|
||
|
||
epreuve_micro(ouvrir_flux=ouvrir_flux, duree_s=2.5, dormir=dormir)
|
||
assert appels == [2.5]
|
||
|
||
|
||
# --- epreuve_transcription -----------------------------------------------------
|
||
|
||
class _FauxTranscripteur:
|
||
def __init__(self, texte="bonjour", leve=False):
|
||
self._texte = texte
|
||
self._leve = leve
|
||
|
||
def transcrire(self, audio):
|
||
if self._leve:
|
||
raise RuntimeError("modèle indisponible")
|
||
return self._texte
|
||
|
||
|
||
def test_epreuve_transcription_rapporte_duree_et_texte():
|
||
horloges = iter([10.0, 10.8])
|
||
r = epreuve_transcription(
|
||
_FauxTranscripteur(texte="bonjour le monde"),
|
||
generer_parole=lambda: np.zeros(16000, dtype="float32"),
|
||
horloge=lambda: next(horloges),
|
||
)
|
||
assert r.reussite is True
|
||
assert "0.80 s" in r.detail
|
||
assert "bonjour le monde" in r.detail
|
||
|
||
|
||
def test_epreuve_transcription_retombe_sur_le_silence_si_la_synthese_echoue():
|
||
def leve():
|
||
raise RuntimeError("`say` absent")
|
||
r = epreuve_transcription(
|
||
_FauxTranscripteur(texte=""),
|
||
generer_parole=leve,
|
||
horloge=lambda: 0.0,
|
||
)
|
||
assert r.reussite is True
|
||
assert "silence" in r.detail
|
||
|
||
|
||
def test_epreuve_transcription_non_concluante_si_la_transcription_echoue():
|
||
r = epreuve_transcription(
|
||
_FauxTranscripteur(leve=True),
|
||
generer_parole=lambda: np.zeros(16000, dtype="float32"),
|
||
horloge=lambda: 0.0,
|
||
)
|
||
assert r.reussite is None
|
||
|
||
|
||
# --- epreuve_presse_papier -------------------------------------------------------
|
||
|
||
def test_epreuve_presse_papier_reussit_si_la_relecture_est_identique():
|
||
stockage = {}
|
||
r = epreuve_presse_papier(
|
||
copier=lambda t: stockage.__setitem__("v", t),
|
||
lire=lambda: stockage["v"],
|
||
)
|
||
assert r.reussite is True
|
||
|
||
|
||
def test_epreuve_presse_papier_echoue_si_la_relecture_differe():
|
||
r = epreuve_presse_papier(copier=lambda t: None, lire=lambda: "autre chose")
|
||
assert r.reussite is False
|
||
|
||
|
||
def test_epreuve_presse_papier_non_concluante_si_la_copie_leve():
|
||
def leve(t):
|
||
raise RuntimeError("presse-papier indisponible")
|
||
r = epreuve_presse_papier(copier=leve, lire=lambda: "")
|
||
assert r.reussite is None
|
||
|
||
|
||
# --- epreuve_raccourci -----------------------------------------------------------
|
||
|
||
class _FauxSysteme:
|
||
def __init__(self, raccourci="<ctrl>+<alt>+d", verrou_sequence=None):
|
||
self._raccourci = raccourci
|
||
self._verrou_sequence = iter(verrou_sequence or [])
|
||
|
||
def raccourci_defaut(self):
|
||
return self._raccourci
|
||
|
||
def acquerir_verrou_instance(self, chemin):
|
||
return next(self._verrou_sequence)
|
||
|
||
|
||
def test_epreuve_raccourci_rapporte_la_valeur_de_la_plateforme():
|
||
r = epreuve_raccourci(_FauxSysteme(raccourci="<ctrl>+<alt>+d"))
|
||
assert r.reussite is True
|
||
assert r.detail == "<ctrl>+<alt>+d"
|
||
|
||
|
||
# --- epreuve_verrou ---------------------------------------------------------------
|
||
|
||
def test_epreuve_verrou_reussit_si_la_seconde_prise_est_refusee():
|
||
r = epreuve_verrou(_FauxSysteme(verrou_sequence=["handle", None]), chemin="/tmp/x.lock")
|
||
assert r.reussite is True
|
||
|
||
|
||
def test_epreuve_verrou_echoue_si_la_seconde_prise_nest_pas_refusee():
|
||
r = epreuve_verrou(_FauxSysteme(verrou_sequence=["h1", "h2"]), chemin="/tmp/x.lock")
|
||
assert r.reussite is False
|
||
|
||
|
||
def test_epreuve_verrou_non_concluante_si_le_premier_verrou_echoue_deja():
|
||
r = epreuve_verrou(_FauxSysteme(verrou_sequence=[None, None]), chemin="/tmp/x.lock")
|
||
assert r.reussite is None
|
||
|
||
|
||
# --- diagnostiquer : enchaînement complet -----------------------------------------
|
||
|
||
def test_diagnostiquer_rend_les_sept_epreuves_dans_lordre():
|
||
resultats = diagnostiquer(
|
||
systeme=_FauxSysteme(verrou_sequence=["h", None]),
|
||
plateforme="win32", machine="AMD64", cuda_disponible=lambda: False,
|
||
detecter_moteur=lambda: "faster-cpu",
|
||
resoudre_modele=lambda nom, forcage: "modele-test",
|
||
creer_transcripteur=lambda **kw: _FauxTranscripteur(texte="ok"),
|
||
ouvrir_flux=lambda **kw: _FauxFlux(kw["callback"], n_blocs=3),
|
||
dormir=lambda s: None,
|
||
generer_parole=lambda: np.zeros(16000, dtype="float32"),
|
||
horloge=lambda: 0.0,
|
||
copier=lambda t: None,
|
||
lire=lambda: "peu importe",
|
||
)
|
||
assert [r.nom for r in resultats] == [
|
||
"Plateforme", "Moteur retenu", "Micro", "Transcription",
|
||
"Presse-papier", "Raccourci", "Verrou",
|
||
]
|
||
|
||
|
||
def test_diagnostiquer_saute_honnetement_la_transcription_si_le_moteur_echoue():
|
||
def leve():
|
||
raise RuntimeError("aucun moteur")
|
||
resultats = diagnostiquer(
|
||
systeme=_FauxSysteme(verrou_sequence=["h", None]),
|
||
detecter_moteur=leve,
|
||
ouvrir_flux=lambda **kw: _FauxFlux(kw["callback"], n_blocs=1),
|
||
dormir=lambda s: None,
|
||
copier=lambda t: None,
|
||
lire=lambda: "peu importe",
|
||
)
|
||
transcription = [r for r in resultats if r.nom == "Transcription"][0]
|
||
# jamais une réussite inventée quand rien n'a pu être mené.
|
||
assert transcription.reussite is None
|
||
assert "sautée" in transcription.detail
|
||
|
||
|
||
class _SystemeTombeToujours:
|
||
"""Doublure dont chaque méthode lève — pour prouver que rien n'est jamais
|
||
inventé même quand la couche système est totalement en panne."""
|
||
|
||
def raccourci_defaut(self):
|
||
raise RuntimeError("tout casse")
|
||
|
||
def acquerir_verrou_instance(self, chemin):
|
||
raise RuntimeError("tout casse")
|
||
|
||
|
||
def test_diagnostiquer_ne_leve_jamais_meme_si_toutes_les_epreuves_echouent():
|
||
def leve(*a, **kw):
|
||
raise RuntimeError("tout casse")
|
||
resultats = diagnostiquer(
|
||
systeme=_SystemeTombeToujours(),
|
||
plateforme="win32", machine="AMD64", cuda_disponible=leve,
|
||
detecter_moteur=leve,
|
||
ouvrir_flux=leve,
|
||
dormir=lambda s: None,
|
||
copier=leve,
|
||
lire=leve,
|
||
)
|
||
assert len(resultats) == 7
|
||
assert all(r.reussite is None for r in resultats)
|
||
|
||
|
||
# --- mise en forme -----------------------------------------------------------------
|
||
|
||
def test_formater_resultat_aligne_le_nom_et_prefixe_un_symbole():
|
||
r = ResultatEpreuve("Plateforme", True, "darwin / arm64 / CUDA absent")
|
||
ligne = formater_resultat(r)
|
||
assert ligne == "✅ Plateforme : darwin / arm64 / CUDA absent"
|
||
|
||
|
||
def test_formater_resultat_marque_lechec():
|
||
r = ResultatEpreuve("Micro", False, "0 blocs reçus en 1 s")
|
||
assert formater_resultat(r).startswith("❌ ")
|
||
|
||
|
||
def test_formater_resultat_marque_le_non_concluant():
|
||
r = ResultatEpreuve("Transcription", None, "sautée")
|
||
assert formater_resultat(r).startswith("➖ ")
|
||
|
||
|
||
def test_mettre_en_forme_ne_prend_aucun_travail_a_sa_charge():
|
||
# fabriqué à la main : aucune épreuve n'est exécutée ici.
|
||
resultats = [
|
||
ResultatEpreuve("Plateforme", True, "darwin / arm64 / CUDA absent"),
|
||
ResultatEpreuve("Moteur retenu", True, "mlx + mlx-community/whisper-medium"),
|
||
ResultatEpreuve("Micro", False, "0 blocs reçus en 1 s"),
|
||
ResultatEpreuve("Transcription", None, "sautée"),
|
||
]
|
||
rapport = mettre_en_forme(resultats)
|
||
for r in resultats:
|
||
assert formater_resultat(r) in rapport
|
||
assert "1 en échec" in rapport
|
||
assert "1 non concluante" in rapport
|
||
|
||
|
||
def test_mettre_en_forme_sur_zero_resultat_ne_leve_pas():
|
||
assert "0/0" in mettre_en_forme([])
|