209
206
self.assertIsInstance(formats[0], SampleExtraRepositoryFormat)
212
class TestFormat6(TestCaseWithTransport):
214
def test_attribute__fetch_order(self):
215
"""Weaves need topological data insertion."""
216
control = bzrdir.BzrDirFormat6().initialize(self.get_url())
217
repo = weaverepo.RepositoryFormat6().initialize(control)
218
self.assertEqual('topological', repo._format._fetch_order)
220
def test_attribute__fetch_uses_deltas(self):
221
"""Weaves do not reuse deltas."""
222
control = bzrdir.BzrDirFormat6().initialize(self.get_url())
223
repo = weaverepo.RepositoryFormat6().initialize(control)
224
self.assertEqual(False, repo._format._fetch_uses_deltas)
226
def test_attribute__fetch_reconcile(self):
227
"""Weave repositories need a reconcile after fetch."""
228
control = bzrdir.BzrDirFormat6().initialize(self.get_url())
229
repo = weaverepo.RepositoryFormat6().initialize(control)
230
self.assertEqual(True, repo._format._fetch_reconcile)
232
def test_no_ancestry_weave(self):
233
control = bzrdir.BzrDirFormat6().initialize(self.get_url())
234
repo = weaverepo.RepositoryFormat6().initialize(control)
235
# We no longer need to create the ancestry.weave file
236
# since it is *never* used.
237
self.assertRaises(NoSuchFile,
238
control.transport.get,
241
def test_supports_external_lookups(self):
242
control = bzrdir.BzrDirFormat6().initialize(self.get_url())
243
repo = weaverepo.RepositoryFormat6().initialize(control)
244
self.assertFalse(repo._format.supports_external_lookups)
247
class TestFormat7(TestCaseWithTransport):
249
def test_attribute__fetch_order(self):
250
"""Weaves need topological data insertion."""
251
control = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
252
repo = weaverepo.RepositoryFormat7().initialize(control)
253
self.assertEqual('topological', repo._format._fetch_order)
255
def test_attribute__fetch_uses_deltas(self):
256
"""Weaves do not reuse deltas."""
257
control = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
258
repo = weaverepo.RepositoryFormat7().initialize(control)
259
self.assertEqual(False, repo._format._fetch_uses_deltas)
261
def test_attribute__fetch_reconcile(self):
262
"""Weave repositories need a reconcile after fetch."""
263
control = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
264
repo = weaverepo.RepositoryFormat7().initialize(control)
265
self.assertEqual(True, repo._format._fetch_reconcile)
267
def test_disk_layout(self):
268
control = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
269
repo = weaverepo.RepositoryFormat7().initialize(control)
270
# in case of side effects of locking.
274
# format 'Bazaar-NG Repository format 7'
276
# inventory.weave == empty_weave
277
# empty revision-store directory
278
# empty weaves directory
279
t = control.get_repository_transport(None)
280
self.assertEqualDiff('Bazaar-NG Repository format 7',
281
t.get('format').read())
282
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
283
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
284
self.assertEqualDiff('# bzr weave file v5\n'
287
t.get('inventory.weave').read())
288
# Creating a file with id Foo:Bar results in a non-escaped file name on
290
control.create_branch()
291
tree = control.create_workingtree()
292
tree.add(['foo'], ['Foo:Bar'], ['file'])
293
tree.put_file_bytes_non_atomic('Foo:Bar', 'content\n')
295
tree.commit('first post', rev_id='first')
296
except errors.IllegalPath:
297
if sys.platform != 'win32':
299
self.knownFailure('Foo:Bar cannot be used as a file-id on windows'
302
self.assertEqualDiff(
303
'# bzr weave file v5\n'
305
'1 7fe70820e08a1aac0ef224d9c66ab66831cc4ab1\n'
313
t.get('weaves/74/Foo%3ABar.weave').read())
315
def test_shared_disk_layout(self):
316
control = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
317
repo = weaverepo.RepositoryFormat7().initialize(control, shared=True)
319
# format 'Bazaar-NG Repository format 7'
320
# inventory.weave == empty_weave
321
# empty revision-store directory
322
# empty weaves directory
323
# a 'shared-storage' marker file.
324
# lock is not present when unlocked
325
t = control.get_repository_transport(None)
326
self.assertEqualDiff('Bazaar-NG Repository format 7',
327
t.get('format').read())
328
self.assertEqualDiff('', t.get('shared-storage').read())
329
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
330
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
331
self.assertEqualDiff('# bzr weave file v5\n'
334
t.get('inventory.weave').read())
335
self.assertFalse(t.has('branch-lock'))
337
def test_creates_lockdir(self):
338
"""Make sure it appears to be controlled by a LockDir existence"""
339
control = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
340
repo = weaverepo.RepositoryFormat7().initialize(control, shared=True)
341
t = control.get_repository_transport(None)
342
# TODO: Should check there is a 'lock' toplevel directory,
343
# regardless of contents
344
self.assertFalse(t.has('lock/held/info'))
347
self.assertTrue(t.has('lock/held/info'))
349
# unlock so we don't get a warning about failing to do so
352
def test_uses_lockdir(self):
353
"""repo format 7 actually locks on lockdir"""
354
base_url = self.get_url()
355
control = bzrdir.BzrDirMetaFormat1().initialize(base_url)
356
repo = weaverepo.RepositoryFormat7().initialize(control, shared=True)
357
t = control.get_repository_transport(None)
361
# make sure the same lock is created by opening it
362
repo = repository.Repository.open(base_url)
364
self.assertTrue(t.has('lock/held/info'))
366
self.assertFalse(t.has('lock/held/info'))
368
def test_shared_no_tree_disk_layout(self):
369
control = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
370
repo = weaverepo.RepositoryFormat7().initialize(control, shared=True)
371
repo.set_make_working_trees(False)
373
# format 'Bazaar-NG Repository format 7'
375
# inventory.weave == empty_weave
376
# empty revision-store directory
377
# empty weaves directory
378
# a 'shared-storage' marker file.
379
t = control.get_repository_transport(None)
380
self.assertEqualDiff('Bazaar-NG Repository format 7',
381
t.get('format').read())
382
## self.assertEqualDiff('', t.get('lock').read())
383
self.assertEqualDiff('', t.get('shared-storage').read())
384
self.assertEqualDiff('', t.get('no-working-trees').read())
385
repo.set_make_working_trees(True)
386
self.assertFalse(t.has('no-working-trees'))
387
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
388
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
389
self.assertEqualDiff('# bzr weave file v5\n'
392
t.get('inventory.weave').read())
394
def test_supports_external_lookups(self):
395
control = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
396
repo = weaverepo.RepositoryFormat7().initialize(control)
397
self.assertFalse(repo._format.supports_external_lookups)
400
209
class TestFormatKnit1(TestCaseWithTransport):
402
211
def test_attribute__fetch_order(self):
619
428
self.assertGetsDefaultInterRepository(dummy_a, dummy_b)
622
class TestInterWeaveRepo(TestCaseWithTransport):
624
def test_is_compatible_and_registered(self):
625
# InterWeaveRepo is compatible when either side
626
# is a format 5/6/7 branch
627
from bzrlib.repofmt import knitrepo, weaverepo
628
formats = [weaverepo.RepositoryFormat5(),
629
weaverepo.RepositoryFormat6(),
630
weaverepo.RepositoryFormat7()]
631
incompatible_formats = [weaverepo.RepositoryFormat4(),
632
knitrepo.RepositoryFormatKnit1(),
634
repo_a = self.make_repository('a')
635
repo_b = self.make_repository('b')
636
is_compatible = weaverepo.InterWeaveRepo.is_compatible
637
for source in incompatible_formats:
638
# force incompatible left then right
639
repo_a._format = source
640
repo_b._format = formats[0]
641
self.assertFalse(is_compatible(repo_a, repo_b))
642
self.assertFalse(is_compatible(repo_b, repo_a))
643
for source in formats:
644
repo_a._format = source
645
for target in formats:
646
repo_b._format = target
647
self.assertTrue(is_compatible(repo_a, repo_b))
648
self.assertEqual(weaverepo.InterWeaveRepo,
649
repository.InterRepository.get(repo_a,
653
431
class TestRepositoryFormat1(knitrepo.RepositoryFormatKnit1):
655
433
def get_format_string(self):
1696
1474
self.assertTrue(new_pack.signature_index._optimize_for_size)
1477
class TestGCCHKPacker(TestCaseWithTransport):
1479
def make_abc_branch(self):
1480
builder = self.make_branch_builder('source')
1481
builder.start_series()
1482
builder.build_snapshot('A', None, [
1483
('add', ('', 'root-id', 'directory', None)),
1484
('add', ('file', 'file-id', 'file', 'content\n')),
1486
builder.build_snapshot('B', ['A'], [
1487
('add', ('dir', 'dir-id', 'directory', None))])
1488
builder.build_snapshot('C', ['B'], [
1489
('modify', ('file-id', 'new content\n'))])
1490
builder.finish_series()
1491
return builder.get_branch()
1493
def make_branch_with_disjoint_inventory_and_revision(self):
1494
"""a repo with separate packs for a revisions Revision and Inventory.
1496
There will be one pack file that holds the Revision content, and one
1497
for the Inventory content.
1499
:return: (repository,
1500
pack_name_with_rev_A_Revision,
1501
pack_name_with_rev_A_Inventory,
1502
pack_name_with_rev_C_content)
1504
b_source = self.make_abc_branch()
1505
b_base = b_source.bzrdir.sprout('base', revision_id='A').open_branch()
1506
b_stacked = b_base.bzrdir.sprout('stacked', stacked=True).open_branch()
1507
b_stacked.lock_write()
1508
self.addCleanup(b_stacked.unlock)
1509
b_stacked.fetch(b_source, 'B')
1510
# Now re-open the stacked repo directly (no fallbacks) so that we can
1511
# fill in the A rev.
1512
repo_not_stacked = b_stacked.bzrdir.open_repository()
1513
repo_not_stacked.lock_write()
1514
self.addCleanup(repo_not_stacked.unlock)
1515
# Now we should have a pack file with A's inventory, but not its
1517
self.assertEqual([('A',), ('B',)],
1518
sorted(repo_not_stacked.inventories.keys()))
1519
self.assertEqual([('B',)],
1520
sorted(repo_not_stacked.revisions.keys()))
1521
stacked_pack_names = repo_not_stacked._pack_collection.names()
1522
# We have a couple names here, figure out which has A's inventory
1523
for name in stacked_pack_names:
1524
pack = repo_not_stacked._pack_collection.get_pack_by_name(name)
1525
keys = [n[1] for n in pack.inventory_index.iter_all_entries()]
1527
inv_a_pack_name = name
1530
self.fail('Could not find pack containing A\'s inventory')
1531
repo_not_stacked.fetch(b_source.repository, 'A')
1532
self.assertEqual([('A',), ('B',)],
1533
sorted(repo_not_stacked.revisions.keys()))
1534
new_pack_names = set(repo_not_stacked._pack_collection.names())
1535
rev_a_pack_names = new_pack_names.difference(stacked_pack_names)
1536
self.assertEqual(1, len(rev_a_pack_names))
1537
rev_a_pack_name = list(rev_a_pack_names)[0]
1538
# Now fetch 'C', so we have a couple pack files to join
1539
repo_not_stacked.fetch(b_source.repository, 'C')
1540
rev_c_pack_names = set(repo_not_stacked._pack_collection.names())
1541
rev_c_pack_names = rev_c_pack_names.difference(new_pack_names)
1542
self.assertEqual(1, len(rev_c_pack_names))
1543
rev_c_pack_name = list(rev_c_pack_names)[0]
1544
return (repo_not_stacked, rev_a_pack_name, inv_a_pack_name,
1547
def test_pack_with_distant_inventories(self):
1548
# See https://bugs.launchpad.net/bzr/+bug/437003
1549
# When repacking, it is possible to have an inventory in a different
1550
# pack file than the associated revision. An autopack can then come
1551
# along, and miss that inventory, and complain.
1552
(repo, rev_a_pack_name, inv_a_pack_name, rev_c_pack_name
1553
) = self.make_branch_with_disjoint_inventory_and_revision()
1554
a_pack = repo._pack_collection.get_pack_by_name(rev_a_pack_name)
1555
c_pack = repo._pack_collection.get_pack_by_name(rev_c_pack_name)
1556
packer = groupcompress_repo.GCCHKPacker(repo._pack_collection,
1557
[a_pack, c_pack], '.test-pack')
1558
# This would raise ValueError in bug #437003, but should not raise an
1562
def test_pack_with_missing_inventory(self):
1563
# Similar to test_pack_with_missing_inventory, but this time, we force
1564
# the A inventory to actually be gone from the repository.
1565
(repo, rev_a_pack_name, inv_a_pack_name, rev_c_pack_name
1566
) = self.make_branch_with_disjoint_inventory_and_revision()
1567
inv_a_pack = repo._pack_collection.get_pack_by_name(inv_a_pack_name)
1568
repo._pack_collection._remove_pack_from_memory(inv_a_pack)
1569
packer = groupcompress_repo.GCCHKPacker(repo._pack_collection,
1570
repo._pack_collection.all_packs(), '.test-pack')
1571
e = self.assertRaises(ValueError, packer.pack)
1572
packer.new_pack.abort()
1573
self.assertContainsRe(str(e),
1574
r"We are missing inventories for revisions: .*'A'")
1699
1577
class TestCrossFormatPacks(TestCaseWithTransport):
1701
1579
def log_pack(self, hint=None):
added
removed
bzrlib/tests/test_repository.py
