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 (features, 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
self.requireFeature(features.not_running_as_root)
67
osutils.make_readonly('a-file')
68
# Make sure the file is read-only (on all platforms)
69
self.assertRaises(IOError, open, 'a-file', 'rb+')
70
a_lock = self.read_lock('a-file')
73
self.assertRaises(errors.LockFailed, self.write_lock, 'a-file')
75
def test_write_lock(self):
76
"""Smoke test for write locks."""
77
a_lock = self.write_lock('a-file')
78
self.addCleanup(a_lock.unlock)
79
# You should be able to read and write to the lock file.
81
self.assertEqual('contents of a-file\n', txt)
82
# Win32 requires that you call seek() when switching between a read
83
# operation and a write operation.
85
a_lock.f.write('more content\n')
88
self.assertEqual('contents of a-file\nmore content\n', txt)
90
def test_multiple_read_locks(self):
91
"""You can take out more than one read lock on the same file."""
92
a_lock = self.read_lock('a-file')
93
self.addCleanup(a_lock.unlock)
94
b_lock = self.read_lock('a-file')
95
self.addCleanup(b_lock.unlock)
97
def test_multiple_write_locks_exclude(self):
98
"""Taking out more than one write lock should fail."""
99
a_lock = self.write_lock('a-file')
100
self.addCleanup(a_lock.unlock)
101
# Taking out a lock on a locked file should raise LockContention
102
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
104
def _disabled_test_read_then_write_excludes(self):
105
"""If a file is read-locked, taking out a write lock should fail."""
106
a_lock = self.read_lock('a-file')
107
self.addCleanup(a_lock.unlock)
108
# Taking out a lock on a locked file should raise LockContention
109
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
111
def test_read_unlock_write(self):
112
"""Make sure that unlocking allows us to lock write"""
113
a_lock = self.read_lock('a-file')
115
a_lock = self.write_lock('a-file')
118
# TODO: jam 20070319 fcntl read locks are not currently fully
119
# mutually exclusive with write locks. This will be fixed
120
# in the next release.
121
def _disabled_test_write_then_read_excludes(self):
122
"""If a file is write-locked, taking out a read lock should fail.
124
The file is exclusively owned by the write lock, so we shouldn't be
125
able to take out a shared read lock.
127
a_lock = self.write_lock('a-file')
128
self.addCleanup(a_lock.unlock)
129
# Taking out a lock on a locked file should raise LockContention
130
self.assertRaises(errors.LockContention, self.read_lock, 'a-file')
132
# TODO: jam 20070319 fcntl write locks are not currently fully
133
# mutually exclusive with read locks. This will be fixed
134
# in the next release.
135
def _disabled_test_write_unlock_read(self):
136
"""If we have removed the write lock, we can grab a read lock."""
137
a_lock = self.write_lock('a-file')
139
a_lock = self.read_lock('a-file')
142
def _disabled_test_multiple_read_unlock_write(self):
143
"""We can only grab a write lock if all read locks are done."""
144
a_lock = b_lock = c_lock = None
146
a_lock = self.read_lock('a-file')
147
b_lock = self.read_lock('a-file')
148
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
151
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
154
c_lock = self.write_lock('a-file')
159
if a_lock is not None:
161
if b_lock is not None:
163
if c_lock is not None:
167
class TestLockUnicodePath(TestCaseWithLock):
169
_test_needs_features = [UnicodeFilenameFeature]
171
def test_read_lock(self):
172
self.build_tree([u'\u1234'])
173
u_lock = self.read_lock(u'\u1234')
174
self.addCleanup(u_lock.unlock)
176
def test_write_lock(self):
177
self.build_tree([u'\u1234'])
178
u_lock = self.write_lock(u'\u1234')
179
self.addCleanup(u_lock.unlock)