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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Tests for temporarily upgrading to a WriteLock."""
23
from bzrlib.tests.per_lock import TestCaseWithLock
26
class TestTemporaryWriteLock(TestCaseWithLock):
29
super(TestTemporaryWriteLock, self).setUp()
30
self.build_tree(['a-file'])
32
def test_can_upgrade_and_write(self):
33
"""With only one lock, we should be able to write lock and switch back."""
34
a_lock = self.read_lock('a-file')
36
success, t_write_lock = a_lock.temporary_write_lock()
37
self.assertTrue(success, "We failed to grab a write lock.")
39
self.assertEqual('contents of a-file\n',
40
t_write_lock.f.read())
41
# We should be able to write to the file.
42
t_write_lock.f.seek(0)
43
t_write_lock.f.write('new contents for a-file\n')
44
t_write_lock.f.seek(0)
45
self.assertEqual('new contents for a-file\n',
46
t_write_lock.f.read())
48
a_lock = t_write_lock.restore_read_lock()
52
def test_is_write_locked(self):
53
"""With a temporary write lock, we cannot grab another lock."""
54
a_lock = self.read_lock('a-file')
56
success, t_write_lock = a_lock.temporary_write_lock()
57
self.assertTrue(success, "We failed to grab a write lock.")
59
self.assertRaises(errors.LockContention,
60
self.write_lock, 'a-file')
61
# TODO: jam 20070319 fcntl read locks are not currently fully
62
# mutually exclusive with write locks. This will be fixed
63
# in the next release.
64
# self.assertRaises(errors.LockContention,
65
# self.read_lock, 'a-file')
67
a_lock = t_write_lock.restore_read_lock()
68
# Now we only have a read lock, so we should be able to grab
69
# another read lock, but not a write lock
70
# TODO: jam 20070319 fcntl write locks are not currently fully
71
# mutually exclusive with read locks. This will be fixed
72
# in the next release.
73
# self.assertRaises(errors.LockContention,
74
# self.write_lock, 'a-file')
75
b_lock = self.read_lock('a-file')
80
def test_fails_when_locked(self):
81
"""We can't upgrade to a write lock if something else locks."""
82
a_lock = self.read_lock('a-file')
84
b_lock = self.read_lock('a-file')
86
success, alt_lock = a_lock.temporary_write_lock()
87
self.assertFalse(success)
88
# Now, 'alt_lock' should be a read-lock on the file. It should
89
# either be the same object as a_lock, or a new object.
90
# If it is a new object, a_lock should be unlocked. (we don't
91
# want to end up with 2 locks on the file)
92
self.assertTrue((alt_lock is a_lock) or (a_lock.f is None))
93
# set a_lock = alt_lock so that cleanup works correctly
96
# We should not be able to grab a write lock
97
# but we should be able to grab another read lock
98
# TODO: jam 20070319 fcntl write locks are not currently fully
99
# mutually exclusive with read locks. This will be fixed
100
# in the next release.
101
# self.assertRaises(errors.LockContention,
102
# self.write_lock, 'a-file')
103
c_lock = self.read_lock('a-file')
110
# TODO: jam 20070314 to truly test these, we should be spawning an external
111
# process, and having it lock/unlock/try lock on request.
113
# TODO: jam 20070314 Test that the write lock can fail if another process
114
# holds a read lock. And that we recover properly.
added
removed
bzrlib/tests/per_lock/test_temporary_write_lock.py
