← 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: 2006-02-22 04:29:54 UTC
  • mfrom: (1566 +trunk)
  • mto: This revision was merged to the branch mainline in revision 1569.
  • Revision ID: mbp@sourcefrog.net-20060222042954-60333f08dd56a646
[merge] from bzr.dev before integration
Fix undefined ordering in sign_my_revisions breaking tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
18
18
import socket
19
19
import threading
20
20
 
21
 
from bzrlib.branch import Branch
 
21
import bzrlib.bzrdir as bzrdir
22
22
import bzrlib.errors as errors
23
23
from bzrlib.osutils import pathjoin, lexists
24
 
from bzrlib.tests import TestCaseInTempDir, TestCase, TestSkipped
 
24
from bzrlib.tests import TestCaseWithTransport, TestCase, TestSkipped
25
25
import bzrlib.transport
26
26
from bzrlib.workingtree import WorkingTree
27
27
 
32
32
    paramiko_loaded = False
33
33
 
34
34
 
35
 
class TestCaseWithSFTPServer(TestCaseInTempDir):
 
35
class TestCaseWithSFTPServer(TestCaseWithTransport):
36
36
    """A test case base class that provides a sftp server on localhost."""
37
37
 
38
38
    def setUp(self):
39
39
        if not paramiko_loaded:
40
40
            raise TestSkipped('you must have paramiko to run this test')
41
41
        super(TestCaseWithSFTPServer, self).setUp()
42
 
        from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer, SFTPServer
43
 
        if getattr(self, '_full_handshake', False):
44
 
            self.server = SFTPServer()
45
 
        else:
46
 
            self._full_handshake = False
47
 
            if getattr(self, '_get_remote_is_absolute', None) is None:
48
 
                self._get_remote_is_absolute = True
49
 
            if self._get_remote_is_absolute:
50
 
                self.server = SFTPAbsoluteServer()
51
 
            else:
52
 
                self.server = SFTPHomeDirServer()
53
 
        self.server.setUp()
54
 
        self.addCleanup(self.server.tearDown)
55
 
        if self._full_handshake:
56
 
            self._sftp_url = self.server._get_sftp_url("")
57
 
        else:
58
 
            self._sftp_url = self.server.get_url()
59
 
        self._root = self.test_dir
60
 
        # Set to a string in setUp to give sftp server a new homedir.
61
 
        self._override_home = None
62
 
        self._is_setup = False
63
 
        self.sftplogs = []
64
 
 
65
 
    def get_remote_url(self, relpath_to_test_root):
66
 
        # FIXME use urljoin ?
67
 
        return self._sftp_url + '/' + relpath_to_test_root
 
42
        from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer
 
43
        if getattr(self, '_get_remote_is_absolute', None) is None:
 
44
            self._get_remote_is_absolute = True
 
45
        if self._get_remote_is_absolute:
 
46
            self.transport_server = SFTPAbsoluteServer
 
47
        else:
 
48
            self.transport_server = SFTPHomeDirServer
 
49
        self.transport_readonly_server = bzrlib.transport.http.HttpServer
68
50
 
69
51
    def get_transport(self, path=None):
70
52
        """Return a transport relative to self._test_root."""
71
 
        from bzrlib.transport import get_transport
72
 
        transport = get_transport(self._sftp_url)
73
 
        if path is None:
74
 
            return transport
75
 
        else:
76
 
            return transport.clone(path)
 
53
        return bzrlib.transport.get_transport(self.get_url(path))
77
54
 
78
55
 
79
56
class SFTPLockTests (TestCaseWithSFTPServer):
105
82
 
106
83
    def test_multiple_connections(self):
107
84
        t = self.get_transport()
108
 
        self.assertTrue('sftpserver - new connection' in self.server.logs)
109
 
        self.server.logs = []
 
85
        self.assertTrue('sftpserver - new connection' in self.get_server().logs)
 
86
        self.get_server().logs = []
110
87
        # The second request should reuse the first connection
111
88
        # SingleListener only allows for a single connection,
112
89
        # So the next line fails unless the connection is reused
113
90
        t2 = self.get_transport()
114
 
        self.assertEquals(self.server.logs, [])
 
91
        self.assertEquals(self.get_server().logs, [])
115
92
 
116
93
 
117
94
class SFTPTransportTestRelative(TestCaseWithSFTPServer):
121
98
        t = self.get_transport()
122
99
        # try what is currently used:
123
100
        # remote path = self._abspath(relpath)
124
 
        self.assertEqual(self._root + '/relative', t._remote_path('relative'))
 
101
        self.assertEqual(self.test_dir + '/relative', t._remote_path('relative'))
125
102
        # we dont os.path.join because windows gives us the wrong path
126
 
        root_segments = self._root.split('/')
 
103
        root_segments = self.test_dir.split('/')
127
104
        root_parent = '/'.join(root_segments[:-1])
128
105
        # .. should be honoured
129
106
        self.assertEqual(root_parent + '/sibling', t._remote_path('../sibling'))
217
194
 
218
195
    def test_lock_file(self):
219
196
        """Make sure that a Branch accessed over sftp tries to lock itself."""
220
 
        b = Branch.create(self._sftp_url)
 
197
        b = bzrdir.BzrDir.create_branch_and_repo(self.get_url())
221
198
        self.failUnlessExists('.bzr/')
222
199
        self.failUnlessExists('.bzr/branch-format')
223
200
        self.failUnlessExists('.bzr/branch-lock')
228
205
        b.unlock()
229
206
        self.failIf(lexists('.bzr/branch-lock.write-lock'))
230
207
 
231
 
    def test_no_working_tree(self):
232
 
        b = Branch.create(self._sftp_url)
233
 
        self.assertRaises(errors.NoWorkingTree, b.working_tree)
234
 
 
235
208
    def test_push_support(self):
236
209
        self.build_tree(['a/', 'a/foo'])
237
 
        t = WorkingTree.create_standalone('a')
 
210
        t = bzrdir.BzrDir.create_standalone_workingtree('a')
238
211
        b = t.branch
239
212
        t.add('foo')
240
213
        t.commit('foo', rev_id='a1')
241
214
 
242
 
        os.mkdir('b')
243
 
        b2 = Branch.create(self._sftp_url + '/b')
 
215
        b2 = bzrdir.BzrDir.create_branch_and_repo(self.get_url('/b'))
244
216
        b2.pull(b)
245
217
 
246
218
        self.assertEquals(b2.revision_history(), ['a1'])
254
226
 
255
227
class SFTPFullHandshakingTest(TestCaseWithSFTPServer):
256
228
    """Verify that a full-handshake (SSH over loopback TCP) sftp connection works."""
257
 
    _full_handshake = True
258
229
    
259
230
    def test_connection(self):
 
231
        from bzrlib.transport.sftp import SFTPFullAbsoluteServer
 
232
        self.transport_server = SFTPFullAbsoluteServer
260
233
        self.get_transport()