19
19
from cStringIO import StringIO
23
22
from bzrlib import (
23
branch as _mod_branch,
34
revision as _mod_revision,
34
from bzrlib.branch import BzrBranchFormat6
35
from bzrlib.delta import TreeDelta
36
from bzrlib.inventory import CommonInventory, Inventory, InventoryDirectory
37
from bzrlib.repofmt.pack_repo import (
38
RepositoryFormatKnitPack5RichRootBroken,
40
from bzrlib.repofmt.weaverepo import (
45
from bzrlib.revision import NULL_REVISION, Revision
46
from bzrlib.smart import server
42
from bzrlib.repofmt import (
47
46
from bzrlib.tests import (
49
TestCaseWithTransport,
53
from bzrlib.tests.per_repository import TestCaseWithRepository
54
from bzrlib.transport import get_transport
55
from bzrlib.transport.fakevfat import FakeVFATServer
56
from bzrlib.upgrade import upgrade
57
from bzrlib.workingtree import WorkingTree
60
class TestRepositoryMakeBranchAndTree(TestCaseWithRepository):
52
class TestRepositoryMakeBranchAndTree(per_repository.TestCaseWithRepository):
62
54
def test_repository_format(self):
63
55
# make sure the repository on tree.branch is of the desired format,
69
61
self.repository_format.__class__)
72
class TestRepository(TestCaseWithRepository):
64
class TestRepository(per_repository.TestCaseWithRepository):
74
66
def assertFormatAttribute(self, attribute, allowed_values):
75
67
"""Assert that the format has an attribute 'attribute'."""
92
84
"""Test the _fetch_reconcile attribute."""
93
85
self.assertFormatAttribute('_fetch_reconcile', (True, False))
87
def test_attribute_format_experimental(self):
88
self.assertFormatAttribute('experimental', (True, False))
95
90
def test_attribute_format_pack_compresses(self):
96
91
self.assertFormatAttribute('pack_compresses', (True, False))
99
94
"""Test the existence of the inventories attribute."""
100
95
tree = self.make_branch_and_tree('tree')
101
96
repo = tree.branch.repository
102
self.assertIsInstance(repo.inventories,
103
bzrlib.versionedfile.VersionedFiles)
97
self.assertIsInstance(repo.inventories, versionedfile.VersionedFiles)
105
99
def test_attribute_inventories_basics(self):
106
100
"""Test basic aspects of the inventories attribute."""
116
110
tree = self.make_branch_and_tree('tree')
117
111
repo = tree.branch.repository
118
112
self.assertIsInstance(repo.revisions,
119
bzrlib.versionedfile.VersionedFiles)
113
versionedfile.VersionedFiles)
121
115
def test_attribute_revision_store_basics(self):
122
116
"""Test the basic behaviour of the revisions attribute."""
150
144
tree = self.make_branch_and_tree('tree')
151
145
repo = tree.branch.repository
152
146
self.assertIsInstance(repo.signatures,
153
bzrlib.versionedfile.VersionedFiles)
147
versionedfile.VersionedFiles)
155
149
def test_attribute_text_store_basics(self):
156
150
"""Test the basic behaviour of the text store."""
167
161
rev_key = (tree.commit("foo"),)
168
162
except errors.IllegalPath:
169
raise TestNotApplicable('file_id %r cannot be stored on this'
163
raise tests.TestNotApplicable(
164
'file_id %r cannot be stored on this'
170
165
' platform for this repo format' % (file_id,))
171
166
if repo._format.rich_root_data:
172
167
root_commit = (tree.get_root_id(),) + rev_key
212
207
tree = self.make_branch_and_tree('tree')
213
208
repo = tree.branch.repository
214
209
self.assertIsInstance(repo.texts,
215
bzrlib.versionedfile.VersionedFiles)
210
versionedfile.VersionedFiles)
217
212
def test_exposed_versioned_files_are_marked_dirty(self):
218
213
repo = self.make_repository('.')
261
256
invs = tree.branch.repository.iter_inventories(revs)
262
257
for rev_id, inv in zip(revs, invs):
263
258
self.assertEqual(rev_id, inv.revision_id)
264
self.assertIsInstance(inv, CommonInventory)
259
self.assertIsInstance(inv, inventory.CommonInventory)
266
261
def test_supports_rich_root(self):
267
262
tree = self.make_branch_and_tree('a')
286
281
# they may not be initializable.
288
283
# supported formats must be able to init and open
289
t = get_transport(self.get_url())
290
readonly_t = get_transport(self.get_readonly_url())
284
t = transport.get_transport(self.get_url())
285
readonly_t = transport.get_transport(self.get_readonly_url())
291
286
made_control = self.bzrdir_format.initialize(t.base)
292
287
made_repo = self.repository_format.initialize(made_control)
293
288
self.assertEqual(made_control, made_repo.bzrdir)
346
341
# because the default open will not open them and
347
342
# they may not be initializable.
349
t = get_transport(self.get_url())
344
t = transport.get_transport(self.get_url())
350
345
made_control = self.bzrdir_format.initialize(t.base)
351
346
made_repo = made_control.create_repository()
352
347
# Check that we have a repository object.
360
355
# because the default open will not open them and
361
356
# they may not be initializable.
363
t = get_transport(self.get_url())
358
t = transport.get_transport(self.get_url())
364
359
made_control = self.bzrdir_format.initialize(t.base)
366
361
made_repo = made_control.create_repository(shared=True)
383
378
self.assertEqual('revision-1', tree.inventory.root.revision)
384
expected = InventoryDirectory('fixed-root', '', None)
379
expected = inventory.InventoryDirectory('fixed-root', '', None)
385
380
expected.revision = 'revision-1'
386
381
self.assertEqual([('', 'V', 'directory', 'fixed-root', expected)],
387
382
list(tree.list_files(include_root=True)))
412
407
tree_a.commit('rev2', rev_id='rev2')
414
409
delta = tree_a.branch.repository.get_revision_delta('rev1')
415
self.assertIsInstance(delta, TreeDelta)
410
self.assertIsInstance(delta, _mod_delta.TreeDelta)
416
411
self.assertEqual([('foo', 'file1', 'file')], delta.added)
417
412
delta = tree_a.branch.repository.get_revision_delta('rev2')
418
self.assertIsInstance(delta, TreeDelta)
413
self.assertIsInstance(delta, _mod_delta.TreeDelta)
419
414
self.assertEqual([('vla', 'file2', 'file')], delta.added)
421
416
def test_get_revision_delta_filtered(self):
431
426
# Test multiple files
432
427
delta = tree_a.branch.repository.get_revision_delta('rev1',
433
428
specific_fileids=['foo-id', 'baz-id'])
434
self.assertIsInstance(delta, TreeDelta)
429
self.assertIsInstance(delta, _mod_delta.TreeDelta)
435
430
self.assertEqual([
436
431
('baz', 'baz-id', 'file'),
437
432
('foo', 'foo-id', 'file'),
439
434
# Test a directory
440
435
delta = tree_a.branch.repository.get_revision_delta('rev1',
441
436
specific_fileids=['bar-id'])
442
self.assertIsInstance(delta, TreeDelta)
437
self.assertIsInstance(delta, _mod_delta.TreeDelta)
443
438
self.assertEqual([
444
439
('bar', 'bar-id', 'directory'),
445
440
('bar/b1', 'b1-id', 'file'),
448
443
# Test a file in a directory
449
444
delta = tree_a.branch.repository.get_revision_delta('rev1',
450
445
specific_fileids=['b2-id'])
451
self.assertIsInstance(delta, TreeDelta)
446
self.assertIsInstance(delta, _mod_delta.TreeDelta)
452
447
self.assertEqual([
453
448
('bar', 'bar-id', 'directory'),
454
449
('bar/b2', 'b2-id', 'file'),
456
451
# Try another revision
457
452
delta = tree_a.branch.repository.get_revision_delta('rev2',
458
453
specific_fileids=['b3-id'])
459
self.assertIsInstance(delta, TreeDelta)
454
self.assertIsInstance(delta, _mod_delta.TreeDelta)
460
455
self.assertEqual([
461
456
('bar', 'bar-id', 'directory'),
462
457
('bar/b3', 'b3-id', 'file'),
464
459
delta = tree_a.branch.repository.get_revision_delta('rev2',
465
460
specific_fileids=['foo-id'])
466
self.assertIsInstance(delta, TreeDelta)
461
self.assertIsInstance(delta, _mod_delta.TreeDelta)
467
462
self.assertEqual([], delta.added)
469
464
def test_clone_bzrdir_repository_revision(self):
470
465
# make a repository with some revisions,
471
466
# and clone it, this should not have unreferenced revisions.
472
467
# also: test cloning with a revision id of NULL_REVISION -> empty repo.
473
raise TestSkipped('revision limiting is not implemented yet.')
468
raise tests.TestSkipped('revision limiting is not implemented yet.')
475
470
def test_clone_repository_basis_revision(self):
476
raise TestSkipped('the use of a basis should not add noise data to the result.')
471
raise tests.TestSkipped(
472
'the use of a basis should not add noise data to the result.')
478
474
def test_clone_shared_no_tree(self):
479
475
# cloning a shared repository keeps it shared
514
510
# This gives metadir branches something they can convert to.
515
511
# it would be nice to have a 'latest' vs 'default' concept.
516
512
format = bzrdir.format_registry.make_bzrdir('dirstate-with-subtree')
517
upgrade(repo.bzrdir.root_transport.base, format=format)
513
upgrade.upgrade(repo.bzrdir.root_transport.base, format=format)
518
514
except errors.UpToDateFormat:
519
515
# this is in the most current format already.
521
517
except errors.BadConversionTarget, e:
522
raise TestSkipped(str(e))
523
wt = WorkingTree.open(wt.basedir)
518
raise tests.TestSkipped(str(e))
519
wt = workingtree.WorkingTree.open(wt.basedir)
524
520
new_signature = wt.branch.repository.get_signature_text('A')
525
521
self.assertEqual(old_signature, new_signature)
608
604
return # local conversion to/from RemoteObjects is irrelevant.
609
605
if self.repository_format.get_format_description() \
610
606
== "Repository format 4":
611
raise TestSkipped('Cannot convert format-4 to itself')
607
raise tests.TestSkipped('Cannot convert format-4 to itself')
612
608
self.build_tree_contents(_upgrade_dir_template)
613
old_repodir = bzrlib.bzrdir.BzrDir.open_unsupported('.')
609
old_repodir = bzrdir.BzrDir.open_unsupported('.')
614
610
old_repo_format = old_repodir.open_repository()._format
615
611
format = self.repository_format._matchingbzrdir
617
613
format.repository_format = self.repository_format
618
614
except AttributeError:
616
upgrade.upgrade('.', format)
622
618
def test_pointless_commit(self):
623
619
tree = self.make_branch_and_tree('.')
735
731
tree.commit('initial commit', rev_id='rev1')
736
732
tree.commit('next commit', rev_id='rev2')
737
733
graph = tree.branch.repository.get_graph()
738
parents = graph.get_parent_map([NULL_REVISION, 'rev1', 'rev2'])
734
parents = graph.get_parent_map(
735
[_mod_revision.NULL_REVISION, 'rev1', 'rev2'])
739
736
for value in parents.values():
740
737
self.assertIsInstance(value, tuple)
763
760
self.addCleanup(repo.unlock)
764
761
repo.start_write_group()
765
762
self.addCleanup(repo.abort_write_group)
766
inv = Inventory(revision_id='A')
763
inv = inventory.Inventory(revision_id='A')
767
764
inv.root.revision = 'A'
768
765
repo.add_inventory('A', inv, [])
769
repo.add_revision('A', Revision('A', committer='A', timestamp=0,
770
inventory_sha1='', timezone=0, message='A'))
766
repo.add_revision('A', _mod_revision.Revision(
767
'A', committer='A', timestamp=0,
768
inventory_sha1='', timezone=0, message='A'))
771
769
repo.add_signature_text('A', 'This might be a signature')
772
770
self.assertEqual('This might be a signature',
773
771
repo.get_signature_text('A'))
775
773
def test_add_revision_inventory_sha1(self):
776
inv = Inventory(revision_id='A')
774
inv = inventory.Inventory(revision_id='A')
777
775
inv.root.revision = 'A'
778
776
inv.root.file_id = 'fixed-root'
779
777
# Insert the inventory on its own to an identical repository, to get
791
789
repo.start_write_group()
792
790
root_id = inv.root.file_id
793
791
repo.texts.add_lines(('fixed-root', 'A'), [], [])
794
repo.add_revision('A', Revision('A', committer='B', timestamp=0,
795
timezone=0, message='C'), inv=inv)
792
repo.add_revision('A', _mod_revision.Revision(
793
'A', committer='B', timestamp=0,
794
timezone=0, message='C'), inv=inv)
796
795
repo.commit_write_group()
825
824
"""Make a RemoteRepository object backed by a real repository that will
826
825
be created at the given path."""
827
826
repo = self.make_repository(path, shared=shared)
828
smart_server = server.SmartTCPServer_for_testing()
827
smart_server = test_server.SmartTCPServer_for_testing()
829
828
self.start_server(smart_server, self.get_server())
830
remote_transport = get_transport(smart_server.get_url()).clone(path)
829
remote_transport = transport.get_transport(
830
smart_server.get_url()).clone(path)
831
831
remote_bzrdir = bzrdir.BzrDir.open_from_transport(remote_transport)
832
832
remote_repo = remote_bzrdir.open_repository()
833
833
return remote_repo
835
835
def test_sprout_from_hpss_preserves_format(self):
836
836
"""repo.sprout from a smart server preserves the repository format."""
837
if self.repository_format == RepositoryFormat7():
838
raise TestNotApplicable(
837
if self.repository_format == weaverepo.RepositoryFormat7():
838
raise tests.TestNotApplicable(
839
839
"Cannot fetch weaves over smart protocol.")
840
840
remote_repo = self.make_remote_repository('remote')
841
841
local_bzrdir = self.make_bzrdir('local')
843
843
local_repo = remote_repo.sprout(local_bzrdir)
844
844
except errors.TransportNotPossible:
845
raise TestNotApplicable(
845
raise tests.TestNotApplicable(
846
846
"Cannot lock_read old formats like AllInOne over HPSS.")
847
847
remote_backing_repo = bzrdir.BzrDir.open(
848
848
self.get_vfs_only_url('remote')).open_repository()
851
851
def test_sprout_branch_from_hpss_preserves_repo_format(self):
852
852
"""branch.sprout from a smart server preserves the repository format.
854
weave_formats = [RepositoryFormat5(), RepositoryFormat6(),
854
weave_formats = [weaverepo.RepositoryFormat5(),
855
weaverepo.RepositoryFormat6(),
856
weaverepo.RepositoryFormat7()]
856
857
if self.repository_format in weave_formats:
857
raise TestNotApplicable(
858
raise tests.TestNotApplicable(
858
859
"Cannot fetch weaves over smart protocol.")
859
860
remote_repo = self.make_remote_repository('remote')
860
861
remote_branch = remote_repo.bzrdir.create_branch()
862
863
local_bzrdir = remote_branch.bzrdir.sprout('local')
863
864
except errors.TransportNotPossible:
864
raise TestNotApplicable(
865
raise tests.TestNotApplicable(
865
866
"Cannot lock_read old formats like AllInOne over HPSS.")
866
867
local_repo = local_bzrdir.open_repository()
867
868
remote_backing_repo = bzrdir.BzrDir.open(
872
873
"""branch.sprout from a smart server preserves the repository format of
873
874
a branch from a shared repository.
875
weave_formats = [RepositoryFormat5(), RepositoryFormat6(),
876
weave_formats = [weaverepo.RepositoryFormat5(),
877
weaverepo.RepositoryFormat6(),
878
weaverepo.RepositoryFormat7()]
877
879
if self.repository_format in weave_formats:
878
raise TestNotApplicable(
880
raise tests.TestNotApplicable(
879
881
"Cannot fetch weaves over smart protocol.")
880
882
# Make a shared repo
881
883
remote_repo = self.make_remote_repository('remote', shared=True)
895
897
local_bzrdir = remote_branch.bzrdir.sprout('local')
896
898
except errors.TransportNotPossible:
897
raise TestNotApplicable(
899
raise tests.TestNotApplicable(
898
900
"Cannot lock_read old formats like AllInOne over HPSS.")
899
901
local_repo = local_bzrdir.open_repository()
900
902
self.assertEqual(remote_backing_repo._format, local_repo._format)
902
904
def test_clone_to_hpss(self):
903
pre_metadir_formats = [RepositoryFormat5(), RepositoryFormat6()]
905
pre_metadir_formats = [weaverepo.RepositoryFormat5(),
906
weaverepo.RepositoryFormat6()]
904
907
if self.repository_format in pre_metadir_formats:
905
raise TestNotApplicable(
908
raise tests.TestNotApplicable(
906
909
"Cannot lock pre_metadir_formats remotely.")
907
910
remote_transport = self.make_smart_server('remote')
908
911
local_branch = self.make_branch('local')
920
923
repo = self.make_repository('repo', shared=True)
921
924
except errors.IncompatibleFormat:
922
raise TestNotApplicable('Cannot make a shared repository')
925
raise tests.TestNotApplicable('Cannot make a shared repository')
923
926
if isinstance(repo.bzrdir, bzrdir.BzrDirPreSplitOut):
924
raise KnownFailure("pre metadir branches do not upgrade on push "
927
raise tests.KnownFailure(
928
"pre metadir branches do not upgrade on push "
925
929
"with stacking policy")
926
if isinstance(repo._format, RepositoryFormatKnitPack5RichRootBroken):
927
raise TestNotApplicable("unsupported format")
930
if isinstance(repo._format,
931
pack_repo.RepositoryFormatKnitPack5RichRootBroken):
932
raise tests.TestNotApplicable("unsupported format")
928
933
# Make a source branch in 'repo' in an unstackable branch format
929
934
bzrdir_format = self.repository_format._matchingbzrdir
930
935
transport = self.get_transport('repo/branch')
931
936
transport.mkdir('.')
932
937
target_bzrdir = bzrdir_format.initialize_on_transport(transport)
933
branch = BzrBranchFormat6().initialize(target_bzrdir)
938
branch = _mod_branch.BzrBranchFormat6().initialize(target_bzrdir)
934
939
# Ensure that stack_on will be stackable and match the serializer of
936
941
if isinstance(repo, remote.RemoteRepository):
1012
1017
repo = self.make_repository_and_foo_bar(shared=True)
1013
1018
except errors.IncompatibleFormat:
1014
raise TestNotApplicable
1019
raise tests.TestNotApplicable
1015
1020
branches = repo.find_branches(using=True)
1016
1021
self.assertContainsRe(branches[-1].base, 'repository/foo/$')
1017
1022
# in some formats, creating a repo creates a branch
1046
1051
repo.set_make_working_trees(True)
1047
1052
except errors.RepositoryUpgradeRequired, e:
1048
raise TestNotApplicable('Format does not support this flag.')
1053
raise tests.TestNotApplicable('Format does not support this flag.')
1049
1054
self.assertTrue(repo.make_working_trees())
1051
1056
def test_set_get_make_working_trees_false(self):
1054
1059
repo.set_make_working_trees(False)
1055
1060
except errors.RepositoryUpgradeRequired, e:
1056
raise TestNotApplicable('Format does not support this flag.')
1061
raise tests.TestNotApplicable('Format does not support this flag.')
1057
1062
self.assertFalse(repo.make_working_trees())
1060
class TestRepositoryLocking(TestCaseWithRepository):
1065
class TestRepositoryLocking(per_repository.TestCaseWithRepository):
1062
1067
def test_leave_lock_in_place(self):
1063
1068
repo = self.make_repository('r')
1149
1154
tree_a.commit('rev3', rev_id='rev3', allow_pointless=True)
1150
1155
except errors.RevisionNotPresent:
1151
raise TestNotApplicable("Cannot test with ghosts for this format.")
1156
raise tests.TestNotApplicable(
1157
"Cannot test with ghosts for this format.")
1152
1158
# add another reference to a ghost, and a second ghost.
1153
1159
tree_a.add_parent_tree_id('ghost1')
1154
1160
tree_a.add_parent_tree_id('ghost2')
1217
1223
repo = self.make_repository('inventory_with_unnecessary_ghost')
1218
1224
repo.lock_write()
1219
1225
repo.start_write_group()
1220
inv = Inventory(revision_id = 'ghost')
1226
inv = inventory.Inventory(revision_id = 'ghost')
1221
1227
inv.root.revision = 'ghost'
1222
1228
if repo.supports_rich_root():
1223
1229
root_id = inv.root.file_id
1224
1230
repo.texts.add_lines((root_id, 'ghost'), [], [])
1225
1231
sha1 = repo.add_inventory('ghost', inv, [])
1226
rev = bzrlib.revision.Revision(timestamp=0,
1228
committer="Foo Bar <foo@example.com>",
1230
inventory_sha1=sha1,
1231
revision_id='ghost')
1232
rev = _mod_revision.Revision(
1233
timestamp=0, timezone=None, committer="Foo Bar <foo@example.com>",
1234
message="Message", inventory_sha1=sha1, revision_id='ghost')
1232
1235
rev.parent_ids = ['the_ghost']
1234
1237
repo.add_revision('ghost', rev)
1235
1238
except (errors.NoSuchRevision, errors.RevisionNotPresent):
1236
raise TestNotApplicable("Cannot test with ghosts for this format.")
1239
raise tests.TestNotApplicable(
1240
"Cannot test with ghosts for this format.")
1238
inv = Inventory(revision_id = 'the_ghost')
1242
inv = inventory.Inventory(revision_id = 'the_ghost')
1239
1243
inv.root.revision = 'the_ghost'
1240
1244
if repo.supports_rich_root():
1241
1245
root_id = inv.root.file_id
1242
1246
repo.texts.add_lines((root_id, 'the_ghost'), [], [])
1243
1247
sha1 = repo.add_inventory('the_ghost', inv, [])
1244
rev = bzrlib.revision.Revision(timestamp=0,
1246
committer="Foo Bar <foo@example.com>",
1248
inventory_sha1=sha1,
1249
revision_id='the_ghost')
1248
rev = _mod_revision.Revision(
1249
timestamp=0, timezone=None, committer="Foo Bar <foo@example.com>",
1250
message="Message", inventory_sha1=sha1, revision_id='the_ghost')
1250
1251
rev.parent_ids = []
1251
1252
repo.add_revision('the_ghost', rev)
1252
1253
# check its setup usefully
1280
1281
# FIXME: document why this is a TestCaseWithTransport rather than a
1281
1282
# TestCaseWithRepository
1282
class TestEscaping(TestCaseWithTransport):
1283
class TestEscaping(tests.TestCaseWithTransport):
1283
1284
"""Test that repositories can be stored correctly on VFAT transports.
1285
1286
Makes sure we have proper escaping of invalid characters, etc.
1293
1294
def test_on_vfat(self):
1294
1295
# dont bother with remote repository testing, because this test is
1295
1296
# about local disk layout/support.
1296
from bzrlib.remote import RemoteRepositoryFormat
1297
if isinstance(self.repository_format, RemoteRepositoryFormat):
1297
if isinstance(self.repository_format, remote.RemoteRepositoryFormat):
1299
self.transport_server = FakeVFATServer
1299
self.transport_server = test_server.FakeVFATServer
1300
1300
FOO_ID = 'foo<:>ID'
1301
1301
REV_ID = 'revid-1'
1302
1302
# this makes a default format repository always, which is wrong:
1321
1321
wt.add('file1')
1322
1322
wt.commit('file1', rev_id='rev1')
1323
1323
fileobj = StringIO()
1324
wt.branch.repository.create_bundle('rev1', NULL_REVISION, fileobj)
1324
wt.branch.repository.create_bundle(
1325
'rev1', _mod_revision.NULL_REVISION, fileobj)
added
removed
bzrlib/tests/per_repository/test_repository.py
