17
17
"""Tests for bzrdir implementations - tests a bzrdir format."""
19
19
from cStringIO import StringIO
21
22
from stat import S_ISDIR
24
25
import bzrlib.branch
25
import bzrlib.bzrdir as bzrdir
26
36
from bzrlib.branch import Branch, needs_read_lock, needs_write_lock
27
37
from bzrlib.check import check
28
import bzrlib.errors as errors
29
38
from bzrlib.errors import (FileExists,
32
41
UninitializableFormat,
35
import bzrlib.repository as repository
36
44
import bzrlib.revision
37
45
from bzrlib.tests import (
43
51
from bzrlib.trace import mutter
44
import bzrlib.transactions as transactions
45
import bzrlib.transport as transport
46
52
from bzrlib.transport import get_transport
47
import bzrlib.ui as ui
48
53
from bzrlib.upgrade import upgrade
49
import bzrlib.workingtree as workingtree
52
56
class TestCaseWithBzrDir(TestCaseWithTransport):
150
154
wt = dir.create_workingtree(revision_id=bzrlib.revision.NULL_REVISION)
151
155
self.assertEqual([], wt.get_parent_ids())
157
def test_destroy_workingtree(self):
158
tree = self.make_branch_and_tree('tree')
159
self.build_tree(['tree/file'])
161
tree.commit('first commit')
164
bzrdir.destroy_workingtree()
165
except errors.UnsupportedOperation:
166
raise TestSkipped('Format does not support destroying tree')
167
self.failIfExists('tree/file')
168
self.assertRaises(errors.NoWorkingTree, bzrdir.open_workingtree)
169
bzrdir.create_workingtree()
170
self.failUnlessExists('tree/file')
171
bzrdir.destroy_workingtree_metadata()
172
self.failUnlessExists('tree/file')
173
self.assertRaises(errors.NoWorkingTree, bzrdir.open_workingtree)
153
175
def test_clone_bzrdir_empty(self):
154
176
dir = self.make_bzrdir('source')
155
177
target = dir.clone(self.get_url('target'))
515
537
repo = dir.create_repository()
516
538
repo.fetch(tree.branch.repository)
517
539
self.assertTrue(repo.has_revision('1'))
541
self.assertIs(dir.open_branch().last_revision(), None)
542
except errors.NotBranchError:
518
544
target = self.sproutOrSkip(dir, self.get_url('target'))
519
545
self.assertNotEqual(dir.transport.base, target.transport.base)
546
# testing inventory isn't reasonable for repositories
520
547
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
521
548
['./.bzr/repository/inventory.knit',
552
# If we happen to have a tree, we'll guarantee everything
553
# except for the tree root is the same.
554
inventory_f = file(dir.transport.base+'inventory', 'rb')
555
self.assertContainsRe(inventory_f.read(),
556
'<inventory file_id="TREE_ROOT[^"]*"'
557
' format="5">\n</inventory>\n')
560
if e.errno != errno.ENOENT:
524
563
def test_sprout_bzrdir_with_repository_to_shared(self):
525
564
tree = self.make_branch_and_tree('commit_tree')
1317
1356
unused_repo = thisdir.create_repository()
1318
1357
master.lock_write()
1319
1358
unused_repo.lock_write()
1320
# two yes's : branch and repository. If the repo in this
1321
# dir is inappropriately accessed, 3 will be needed, and
1322
# we'll see that because the stream will be fully consumed
1323
bzrlib.ui.ui_factory.stdin = StringIO("y\ny\ny\n")
1324
master.bzrdir.break_lock()
1325
# only two ys should have been read
1326
self.assertEqual("y\n", bzrlib.ui.ui_factory.stdin.read())
1327
# we should be able to lock a newly opened branch now
1328
branch = master.bzrdir.open_branch()
1331
# we should not be able to lock the repository in thisdir as its still
1332
# held by the explicit lock we took, and the break lock should not have
1334
repo = thisdir.open_repository()
1335
self.assertRaises(errors.LockContention, repo.lock_write)
1336
unused_repo.unlock()
1360
# two yes's : branch and repository. If the repo in this
1361
# dir is inappropriately accessed, 3 will be needed, and
1362
# we'll see that because the stream will be fully consumed
1363
bzrlib.ui.ui_factory.stdin = StringIO("y\ny\ny\n")
1364
master.bzrdir.break_lock()
1365
# only two ys should have been read
1366
self.assertEqual("y\n", bzrlib.ui.ui_factory.stdin.read())
1367
# we should be able to lock a newly opened branch now
1368
branch = master.bzrdir.open_branch()
1371
# we should not be able to lock the repository in thisdir as its still
1372
# held by the explicit lock we took, and the break lock should not have
1374
repo = thisdir.open_repository()
1375
orig_default = lockdir._DEFAULT_TIMEOUT_SECONDS
1377
lockdir._DEFAULT_TIMEOUT_SECONDS = 1
1378
self.assertRaises(errors.LockContention, repo.lock_write)
1380
lockdir._DEFAULT_TIMEOUT_SECONDS = orig_default
1382
unused_repo.unlock()
1337
1383
self.assertRaises(errors.LockBroken, master.unlock)
1339
1385
def test_break_lock_tree(self):