← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp.py

Trim duplicate sftp tests.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp.py
26
26
 
27
27
try:
28
28
    import paramiko
29
 
    from stub_sftp import StubServer, StubSFTPServer
30
29
    paramiko_loaded = True
31
30
except ImportError:
32
31
    paramiko_loaded = False
33
32
 
34
 
# XXX: 20051124 jamesh
35
 
# The tests currently pop up a password prompt when an external ssh
36
 
# is used.  This forces the use of the paramiko implementation.
37
 
if paramiko_loaded:
38
 
    import bzrlib.transport.sftp
39
 
    bzrlib.transport.sftp._ssh_vendor = 'none'
40
 
 
41
 
 
42
 
STUB_SERVER_KEY = """
43
 
-----BEGIN RSA PRIVATE KEY-----
44
 
MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
45
 
oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
46
 
d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB
47
 
gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0
48
 
EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon
49
 
soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H
50
 
tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU
51
 
avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA
52
 
4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g
53
 
H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv
54
 
qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV
55
 
HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc
56
 
nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7
57
 
-----END RSA PRIVATE KEY-----
58
 
"""
59
 
    
60
 
 
61
 
class SingleListener (threading.Thread):
62
 
 
63
 
    def __init__(self, callback):
64
 
        threading.Thread.__init__(self)
65
 
        self._callback = callback
66
 
        self._socket = socket.socket()
67
 
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
68
 
        self._socket.bind(('localhost', 0))
69
 
        self._socket.listen(1)
70
 
        self.port = self._socket.getsockname()[1]
71
 
        self.stop_event = threading.Event()
72
 
 
73
 
    def run(self):
74
 
        s, _ = self._socket.accept()
75
 
        # now close the listen socket
76
 
        self._socket.close()
77
 
        self._callback(s, self.stop_event)
78
 
    
79
 
    def stop(self):
80
 
        self.stop_event.set()
81
 
        # We should consider waiting for the other thread
82
 
        # to stop, because otherwise we get spurious
83
 
        #   bzr: ERROR: Socket exception: Connection reset by peer (54)
84
 
        # because the test suite finishes before the thread has a chance
85
 
        # to close. (Especially when only running a few tests)
86
 
        
87
 
        
 
33
 
88
34
class TestCaseWithSFTPServer(TestCaseInTempDir):
89
 
    """
90
 
    Execute a test case with a stub SFTP server, serving files from the local
91
 
    filesystem over the loopback network.
92
 
    """
93
 
 
94
 
    def _run_server(self, s, stop_event):
95
 
        ssh_server = paramiko.Transport(s)
96
 
        key_file = os.path.join(self._root, 'test_rsa.key')
97
 
        file(key_file, 'w').write(STUB_SERVER_KEY)
98
 
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
99
 
        ssh_server.add_server_key(host_key)
100
 
        server = StubServer(self)
101
 
        if self._override_home is None:
102
 
            home = self._root
103
 
        else:
104
 
            home = self._override_home
105
 
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer, 
106
 
                                         StubSFTPServer, root='/',
107
 
                                         home=home)
108
 
        event = threading.Event()
109
 
        ssh_server.start_server(event, server)
110
 
        event.wait(5.0)
111
 
        stop_event.wait(30.0)
 
35
    """A test case base class that provides a sftp server on localhost."""
112
36
 
113
37
    def setUp(self):
114
38
        super(TestCaseWithSFTPServer, self).setUp()
 
39
        from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer
 
40
        if getattr(self, '_get_remote_is_absolute', None) is None:
 
41
            self._get_remote_is_absolute = True
 
42
        if self._get_remote_is_absolute:
 
43
            self.server = SFTPAbsoluteServer()
 
44
        else:
 
45
            self.server = SFTPHomeDirServer()
 
46
        self.server.setUp()
 
47
        self._sftp_url = self.server.get_url()
115
48
        self._root = self.test_dir
116
49
        # Set to a string in setUp to give sftp server a new homedir.
117
50
        self._override_home = None
118
51
        self._is_setup = False
119
 
        self._get_remote_is_absolute = True
120
52
        self.sftplogs = []
121
53
 
122
 
    def delayed_setup(self):
123
 
        # some tests are just stubs that call setUp and then immediately call
124
 
        # tearDwon.  so don't create the port listener until get_transport is
125
 
        # called and we know we're in an actual test.
126
 
        if self._is_setup:
127
 
            return
128
 
        self._listener = SingleListener(self._run_server)
129
 
        self._listener.setDaemon(True)
130
 
        self._listener.start()        
131
 
        self._sftp_url = self._get_sftp_url("%%2f%s" % 
132
 
            bzrlib.transport.urlescape(self._root[1:]))
133
 
        self._is_setup = True
134
 
        
135
 
    def _get_sftp_url(self, path):
136
 
        """Calculate a sftp url to this server for path."""
137
 
        return 'sftp://foo:bar@localhost:%d/%s' % (self._listener.port, path)
138
 
 
139
54
    def get_remote_url(self, relpath_to_test_root):
140
 
        self.delayed_setup()
141
 
        if self._get_remote_is_absolute:
142
 
            return self._get_sftp_url("%%2f%s" % 
143
 
                bzrlib.transport.urlescape(self._root[1:] 
144
 
                                           + '/'
145
 
                                           + relpath_to_test_root))
146
 
        else:
147
 
            return self._get_sftp_url(
148
 
                bzrlib.transport.urlescape(relpath_to_test_root))
 
55
        # FIXME use urljoin ?
 
