Protection multi-instance GUI V6 : mutex kernel nommé sur Windows (partagé avec l'installeur Inno via AppMutex), fcntl exclusif sur POSIX (dev/test). 3 tests unitaires, self-test OK, 0 régression gui_v6 (145 passed). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
"""Protection multi-instance de la GUI V6 (P0-7).
|
|
|
|
- Windows (frozen) : mutex nommé kernel via ctypes — c'est CE nom que l'installeur
|
|
Inno détecte (``AppMutex``) pour fermer l'app avant une mise à jour (D8).
|
|
- POSIX (dev/test) : verrou ``fcntl`` exclusif sur un fichier dans le dossier app.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Nom partagé avec installer/Anonymisation.iss (AppMutex). NE PAS modifier seul.
|
|
APP_MUTEX_NAME = "AivanonymAnonymisationV6"
|
|
|
|
|
|
class AlreadyRunningError(RuntimeError):
|
|
"""Une autre instance de l'application est déjà en cours d'exécution."""
|
|
|
|
|
|
def _lock_dir() -> Path:
|
|
base = os.environ.get("LOCALAPPDATA")
|
|
root = Path(base) if base else Path.home() / ".local" / "share"
|
|
d = root / "Aivanov" / "Anonymisation"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
class SingleInstance:
|
|
def __init__(self) -> None:
|
|
self._handle = None # mutex Windows
|
|
self._fh = None # file handle POSIX
|
|
|
|
def acquire(self) -> None:
|
|
if sys.platform.startswith("win"):
|
|
self._acquire_windows()
|
|
else:
|
|
self._acquire_posix()
|
|
|
|
def _acquire_windows(self) -> None: # pragma: no cover (exécuté sur Windows)
|
|
import ctypes
|
|
|
|
ERROR_ALREADY_EXISTS = 183
|
|
handle = ctypes.windll.kernel32.CreateMutexW(None, False, APP_MUTEX_NAME)
|
|
if not handle or ctypes.windll.kernel32.GetLastError() == ERROR_ALREADY_EXISTS:
|
|
raise AlreadyRunningError("L'application est déjà ouverte.")
|
|
self._handle = handle
|
|
|
|
def _acquire_posix(self) -> None:
|
|
import fcntl
|
|
|
|
path = _lock_dir() / "instance.lock"
|
|
fh = open(path, "w")
|
|
try:
|
|
fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError:
|
|
fh.close()
|
|
raise AlreadyRunningError("L'application est déjà ouverte.")
|
|
self._fh = fh
|
|
|
|
def release(self) -> None:
|
|
if self._handle is not None: # pragma: no cover
|
|
import ctypes
|
|
|
|
ctypes.windll.kernel32.CloseHandle(self._handle)
|
|
self._handle = None
|
|
if self._fh is not None:
|
|
self._fh.close()
|
|
self._fh = None
|