# Copyright (C) 2007 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 temporarily upgrading to a WriteLock."""
from bzrlib import (
errors,
)
from bzrlib.tests.per_lock import TestCaseWithLock
class TestTemporaryWriteLock(TestCaseWithLock):
def setUp(self):
super(TestTemporaryWriteLock, self).setUp()
self.build_tree(['a-file'])
def test_can_upgrade_and_write(self):
"""With only one lock, we should be able to write lock and switch back."""
a_lock = self.read_lock('a-file')
try:
success, t_write_lock = a_lock.temporary_write_lock()
self.assertTrue(success, "We failed to grab a write lock.")
try:
self.assertEqual('contents of a-file\n',
t_write_lock.f.read())
# We should be able to write to the file.
t_write_lock.f.seek(0)
t_write_lock.f.write('new contents for a-file\n')
t_write_lock.f.seek(0)
self.assertEqual('new contents for a-file\n',
t_write_lock.f.read())
finally:
a_lock = t_write_lock.restore_read_lock()
finally:
a_lock.unlock()
def test_is_write_locked(self):
"""With a temporary write lock, we cannot grab another lock."""
a_lock = self.read_lock('a-file')
try:
success, t_write_lock = a_lock.temporary_write_lock()
self.assertTrue(success, "We failed to grab a write lock.")
try:
self.assertRaises(errors.LockContention,
self.write_lock, 'a-file')
# TODO: jam 20070319 fcntl read locks are not currently fully
# mutually exclusive with write locks. This will be fixed
# in the next release.
# self.assertRaises(errors.LockContention,
# self.read_lock, 'a-file')
finally:
a_lock = t_write_lock.restore_read_lock()
# Now we only have a read lock, so we should be able to grab
# another read lock, but not a write lock
# TODO: jam 20070319 fcntl write locks are not currently fully
# mutually exclusive with read locks. This will be fixed
# in the next release.
# self.assertRaises(errors.LockContention,
# self.write_lock, 'a-file')
b_lock = self.read_lock('a-file')
b_lock.unlock()
finally:
a_lock.unlock()
def test_fails_when_locked(self):
"""We can't upgrade to a write lock if something else locks."""
a_lock = self.read_lock('a-file')
try:
b_lock = self.read_lock('a-file')
try:
success, alt_lock = a_lock.temporary_write_lock()
self.assertFalse(success)
# Now, 'alt_lock' should be a read-lock on the file. It should
# either be the same object as a_lock, or a new object.
# If it is a new object, a_lock should be unlocked. (we don't
# want to end up with 2 locks on the file)
self.assertTrue((alt_lock is a_lock) or (a_lock.f is None))
# set a_lock = alt_lock so that cleanup works correctly
a_lock = alt_lock
# We should not be able to grab a write lock
# but we should be able to grab another read lock
# TODO: jam 20070319 fcntl write locks are not currently fully
# mutually exclusive with read locks. This will be fixed
# in the next release.
# self.assertRaises(errors.LockContention,
# self.write_lock, 'a-file')
c_lock = self.read_lock('a-file')
c_lock.unlock()
finally:
b_lock.unlock()
finally:
a_lock.unlock()
# TODO: jam 20070314 to truly test these, we should be spawning an external
# process, and having it lock/unlock/try lock on request.
# TODO: jam 20070314 Test that the write lock can fail if another process
# holds a read lock. And that we recover properly.
|