535
536
def test_probe_transport(self):
536
537
t = self.get_transport()
537
self.assertIsInstance(t, remote.SmartTransport)
538
self.assertIsInstance(t, remote.RemoteTransport)
539
540
def test_get_medium_from_transport(self):
540
541
"""Remote transport has a medium always, which it can return."""
793
794
smart_server = server.SmartTCPServer(backing_transport=FlakyTransport())
794
795
smart_server.start_background_thread()
796
transport = remote.SmartTCPTransport(smart_server.get_url())
797
transport = remote.RemoteTCPTransport(smart_server.get_url())
798
799
transport.get('something')
799
800
except errors.TransportError, e:
824
825
self.backing_transport = get_transport("readonly+" + self.backing_transport.abspath('.'))
825
826
self.server = server.SmartTCPServer(self.backing_transport)
826
827
self.server.start_background_thread()
827
self.transport = remote.SmartTCPTransport(self.server.get_url())
828
self.transport = remote.RemoteTCPTransport(self.server.get_url())
828
829
self.addCleanup(self.tearDownServer)
830
831
def tearDownServer(self):
854
855
self.tearDownServer()
855
856
# if the listening socket has closed, we should get a BADFD error
856
857
# when connecting, rather than a hang.
857
transport = remote.SmartTCPTransport(server.get_url())
858
transport = remote.RemoteTCPTransport(server.get_url())
858
859
self.assertRaises(errors.ConnectionError, transport.has, '.')
1112
1113
def test_registration(self):
1113
1114
t = get_transport('bzr+ssh://example.com/path')
1114
self.assertIsInstance(t, remote.SmartSSHTransport)
1115
self.assertIsInstance(t, remote.RemoteSSHTransport)
1115
1116
self.assertEqual('example.com', t._host)
1122
1123
input = StringIO("ok\n3\nbardone\n")
1123
1124
output = StringIO()
1124
1125
client_medium = medium.SmartSimplePipesClientMedium(input, output)
1125
transport = remote.SmartTransport(
1126
transport = remote.RemoteTransport(
1126
1127
'bzr://localhost/', medium=client_medium)
1128
1129
# We want to make sure the client is used when the first remote
1142
1143
def test__translate_error_readonly(self):
1143
1144
"""Sending a ReadOnlyError to _translate_error raises TransportNotPossible."""
1144
1145
client_medium = medium.SmartClientMedium()
1145
transport = remote.SmartTransport(
1146
transport = remote.RemoteTransport(
1146
1147
'bzr://localhost/', medium=client_medium)
1147
1148
self.assertRaises(errors.TransportNotPossible,
1148
1149
transport._translate_error, ("ReadOnlyError", ))
1451
1452
errors.ReadingCompleted, smart_protocol.read_body_bytes)
1455
class TestSmartClientUnicode(tests.TestCase):
1456
"""_SmartClient tests for unicode arguments.
1458
Unicode arguments to call_with_body_bytes are not correct (remote method
1459
names, arguments, and bodies must all be expressed as byte strings), but
1460
_SmartClient should gracefully reject them, rather than getting into a
1461
broken state that prevents future correct calls from working. That is, it
1462
should be possible to issue more requests on the medium afterwards, rather
1463
than allowing one bad call to call_with_body_bytes to cause later calls to
1464
mysteriously fail with TooManyConcurrentRequests.
1467
def assertCallDoesNotBreakMedium(self, method, args, body):
1468
"""Call a medium with the given method, args and body, then assert that
1469
the medium is left in a sane state, i.e. is capable of allowing further
1472
input = StringIO("\n")
1474
client_medium = medium.SmartSimplePipesClientMedium(input, output)
1475
smart_client = client._SmartClient(client_medium)
1476
self.assertRaises(TypeError,
1477
smart_client.call_with_body_bytes, method, args, body)
1478
self.assertEqual("", output.getvalue())
1479
self.assertEqual(None, client_medium._current_request)
1481
def test_call_with_body_bytes_unicode_method(self):
1482
self.assertCallDoesNotBreakMedium(u'method', ('args',), 'body')
1484
def test_call_with_body_bytes_unicode_args(self):
1485
self.assertCallDoesNotBreakMedium('method', (u'args',), 'body')
1486
self.assertCallDoesNotBreakMedium('method', ('arg1', u'arg2'), 'body')
1488
def test_call_with_body_bytes_unicode_body(self):
1489
self.assertCallDoesNotBreakMedium('method', ('args',), u'body')
1454
1492
class LengthPrefixedBodyDecoder(tests.TestCase):
1456
1494
# XXX: TODO: make accept_reading_trailer invoke translate_response or
1544
1582
http_transport = self.get_readonly_transport()
1545
1583
medium = http_transport.get_smart_medium()
1546
1584
#remote_transport = RemoteTransport('fake_url', medium)
1547
remote_transport = remote.SmartTransport('/', medium=medium)
1585
remote_transport = remote.RemoteTransport('/', medium=medium)
1548
1586
self.assertEqual(
1549
1587
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
added
removed
bzrlib/tests/test_smart_transport.py
