1
# Copyright (C) 2007 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for OS level locks."""
24
from bzrlib.tests import UnicodeFilenameFeature
25
from bzrlib.tests.per_lock import TestCaseWithLock
28
class TestLock(TestCaseWithLock):
31
super(TestLock, self).setUp()
32
self.build_tree(['a-file'])
34
def test_read_lock(self):
35
"""Smoke test for read locks."""
36
a_lock = self.read_lock('a-file')
37
self.addCleanup(a_lock.unlock)
38
# The lock file should be opened for reading
40
self.assertEqual('contents of a-file\n', txt)
42
def test_create_if_needed_read(self):
43
"""We will create the file if it doesn't exist yet."""
44
a_lock = self.read_lock('other-file')
45
self.addCleanup(a_lock.unlock)
47
self.assertEqual('', txt)
49
def test_create_if_needed_write(self):
50
"""We will create the file if it doesn't exist yet."""
51
a_lock = self.write_lock('other-file')
52
self.addCleanup(a_lock.unlock)
54
self.assertEqual('', txt)
56
a_lock.f.write('foo\n')
59
self.assertEqual('foo\n', txt)
61
def test_readonly_file(self):
62
"""If the file is readonly, we can take a read lock.
64
But we shouldn't be able to take a write lock.
66
osutils.make_readonly('a-file')
67
# Make sure the file is read-only (on all platforms)
68
self.assertRaises(IOError, open, 'a-file', 'rb+')
69
a_lock = self.read_lock('a-file')
72
self.assertRaises(errors.LockFailed, self.write_lock, 'a-file')
74
def test_write_lock(self):
75
"""Smoke test for write locks."""
76
a_lock = self.write_lock('a-file')
77
self.addCleanup(a_lock.unlock)
78
# You should be able to read and write to the lock file.
80
self.assertEqual('contents of a-file\n', txt)
81
# Win32 requires that you call seek() when switching between a read
82
# operation and a write operation.
84
a_lock.f.write('more content\n')
87
self.assertEqual('contents of a-file\nmore content\n', txt)
89
def test_multiple_read_locks(self):
90
"""You can take out more than one read lock on the same file."""
91
a_lock = self.read_lock('a-file')
92
self.addCleanup(a_lock.unlock)
93
b_lock = self.read_lock('a-file')
94
self.addCleanup(b_lock.unlock)
96
def test_multiple_write_locks_exclude(self):
97
"""Taking out more than one write lock should fail."""
98
a_lock = self.write_lock('a-file')
99
self.addCleanup(a_lock.unlock)
100
# Taking out a lock on a locked file should raise LockContention
101
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
103
def _disabled_test_read_then_write_excludes(self):
104
"""If a file is read-locked, taking out a write lock should fail."""
105
a_lock = self.read_lock('a-file')
106
self.addCleanup(a_lock.unlock)
107
# Taking out a lock on a locked file should raise LockContention
108
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
110
def test_read_unlock_write(self):
111
"""Make sure that unlocking allows us to lock write"""
112
a_lock = self.read_lock('a-file')
114
a_lock = self.write_lock('a-file')
117
# TODO: jam 20070319 fcntl read locks are not currently fully
118
# mutually exclusive with write locks. This will be fixed
119
# in the next release.
120
def _disabled_test_write_then_read_excludes(self):
121
"""If a file is write-locked, taking out a read lock should fail.
123
The file is exclusively owned by the write lock, so we shouldn't be
124
able to take out a shared read lock.
126
a_lock = self.write_lock('a-file')
127
self.addCleanup(a_lock.unlock)
128
# Taking out a lock on a locked file should raise LockContention
129
self.assertRaises(errors.LockContention, self.read_lock, 'a-file')
131
# TODO: jam 20070319 fcntl write locks are not currently fully
132
# mutually exclusive with read locks. This will be fixed
133
# in the next release.
134
def _disabled_test_write_unlock_read(self):
135
"""If we have removed the write lock, we can grab a read lock."""
136
a_lock = self.write_lock('a-file')
138
a_lock = self.read_lock('a-file')
141
def _disabled_test_multiple_read_unlock_write(self):
142
"""We can only grab a write lock if all read locks are done."""
143
a_lock = b_lock = c_lock = None
145
a_lock = self.read_lock('a-file')
146
b_lock = self.read_lock('a-file')
147
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
150
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
153
c_lock = self.write_lock('a-file')
158
if a_lock is not None:
160
if b_lock is not None:
162
if c_lock is not None:
166
class TestLockUnicodePath(TestCaseWithLock):
168
_test_needs_features = [UnicodeFilenameFeature]
170
def test_read_lock(self):
171
self.build_tree([u'\u1234'])
172
u_lock = self.read_lock(u'\u1234')
173
self.addCleanup(u_lock.unlock)
175
def test_write_lock(self):
176
self.build_tree([u'\u1234'])
177
u_lock = self.write_lock(u'\u1234')
178
self.addCleanup(u_lock.unlock)