19
19
from bzrlib import (
27
class TestLock(tests.TestCaseInTempDir):
24
from bzrlib.tests.per_lock import TestCaseWithLock
27
class TestLock(TestCaseWithLock):
30
30
super(TestLock, self).setUp()
33
33
def test_read_lock(self):
34
34
"""Smoke test for read locks."""
35
a_lock = lock.ReadLock('a-file')
35
a_lock = self.read_lock('a-file')
36
36
self.addCleanup(a_lock.unlock)
37
37
# The lock file should be opened for reading
38
38
txt = a_lock.f.read()
41
41
def test_create_if_needed_read(self):
42
42
"""We will create the file if it doesn't exist yet."""
43
a_lock = lock.ReadLock('other-file')
43
a_lock = self.read_lock('other-file')
44
44
self.addCleanup(a_lock.unlock)
45
45
txt = a_lock.f.read()
46
46
self.assertEqual('', txt)
48
48
def test_create_if_needed_write(self):
49
49
"""We will create the file if it doesn't exist yet."""
50
a_lock = lock.WriteLock('other-file')
50
a_lock = self.write_lock('other-file')
51
51
self.addCleanup(a_lock.unlock)
52
52
txt = a_lock.f.read()
53
53
self.assertEqual('', txt)
64
64
osutils.make_readonly('a-file')
65
65
# Make sure the file is read-only (on all platforms)
66
66
self.assertRaises(IOError, open, 'a-file', 'rb+')
67
a_lock = lock.ReadLock('a-file')
67
a_lock = self.read_lock('a-file')
70
70
# TODO: jam 20070313 This should be a specific subclass
71
self.assertRaises(errors.ReadOnlyLockError, lock.WriteLock, 'a-file')
71
self.assertRaises(errors.ReadOnlyLockError, self.write_lock, 'a-file')
73
73
def test_write_lock(self):
74
74
"""Smoke test for write locks."""
75
a_lock = lock.WriteLock('a-file')
75
a_lock = self.write_lock('a-file')
76
76
self.addCleanup(a_lock.unlock)
77
77
# You should be able to read and write to the lock file.
78
78
txt = a_lock.f.read()
88
88
def test_multiple_read_locks(self):
89
89
"""You can take out more than one read lock on the same file."""
90
a_lock = lock.ReadLock('a-file')
90
a_lock = self.read_lock('a-file')
91
91
self.addCleanup(a_lock.unlock)
92
b_lock = lock.ReadLock('a-file')
92
b_lock = self.read_lock('a-file')
93
93
self.addCleanup(b_lock.unlock)
95
95
def test_multiple_write_locks_exclude(self):
96
96
"""Taking out more than one write lock should fail."""
97
a_lock = lock.WriteLock('a-file')
97
a_lock = self.write_lock('a-file')
98
98
self.addCleanup(a_lock.unlock)
99
99
# Taking out a lock on a locked file should raise LockContention
100
self.assertRaises(errors.LockContention, lock.WriteLock, 'a-file')
100
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
102
102
def _disabled_test_read_then_write_excludes(self):
103
103
"""If a file is read-locked, taking out a write lock should fail."""
104
a_lock = lock.ReadLock('a-file')
104
a_lock = self.read_lock('a-file')
105
105
self.addCleanup(a_lock.unlock)
106
106
# Taking out a lock on a locked file should raise LockContention
107
self.assertRaises(errors.LockContention, lock.WriteLock, 'a-file')
107
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
109
109
def _disabled_test_write_then_read_excludes(self):
110
110
"""If a file is write-locked, taking out a read lock should fail.
112
112
The file is exclusively owned by the write lock, so we shouldn't be
113
113
able to take out a shared read lock.
115
a_lock = lock.WriteLock('a-file')
115
a_lock = self.write_lock('a-file')
116
116
self.addCleanup(a_lock.unlock)
117
117
# Taking out a lock on a locked file should raise LockContention
118
self.assertRaises(errors.LockContention, lock.ReadLock, 'a-file')
118
self.assertRaises(errors.LockContention, self.read_lock, 'a-file')