« back to all changes in this revision
Viewing changes to bzrlib/tests/per_lock/test_lock.py
- browse files at revision 2353.4.2
- compare with another revision
- compare with revision 489
- download diff
- download tarball
- view history from revision 2353.4.2
- Committer: John Arbash Meinel
- Date: 2007-03-14 22:31:38 UTC
- mfrom: (2353.3.11 lock_cleanup)
- mto: (2363.3.1 simple_locking) (2353.4.10 locking)
- mto: This revision was merged to the branch mainline in revision 2367.
- Revision ID: john@arbash-meinel.com-20070314223138-wfmodqhruo0g5l0e
[merge] LockCleanup changes.
- files added:
- bzrlib/tests/per_lock
- files renamed:
- bzrlib/tests/test_lock.py => bzrlib/tests/per_lock/test_lock.py
- files modified:
- bzrlib/errors.py
added
removed
bzrlib/tests/per_lock/test_lock.py
18
18
19
19
from bzrlib import (
20
20
errors,
21
lock,
22
21
osutils,
23
tests,
24
22
)
25
23
26
27
class TestLock(tests.TestCaseInTempDir):
24
from bzrlib.tests.per_lock import TestCaseWithLock
25
26
27
class TestLock(TestCaseWithLock):
28
28
29
29
def setUp(self):
30
30
super(TestLock, self).setUp()
32
32
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()
40
40
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)
47
47
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')
68
68
a_lock.unlock()
69
69
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')
72
72
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()
87
87
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)
94
94
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')
101
101
102
102
def 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')
108
108
109
109
def test_read_unlock_write(self):
110
110
"""Make sure that unlocking allows us to lock write"""
111
a_lock = lock.ReadLock('a-file')
111
a_lock = self.read_lock('a-file')
112
112
a_lock.unlock()
113
a_lock = lock.WriteLock('a-file')
113
a_lock = self.write_lock('a-file')
114
114
a_lock.unlock()
115
115
116
116
def test_write_then_read_excludes(self):
119
119
The file is exclusively owned by the write lock, so we shouldn't be
120
120
able to take out a shared read lock.
121
121
"""
122
a_lock = lock.WriteLock('a-file')
122
a_lock = self.write_lock('a-file')
123
123
self.addCleanup(a_lock.unlock)
124
124
# Taking out a lock on a locked file should raise LockContention
125
self.assertRaises(errors.LockContention, lock.ReadLock, 'a-file')
125
self.assertRaises(errors.LockContention, self.read_lock, 'a-file')
126
126
127
127
def test_write_unlock_read(self):
128
128
"""If we have removed the write lock, we can grab a read lock."""
129
a_lock = lock.WriteLock('a-file')
129
a_lock = self.write_lock('a-file')
130
130
a_lock.unlock()
131
a_lock = lock.ReadLock('a-file')
131
a_lock = self.read_lock('a-file')
132
132
a_lock.unlock()
133
133
134
134
def test_multiple_read_unlock_write(self):
135
135
"""We can only grab a write lock if all read locks are done."""
136
136
a_lock = b_lock = c_lock = None
137
137
try:
138
a_lock = lock.ReadLock('a-file')
139
b_lock = lock.ReadLock('a-file')
140
self.assertRaises(errors.LockContention, lock.WriteLock, 'a-file')
138
a_lock = self.read_lock('a-file')
139
b_lock = self.read_lock('a-file')
140
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
141
141
a_lock.unlock()
142
142
a_lock = None
143
self.assertRaises(errors.LockContention, lock.WriteLock, 'a-file')
143
self.assertRaises(errors.LockContention, self.write_lock, 'a-file')
144
144
b_lock.unlock()
145
145
b_lock = None
146
c_lock = lock.WriteLock('a-file')
146
c_lock = self.write_lock('a-file')
147
147
c_lock.unlock()
148
148
c_lock = None
149
149
finally:
Loggerhead is a web-based interface for Breezy
Version: 2.0.1