← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp_transport.py

  • Committer: Martin Pool
  • Date: 2010-02-25 06:17:27 UTC
  • mfrom: (5055 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5057.
  • Revision ID: mbp@sourcefrog.net-20100225061727-4sd9lt0qmdc6087t
merge news

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
1
 
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
1
# Copyright (C) 2005-2010 Robey Pointer <robey@lag.net>
2
2
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
3
3
#
4
4
# This program is free software; you can redistribute it and/or modify
46
46
 
47
47
if features.paramiko.available():
48
48
    from bzrlib.transport import sftp as _mod_sftp
49
 
    from bzrlib.transport.sftp import (
50
 
        SFTPAbsoluteServer,
51
 
        SFTPHomeDirServer,
52
 
        SFTPTransport,
53
 
        )
 
49
    from bzrlib.tests import stub_sftp
54
50
 
55
51
from bzrlib.workingtree import WorkingTree
56
52
 
60
56
    if getattr(testcase, '_get_remote_is_absolute', None) is None:
61
57
        testcase._get_remote_is_absolute = True
62
58
    if testcase._get_remote_is_absolute:
63
 
        testcase.transport_server = SFTPAbsoluteServer
 
59
        testcase.transport_server = stub_sftp.SFTPAbsoluteServer
64
60
    else:
65
 
        testcase.transport_server = SFTPHomeDirServer
 
61
        testcase.transport_server = stub_sftp.SFTPHomeDirServer
66
62
    testcase.transport_readonly_server = HttpServer
67
63
 
68
64
 
162
158
        self.requireFeature(features.paramiko)
163
159
 
164
160
    def test_parse_url_with_home_dir(self):
165
 
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative')
 
161
        s = _mod_sftp.SFTPTransport(
 
162
            'sftp://ro%62ey:h%40t@example.com:2222/~/relative')
166
163
        self.assertEquals(s._host, 'example.com')
167
164
        self.assertEquals(s._port, 2222)
168
165
        self.assertEquals(s._user, 'robey')
170
167
        self.assertEquals(s._path, '/~/relative/')
171
168
 
172
169
    def test_relpath(self):
173
 
        s = SFTPTransport('sftp://user@host.com/abs/path')
 
170
        s = _mod_sftp.SFTPTransport('sftp://user@host.com/abs/path')
174
171
        self.assertRaises(errors.PathNotChild, s.relpath,
175
172
                          'sftp://user@host.com/~/rel/path/sub')
176
173
 
190
187
            ssh._ssh_vendor_manager._cached_ssh_vendor = orig_vendor
191
188
 
192
189
    def test_abspath_root_sibling_server(self):
193
 
        from bzrlib.transport.sftp import SFTPSiblingAbsoluteServer
194
 
        server = SFTPSiblingAbsoluteServer()
 
190
        server = stub_sftp.SFTPSiblingAbsoluteServer()
195
191
        server.start_server()
196
192
        try:
197
193
            transport = get_transport(server.get_url())
255
251
 
256
252
    def setUp(self):
257
253
        super(SSHVendorConnection, self).setUp()
258
 
        from bzrlib.transport.sftp import SFTPFullAbsoluteServer
259
254
 
260
255
        def create_server():
261
256
            """Just a wrapper so that when created, it will set _vendor"""
262
257
            # SFTPFullAbsoluteServer can handle any vendor,
263
258
            # it just needs to be set between the time it is instantiated
264
259
            # and the time .setUp() is called
265
 
            server = SFTPFullAbsoluteServer()
 
260
            server = stub_sftp.SFTPFullAbsoluteServer()
266
261
            server._vendor = self._test_vendor
267
262
            return server
268
263
        self._test_vendor = 'loopback'
409
404
        self.requireFeature(features.paramiko)
410
405
 
411
406
    def test_delay(self):
412
 
        from bzrlib.transport.sftp import SocketDelay
413
407
        sending = FakeSocket()
414
 
        receiving = SocketDelay(sending, 0.1, bandwidth=1000000,
415
 
                                really_sleep=False)
 
408
        receiving = stub_sftp.SocketDelay(sending, 0.1, bandwidth=1000000,
 
409
                                          really_sleep=False)
416
410
        # check that simulated time is charged only per round-trip:
417
 
        t1 = SocketDelay.simulated_time
 
411
        t1 = stub_sftp.SocketDelay.simulated_time
418
412
        receiving.send("connect1")
419
413
        self.assertEqual(sending.recv(1024), "connect1")
420
 
        t2 = SocketDelay.simulated_time
 
414
        t2 = stub_sftp.SocketDelay.simulated_time
421
415
        self.assertAlmostEqual(t2 - t1, 0.1)
422
416
        receiving.send("connect2")
423
417
        self.assertEqual(sending.recv(1024), "connect2")
424
418
        sending.send("hello")
425
419
        self.assertEqual(receiving.recv(1024), "hello")
426
 
        t3 = SocketDelay.simulated_time
 
420
        t3 = stub_sftp.SocketDelay.simulated_time
427
421
        self.assertAlmostEqual(t3 - t2, 0.1)
428
422
        sending.send("hello")
429
423
        self.assertEqual(receiving.recv(1024), "hello")
431
425
        self.assertEqual(receiving.recv(1024), "hello")
432
426
        sending.send("hello")
433
427
        self.assertEqual(receiving.recv(1024), "hello")
434
 
        t4 = SocketDelay.simulated_time
 
428
        t4 = stub_sftp.SocketDelay.simulated_time
435
429
        self.assertAlmostEqual(t4, t3)
436
430
 
437
431
    def test_bandwidth(self):
438
 
        from bzrlib.transport.sftp import SocketDelay
439
432
        sending = FakeSocket()
440
 
        receiving = SocketDelay(sending, 0, bandwidth=8.0/(1024*1024),
441
 
                                really_sleep=False)
 
433
        receiving = stub_sftp.SocketDelay(sending, 0, bandwidth=8.0/(1024*1024),
 
434
                                          really_sleep=False)
442
435
        # check that simulated time is charged only per round-trip:
443
 
        t1 = SocketDelay.simulated_time
 
436
        t1 = stub_sftp.SocketDelay.simulated_time
444
437
        receiving.send("connect")
445
438
        self.assertEqual(sending.recv(1024), "connect")
446
439
        sending.send("a" * 100)
447
440
        self.assertEqual(receiving.recv(1024), "a" * 100)
448
 
        t2 = SocketDelay.simulated_time
 
441
        t2 = stub_sftp.SocketDelay.simulated_time
449
442
        self.assertAlmostEqual(t2 - t1, 100 + 7)
450
443
 
451
444