128 lines
4.7 KiB
Python
128 lines
4.7 KiB
Python
"""Cycle de vie de l'overlay, et interdiction de bloquer le thread principal.
|
|
|
|
Bug du 2026-08-21 (deuxieme occurrence) : le halo est reste fige sur « Je vous
|
|
ecoute… », forme d'onde plate, process a 0 % CPU. Une autre application (Loom)
|
|
avait pris le micro ; `stream.stop()` de PortAudio s'est BLOQUE. Comme toggle()
|
|
s'execute sur le thread principal Cocoa, toute l'interface a gele — et le garde-fou
|
|
NSTimer pose le matin meme etait inoperant PAR CONSTRUCTION, puisqu'il vit sur cette
|
|
meme boucle d'evenements bloquee.
|
|
|
|
D'ou la regle verrouillee ici : le chemin appele depuis le thread principal ne
|
|
touche JAMAIS au peripherique audio. Il change l'etat, met a jour l'overlay, et
|
|
delegue.
|
|
"""
|
|
import numpy as np
|
|
|
|
from controller import ZonzaController
|
|
from core import CONFIG
|
|
|
|
|
|
class FauxPulse:
|
|
def __init__(self):
|
|
self.statuts = []
|
|
self.visible = False
|
|
|
|
def show(self): self.visible = True
|
|
def hide(self): self.visible = False
|
|
def set_level(self, level): pass
|
|
def set_status(self, text): self.statuts.append(text)
|
|
def rearmer_garde(self, duree_s): self.garde = duree_s
|
|
|
|
|
|
class RecorderQuiBloque:
|
|
"""Reproduit un peripherique accapare : start/stop ne rendent jamais la main."""
|
|
def __init__(self):
|
|
self.start_appele = False
|
|
self.stop_appele = False
|
|
|
|
def start(self):
|
|
self.start_appele = True
|
|
raise AssertionError("start() ne doit pas etre appele sur le thread principal")
|
|
|
|
def stop(self):
|
|
self.stop_appele = True
|
|
raise AssertionError("stop() ne doit pas etre appele sur le thread principal")
|
|
|
|
|
|
class FauxRecorder:
|
|
def __init__(self, audio): self.audio = audio
|
|
def start(self): pass
|
|
def stop(self): return self.audio
|
|
|
|
|
|
class FauxTranscripteur:
|
|
def warmup(self): pass
|
|
def transcrire(self, audio): return "bonjour"
|
|
|
|
|
|
class FauxSysteme:
|
|
def __init__(self): self.sons = []
|
|
def jouer_son(self, nom): self.sons.append(nom)
|
|
def modificateurs_enfonces(self): return False
|
|
def raccourci_defaut(self): return "<cmd>+<ctrl>+d"
|
|
def chemin_log(self): return "/tmp/zonza-test.log"
|
|
def dossier_donnees(self): return "/tmp/zonza-test"
|
|
def acquerir_verrou_instance(self, chemin): return object()
|
|
|
|
|
|
def _ctrl(recorder, taches=None):
|
|
pulse = FauxPulse()
|
|
c = ZonzaController(CONFIG, pulse, FauxTranscripteur(), FauxSysteme(), lancer_tache=taches)
|
|
c.recorder = recorder
|
|
return c, pulse
|
|
|
|
|
|
def test_le_controleur_n_importe_aucun_module_de_plateforme():
|
|
"""Le noyau ne doit connaître que des contrats. Si cette assertion tombe, une
|
|
dépendance de plateforme s'est glissée dans le chemin portable."""
|
|
import controller
|
|
source = open(controller.__file__, encoding="utf-8").read()
|
|
for interdit in ("AppKit", "Foundation", "objc", "Quartz", "mlx_whisper",
|
|
"afplay", "from engine import"):
|
|
assert interdit not in source, f"{interdit} n'a rien à faire dans controller.py"
|
|
|
|
|
|
def test_demarrer_ne_touche_pas_l_audio_sur_le_thread_principal():
|
|
differees = []
|
|
c, pulse = _ctrl(RecorderQuiBloque(), taches=differees.append)
|
|
c.toggle() # ne doit rien lever
|
|
assert pulse.visible
|
|
assert any("écoute" in s.lower() for s in pulse.statuts)
|
|
assert len(differees) == 1, "l'ouverture du flux doit etre deleguee"
|
|
|
|
|
|
def test_arreter_ne_touche_pas_l_audio_sur_le_thread_principal():
|
|
differees = []
|
|
c, pulse = _ctrl(RecorderQuiBloque(), taches=differees.append)
|
|
c.toggle() # demarre
|
|
differees.clear()
|
|
c.toggle() # arrete : ne doit rien lever non plus
|
|
assert len(differees) == 1, "la fermeture du flux doit etre deleguee"
|
|
assert any("transcription" in s.lower() for s in pulse.statuts)
|
|
|
|
|
|
def test_la_bulle_annonce_l_ecoute_puis_la_transcription():
|
|
differees = []
|
|
c, pulse = _ctrl(FauxRecorder(np.ones(16000, dtype="float32") * 0.1), taches=differees.append)
|
|
c.toggle(); c.toggle()
|
|
assert [s.lower() for s in pulse.statuts][:2] == ["je vous écoute…", "transcription…"]
|
|
|
|
|
|
def test_la_tache_differee_referme_l_overlay_a_la_fin():
|
|
differees = []
|
|
c, pulse = _ctrl(FauxRecorder(np.ones(16000, dtype="float32") * 0.1), taches=differees.append)
|
|
c.toggle(); differees.clear(); c.toggle()
|
|
assert pulse.visible, "l'overlay reste tant que la tache n'a pas tourne"
|
|
for t in differees:
|
|
t() # on execute la tache differee
|
|
assert not pulse.visible, "l'overlay doit etre referme par la tache"
|
|
|
|
|
|
def test_un_enregistrement_trop_court_referme_aussi():
|
|
differees = []
|
|
c, pulse = _ctrl(FauxRecorder(np.zeros(10, dtype="float32")), taches=differees.append)
|
|
c.toggle(); differees.clear(); c.toggle()
|
|
for t in differees:
|
|
t()
|
|
assert not pulse.visible
|