Extrait les appels macOS derriere le contrat Systeme

- systeme/macos.py : SystemeMacOS (sons afplay, modificateurs Quartz, chemins,
  verrou d'instance fcntl) implementant le contrat Systeme.
- systeme/__init__.py : choisir_systeme() fabrique selon sys.platform,
  refuse clairement une plateforme inconnue.
- core.py : verrou d'instance (fcntl, LOCK_PATH, acquire_single_instance_lock)
  retire ; import os retire aussi, verifie inutilise ailleurs dans le fichier.
- app.py : passe par choisir_systeme().acquerir_verrou_instance(dossier_donnees()
  + instance.lock) au lieu de core.acquire_single_instance_lock. Seul changement
  dans ce fichier, nom de fichier inchange.
- tests/test_single_instance.py : reoriente vers choisir_systeme(), memes
  assertions et meme nombre de tests.
- tests/test_systeme.py (nouveau) : verrou du test_le_verrou_refuse_une_seconde_prise
  garde desormais la reference au premier verrou — sans variable, CPython le
  garbage-collectait entre les deux assert (refcount 0 -> fermeture -> verrou
  relache), le test passait pour la mauvaise raison.

89 tests verts (84 + 5).
This commit is contained in:
Ralph Mayola 2026-08-23 17:57:52 +02:00
parent 434944def6
commit 5c4477fe1d
6 changed files with 125 additions and 43 deletions

11
app.py
View File

@ -12,6 +12,7 @@ native est de toute façon minuscule pour nos besoins : une icône, deux items d
Zonza vit dans la barre de menu, jamais dans le Dock : c'est le LSUIElement de Zonza vit dans la barre de menu, jamais dans le Dock : c'est le LSUIElement de
l'Info.plist (posé par build_app.sh) qui s'en charge. l'Info.plist (posé par build_app.sh) qui s'en charge.
""" """
import os
import threading import threading
from objc import super from objc import super
@ -28,8 +29,9 @@ from PyObjCTools import AppHelper
from pynput import keyboard from pynput import keyboard
from controller import ZonzaController from controller import ZonzaController
from core import CONFIG, acquire_single_instance_lock, validate_config from core import CONFIG, validate_config
from pulse_window import PulseWindow from pulse_window import PulseWindow
from systeme import choisir_systeme
_STATE_TITLES = { _STATE_TITLES = {
"idle": "🎙️", "idle": "🎙️",
@ -128,8 +130,11 @@ class ZonzaApp(NSObject):
def main(): def main():
# une seule instance a la fois : sinon deux modeles en RAM et la meme # une seule instance a la fois : sinon deux modeles en RAM et la meme
# hotkey captee deux fois (cf acquire_single_instance_lock). # hotkey captee deux fois (cf systeme.acquerir_verrou_instance).
lock = acquire_single_instance_lock() systeme = choisir_systeme()
lock = systeme.acquerir_verrou_instance(
os.path.join(systeme.dossier_donnees(), "instance.lock")
)
if lock is None: if lock is None:
print("[garde] une autre instance de Zonza tourne deja — arret.") print("[garde] une autre instance de Zonza tourne deja — arret.")
return return

31
core.py
View File

@ -3,8 +3,6 @@
Tout ce qui est ici est testable sans micro, sans clavier, sans modèle. Tout ce qui est ici est testable sans micro, sans clavier, sans modèle.
""" """
import fcntl
import os
import re import re
CONFIG = { CONFIG = {
@ -196,33 +194,4 @@ def clean_transcript(text):
return re.sub(r"\s+", " ", text).strip() return re.sub(r"\s+", " ", text).strip()
LOCK_PATH = os.path.expanduser("~/Library/Application Support/Zonza/instance.lock")
def acquire_single_instance_lock(path=LOCK_PATH):
"""Prend un verrou exclusif non bloquant ; renvoie le fichier, ou None si une
autre instance le detient deja.
Pourquoi : deux bundles Zonza de meme identifiant (dont un fantome reste dans
la Corbeille mais toujours enregistre aupres de LaunchServices) pouvaient
tourner en meme temps. Chacun chargeait son propre modele MLX-Whisper et
captait la MEME hotkey globale : un appui, deux transcriptions, machine
saturee. Purger le fantome corrige l'incident ; ce verrou corrige la classe.
Le handle doit rester vivant tant que l'app tourne : le verrou tombe a la
fermeture du fichier, et le noyau le libere si le processus meurt brutalement
(donc pas de verrou orphelin apres un plantage).
"""
parent = os.path.dirname(path)
if parent:
os.makedirs(parent, exist_ok=True)
handle = open(path, "a")
try:
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
handle.close()
return None
return handle
CONFIG["initial_prompt"] = build_initial_prompt() CONFIG["initial_prompt"] = build_initial_prompt()

15
systeme/__init__.py Normal file
View File

@ -0,0 +1,15 @@
"""Choisit la couche système selon la plateforme."""
import sys
def choisir_systeme(plateforme=None):
"""Rend l'implémentation du contrat Systeme pour cette machine.
`plateforme` est injectable pour tester les branches absentes de la machine
de développement.
"""
plateforme = plateforme or sys.platform
if plateforme.startswith("darwin"):
from systeme.macos import SystemeMacOS
return SystemeMacOS()
raise NotImplementedError(f"Plateforme non prise en charge : {plateforme}")

54
systeme/macos.py Normal file
View File

@ -0,0 +1,54 @@
"""Appels propres à macOS. Rien ici ne doit être importé directement par le noyau."""
import fcntl
import os
import subprocess
_SONS = {
"start": "/System/Library/Sounds/Pop.aiff",
"done": "/System/Library/Sounds/Glass.aiff",
"error": "/System/Library/Sounds/Basso.aiff",
}
# Masque des modificateurs (ctrl, shift, alt, cmd) dans les drapeaux Quartz.
_MASQUE_MODIFICATEURS = 0x000F0000
class SystemeMacOS:
"""Implémentation du contrat Systeme pour macOS."""
def jouer_son(self, nom):
chemin = _SONS.get(nom)
if not chemin:
return
try:
subprocess.Popen(["afplay", chemin])
except Exception:
pass # le son ne doit jamais casser la dictée
def modificateurs_enfonces(self):
try:
from Quartz import CGEventSourceFlagsState
return bool(CGEventSourceFlagsState(1) & _MASQUE_MODIFICATEURS)
except Exception:
return False # dans le doute, ne jamais bloquer le collage
def raccourci_defaut(self):
return "<cmd>+<ctrl>+d"
def chemin_log(self):
return os.path.expanduser("~/Library/Logs/Zonza.log")
def dossier_donnees(self):
return os.path.expanduser("~/Library/Application Support/Zonza")
def acquerir_verrou_instance(self, chemin):
parent = os.path.dirname(chemin)
if parent:
os.makedirs(parent, exist_ok=True)
fichier = open(chemin, "a")
try:
fcntl.flock(fichier, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
fichier.close()
return None
return fichier

View File

@ -13,7 +13,7 @@ bug — quel que soit le nombre de bundles présents, la 2e instance renonce.
import os import os
import tempfile import tempfile
from core import acquire_single_instance_lock from systeme import choisir_systeme
def _tmp_lock(): def _tmp_lock():
@ -21,31 +21,31 @@ def _tmp_lock():
def test_premiere_instance_obtient_le_verrou(): def test_premiere_instance_obtient_le_verrou():
handle = acquire_single_instance_lock(_tmp_lock()) handle = choisir_systeme().acquerir_verrou_instance(_tmp_lock())
assert handle is not None assert handle is not None
def test_deuxieme_instance_est_refusee(): def test_deuxieme_instance_est_refusee():
path = _tmp_lock() path = _tmp_lock()
premier = acquire_single_instance_lock(path) premier = choisir_systeme().acquerir_verrou_instance(path)
assert premier is not None assert premier is not None
assert acquire_single_instance_lock(path) is None assert choisir_systeme().acquerir_verrou_instance(path) is None
def test_verrou_repris_apres_liberation(): def test_verrou_repris_apres_liberation():
path = _tmp_lock() path = _tmp_lock()
premier = acquire_single_instance_lock(path) premier = choisir_systeme().acquerir_verrou_instance(path)
premier.close() premier.close()
assert acquire_single_instance_lock(path) is not None assert choisir_systeme().acquerir_verrou_instance(path) is not None
def test_cree_le_dossier_parent_manquant(): def test_cree_le_dossier_parent_manquant():
path = _tmp_lock() path = _tmp_lock()
assert not os.path.isdir(os.path.dirname(path)) assert not os.path.isdir(os.path.dirname(path))
acquire_single_instance_lock(path) choisir_systeme().acquerir_verrou_instance(path)
assert os.path.isdir(os.path.dirname(path)) assert os.path.isdir(os.path.dirname(path))
def test_deux_chemins_differents_ne_se_genent_pas(): def test_deux_chemins_differents_ne_se_genent_pas():
assert acquire_single_instance_lock(_tmp_lock()) is not None assert choisir_systeme().acquerir_verrou_instance(_tmp_lock()) is not None
assert acquire_single_instance_lock(_tmp_lock()) is not None assert choisir_systeme().acquerir_verrou_instance(_tmp_lock()) is not None

39
tests/test_systeme.py Normal file
View File

@ -0,0 +1,39 @@
"""La fabrique rend une implémentation conforme au contrat Systeme."""
import os
import tempfile
import pytest
from systeme import choisir_systeme
def test_la_fabrique_rend_une_implementation_complete():
s = choisir_systeme()
for methode in ("jouer_son", "modificateurs_enfonces", "raccourci_defaut",
"chemin_log", "dossier_donnees", "acquerir_verrou_instance"):
assert callable(getattr(s, methode)), f"{methode} manque"
def test_une_plateforme_inconnue_est_refusee_clairement():
with pytest.raises(NotImplementedError) as e:
choisir_systeme(plateforme="haiku-os")
assert "haiku-os" in str(e.value)
def test_le_raccourci_macos_est_celui_d_aujourd_hui():
assert choisir_systeme(plateforme="darwin").raccourci_defaut() == "<cmd>+<ctrl>+d"
def test_le_verrou_refuse_une_seconde_prise():
s = choisir_systeme()
chemin = os.path.join(tempfile.mkdtemp(), "sous-dossier", "z.lock")
# Référence gardée : sinon CPython ramasse l'objet fichier dès la fin de
# l'assert (refcount à 0), ce qui referme le fichier et relâche le verrou
# AVANT la seconde prise — le test passerait pour la mauvaise raison.
verrou = s.acquerir_verrou_instance(chemin)
assert verrou is not None
assert s.acquerir_verrou_instance(chemin) is None
def test_aucun_modificateur_enfonce_pendant_les_tests():
assert choisir_systeme().modificateurs_enfonces() is False