56
        return self._sftp_url + '/' + relpath_to_test_root
149
57
 
150
58
    def tearDown(self):
151
59
        try:
152
 
            self._listener.stop()
153
 
        except AttributeError:
154
 
            pass
155
 
        TestCaseInTempDir.tearDown(self)
 
60
            self.server.tearDown()
 
61
        finally:
 
62
            TestCaseInTempDir.tearDown(self)
156
63
 
157
64
    def get_transport(self, path=None):
158
65
        """Return a transport relative to self._test_root."""
159
 
        self.delayed_setup()
160
 
        from bzrlib.transport.sftp import SFTPTransport
 
66
        from bzrlib.transport import get_transport
 
67
        transport = get_transport(self._sftp_url)
161
68
        if path is None:
162
 
            url = self._sftp_url
163
 
            if self._get_remote_is_absolute:
164
 
                url = self._sftp_url
165
 
            else:
166
 
                url = self._get_sftp_url('')
 
69
            return transport
167
70
        else:
168
 
            url = self._get_sftp_url(path)
169
 
        return SFTPTransport(url)
170
 
 
171
 
    def log(self, *args):
172
 
        """Override the default log to grab sftp server messages"""
173
 
        TestCaseInTempDir.log(self, *args)
174
 
        if args and args[0].startswith('sftpserver'):
175
 
            self.sftplogs.append(args[0])
176
 
 
177
 
 
178
 
class SFTPTransportTestMixin (TestTransportMixIn):
179
 
    readonly = False
 
71
            return transport.clone(path)
 
72
 
 
73
 
 
74
class SFTPLockTests (TestCaseWithSFTPServer):
180
75
 
181
76
    def test_sftp_locks(self):
182
77
        from bzrlib.errors import LockError
205
100
 
206
101
    def test_multiple_connections(self):
207
102
        t = self.get_transport()
208
 
        self.assertEquals(self.sftplogs, 
 
103
        self.assertEquals(self.server.logs, 
209
104
                ['sftpserver - authorizing: foo'
210
105
               , 'sftpserver - channel request: session, 1'])
211
 
        self.sftplogs = []
 
106
        self.server.logs = []
212
107
        # The second request should reuse the first connection
213
108
        # SingleListener only allows for a single connection,
214
109
        # So the next line fails unless the connection is reused
215
110
        t2 = self.get_transport()
216
 
        self.assertEquals(self.sftplogs, [])
217
 
 
218
 
 
219
 
class SFTPTransportTestAbsolute(TestCaseWithSFTPServer, SFTPTransportTestMixin):
220
 
    """Test the SFTP transport with %2f based abs paths."""
 
111
        self.assertEquals(self.server.logs, [])
221
112
 
222
113
 
223
114
class SFTPTransportTestAbsoluteSibling(TestCaseWithSFTPServer, 
224
 
                                       SFTPTransportTestMixin):
 
115
                                       TestTransportMixIn):
225
116
    """Test the SFTP transport with %2f based sibling paths to $HOME."""
226
117
 
227
118
    def setUp(self):
 
119
        self._override_home = '/dev/noone/runs/tests/here'
228
120
        super(SFTPTransportTestAbsoluteSibling, self).setUp()
229
 
        self._override_home = '/dev/noone/runs/tests/here'
230
 
 
231
 
 
232
 
class SFTPTransportTestRelative(TestCaseWithSFTPServer, SFTPTransportTestMixin):
 
121
 
 
122
 
 
123
class SFTPTransportTestRelative(TestCaseWithSFTPServer):
233
124
    """Test the SFTP transport with homedir based relative paths."""
234
125
 
235
 
    def setUp(self):
236
 
        super(SFTPTransportTestRelative, self).setUp()
237
 
        self._get_remote_is_absolute = False
238
 
 
239
 
 
240
 
class SFTPTransportRemotePathTest(TestCaseWithSFTPServer):
241
 
    
242
126
    def test__remote_path(self):
243
127
        t = self.get_transport()
244
128
        # try what is currently used:
252
136
        # /  should be illegal ?
253
137
        ### FIXME decide and then test for all transports. RBC20051208
254
138
 
 
139
 
 
140
class SFTPTransportTestRelative(TestCaseWithSFTPServer):
 
141
    """Test the SFTP transport with homedir based relative paths."""
 
142
 
 
143
    def setUp(self):
 
144
        self._get_remote_is_absolute = False
 
145
        super(SFTPTransportTestRelative, self).setUp()
 
146
 
255
147
    def test__remote_path_relative_root(self):
256
148
        # relative paths are preserved
257
149
        t = self.get_transport('')
337
229
 
338
230
    def test_lock_file(self):
339
231
        """Make sure that a Branch accessed over sftp tries to lock itself."""
340
 
        self.delayed_setup()
341
232
        b = Branch.initialize(self._sftp_url)
342
233
        self.failUnlessExists('.bzr/')
343
234
        self.failUnlessExists('.bzr/branch-format')
350
241
        self.failIf(os.path.lexists('.bzr/branch-lock.write-lock'))
351
242
 
352
243
    def test_no_working_tree(self):
353
 
        self.delayed_setup()
354
244
        b = Branch.initialize(self._sftp_url)
355
245
        self.assertRaises(errors.NoWorkingTree, b.working_tree)
356
246
 
357
247
    def test_push_support(self):
358
 
        self.delayed_setup()
359
 
 
360
248
        self.build_tree(['a/', 'a/foo'])
361
249
        b = Branch.initialize('a')
362
250
        t = b.working_tree()