← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_transport.py

  • Committer: John Arbash Meinel
  • Date: 2010-02-10 17:52:08 UTC
  • mfrom: (5021 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5023.
  • Revision ID: john@arbash-meinel.com-20100210175208-bubuwav4uqigu291
Merge bzr.dev 5021 to resolve NEWS

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
1
 
# Copyright (C) 2006, 2007, 2008, 2009 Canonical Ltd
 
1
# Copyright (C) 2006-2010 Canonical Ltd
2
2
#
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
28
28
        errors,
29
29
        osutils,
30
30
        tests,
 
31
        transport,
31
32
        urlutils,
32
33
        )
33
34
from bzrlib.smart import (
39
40
        server,
40
41
        vfs,
41
42
)
42
 
from bzrlib.tests.test_smart import TestCaseWithSmartMedium
 
43
from bzrlib.tests import test_smart
43
44
from bzrlib.transport import (
44
 
        get_transport,
 
45
        http,
45
46
        local,
46
47
        memory,
47
48
        remote,
48
49
        )
49
 
from bzrlib.transport.http import SmartClientHTTPMediumRequest
50
50
 
51
51
 
52
52
class StringIOSSHVendor(object):
557
557
        self.assertRaises(errors.ReadingCompleted, request.read_bytes, None)
558
558
 
559
559
 
560
 
class RemoteTransportTests(TestCaseWithSmartMedium):
 
560
class RemoteTransportTests(test_smart.TestCaseWithSmartMedium):
561
561
 
562
562
    def test_plausible_url(self):
563
563
        self.assert_(self.get_url().startswith('bzr://'))
1003
1003
            mem_server.start_server()
1004
1004
            self.addCleanup(mem_server.stop_server)
1005
1005
            self.permit_url(mem_server.get_url())
1006
 
            self.backing_transport = get_transport(mem_server.get_url())
 
1006
            self.backing_transport = transport.get_transport(
 
1007
                mem_server.get_url())
1007
1008
        else:
1008
1009
            self.backing_transport = backing_transport
1009
1010
        if readonly:
1010
1011
            self.real_backing_transport = self.backing_transport
1011
 
            self.backing_transport = get_transport("readonly+" + self.backing_transport.abspath('.'))
 
1012
            self.backing_transport = transport.get_transport(
 
1013
                "readonly+" + self.backing_transport.abspath('.'))
1012
1014
        self.server = server.SmartTCPServer(self.backing_transport)
1013
1015
        self.server.start_background_thread('-' + self.id())
1014
1016
        self.transport = remote.RemoteTCPTransport(self.server.get_url())
1161
1163
        self.hook_calls = []
1162
1164
        server.SmartTCPServer.hooks.install_named_hook('server_started',
1163
1165
            self.capture_server_call, None)
1164
 
        self.start_server(backing_transport=get_transport("."))
 
1166
        self.start_server(backing_transport=transport.get_transport("."))
1165
1167
        # at this point, the server will be starting a thread up.
1166
1168
        # there is no indicator at the moment, so bodge it by doing a request.
1167
1169
        self.transport.has('.')
1194
1196
        self.hook_calls = []
1195
1197
        server.SmartTCPServer.hooks.install_named_hook('server_stopped',
1196
1198
            self.capture_server_call, None)
1197
 
        self.start_server(backing_transport=get_transport("."))
 
1199
        self.start_server(backing_transport=transport.get_transport("."))
1198
1200
        result = [(
1199
1201
            [self.backing_transport.base, self.backing_transport.external_url()]
1200
1202
            , self.transport.base)]
1339
1341
class RemoteTransportRegistration(tests.TestCase):
1340
1342
 
1341
1343
    def test_registration(self):
1342
 
        t = get_transport('bzr+ssh://example.com/path')
 
1344
        t = transport.get_transport('bzr+ssh://example.com/path')
1343
1345
        self.assertIsInstance(t, remote.RemoteSSHTransport)
1344
1346
        self.assertEqual('example.com', t._host)
1345
1347
 
1346
1348
    def test_bzr_https(self):
1347
1349
        # https://bugs.launchpad.net/bzr/+bug/128456
1348
 
        t = get_transport('bzr+https://example.com/path')
 
1350
        t = transport.get_transport('bzr+https://example.com/path')
1349
1351
        self.assertIsInstance(t, remote.RemoteHTTPTransport)
1350
1352
        self.assertStartsWith(
1351
1353
            t._http_transport.base,
3516
3518
 
3517
3519
    def test_smart_http_medium_request_accept_bytes(self):
3518
3520
        medium = FakeHTTPMedium()
3519
 
        request = SmartClientHTTPMediumRequest(medium)
 
3521
        request = http.SmartClientHTTPMediumRequest(medium)
3520
3522
        request.accept_bytes('abc')
3521
3523
        request.accept_bytes('def')
3522
3524
        self.assertEqual(None, medium.written_request)