1
# Copyright (C) 2010 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for bzrdir implementations - tests a bzrdir format."""
20
from stat import S_ISDIR
25
revision as _mod_revision,
28
from bzrlib.tests import (
31
from bzrlib.tests.per_bzrdir import TestCaseWithBzrDir
32
from bzrlib.transport.local import (
37
class TestBzrDir(TestCaseWithBzrDir):
39
# Many of these tests test for disk equality rather than checking
40
# for semantic equivalence. This works well for some tests but
41
# is not good at handling changes in representation or the addition
42
# or removal of control data. It would be nice to for instance:
43
# sprout a new branch, check that the nickname has been reset by hand
44
# and then set the nickname to match the source branch, at which point
45
# a semantic equivalence should pass
47
def assertDirectoriesEqual(self, source, target, ignore_list=[]):
48
"""Assert that the content of source and target are identical.
50
paths in ignore list will be completely ignored.
52
We ignore paths that represent data which is allowed to change during
53
a clone or sprout: for instance, inventory.knit contains gzip fragements
54
which have timestamps in them, and as we have read the inventory from
55
the source knit, the already-read data is recompressed rather than
56
reading it again, which leads to changed timestamps. This is ok though,
57
because the inventory.kndx file is not ignored, and the integrity of
58
knit joins is tested by test_knit and test_versionedfile.
60
:seealso: Additionally, assertRepositoryHasSameItems provides value
61
rather than representation checking of repositories for
67
dir = directories.pop()
68
for path in set(source.list_dir(dir) + target.list_dir(dir)):
69
path = dir + '/' + path
70
if path in ignore_list:
73
stat = source.stat(path)
74
except errors.NoSuchFile:
75
self.fail('%s not in source' % path)
76
if S_ISDIR(stat.st_mode):
77
self.assertTrue(S_ISDIR(target.stat(path).st_mode))
78
directories.append(path)
80
self.assertEqualDiff(source.get(path).read(),
81
target.get(path).read(),
82
"text for file %r differs:\n" % path)
84
def assertRepositoryHasSameItems(self, left_repo, right_repo):
85
"""require left_repo and right_repo to contain the same data."""
86
# XXX: TODO: Doesn't work yet, because we need to be able to compare
87
# local repositories to remote ones... but this is an as-yet unsolved
88
# aspect of format management and the Remote protocols...
89
# self.assertEqual(left_repo._format.__class__,
90
# right_repo._format.__class__)
93
right_repo.lock_read()
96
all_revs = left_repo.all_revision_ids()
97
self.assertEqual(left_repo.all_revision_ids(),
98
right_repo.all_revision_ids())
99
for rev_id in left_repo.all_revision_ids():
100
self.assertEqual(left_repo.get_revision(rev_id),
101
right_repo.get_revision(rev_id))
102
# Assert the revision trees (and thus the inventories) are equal
103
sort_key = lambda rev_tree: rev_tree.get_revision_id()
104
rev_trees_a = sorted(
105
left_repo.revision_trees(all_revs), key=sort_key)
106
rev_trees_b = sorted(
107
right_repo.revision_trees(all_revs), key=sort_key)
108
for tree_a, tree_b in zip(rev_trees_a, rev_trees_b):
109
self.assertEqual([], list(tree_a.iter_changes(tree_b)))
111
text_index = left_repo._generate_text_key_index()
112
self.assertEqual(text_index,
113
right_repo._generate_text_key_index())
115
for file_id, revision_id in text_index.iterkeys():
116
desired_files.append(
117
(file_id, revision_id, (file_id, revision_id)))
118
left_texts = list(left_repo.iter_files_bytes(desired_files))
119
right_texts = list(right_repo.iter_files_bytes(desired_files))
122
self.assertEqual(left_texts, right_texts)
124
for rev_id in all_revs:
126
left_text = left_repo.get_signature_text(rev_id)
127
except errors.NoSuchRevision:
129
right_text = right_repo.get_signature_text(rev_id)
130
self.assertEqual(left_text, right_text)
136
def sproutOrSkip(self, from_bzrdir, to_url, revision_id=None,
137
force_new_repo=False, accelerator_tree=None,
138
create_tree_if_local=True):
139
"""Sprout from_bzrdir into to_url, or raise TestSkipped.
141
A simple wrapper for from_bzrdir.sprout that translates NotLocalUrl into
142
TestSkipped. Returns the newly sprouted bzrdir.
144
to_transport = transport.get_transport(to_url)
145
if not isinstance(to_transport, LocalTransport):
146
raise TestSkipped('Cannot sprout to remote bzrdirs.')
147
target = from_bzrdir.sprout(to_url, revision_id=revision_id,
148
force_new_repo=force_new_repo,
149
possible_transports=[to_transport],
150
accelerator_tree=accelerator_tree,
151
create_tree_if_local=create_tree_if_local)
154
def skipIfNoWorkingTree(self, a_bzrdir):
155
"""Raises TestSkipped if a_bzrdir doesn't have a working tree.
157
If the bzrdir does have a workingtree, this is a no-op.
160
a_bzrdir.open_workingtree()
161
except (errors.NotLocalUrl, errors.NoWorkingTree):
162
raise TestSkipped("bzrdir on transport %r has no working tree"
163
% a_bzrdir.transport)
165
def createWorkingTreeOrSkip(self, a_bzrdir):
166
"""Create a working tree on a_bzrdir, or raise TestSkipped.
168
A simple wrapper for create_workingtree that translates NotLocalUrl into
169
TestSkipped. Returns the newly created working tree.
172
# This passes in many named options to make sure they're
173
# understood by subclasses: see
174
# <https://bugs.launchpad.net/bzr/+bug/524627>.
175
return a_bzrdir.create_workingtree(
178
accelerator_tree=None,
180
except errors.NotLocalUrl:
181
raise TestSkipped("cannot make working tree with transport %r"
182
% a_bzrdir.transport)
184
def test_clone_bzrdir_repository_under_shared_force_new_repo(self):
185
tree = self.make_branch_and_tree('commit_tree')
186
self.build_tree(['commit_tree/foo'])
188
tree.commit('revision 1', rev_id='1')
189
dir = self.make_bzrdir('source')
190
repo = dir.create_repository()
191
repo.fetch(tree.branch.repository)
192
self.assertTrue(repo.has_revision('1'))
194
self.make_repository('target', shared=True)
195
except errors.IncompatibleFormat:
197
target = dir.clone(self.get_url('target/child'), force_new_repo=True)
198
self.assertNotEqual(dir.transport.base, target.transport.base)
199
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
200
['./.bzr/repository',
202
self.assertRepositoryHasSameItems(tree.branch.repository, repo)
204
def test_clone_bzrdir_branch_and_repo(self):
205
tree = self.make_branch_and_tree('commit_tree')
206
self.build_tree(['commit_tree/foo'])
208
tree.commit('revision 1')
209
source = self.make_branch('source')
210
tree.branch.repository.copy_content_into(source.repository)
211
tree.branch.copy_content_into(source)
213
target = dir.clone(self.get_url('target'))
214
self.assertNotEqual(dir.transport.base, target.transport.base)
215
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
217
'./.bzr/basis-inventory-cache',
218
'./.bzr/checkout/stat-cache',
219
'./.bzr/merge-hashes',
223
self.assertRepositoryHasSameItems(
224
tree.branch.repository, target.open_repository())
226
def test_clone_on_transport(self):
227
a_dir = self.make_bzrdir('source')
228
target_transport = a_dir.root_transport.clone('..').clone('target')
229
target = a_dir.clone_on_transport(target_transport)
230
self.assertNotEqual(a_dir.transport.base, target.transport.base)
231
self.assertDirectoriesEqual(a_dir.root_transport, target.root_transport,
232
['./.bzr/merge-hashes'])
234
def test_clone_bzrdir_empty(self):
235
dir = self.make_bzrdir('source')
236
target = dir.clone(self.get_url('target'))
237
self.assertNotEqual(dir.transport.base, target.transport.base)
238
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
239
['./.bzr/merge-hashes'])
241
def test_clone_bzrdir_empty_force_new_ignored(self):
242
# the force_new_repo parameter should have no effect on an empty
243
# bzrdir's clone logic
244
dir = self.make_bzrdir('source')
245
target = dir.clone(self.get_url('target'), force_new_repo=True)
246
self.assertNotEqual(dir.transport.base, target.transport.base)
247
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
248
['./.bzr/merge-hashes'])
250
def test_clone_bzrdir_repository(self):
251
tree = self.make_branch_and_tree('commit_tree')
252
self.build_tree(['foo'], transport=tree.bzrdir.transport.clone('..'))
254
tree.commit('revision 1', rev_id='1')
255
dir = self.make_bzrdir('source')
256
repo = dir.create_repository()
257
repo.fetch(tree.branch.repository)
258
self.assertTrue(repo.has_revision('1'))
259
target = dir.clone(self.get_url('target'))
260
self.assertNotEqual(dir.transport.base, target.transport.base)
261
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
263
'./.bzr/merge-hashes',
266
self.assertRepositoryHasSameItems(tree.branch.repository,
267
target.open_repository())
269
def test_clone_bzrdir_tree_branch_repo(self):
270
tree = self.make_branch_and_tree('source')
271
self.build_tree(['source/foo'])
273
tree.commit('revision 1')
275
target = dir.clone(self.get_url('target'))
276
self.skipIfNoWorkingTree(target)
277
self.assertNotEqual(dir.transport.base, target.transport.base)
278
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
279
['./.bzr/stat-cache',
280
'./.bzr/checkout/dirstate',
281
'./.bzr/checkout/stat-cache',
282
'./.bzr/checkout/merge-hashes',
283
'./.bzr/merge-hashes',
286
self.assertRepositoryHasSameItems(tree.branch.repository,
287
target.open_repository())
288
target.open_workingtree().revert()
290
def test_revert_inventory(self):
291
tree = self.make_branch_and_tree('source')
292
self.build_tree(['source/foo'])
294
tree.commit('revision 1')
296
target = dir.clone(self.get_url('target'))
297
self.skipIfNoWorkingTree(target)
298
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
299
['./.bzr/stat-cache',
300
'./.bzr/checkout/dirstate',
301
'./.bzr/checkout/stat-cache',
302
'./.bzr/checkout/merge-hashes',
303
'./.bzr/merge-hashes',
306
self.assertRepositoryHasSameItems(tree.branch.repository,
307
target.open_repository())
309
target.open_workingtree().revert()
310
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
311
['./.bzr/stat-cache',
312
'./.bzr/checkout/dirstate',
313
'./.bzr/checkout/stat-cache',
314
'./.bzr/checkout/merge-hashes',
315
'./.bzr/merge-hashes',
318
self.assertRepositoryHasSameItems(tree.branch.repository,
319
target.open_repository())
321
def test_clone_bzrdir_tree_branch_reference(self):
322
# a tree with a branch reference (aka a checkout)
323
# should stay a checkout on clone.
324
referenced_branch = self.make_branch('referencced')
325
dir = self.make_bzrdir('source')
327
reference = bzrlib.branch.BranchReferenceFormat().initialize(dir,
328
target_branch=referenced_branch)
329
except errors.IncompatibleFormat:
330
# this is ok too, not all formats have to support references.
332
self.createWorkingTreeOrSkip(dir)
333
target = dir.clone(self.get_url('target'))
334
self.skipIfNoWorkingTree(target)
335
self.assertNotEqual(dir.transport.base, target.transport.base)
336
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
337
['./.bzr/stat-cache',
338
'./.bzr/checkout/stat-cache',
339
'./.bzr/checkout/merge-hashes',
340
'./.bzr/merge-hashes',
341
'./.bzr/repository/inventory.knit',
344
def test_clone_bzrdir_branch_and_repo_into_shared_repo_force_new_repo(self):
345
# by default cloning into a shared repo uses the shared repo.
346
tree = self.make_branch_and_tree('commit_tree')
347
self.build_tree(['commit_tree/foo'])
349
tree.commit('revision 1')
350
source = self.make_branch('source')
351
tree.branch.repository.copy_content_into(source.repository)
352
tree.branch.copy_content_into(source)
354
self.make_repository('target', shared=True)
355
except errors.IncompatibleFormat:
358
target = dir.clone(self.get_url('target/child'), force_new_repo=True)
359
self.assertNotEqual(dir.transport.base, target.transport.base)
360
repo = target.open_repository()
361
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
362
['./.bzr/repository',
364
self.assertRepositoryHasSameItems(tree.branch.repository, repo)
366
def test_clone_bzrdir_branch_reference(self):
367
# cloning should preserve the reference status of the branch in a bzrdir
368
referenced_branch = self.make_branch('referencced')
369
dir = self.make_bzrdir('source')
371
reference = bzrlib.branch.BranchReferenceFormat().initialize(dir,
372
target_branch=referenced_branch)
373
except errors.IncompatibleFormat:
374
# this is ok too, not all formats have to support references.
376
target = dir.clone(self.get_url('target'))
377
self.assertNotEqual(dir.transport.base, target.transport.base)
378
self.assertDirectoriesEqual(dir.root_transport, target.root_transport)
380
def test_sprout_bzrdir_repository(self):
381
tree = self.make_branch_and_tree('commit_tree')
382
self.build_tree(['foo'], transport=tree.bzrdir.transport.clone('..'))
384
tree.commit('revision 1', rev_id='1')
385
dir = self.make_bzrdir('source')
386
repo = dir.create_repository()
387
repo.fetch(tree.branch.repository)
388
self.assertTrue(repo.has_revision('1'))
391
_mod_revision.is_null(_mod_revision.ensure_null(
392
dir.open_branch().last_revision())))
393
except errors.NotBranchError:
395
target = dir.sprout(self.get_url('target'))
396
self.assertNotEqual(dir.transport.base, target.transport.base)
397
# testing inventory isn't reasonable for repositories
398
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
404
'./.bzr/repository/inventory.knit',
407
local_inventory = dir.transport.local_abspath('inventory')
408
except errors.NotLocalUrl:
411
# If we happen to have a tree, we'll guarantee everything
412
# except for the tree root is the same.
413
inventory_f = file(local_inventory, 'rb')
414
self.addCleanup(inventory_f.close)
415
self.assertContainsRe(inventory_f.read(),
416
'<inventory format="5">\n</inventory>\n')
418
if e.errno != errno.ENOENT:
421
def test_sprout_bzrdir_branch_and_repo(self):
422
tree = self.make_branch_and_tree('commit_tree')
423
self.build_tree(['commit_tree/foo'])
425
tree.commit('revision 1')
426
source = self.make_branch('source')
427
tree.branch.repository.copy_content_into(source.repository)
428
tree.bzrdir.open_branch().copy_content_into(source)
430
target = dir.sprout(self.get_url('target'))
431
self.assertNotEqual(dir.transport.base, target.transport.base)
432
target_repo = target.open_repository()
433
self.assertRepositoryHasSameItems(source.repository, target_repo)
434
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
436
'./.bzr/basis-inventory-cache',
437
'./.bzr/branch/branch.conf',
438
'./.bzr/branch/parent',
440
'./.bzr/checkout/inventory',
441
'./.bzr/checkout/stat-cache',
449
def test_sprout_bzrdir_tree_branch_repo(self):
450
tree = self.make_branch_and_tree('source')
451
self.build_tree(['foo'], transport=tree.bzrdir.transport.clone('..'))
453
tree.commit('revision 1')
455
target = self.sproutOrSkip(dir, self.get_url('target'))
456
self.assertNotEqual(dir.transport.base, target.transport.base)
457
self.assertDirectoriesEqual(dir.root_transport, target.root_transport,
459
'./.bzr/branch/branch.conf',
460
'./.bzr/branch/parent',
461
'./.bzr/checkout/dirstate',
462
'./.bzr/checkout/stat-cache',
463
'./.bzr/checkout/inventory',
469
self.assertRepositoryHasSameItems(
470
tree.branch.repository, target.open_repository())
473
def test_retire_bzrdir(self):
474
bd = self.make_bzrdir('.')
475
transport = bd.root_transport
476
# must not overwrite existing directories
477
self.build_tree(['.bzr.retired.0/', '.bzr.retired.0/junk',],
479
self.failUnless(transport.has('.bzr'))
481
self.failIf(transport.has('.bzr'))
482
self.failUnless(transport.has('.bzr.retired.1'))
484
def test_retire_bzrdir_limited(self):
485
bd = self.make_bzrdir('.')
486
transport = bd.root_transport
487
# must not overwrite existing directories
488
self.build_tree(['.bzr.retired.0/', '.bzr.retired.0/junk',],
490
self.failUnless(transport.has('.bzr'))
491
self.assertRaises((errors.FileExists, errors.DirectoryNotEmpty),
492
bd.retire_bzrdir, limit=0)