1
# Copyright (C) 2006-2014, 2016 Canonical Ltd
1
# Copyright (C) 2006-2016 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
503
503
def test_ssh_client_repr(self):
504
504
client_medium = medium.SmartSSHClientMedium(
505
505
'base', medium.SSHParams("example.com", "4242", "username"))
507
507
"SmartSSHClientMedium(bzr+ssh://username@example.com:4242/)",
508
508
repr(client_medium))
510
510
def test_ssh_client_repr_no_port(self):
511
511
client_medium = medium.SmartSSHClientMedium(
512
512
'base', medium.SSHParams("example.com", None, "username"))
514
514
"SmartSSHClientMedium(bzr+ssh://username@example.com/)",
515
515
repr(client_medium))
517
517
def test_ssh_client_repr_no_username(self):
518
518
client_medium = medium.SmartSSHClientMedium(
519
519
'base', medium.SSHParams("example.com", None, None))
521
521
"SmartSSHClientMedium(bzr+ssh://example.com/)",
522
522
repr(client_medium))
773
773
class RemoteTransportTests(test_smart.TestCaseWithSmartMedium):
775
775
def test_plausible_url(self):
776
self.assert_(self.get_url().startswith('bzr://'))
776
self.assertTrue(self.get_url().startswith('bzr://'))
778
778
def test_probe_transport(self):
779
779
t = self.get_transport()
1595
1595
conn2.get_smart_medium())
1597
1597
def test__remote_path(self):
1598
self.assertEquals('/foo/bar',
1598
self.assertEqual('/foo/bar',
1599
1599
self.transport._remote_path('foo/bar'))
1601
1601
def test_clone_changes_base(self):
1602
1602
"""Cloning transport produces one with a new base location"""
1603
1603
conn2 = self.transport.clone('subdir')
1604
self.assertEquals(self.transport.base + 'subdir/',
1604
self.assertEqual(self.transport.base + 'subdir/',
1607
1607
def test_open_dir(self):
1610
1610
transport = self.transport
1611
1611
self.backing_transport.mkdir('toffee')
1612
1612
self.backing_transport.mkdir('toffee/apple')
1613
self.assertEquals('/toffee', transport._remote_path('toffee'))
1613
self.assertEqual('/toffee', transport._remote_path('toffee'))
1614
1614
toffee_trans = transport.clone('toffee')
1615
1615
# Check that each transport has only the contents of its directory
1616
1616
# directly visible. If state was being held in the wrong object, it's
3199
3199
requester, output = self.make_client_encoder_and_output()
3200
3200
requester.set_headers({'header name': 'header value'})
3201
3201
requester.call('one arg')
3203
3203
'bzr message 3 (bzr 1.6)\n' # protocol version
3204
3204
'\x00\x00\x00\x1fd11:header name12:header valuee' # headers
3205
3205
's\x00\x00\x00\x0bl7:one arge' # args
3215
3215
requester, output = self.make_client_encoder_and_output()
3216
3216
requester.set_headers({'header name': 'header value'})
3217
3217
requester.call_with_body_bytes(('one arg',), 'body bytes')
3219
3219
'bzr message 3 (bzr 1.6)\n' # protocol version
3220
3220
'\x00\x00\x00\x1fd11:header name12:header valuee' # headers
3221
3221
's\x00\x00\x00\x0bl7:one arge' # args
3250
3250
requester.set_headers({'header name': 'header value'})
3251
3251
stream = ['chunk 1', 'chunk two']
3252
3252
requester.call_with_body_stream(('one arg',), stream)
3254
3254
'bzr message 3 (bzr 1.6)\n' # protocol version
3255
3255
'\x00\x00\x00\x1fd11:header name12:header valuee' # headers
3256
3256
's\x00\x00\x00\x0bl7:one arge' # args
3265
3265
requester.set_headers({})
3267
3267
requester.call_with_body_stream(('one arg',), stream)
3269
3269
'bzr message 3 (bzr 1.6)\n' # protocol version
3270
3270
'\x00\x00\x00\x02de' # headers
3271
3271
's\x00\x00\x00\x0bl7:one arge' # args
3287
3287
raise Exception('Boom!')
3288
3288
self.assertRaises(Exception, requester.call_with_body_stream,
3289
3289
('one arg',), stream_that_fails())
3291
3291
'bzr message 3 (bzr 1.6)\n' # protocol version
3292
3292
'\x00\x00\x00\x02de' # headers
3293
3293
's\x00\x00\x00\x0bl7:one arge' # args
4285
4285
t = remote.RemoteHTTPTransport('bzr+http://www.example.com/foo')
4286
4286
r = t._redirected_to('http://www.example.com/foo',
4287
4287
'http://www.example.com/bar')
4288
self.assertEquals(type(r), type(t))
4288
self.assertEqual(type(r), type(t))
4290
4290
def test__redirect_sibling_protocol(self):
4291
4291
t = remote.RemoteHTTPTransport('bzr+http://www.example.com/foo')
4292
4292
r = t._redirected_to('http://www.example.com/foo',
4293
4293
'https://www.example.com/bar')
4294
self.assertEquals(type(r), type(t))
4294
self.assertEqual(type(r), type(t))
4295
4295
self.assertStartsWith(r.base, 'bzr+https')
4297
4297
def test__redirect_to_with_user(self):
4298
4298
t = remote.RemoteHTTPTransport('bzr+http://joe@www.example.com/foo')
4299
4299
r = t._redirected_to('http://www.example.com/foo',
4300
4300
'http://www.example.com/bar')
4301
self.assertEquals(type(r), type(t))
4302
self.assertEquals('joe', t._parsed_url.user)
4303
self.assertEquals(t._parsed_url.user, r._parsed_url.user)
4301
self.assertEqual(type(r), type(t))
4302
self.assertEqual('joe', t._parsed_url.user)
4303
self.assertEqual(t._parsed_url.user, r._parsed_url.user)
4305
4305
def test_redirected_to_same_host_different_protocol(self):
4306
4306
t = remote.RemoteHTTPTransport('bzr+http://joe@www.example.com/foo')
4307
4307
r = t._redirected_to('http://www.example.com/foo',
4308
4308
'ftp://www.example.com/foo')
4309
self.assertNotEquals(type(r), type(t))
4309
self.assertNotEqual(type(r), type(t))