"""`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="++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="++d")) assert r.reussite is True assert r.detail == "++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([])