diff --git a/PythonProj/ScanOCR/スクリプト/lock_manager.py b/PythonProj/ScanOCR/スクリプト/lock_manager.py new file mode 100644 index 00000000..606dbcf2 --- /dev/null +++ b/PythonProj/ScanOCR/スクリプト/lock_manager.py @@ -0,0 +1,44 @@ +# -*- coding: utf-8 -*- +"""二重起動防止ロックファイル管理モジュール(Windows専用)。""" +from __future__ import annotations + +import ctypes +import os +from pathlib import Path +from typing import Callable + + +class LockAcquisitionError(Exception): + pass + + +def _is_pid_running_windows(pid: int) -> bool: + PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + handle = ctypes.windll.kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if handle: + ctypes.windll.kernel32.CloseHandle(handle) + return True + return False + + +def acquire_lock( + lock_path: Path, + pid_checker: Callable[[int], bool] = _is_pid_running_windows, +) -> None: + if lock_path.exists(): + try: + recorded_pid = int(lock_path.read_text(encoding="utf-8").strip()) + except ValueError: + recorded_pid = None + + if recorded_pid is not None and pid_checker(recorded_pid): + raise LockAcquisitionError(f"既に実行中です(PID={recorded_pid})") + + lock_path.unlink() + + lock_path.write_text(str(os.getpid()), encoding="utf-8") + + +def release_lock(lock_path: Path) -> None: + if lock_path.exists(): + lock_path.unlink() diff --git a/PythonProj/ScanOCR/スクリプト/tests/test_lock_manager.py b/PythonProj/ScanOCR/スクリプト/tests/test_lock_manager.py new file mode 100644 index 00000000..e803b1fc --- /dev/null +++ b/PythonProj/ScanOCR/スクリプト/tests/test_lock_manager.py @@ -0,0 +1,46 @@ +import os + +import pytest + +from lock_manager import LockAcquisitionError, acquire_lock, release_lock + + +def test_acquire_lock_creates_file_with_pid(tmp_path): + lock_path = tmp_path / ".lock" + + acquire_lock(lock_path, pid_checker=lambda pid: True) + + assert lock_path.exists() + assert lock_path.read_text(encoding="utf-8").strip() == str(os.getpid()) + + +def test_acquire_lock_raises_when_existing_pid_running(tmp_path): + lock_path = tmp_path / ".lock" + lock_path.write_text("12345", encoding="utf-8") + + with pytest.raises(LockAcquisitionError): + acquire_lock(lock_path, pid_checker=lambda pid: True) + + +def test_acquire_lock_replaces_stale_lock(tmp_path): + lock_path = tmp_path / ".lock" + lock_path.write_text("12345", encoding="utf-8") + + acquire_lock(lock_path, pid_checker=lambda pid: False) + + assert lock_path.read_text(encoding="utf-8").strip() == str(os.getpid()) + + +def test_release_lock_removes_file(tmp_path): + lock_path = tmp_path / ".lock" + lock_path.write_text("12345", encoding="utf-8") + + release_lock(lock_path) + + assert not lock_path.exists() + + +def test_release_lock_missing_file_no_error(tmp_path): + lock_path = tmp_path / ".lock" + + release_lock(lock_path) # 例外が出ないことを確認するだけ