1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
from __future__ import annotations
import sys
from threading import Thread
from typing import TYPE_CHECKING
import pytest
from filelock import FileLock, SoftFileLock, Timeout
if TYPE_CHECKING:
from pathlib import Path
unix_only = pytest.mark.skipif(sys.platform == "win32", reason="unix-only symlink test")
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_same_thread_different_instances_raises(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock_path = tmp_path / "test.lock"
lock1 = lock_type(lock_path)
with lock1:
lock2 = lock_type(lock_path)
with pytest.raises(RuntimeError, match="Deadlock"):
lock2.acquire()
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_finite_timeout_gives_timeout_not_deadlock(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock_path = tmp_path / "test.lock"
lock1 = lock_type(lock_path)
with lock1:
lock2 = lock_type(lock_path, timeout=0.1)
with pytest.raises(Timeout):
lock2.acquire()
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_non_blocking_gives_timeout_not_deadlock(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock_path = tmp_path / "test.lock"
lock1 = lock_type(lock_path)
with lock1:
lock2 = lock_type(lock_path, blocking=False)
with pytest.raises(Timeout):
lock2.acquire()
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_different_paths_no_conflict(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock1 = lock_type(tmp_path / "a.lock")
lock2 = lock_type(tmp_path / "b.lock")
with lock1, lock2:
assert lock1.is_locked
assert lock2.is_locked
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_same_instance_reentrant_works(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock = lock_type(tmp_path / "test.lock")
with lock:
with lock:
assert lock.is_locked
assert lock.is_locked
assert not lock.is_locked
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_singleton_avoids_deadlock(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock_path = tmp_path / "test.lock"
lock1 = lock_type(lock_path, is_singleton=True)
with lock1:
lock2 = lock_type(lock_path, is_singleton=True)
assert lock1 is lock2
with lock2:
assert lock2.is_locked
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_different_threads_no_false_positive(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock_path = tmp_path / "test.lock"
lock1 = lock_type(lock_path, timeout=0)
lock1.acquire()
error: BaseException | None = None
def acquire_in_thread() -> None:
nonlocal error
lock2 = lock_type(lock_path, timeout=0)
try:
lock2.acquire()
except BaseException as exc:
error = exc
thread = Thread(target=acquire_in_thread)
thread.start()
thread.join()
lock1.release()
assert not isinstance(error, RuntimeError), "Should not raise RuntimeError in different thread"
@unix_only
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_symlink_same_canonical_path(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock_path = tmp_path / "test.lock"
symlink_path = tmp_path / "link.lock"
symlink_path.symlink_to(lock_path)
lock1 = lock_type(lock_path)
with lock1:
lock2 = lock_type(symlink_path)
with pytest.raises(RuntimeError, match="Deadlock"):
lock2.acquire()
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_cleanup_on_release(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock_path = tmp_path / "test.lock"
lock1 = lock_type(lock_path)
lock1.acquire()
lock1.release()
lock2 = lock_type(lock_path)
with lock2:
assert lock2.is_locked
@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
def test_force_release_cleans_registry(tmp_path: Path, lock_type: type[FileLock]) -> None:
lock_path = tmp_path / "test.lock"
lock1 = lock_type(lock_path)
with lock1:
lock1.acquire()
assert lock1.lock_counter == 2
lock1.release(force=True)
lock2 = lock_type(lock_path)
with lock2:
assert lock2.is_locked
|
