1
# Copyright (C) 2005-2011 Canonical Ltd
1
# Copyright (C) 2005-2010 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
from StringIO import StringIO
24
from ..lockable_files import LockableFiles, TransportLock
26
from bzrlib.errors import BzrBadParameterNotString, NoSuchFile, ReadOnlyError
27
from bzrlib.lockable_files import LockableFiles, TransportLock
28
from bzrlib.symbol_versioning import (
31
from bzrlib.tests import (
29
from .test_smart import TestCaseWithSmartMedium
30
from .test_transactions import DummyWeave
31
from ..transactions import (PassThroughTransaction,
35
from bzrlib.tests.test_smart import TestCaseWithSmartMedium
36
from bzrlib.tests.test_transactions import DummyWeave
37
from bzrlib.transactions import (PassThroughTransaction,
37
43
# these tests are applied in each parameterized suite for LockableFiles
76
82
self.lockable.unlock()
77
83
raise TestNotApplicable("%r is not breakable" % (self.lockable,))
78
84
l2 = self.get_lockable()
79
orig_factory = breezy.ui.ui_factory
85
orig_factory = bzrlib.ui.ui_factory
80
86
# silent ui - no need for stdout
81
breezy.ui.ui_factory = breezy.ui.CannedInputUIFactory([True])
87
bzrlib.ui.ui_factory = bzrlib.ui.CannedInputUIFactory([True])
85
breezy.ui.ui_factory = orig_factory
91
bzrlib.ui.ui_factory = orig_factory
119
125
# This test does not apply, because this lockable refuses
122
different_token = token + b'xxx'
128
different_token = token + 'xxx'
123
129
# Re-using the same lockable instance with a different token will
124
130
# raise TokenMismatch.
125
131
self.assertRaises(errors.TokenMismatch,
272
278
_TestLockableFiles_mixin):
275
super(TestLockableFiles_TransportLock, self).setUp()
276
t = transport.get_transport_from_path('.')
281
TestCaseInTempDir.setUp(self)
282
t = transport.get_transport('.')
278
284
self.sub_transport = t.clone('.bzr')
279
285
self.lockable = self.get_lockable()
295
301
class TestLockableFiles_LockDir(TestCaseInTempDir,
296
_TestLockableFiles_mixin):
302
_TestLockableFiles_mixin):
297
303
"""LockableFile tests run with LockDir underneath"""
300
super(TestLockableFiles_LockDir, self).setUp()
301
self.transport = transport.get_transport_from_path('.')
306
TestCaseInTempDir.setUp(self)
307
self.transport = transport.get_transport('.')
302
308
self.lockable = self.get_lockable()
303
309
# the lock creation here sets mode - test_permissions on branch
304
310
# tests that implicitly, but it might be a good idea to factor
323
329
lockable = LockableFiles(self.transport.clone('readonly'), 'test-lock',
325
331
# The directory mode should be read-write-execute for the current user
326
self.assertEqual(0o0700, lockable._dir_mode & 0o0700)
332
self.assertEqual(00700, lockable._dir_mode & 00700)
327
333
# Files should be read-write for the current user
328
self.assertEqual(0o0600, lockable._file_mode & 0o0700)
334
self.assertEqual(00600, lockable._file_mode & 00700)
331
337
class TestLockableFiles_RemoteLockDir(TestCaseWithSmartMedium,
332
_TestLockableFiles_mixin):
338
_TestLockableFiles_mixin):
333
339
"""LockableFile tests run with RemoteLockDir on a branch."""
336
super(TestLockableFiles_RemoteLockDir, self).setUp()
342
TestCaseWithSmartMedium.setUp(self)
337
343
# can only get a RemoteLockDir with some RemoteObject...
338
344
# use a branch as thats what we want. These mixin tests test the end
339
345
# to end behaviour, so stubbing out the backend and simulating would
340
346
# defeat the purpose. We test the protocol implementation separately
341
347
# in test_remote and test_smart as usual.
342
348
b = self.make_branch('foo')
343
self.addCleanup(b.controldir.transport.disconnect)
344
self.transport = transport.get_transport_from_path('.')
349
self.addCleanup(b.bzrdir.transport.disconnect)
350
self.transport = transport.get_transport('.')
345
351
self.lockable = self.get_lockable()
347
353
def get_lockable(self):
348
354
# getting a new lockable involves opening a new instance of the branch
349
branch = breezy.branch.Branch.open(self.get_url('foo'))
350
self.addCleanup(branch.controldir.transport.disconnect)
355
branch = bzrlib.branch.Branch.open(self.get_url('foo'))
356
self.addCleanup(branch.bzrdir.transport.disconnect)
351
357
return branch.control_files