47 lines
1.2 KiB
Python
47 lines
1.2 KiB
Python
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) # 例外が出ないことを確認するだけ
|