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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
# Copyright (C) 2009, 2011 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
"""Tests for OS Locks."""
from bzrlib import (
debug,
errors,
lock,
tests,
)
from bzrlib.tests.scenarios import load_tests_apply_scenarios
load_tests = load_tests_apply_scenarios
class TestOSLock(tests.TestCaseInTempDir):
scenarios = [(
name, {
'write_lock': write_lock,
'read_lock': read_lock})
for name, write_lock, read_lock in lock._lock_classes]
read_lock = None
write_lock = None
def setUp(self):
super(TestOSLock, self).setUp()
self.build_tree(['a-lock-file'])
def test_create_read_lock(self):
r_lock = self.read_lock('a-lock-file')
r_lock.unlock()
def test_create_write_lock(self):
w_lock = self.write_lock('a-lock-file')
w_lock.unlock()
def test_read_locks_share(self):
r_lock = self.read_lock('a-lock-file')
try:
lock2 = self.read_lock('a-lock-file')
lock2.unlock()
finally:
r_lock.unlock()
def test_write_locks_are_exclusive(self):
w_lock = self.write_lock('a-lock-file')
try:
self.assertRaises(errors.LockContention,
self.write_lock, 'a-lock-file')
finally:
w_lock.unlock()
def test_read_locks_block_write_locks(self):
r_lock = self.read_lock('a-lock-file')
try:
if lock.have_fcntl and self.write_lock is lock._fcntl_WriteLock:
# With -Dlock, fcntl locks are properly exclusive
debug.debug_flags.add('strict_locks')
self.assertRaises(errors.LockContention,
self.write_lock, 'a-lock-file')
# But not without it
debug.debug_flags.remove('strict_locks')
try:
w_lock = self.write_lock('a-lock-file')
except errors.LockContention:
self.fail('Unexpected success. fcntl read locks'
' do not usually block write locks')
else:
w_lock.unlock()
self.knownFailure('fcntl read locks don\'t'
' block write locks without -Dlock')
else:
self.assertRaises(errors.LockContention,
self.write_lock, 'a-lock-file')
finally:
r_lock.unlock()
def test_write_locks_block_read_lock(self):
w_lock = self.write_lock('a-lock-file')
try:
if lock.have_fcntl and self.read_lock is lock._fcntl_ReadLock:
# With -Dlock, fcntl locks are properly exclusive
debug.debug_flags.add('strict_locks')
self.assertRaises(errors.LockContention,
self.read_lock, 'a-lock-file')
# But not without it
debug.debug_flags.remove('strict_locks')
try:
r_lock = self.read_lock('a-lock-file')
except errors.LockContention:
self.fail('Unexpected success. fcntl write locks'
' do not usually block read locks')
else:
r_lock.unlock()
self.knownFailure('fcntl write locks don\'t'
' block read locks without -Dlock')
else:
self.assertRaises(errors.LockContention,
self.read_lock, 'a-lock-file')
finally:
w_lock.unlock()
def test_temporary_write_lock(self):
r_lock = self.read_lock('a-lock-file')
try:
status, w_lock = r_lock.temporary_write_lock()
self.assertTrue(status)
# This should block another write lock
try:
self.assertRaises(errors.LockContention,
self.write_lock, 'a-lock-file')
finally:
r_lock = w_lock.restore_read_lock()
# We should be able to take a read lock now
r_lock2 = self.read_lock('a-lock-file')
r_lock2.unlock()
finally:
r_lock.unlock()
def test_temporary_write_lock_fails(self):
r_lock = self.read_lock('a-lock-file')
try:
r_lock2 = self.read_lock('a-lock-file')
try:
status, w_lock = r_lock.temporary_write_lock()
self.assertFalse(status)
# Taking out the lock requires unlocking and locking again, so
# we have to replace the original object
r_lock = w_lock
finally:
r_lock2.unlock()
# We should be able to take a read lock now
r_lock2 = self.read_lock('a-lock-file')
r_lock2.unlock()
finally:
r_lock.unlock()
|