← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockable_files.py

(mbp) merge bzr.dev to 0.8, prepare for release

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockable_files.py
16
16
 
17
17
from StringIO import StringIO
18
18
 
 
19
import bzrlib
19
20
from bzrlib.branch import Branch
 
21
import bzrlib.errors as errors
20
22
from bzrlib.errors import BzrBadParameterNotString, NoSuchFile, ReadOnlyError
21
23
from bzrlib.lockable_files import LockableFiles, TransportLock
22
24
from bzrlib.lockdir import LockDir
95
97
    def test__escape_empty(self):
96
98
        self.assertEqual('', self.lockable._escape(''))
97
99
 
 
100
    def test_break_lock(self):
 
101
        # some locks are not breakable
 
102
        self.lockable.lock_write()
 
103
        try:
 
104
            self.assertRaises(AssertionError, self.lockable.break_lock)
 
105
        except NotImplementedError:
 
106
            # this lock cannot be broken
 
107
            self.lockable.unlock()
 
108
            return
 
109
        l2 = self.get_lockable()
 
110
        orig_factory = bzrlib.ui.ui_factory
 
111
        # silent ui - no need for stdout
 
112
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
 
113
        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
 
114
        try:
 
115
            l2.break_lock()
 
116
        finally:
 
117
            bzrlib.ui.ui_factory = orig_factory
 
118
        try:
 
119
            l2.lock_write()
 
120
            l2.unlock()
 
121
        finally:
 
122
            self.assertRaises(errors.LockBroken, self.lockable.unlock)
 
123
            self.assertFalse(self.lockable.is_locked())
 
124
 
98
125
 
99
126
# This method of adapting tests to parameters is different to 
100
127
# the TestProviderAdapters used elsewhere, but seems simpler for this 
106
133
        super(TestLockableFiles_TransportLock, self).setUp()
107
134
        transport = get_transport('.')
108
135
        transport.mkdir('.bzr')
109
 
        sub_transport = transport.clone('.bzr')
110
 
        self.lockable = LockableFiles(sub_transport, 'my-lock', TransportLock)
 
136
        self.sub_transport = transport.clone('.bzr')
 
137
        self.lockable = self.get_lockable()
111
138
        self.lockable.create_lock()
 
139
 
 
140
    def tearDown(self):
 
141
        super(TestLockableFiles_TransportLock, self).tearDown()
 
142
        # free the subtransport so that we do not get a 5 second
 
143
        # timeout due to the SFTP connection cache.
 
144
        del self.sub_transport
 
145
 
 
146
    def get_lockable(self):
 
147
        return LockableFiles(self.sub_transport, 'my-lock', TransportLock)
112
148
        
113
149
 
114
150
class TestLockableFiles_LockDir(TestCaseInTempDir,
118
154
    def setUp(self):
119
155
        super(TestLockableFiles_LockDir, self).setUp()
120
156
        self.transport = get_transport('.')
121
 
        self.lockable = LockableFiles(self.transport, 'my-lock', LockDir)
 
157
        self.lockable = self.get_lockable()
 
158
        # the lock creation here sets mode - test_permissions on branch 
 
159
        # tests that implicitly, but it might be a good idea to factor 
 
160
        # out the mode checking logic and have it applied to loackable files
 
161
        # directly. RBC 20060418
122
162
        self.lockable.create_lock()
123
163
 
124
 
    def test_lock_is_lockdir(self):
125
 
        """Created instance should use a LockDir.
126
 
        
127
 
        This primarily tests the mixin adapter works properly.
128
 
        """
129
 
        ## self.assertIsInstance(self.lockable, LockableFiles)
130
 
        ## self.assertIsInstance(self.lockable._lock_strategy,
131
 
                              ## LockDirStrategy)
 
164
    def get_lockable(self):
 
165
        return LockableFiles(self.transport, 'my-lock', LockDir)
132
166
 
133
167
    def test_lock_created(self):
134
168
        self.assertTrue(self.transport.has('my-lock'))