-
Notifications
You must be signed in to change notification settings - Fork 0
/
tests.py
133 lines (108 loc) · 4.28 KB
/
tests.py
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
import contextlib
import logging
import math
import subprocess
import threading
import time
import unittest
import fslock
logging.basicConfig(level=logging.DEBUG)
class TestLocks(unittest.TestCase):
def test_sharedlock_nonexistent(self):
stack = contextlib.ExitStack()
with self.assertRaises(FileNotFoundError):
stack.enter_context(fslock.FSLockShared('/tmp/nonex_shared'))
def test_exclusivelock_nonexistent(self):
stack = contextlib.ExitStack()
with self.assertRaises(FileNotFoundError):
stack.enter_context(
fslock.FSLockExclusive('/tmp/nonex_exclusive_folder/file')
)
with fslock.FSLockExclusive('/tmp/nonex_exclusive'):
pass
def test_shared(self):
stack = contextlib.ExitStack()
with stack:
# Create a file
with open('/tmp/shared', 'w'):
pass
# Get shared lock
stack.enter_context(fslock.FSLockShared('/tmp/shared'))
# Get another shared lock
with fslock.FSLockShared('/tmp/shared'):
pass
# Check that it's locked using the shell
res = subprocess.call([
'flock', '--exclusive', '--nonblock',
'--conflict-exit-code', '43',
'/tmp/shared',
'-c', 'true',
])
self.assertEqual(res, 43)
# Try to get an exclusive lock
def try_exclusive():
with fslock.FSLockExclusive('/tmp/shared'):
try_exclusive.got_it = True
try_exclusive.got_it = False
t = threading.Thread(target=try_exclusive)
t.setDaemon(True)
t.start()
time.sleep(1)
self.assertFalse(try_exclusive.got_it)
# Get a shared lock with timeout
stack.enter_context(fslock.FSLockShared('/tmp/shared', timeout=2))
# Try to get an exclusive lock with timeout
start = time.perf_counter()
with self.assertRaises(TimeoutError):
stack.enter_context(
fslock.FSLockExclusive('/tmp/shared', timeout=2)
)
self.assertTrue(math.fabs(time.perf_counter() - start - 2) < 0.2)
def test_exclusive(self):
with fslock.FSLockExclusive('/tmp/exclusive'):
# Check that it's locked using the shell
res = subprocess.call([
'flock', '--shared', '--nonblock',
'--conflict-exit-code', '43',
'/tmp/exclusive',
'-c', 'true',
])
self.assertEqual(res, 43)
# Try to get a shared lock
def try_shared():
with fslock.FSLockShared('/tmp/exclusive'):
try_shared.got_it = True
try_shared.got_it = False
t = threading.Thread(target=try_shared)
t.setDaemon(True)
t.start()
time.sleep(1)
self.assertFalse(try_shared.got_it)
# Try to get an exclusive lock
def try_exclusive():
with fslock.FSLockExclusive('/tmp/exclusive'):
try_exclusive.got_it = True
try_exclusive.got_it = False
t = threading.Thread(target=try_exclusive)
t.setDaemon(True)
t.start()
time.sleep(1)
self.assertFalse(try_exclusive.got_it)
stack = contextlib.ExitStack()
with fslock.FSLockExclusive('/tmp/exclusive2', timeout=2):
# Try to get a shared lock with timeout
start = time.perf_counter()
with self.assertRaises(TimeoutError):
stack.enter_context(
fslock.FSLockShared('/tmp/exclusive2', timeout=2)
)
self.assertTrue(math.fabs(time.perf_counter() - start - 2) < 0.2)
# Try to get an exclusive lock with timeout
start = time.perf_counter()
with self.assertRaises(TimeoutError):
stack.enter_context(
fslock.FSLockExclusive('/tmp/exclusive2', timeout=2)
)
self.assertTrue(math.fabs(time.perf_counter() - start - 2) < 0.2)
if __name__ == '__main__':
unittest.main()