← 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: Robert Collins
  • Date: 2007-07-04 08:08:13 UTC
  • mfrom: (2572 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2587.
  • Revision ID: robertc@robertcollins.net-20070704080813-wzebx0r88fvwj5rq
Merge bzr.dev.

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>, Canonical Ltd
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
2
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
2
3
#
3
4
# This program is free software; you can redistribute it and/or modify
4
5
# it under the terms of the GNU General Public License as published by
16
17
 
17
18
import os
18
19
import socket
 
20
import sys
19
21
import threading
20
22
import time
21
23
 
23
25
import bzrlib.errors as errors
24
26
from bzrlib.osutils import pathjoin, lexists, set_or_unset_env
25
27
from bzrlib.tests import TestCaseWithTransport, TestCase, TestSkipped
 
28
from bzrlib.tests.HttpServer import HttpServer
26
29
import bzrlib.transport
27
30
from bzrlib.transport import get_transport
28
31
import bzrlib.transport.http
44
47
        testcase.transport_server = SFTPAbsoluteServer
45
48
    else:
46
49
        testcase.transport_server = SFTPHomeDirServer
47
 
    testcase.transport_readonly_server = bzrlib.transport.http.HttpServer
 
50
    testcase.transport_readonly_server = HttpServer
48
51
 
49
52
 
50
53
class TestCaseWithSFTPServer(TestCaseWithTransport):
54
57
        super(TestCaseWithSFTPServer, self).setUp()
55
58
        if not paramiko_loaded:
56
59
            raise TestSkipped('you must have paramiko to run this test')
57
 
        set_test_transport_to_sftp(self) 
 
60
        set_test_transport_to_sftp(self)
58
61
 
59
62
    def get_transport(self, path=None):
60
63
        """Return a transport relative to self._test_root."""
104
107
 
105
108
    def test__remote_path(self):
106
109
        t = self.get_transport()
 
110
        # This test require unix-like absolute path
 
111
        test_dir = self.test_dir
 
112
        if sys.platform == 'win32':
 
113
            # using hack suggested by John Meinel.
 
114
            # TODO: write another mock server for this test
 
115
            #       and use absolute path without drive letter
 
116
            test_dir = '/' + test_dir
107
117
        # try what is currently used:
108
118
        # remote path = self._abspath(relpath)
109
 
        self.assertEqual(self.test_dir + '/relative', t._remote_path('relative'))
 
119
        self.assertEqual(test_dir + '/relative', t._remote_path('relative'))
110
120
        # we dont os.path.join because windows gives us the wrong path
111
 
        root_segments = self.test_dir.split('/')
 
121
        root_segments = test_dir.split('/')
112
122
        root_parent = '/'.join(root_segments[:-1])
113
123
        # .. should be honoured
114
124
        self.assertEqual(root_parent + '/sibling', t._remote_path('../sibling'))
204
214
        """Test that if no 'ssh' is available we get builtin paramiko"""
205
215
        from bzrlib.transport import ssh
206
216
        # set '.' as the only location in the path, forcing no 'ssh' to exist
207
 
        orig_vendor = ssh._ssh_vendor
 
217
        orig_vendor = ssh._ssh_vendor_manager._cached_ssh_vendor
208
218
        orig_path = set_or_unset_env('PATH', '.')
209
219
        try:
210
220
            # No vendor defined yet, query for one
211
 
            ssh._ssh_vendor = None
 
221
            ssh._ssh_vendor_manager.clear_cache()
212
222
            vendor = ssh._get_ssh_vendor()
213
223
            self.assertIsInstance(vendor, ssh.ParamikoVendor)
214
224
        finally:
215
225
            set_or_unset_env('PATH', orig_path)
216
 
            ssh._ssh_vendor = orig_vendor
 
226
            ssh._ssh_vendor_manager._cached_ssh_vendor = orig_vendor
217
227
 
218
228
    def test_abspath_root_sibling_server(self):
219
229
        from bzrlib.transport.sftp import SFTPSiblingAbsoluteServer
292
302
            server._vendor = self._test_vendor
293
303
            return server
294
304
        self._test_vendor = 'loopback'
295
 
        self.transport_server = create_server
 
305
        self.vfs_transport_server = create_server
296
306
        f = open('a_file', 'wb')
297
307
        try:
298
308
            f.write('foobar\n')
337
347
        s.bind(('localhost', 0))
338
348
        self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
339
349
 
340
 
        orig_vendor = bzrlib.transport.ssh._ssh_vendor
 
350
        orig_vendor = bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor
341
351
        def reset():
342
 
            bzrlib.transport.ssh._ssh_vendor = orig_vendor
 
352
            bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor = orig_vendor
343
353
            s.close()
344
354
        self.addCleanup(reset)
345
355
 
346
356
    def set_vendor(self, vendor):
347
357
        import bzrlib.transport.ssh
348
 
        bzrlib.transport.ssh._ssh_vendor = vendor
 
358
        bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor = vendor
349
359
 
350
360
    def test_bad_connection_paramiko(self):
351
361
        """Test that a real connection attempt raises the right error"""
380
390
            # paramiko, but we know about it, so we don't have to
381
391
            # fail the test
382
392
            raise TestSkipped('Known NameError bug with paramiko-1.6.1')
383
 
        self.assertContainsRe(err, 'Connection error')
 
393
        self.assertContainsRe(err, r'bzr: ERROR: Unable to connect to SSH host'
 
394
                                   r' 127\.0\.0\.1:\d+; ')
384
395
 
385
396
 
386
397
class SFTPLatencyKnob(TestCaseWithSFTPServer):