diff --git a/app.py b/app.py index 0b02808..7a07477 100644 --- a/app.py +++ b/app.py @@ -12,6 +12,19 @@ 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 l'Info.plist (posé par build_app.sh) qui s'en charge. """ +import sys + +# --diagnostic doit s'évaluer AVANT tout import Cocoa/pyobjc : sur une machine +# qui ne les a pas (aujourd'hui aucune Windows n'est disponible pour ce +# portage, mais c'est la raison d'être de cette commande), c'est le seul +# moyen d'éprouver la chaîne. La sortie se fait donc ICI, au chargement du +# module — jamais dans main(), qui serait déjà trop tard pour éviter +# `from AppKit import ...` ci-dessous. +if "--diagnostic" in sys.argv: + from diagnostic import diagnostiquer, mettre_en_forme + print(mettre_en_forme(diagnostiquer())) + sys.exit(0) + import os import threading diff --git a/diagnostic.py b/diagnostic.py new file mode 100644 index 0000000..17812d1 --- /dev/null +++ b/diagnostic.py @@ -0,0 +1,303 @@ +"""`zonza --diagnostic` : éprouve toute la chaîne sur la machine cible et dit où +elle casse. + +Aucune machine Windows n'étant disponible pour ce portage, cette commande est +le SEUL moyen pour l'utilisateur de valider `systeme/windows.py` en conditions +réelles. D'où deux règles strictes suivies ici : + +1. Chaque épreuve attrape SES PROPRES erreurs : une panne (micro absent, + moteur introuvable, `say` manquant...) ne doit jamais faire tomber les + épreuves suivantes ni le rapport entier. +2. Une épreuve qui n'a pas pu être menée est marquée `reussite=None` + (« non concluante ») — JAMAIS `True`. Mentir sur une réussite serait pire + que ne rien dire : mieux vaut « je ne sais pas » qu'un faux vert. + +Deux fonctions séparées, pour rester testable : `diagnostiquer()` rend une +liste de `ResultatEpreuve` (aucun `print`) ; `mettre_en_forme()` la traduit en +texte affichable. Toutes les dépendances matérielles/plateforme sont +injectables, ce qui permet d'éprouver l'ENCHAÎNEMENT des épreuves depuis ce +Mac sans jamais toucher un vrai micro ni un vrai moteur de transcription. +""" +import os +import platform +import shutil +import subprocess +import sys +import tempfile +import time +import uuid +import wave +from dataclasses import dataclass + +import numpy as np + +import core +import moteur +from moteur import _cuda_disponible +from systeme import choisir_systeme + + +@dataclass(frozen=True) +class ResultatEpreuve: + """Le compte-rendu d'une seule épreuve. + + `reussite` : True (menée, et tout va bien), False (menée, et quelque chose + est cassé), ou None (n'a PAS pu être menée — non concluante, jamais + présentée comme une réussite). + """ + nom: str + reussite: bool | None + detail: str + + +# --- épreuve 1 : plateforme ------------------------------------------------- + +def epreuve_plateforme(plateforme=None, machine=None, cuda_disponible=None): + """Décrit la machine : plateforme Python, architecture, présence de CUDA.""" + plateforme = plateforme if plateforme is not None else sys.platform + machine = machine if machine is not None else platform.machine() + sonder_cuda = cuda_disponible if cuda_disponible is not None else _cuda_disponible + try: + cuda = sonder_cuda() + except Exception as e: + return ResultatEpreuve("Plateforme", None, f"sonde CUDA en échec : {e}") + cuda_txt = "présent" if cuda else "absent" + return ResultatEpreuve("Plateforme", True, f"{plateforme} / {machine} / CUDA {cuda_txt}") + + +# --- épreuve 2 : moteur retenu ---------------------------------------------- + +def epreuve_moteur(detecter_moteur=None, resoudre_modele=None, creer_transcripteur=None): + """Décide du moteur, résout son modèle, instancie le transcripteur. + + Rend `(resultat, transcripteur_ou_None)` : le transcripteur sert à + l'épreuve Transcription sans être reconstruit — mais seulement si cette + épreuve a réussi ; sinon la Transcription sera honnêtement sautée. + """ + detecter_moteur = detecter_moteur or moteur.detecter_moteur + resoudre_modele = resoudre_modele or core.modele_retenu + creer_transcripteur = creer_transcripteur or moteur.choisir_transcripteur + try: + nom = detecter_moteur() + nom_modele = resoudre_modele(nom, core.CONFIG["model"]) + transcripteur = creer_transcripteur( + nom=nom, modele=nom_modele, langue=core.CONFIG["language"], + amorce=core.CONFIG["initial_prompt"], + ) + except Exception as e: + return ResultatEpreuve("Moteur retenu", None, f"détection impossible : {e}"), None + return ResultatEpreuve("Moteur retenu", True, f"{nom} + {nom_modele}"), transcripteur + + +# --- épreuve 3 : micro ------------------------------------------------------- + +def epreuve_micro(ouvrir_flux=None, sample_rate=16000, duree_s=1.0, dormir=None): + """Ouvre VRAIMENT le micro pendant `duree_s` et compte les blocs reçus. + + `ouvrir_flux` suit le même contrat que `Recorder._ouvrir_flux` dans + `capture.py` : injectable, pour que les tests ne touchent à aucun + périphérique réel. + """ + dormir = dormir or time.sleep + if ouvrir_flux is None: + try: + import sounddevice as sd + except Exception as e: + return ResultatEpreuve("Micro", None, f"sounddevice indisponible : {e}") + ouvrir_flux = lambda **kw: sd.InputStream(**kw) # noqa: E731 + + compte = {"n": 0} + + def on_block(indata, frames, time_info, status): + compte["n"] += 1 + + try: + flux = ouvrir_flux(samplerate=sample_rate, channels=1, dtype="float32", + callback=on_block) + flux.start() + except Exception as e: + return ResultatEpreuve("Micro", None, f"ouverture du micro impossible : {e}") + + dormir(duree_s) + + for action in ("stop", "close"): + try: + getattr(flux, action)() + except Exception: + pass # meilleur effort : le compte déjà reçu reste valable + + n = compte["n"] + return ResultatEpreuve("Micro", n > 0, f"{n} blocs reçus en {duree_s:g} s") + + +# --- épreuve 4 : transcription ----------------------------------------------- + +def _generer_echantillon_parole(texte="Bonjour, ceci est un test de dictée Zonza.", + sample_rate=16000): + """Synthétise un court échantillon de parole via la commande `say` de macOS. + + LÈVE si `say` est absent, échoue, ou si le WAV produit est illisible — + c'est volontaire : l'appelant retombe alors sur du silence plutôt que + d'inventer un résultat. Sur une machine sans `say` (Windows, Linux), + c'est TOUJOURS ce chemin qui est emprunté. + """ + if shutil.which("say") is None: + raise RuntimeError("commande `say` absente (probablement pas macOS)") + chemin = tempfile.NamedTemporaryFile(suffix=".wav", delete=False).name + try: + subprocess.run( + ["say", "-o", chemin, f"--data-format=LEI16@{sample_rate}", texte], + check=True, capture_output=True, timeout=15, + ) + with wave.open(chemin, "rb") as w: + brut = w.readframes(w.getnframes()) + return np.frombuffer(brut, dtype=" n'existe pas sur Windows : pas d'équivalent à la touche Commande. + return "++d" + + # --- chemins ------------------------------------------------------------ + def chemin_log(self): + return os.path.join(self._dossier_local_appdata(), "Zonza", "Zonza.log") + + def dossier_donnees(self): + return os.path.join(self._dossier_local_appdata(), "Zonza") + + @staticmethod + def _dossier_local_appdata(): + # os.environ est standard : lisible (et falsifiable par les tests via + # monkeypatch.setenv) depuis n'importe quelle plateforme, contrairement + # à winsound/ctypes.windll/msvcrt. + return os.environ.get("LOCALAPPDATA", os.path.expanduser("~")) + + # --- verrou d'instance -------------------------------------------------- + def acquerir_verrou_instance(self, chemin): + parent = os.path.dirname(chemin) + if parent: + os.makedirs(parent, exist_ok=True) + fichier = open(chemin, "a+") + fichier.seek(0) + # msvcrt.locking verrouille un OCTET à partir de la position courante ; + # il exige que cet octet existe déjà dans le fichier, d'où cette garde. + if os.fstat(fichier.fileno()).st_size == 0: + fichier.write("0") + fichier.flush() + fichier.seek(0) + try: + (self._verrouiller or self._verrouiller_reel)(fichier) + except OSError: + fichier.close() + return None + return fichier + + @staticmethod + def _verrouiller_reel(fichier): + import msvcrt + fichier.seek(0) + msvcrt.locking(fichier.fileno(), msvcrt.LK_NBLCK, 1) diff --git a/tests/test_diagnostic.py b/tests/test_diagnostic.py new file mode 100644 index 0000000..3861c31 --- /dev/null +++ b/tests/test_diagnostic.py @@ -0,0 +1,326 @@ +"""`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([]) diff --git a/tests/test_systeme_windows.py b/tests/test_systeme_windows.py new file mode 100644 index 0000000..b52f504 --- /dev/null +++ b/tests/test_systeme_windows.py @@ -0,0 +1,154 @@ +"""Couche Windows du contrat Systeme — éprouvée depuis macOS par injection. + +`winsound`, `msvcrt` et `ctypes.windll` n'existent pas sur macOS : c'est +pourquoi `SystemeWindows` les reçoit en paramètres injectables (jouer, clavier, +verrouiller) plutôt que de les importer en tête de module. Ces tests éprouvent +donc la LOGIQUE française de la classe — quel alias pour quel son, quel bit +pour quelle touche, comment les chemins et le verrou sont construits — jamais +le comportement réel de winsound.PlaySound, GetAsyncKeyState ou msvcrt.locking : +aucune machine ici ne peut le vérifier. Seule une vraie machine Windows le peut. +""" +import os + +import pytest + +from systeme.windows import SystemeWindows + + +# --- jouer_son --------------------------------------------------------- + +def test_jouer_son_start_produit_lalias_systemasterisk(): + appels = [] + s = SystemeWindows(jouer=appels.append) + s.jouer_son("start") + assert appels == ["SystemAsterisk"] + + +def test_jouer_son_done_produit_lalias_systemexclamation(): + appels = [] + s = SystemeWindows(jouer=appels.append) + s.jouer_son("done") + assert appels == ["SystemExclamation"] + + +def test_jouer_son_error_produit_lalias_systemhand(): + appels = [] + s = SystemeWindows(jouer=appels.append) + s.jouer_son("error") + assert appels == ["SystemHand"] + + +def test_jouer_son_nom_inconnu_ne_produit_aucun_appel(): + appels = [] + s = SystemeWindows(jouer=appels.append) + s.jouer_son("inconnu") + assert appels == [] + + +def test_jouer_son_qui_echoue_ne_leve_pas(): + def echoue(alias): + raise OSError("winsound indisponible") + s = SystemeWindows(jouer=echoue) + s.jouer_son("start") # ne doit jamais lever : le son ne casse pas la dictée + + +# --- modificateurs_enfonces -------------------------------------------- + +@pytest.mark.parametrize("vk", [0x11, 0x10, 0x12, 0x5B, 0x5C]) +def test_modificateurs_enfonces_vrai_pour_chacune_des_cinq_touches(vk): + # bit de poids fort posé UNIQUEMENT pour la touche ciblée par ce cas. + s = SystemeWindows(clavier=lambda sonde, cible=vk: 0x8000 if sonde == cible else 0) + assert s.modificateurs_enfonces() is True + + +def test_modificateurs_enfonces_faux_si_aucun_bit_de_poids_fort_pose(): + # une valeur non nulle mais sans le bit 0x8000 ne doit pas compter. + s = SystemeWindows(clavier=lambda vk: 0x0001) + assert s.modificateurs_enfonces() is False + + +def test_modificateurs_enfonces_faux_si_toutes_les_touches_sont_relachees(): + s = SystemeWindows(clavier=lambda vk: 0) + assert s.modificateurs_enfonces() is False + + +def test_modificateurs_enfonces_faux_si_la_sonde_leve(): + def leve(vk): + raise OSError("user32 indisponible") + s = SystemeWindows(clavier=leve) + assert s.modificateurs_enfonces() is False + + +# --- raccourci_defaut ---------------------------------------------------- + +def test_raccourci_defaut_est_ctrl_alt_d(): + # n'existe pas sur Windows, contrairement à macOS. + assert SystemeWindows().raccourci_defaut() == "++d" + + +# --- chemins --------------------------------------------------------------- + +def test_chemin_log_derive_de_localappdata(monkeypatch, tmp_path): + monkeypatch.setenv("LOCALAPPDATA", str(tmp_path)) + assert SystemeWindows().chemin_log() == os.path.join(str(tmp_path), "Zonza", "Zonza.log") + + +def test_dossier_donnees_derive_de_localappdata(monkeypatch, tmp_path): + monkeypatch.setenv("LOCALAPPDATA", str(tmp_path)) + assert SystemeWindows().dossier_donnees() == os.path.join(str(tmp_path), "Zonza") + + +def test_chemin_log_retombe_sur_le_dossier_utilisateur_si_localappdata_absent(monkeypatch): + monkeypatch.delenv("LOCALAPPDATA", raising=False) + attendu = os.path.join(os.path.expanduser("~"), "Zonza", "Zonza.log") + assert SystemeWindows().chemin_log() == attendu + + +# --- verrou d'instance ------------------------------------------------------- + +def test_le_verrou_rend_un_objet_la_premiere_fois(tmp_path): + chemin = str(tmp_path / "sous-dossier" / "z.lock") + s = SystemeWindows(verrouiller=lambda f: None) + verrou = s.acquerir_verrou_instance(chemin) + assert verrou is not None + verrou.close() + + +def test_le_verrou_cree_le_dossier_parent_manquant(tmp_path): + chemin = str(tmp_path / "sous-dossier" / "z.lock") + s = SystemeWindows(verrouiller=lambda f: None) + assert not os.path.isdir(os.path.dirname(chemin)) + verrou = s.acquerir_verrou_instance(chemin) + assert os.path.isdir(os.path.dirname(chemin)) + verrou.close() + + +def test_le_verrou_refuse_une_seconde_prise(tmp_path): + chemin = str(tmp_path / "z.lock") + pris = {"valeur": False} + + def verrouiller(f): + if pris["valeur"]: + raise OSError("verrou déjà pris") + pris["valeur"] = True + + s = SystemeWindows(verrouiller=verrouiller) + premier = s.acquerir_verrou_instance(chemin) + assert premier is not None + second = s.acquerir_verrou_instance(chemin) + assert second is None + premier.close() + + +def test_un_verrou_dont_la_prise_echoue_referme_le_fichier(tmp_path): + chemin = str(tmp_path / "z.lock") + captures = {} + + def echoue(f): + captures["fichier"] = f + raise OSError("verrou pris") + + s = SystemeWindows(verrouiller=echoue) + resultat = s.acquerir_verrou_instance(chemin) + assert resultat is None + assert captures["fichier"].closed