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
|
import os
import sys
from dogpile.cache.backends.file import AbstractFileLock
from dogpile.cache.proxy import ProxyBackend
from dogpile.testing import assert_raises_message
from dogpile.testing.fixtures import _GenericBackendTestSuite
from dogpile.testing.fixtures import _GenericMutexTestSuite
from dogpile.testing.fixtures import _GenericSerializerTestSuite
from dogpile.util.readwrite_lock import ReadWriteMutex
try:
import fcntl # noqa
has_fcntl = True
except ImportError:
has_fcntl = False
class MutexLock(AbstractFileLock):
def __init__(self, filename):
self.mutex = ReadWriteMutex()
def acquire_read_lock(self, wait):
ret = self.mutex.acquire_read_lock(wait)
return wait or ret
def acquire_write_lock(self, wait):
ret = self.mutex.acquire_write_lock(wait)
return wait or ret
def release_read_lock(self):
return self.mutex.release_read_lock()
def release_write_lock(self):
return self.mutex.release_write_lock()
test_fname = "test_%s.db" % sys.hexversion
if has_fcntl:
class DBMBackendTest(_GenericBackendTestSuite):
backend = "dogpile.cache.dbm"
config_args = {"arguments": {"filename": test_fname}}
class DBMBackendConditionTest(_GenericBackendTestSuite):
backend = "dogpile.cache.dbm"
config_args = {
"arguments": {"filename": test_fname, "lock_factory": MutexLock}
}
class DBMBackendProxyTest(_GenericBackendTestSuite):
backend = "dogpile.cache.dbm"
config_args = {
"arguments": {"filename": test_fname, "lock_factory": MutexLock},
"wrap": [ProxyBackend],
}
class DBMBackendSerializerTest(
_GenericSerializerTestSuite, DBMBackendConditionTest
):
pass
class DBMBackendNoLockTest(_GenericBackendTestSuite):
backend = "dogpile.cache.dbm"
config_args = {
"arguments": {
"filename": test_fname,
"rw_lockfile": False,
"dogpile_lockfile": False,
}
}
class _DBMMutexTestSuite(_GenericMutexTestSuite):
backend = "dogpile.cache.dbm"
def test_release_assertion_thread(self):
backend = self._backend()
m1 = backend.get_mutex("foo")
assert_raises_message(
AssertionError, "this thread didn't do the acquire", m1.release
)
def test_release_assertion_key(self):
backend = self._backend()
m1 = backend.get_mutex("foo")
m2 = backend.get_mutex("bar")
m1.acquire()
try:
assert_raises_message(
AssertionError, "No acquire held for key 'bar'", m2.release
)
finally:
m1.release()
if has_fcntl:
class DBMMutexFileTest(_DBMMutexTestSuite):
config_args = {"arguments": {"filename": test_fname}}
class DBMMutexConditionTest(_DBMMutexTestSuite):
config_args = {
"arguments": {"filename": test_fname, "lock_factory": MutexLock}
}
def teardown():
for fname in os.listdir(os.curdir):
if fname.startswith(test_fname):
os.unlink(fname)
|
