69
69
from bzrlib.smart.repository import (
70
70
SmartServerRepositoryGetParentMap,
71
71
SmartServerRepositoryGetStream_1_19,
72
_stream_to_byte_stream,
73
74
from bzrlib.symbol_versioning import deprecated_in
74
75
from bzrlib.tests import (
174
175
def test_remote_branch_set_append_revisions_only(self):
175
176
# Make a format 1.9 branch, which supports append_revisions_only
176
177
branch = self.make_branch('branch', format='1.9')
177
config = branch.get_config()
178
178
branch.set_append_revisions_only(True)
179
config = branch.get_config_stack()
179
180
self.assertEqual(
180
'True', config.get_user_option('append_revisions_only'))
181
True, config.get('append_revisions_only'))
181
182
branch.set_append_revisions_only(False)
183
config = branch.get_config_stack()
182
184
self.assertEqual(
183
'False', config.get_user_option('append_revisions_only'))
185
False, config.get('append_revisions_only'))
185
187
def test_remote_branch_set_append_revisions_only_upgrade_reqd(self):
186
188
branch = self.make_branch('branch', format='knit')
187
config = branch.get_config()
188
189
self.assertRaises(
189
190
errors.UpgradeRequired, branch.set_append_revisions_only, True)
503
504
self.assertRaises(errors.UnknownFormatError, a_bzrdir.cloning_metadir)
507
class TestBzrDirCheckoutMetaDir(TestRemote):
509
def test__get_checkout_format(self):
510
transport = MemoryTransport()
511
client = FakeClient(transport.base)
512
reference_bzrdir_format = bzrdir.format_registry.get('default')()
513
control_name = reference_bzrdir_format.network_name()
514
client.add_expected_call(
515
'BzrDir.checkout_metadir', ('quack/', ),
516
'success', (control_name, '', ''))
517
transport.mkdir('quack')
518
transport = transport.clone('quack')
519
a_bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
521
result = a_bzrdir.checkout_metadir()
522
# We should have got a reference control dir with default branch and
523
# repository formats.
524
self.assertEqual(bzrdir.BzrDirMetaFormat1, type(result))
525
self.assertEqual(None, result._repository_format)
526
self.assertEqual(None, result._branch_format)
527
self.assertFinished(client)
529
def test_unknown_format(self):
530
transport = MemoryTransport()
531
client = FakeClient(transport.base)
532
client.add_expected_call(
533
'BzrDir.checkout_metadir', ('quack/',),
534
'success', ('dontknow', '', ''))
535
transport.mkdir('quack')
536
transport = transport.clone('quack')
537
a_bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
539
self.assertRaises(errors.UnknownFormatError,
540
a_bzrdir.checkout_metadir)
541
self.assertFinished(client)
506
544
class TestBzrDirDestroyBranch(TestRemote):
508
546
def test_destroy_default(self):
4196
4234
'Repository.unlock', ('quack/', 'token', 'False'),
4197
4235
'success', ('ok', ))
4198
4236
repo.pack(['hinta', 'hintb'])
4239
class TestRepositoryIterInventories(TestRemoteRepository):
4240
"""Test Repository.iter_inventories."""
4242
def _serialize_inv_delta(self, old_name, new_name, delta):
4243
serializer = inventory_delta.InventoryDeltaSerializer(True, False)
4244
return "".join(serializer.delta_to_lines(old_name, new_name, delta))
4246
def test_single_empty(self):
4247
transport_path = 'quack'
4248
repo, client = self.setup_fake_client_and_repository(transport_path)
4249
fmt = bzrdir.format_registry.get('2a')().repository_format
4251
stream = [('inventory-deltas', [
4252
versionedfile.FulltextContentFactory('somerevid', None, None,
4253
self._serialize_inv_delta('null:', 'somerevid', []))])]
4254
client.add_expected_call(
4255
'VersionedFileRepository.get_inventories', ('quack/', 'unordered'),
4256
'success', ('ok', ),
4257
_stream_to_byte_stream(stream, fmt))
4258
ret = list(repo.iter_inventories(["somerevid"]))
4259
self.assertLength(1, ret)
4261
self.assertEquals("somerevid", inv.revision_id)
4263
def test_empty(self):
4264
transport_path = 'quack'
4265
repo, client = self.setup_fake_client_and_repository(transport_path)
4266
ret = list(repo.iter_inventories([]))
4267
self.assertEquals(ret, [])
4269
def test_missing(self):
4270
transport_path = 'quack'
4271
repo, client = self.setup_fake_client_and_repository(transport_path)
4272
client.add_expected_call(
4273
'VersionedFileRepository.get_inventories', ('quack/', 'unordered'),
4274
'success', ('ok', ), iter([]))
4275
self.assertRaises(errors.NoSuchRevision, list, repo.iter_inventories(