171
172
"""Lookalike for _SmartClient allowing testing."""
173
174
def __init__(self, fake_medium_base='fake base'):
174
"""Create a FakeClient.
176
:param responses: A list of response-tuple, body-data pairs to be sent
177
back to callers. A special case is if the response-tuple is
178
'unknown verb', then a UnknownSmartMethod will be raised for that
179
call, using the second element of the tuple as the verb in the
175
"""Create a FakeClient."""
182
176
self.responses = []
184
178
self.expecting_body = False
1016
class TestTransportMkdir(tests.TestCase):
1018
def test_permissiondenied(self):
1019
client = FakeClient()
1020
client.add_error_response('PermissionDenied', 'remote path', 'extra')
1021
transport = RemoteTransport('bzr://example.com/', medium=False,
1023
exc = self.assertRaises(
1024
errors.PermissionDenied, transport.mkdir, 'client path')
1025
expected_error = errors.PermissionDenied('/client path', 'extra')
1026
self.assertEqual(expected_error, exc)
1022
1029
class TestRemoteSSHTransportAuthentication(tests.TestCaseInTempDir):
1024
1031
def test_defaults_to_none(self):
1123
1130
class TestRepositoryGetGraph(TestRemoteRepository):
1125
1132
def test_get_graph(self):
1126
# get_graph returns a graph with the repository as the
1133
# get_graph returns a graph with a custom parents provider.
1128
1134
transport_path = 'quack'
1129
1135
repo, client = self.setup_fake_client_and_repository(transport_path)
1130
1136
graph = repo.get_graph()
1131
self.assertEqual(graph._parents_provider, repo)
1137
self.assertNotEqual(graph._parents_provider, repo)
1134
1140
class TestRepositoryGetParentMap(TestRemoteRepository):
1450
1456
src_repo.copy_content_into(dest_repo)
1459
class _StubRealPackRepository(object):
1461
def __init__(self, calls):
1462
self._pack_collection = _StubPackCollection(calls)
1465
class _StubPackCollection(object):
1467
def __init__(self, calls):
1471
self.calls.append(('pack collection autopack',))
1473
def reload_pack_names(self):
1474
self.calls.append(('pack collection reload_pack_names',))
1477
class TestRemotePackRepositoryAutoPack(TestRemoteRepository):
1478
"""Tests for RemoteRepository.autopack implementation."""
1481
"""When the server returns 'ok' and there's no _real_repository, then
1482
nothing else happens: the autopack method is done.
1484
transport_path = 'quack'
1485
repo, client = self.setup_fake_client_and_repository(transport_path)
1486
client.add_expected_call(
1487
'PackRepository.autopack', ('quack/',), 'success', ('ok',))
1489
client.finished_test()
1491
def test_ok_with_real_repo(self):
1492
"""When the server returns 'ok' and there is a _real_repository, then
1493
the _real_repository's reload_pack_name's method will be called.
1495
transport_path = 'quack'
1496
repo, client = self.setup_fake_client_and_repository(transport_path)
1497
client.add_expected_call(
1498
'PackRepository.autopack', ('quack/',),
1500
repo._real_repository = _StubRealPackRepository(client._calls)
1503
[('call', 'PackRepository.autopack', ('quack/',)),
1504
('pack collection reload_pack_names',)],
1507
def test_backwards_compatibility(self):
1508
"""If the server does not recognise the PackRepository.autopack verb,
1509
fallback to the real_repository's implementation.
1511
transport_path = 'quack'
1512
repo, client = self.setup_fake_client_and_repository(transport_path)
1513
client.add_unknown_method_response('PackRepository.autopack')
1514
def stub_ensure_real():
1515
client._calls.append(('_ensure_real',))
1516
repo._real_repository = _StubRealPackRepository(client._calls)
1517
repo._ensure_real = stub_ensure_real
1520
[('call', 'PackRepository.autopack', ('quack/',)),
1522
('pack collection autopack',)],
1453
1526
class TestErrorTranslationBase(tests.TestCaseWithMemoryTransport):
1454
1527
"""Base class for unit tests for bzrlib.remote._translate_error."""
1551
1624
expected_error = errors.DivergedBranches(branch, other_branch)
1552
1625
self.assertEqual(expected_error, translated_error)
1627
def test_ReadError_no_args(self):
1629
translated_error = self.translateTuple(('ReadError',), path=path)
1630
expected_error = errors.ReadError(path)
1631
self.assertEqual(expected_error, translated_error)
1633
def test_ReadError(self):
1635
translated_error = self.translateTuple(('ReadError', path))
1636
expected_error = errors.ReadError(path)
1637
self.assertEqual(expected_error, translated_error)
1639
def test_PermissionDenied_no_args(self):
1641
translated_error = self.translateTuple(('PermissionDenied',), path=path)
1642
expected_error = errors.PermissionDenied(path)
1643
self.assertEqual(expected_error, translated_error)
1645
def test_PermissionDenied_one_arg(self):
1647
translated_error = self.translateTuple(('PermissionDenied', path))
1648
expected_error = errors.PermissionDenied(path)
1649
self.assertEqual(expected_error, translated_error)
1651
def test_PermissionDenied_one_arg_and_context(self):
1652
"""Given a choice between a path from the local context and a path on
1653
the wire, _translate_error prefers the path from the local context.
1655
local_path = 'local path'
1656
remote_path = 'remote path'
1657
translated_error = self.translateTuple(
1658
('PermissionDenied', remote_path), path=local_path)
1659
expected_error = errors.PermissionDenied(local_path)
1660
self.assertEqual(expected_error, translated_error)
1662
def test_PermissionDenied_two_args(self):
1664
extra = 'a string with extra info'
1665
translated_error = self.translateTuple(
1666
('PermissionDenied', path, extra))
1667
expected_error = errors.PermissionDenied(path, extra)
1668
self.assertEqual(expected_error, translated_error)
1555
1671
class TestErrorTranslationRobustness(TestErrorTranslationBase):
1556
1672
"""Unit tests for bzrlib.remote._translate_error's robustness.
1589
1705
self._get_log(keep_log_file=True),
1590
1706
"Missing key 'branch' in context")
1708
def test_path_missing(self):
1709
"""Some translations (PermissionDenied, ReadError) can determine the
1710
'path' variable from either the wire or the local context. If neither
1711
has it, then an error is raised.
1713
error_tuple = ('ReadError',)
1714
server_error = errors.ErrorFromSmartServer(error_tuple)
1715
translated_error = self.translateErrorFromSmartServer(server_error)
1716
self.assertEqual(server_error, translated_error)
1717
# In addition to re-raising ErrorFromSmartServer, some debug info has
1718
# been muttered to the log file for developer to look at.
1719
self.assertContainsRe(
1720
self._get_log(keep_log_file=True), "Missing key 'path' in context")
1593
1723
class TestStacking(tests.TestCaseWithTransport):
1594
1724
"""Tests for operations on stacked remote repositories.
1635
1765
remote_repo.unlock()
1767
def prepare_stacked_remote_branch(self):
1768
smart_server = server.SmartTCPServer_for_testing()
1769
smart_server.setUp()
1770
self.addCleanup(smart_server.tearDown)
1771
tree1 = self.make_branch_and_tree('tree1')
1772
tree1.commit('rev1', rev_id='rev1')
1773
tree2 = self.make_branch_and_tree('tree2', format='1.6')
1774
tree2.branch.set_stacked_on_url(tree1.branch.base)
1775
branch2 = Branch.open(smart_server.get_url() + '/tree2')
1777
self.addCleanup(branch2.unlock)
1780
def test_stacked_get_parent_map(self):
1781
# the public implementation of get_parent_map obeys stacking
1782
branch = self.prepare_stacked_remote_branch()
1783
repo = branch.repository
1784
self.assertEqual(['rev1'], repo.get_parent_map(['rev1']).keys())
1786
def test_stacked__get_parent_map(self):
1787
# the private variant of _get_parent_map ignores stacking
1788
branch = self.prepare_stacked_remote_branch()
1789
repo = branch.repository
1790
self.assertEqual([], repo._get_parent_map(['rev1']).keys())
1793
class TestRemoteBranchEffort(tests.TestCaseWithTransport):
1796
super(TestRemoteBranchEffort, self).setUp()
1797
# Create a smart server that publishes whatever the backing VFS server
1799
self.smart_server = server.SmartTCPServer_for_testing()
1800
self.smart_server.setUp(self.get_server())
1801
self.addCleanup(self.smart_server.tearDown)
1802
# Log all HPSS calls into self.hpss_calls.
1803
_SmartClient.hooks.install_named_hook(
1804
'call', self.capture_hpss_call, None)
1805
self.hpss_calls = []
1807
def capture_hpss_call(self, params):
1808
self.hpss_calls.append(params.method)
1810
def test_copy_content_into_avoids_revision_history(self):
1811
local = self.make_branch('local')
1812
remote_backing_tree = self.make_branch_and_tree('remote')
1813
remote_backing_tree.commit("Commit.")
1814
remote_branch_url = self.smart_server.get_url() + 'remote'
1815
remote_branch = bzrdir.BzrDir.open(remote_branch_url).open_branch()
1816
local.repository.fetch(remote_branch.repository)
1817
self.hpss_calls = []
1818
remote_branch.copy_content_into(local)
1819
self.assertFalse('Branch.revision_history' in self.hpss_calls)
added
removed
bzrlib/tests/test_remote.py
