← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testsftp.py

Prepare for 0.6rc1

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testsftp.py
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>, Canonical Ltd
 
2
 
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
 
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
 
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
import os
 
18
import socket
 
19
import threading
 
20
import unittest
 
21
 
 
22
from bzrlib.selftest import TestCaseInTempDir
 
23
from bzrlib.selftest.testtransport import TestTransportMixIn
 
24
 
 
25
try:
 
26
    import paramiko
 
27
    from stub_sftp import StubServer, StubSFTPServer
 
28
    paramiko_loaded = True
 
29
except ImportError:
 
30
    paramiko_loaded = False
 
31
 
 
32
 
 
33
STUB_SERVER_KEY = """
 
34
-----BEGIN RSA PRIVATE KEY-----
 
35
MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
 
36
oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
 
37
d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB
 
38
gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0
 
39
EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon
 
40
soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H
 
41
tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU
 
42
avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA
 
43
4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g
 
44
H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv
 
45
qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV
 
46
HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc
 
47
nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7
 
48
-----END RSA PRIVATE KEY-----
 
49
"""
 
50
    
 
51
 
 
52
class SingleListener (threading.Thread):
 
53
    def __init__(self, callback):
 
54
        threading.Thread.__init__(self)
 
55
        self._callback = callback
 
56
        self._socket = socket.socket()
 
57
        self._socket.listen(1)
 
58
        self.port = self._socket.getsockname()[1]
 
59
        self.stop_event = threading.Event()
 
60
 
 
61
    def run(self):
 
62
        s, _ = self._socket.accept()
 
63
        # now close the listen socket
 
64
        self._socket.close()
 
65
        self._callback(s, self.stop_event)
 
66
    
 
67
    def stop(self):
 
68
        self.stop_event.set()
 
69
        
 
70
        
 
71
class TestCaseWithSFTPServer (TestCaseInTempDir):
 
72
    """
 
73
    Execute a test case with a stub SFTP server, serving files from the local
 
74
    filesystem over the loopback network.
 
75
    """
 
76
    
 
77
    def _run_server(self, s, stop_event):
 
78
        ssh_server = paramiko.Transport(s)
 
79
        key_file = os.path.join(self._root, 'test_rsa.key')
 
80
        file(key_file, 'w').write(STUB_SERVER_KEY)
 
81
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
 
82
        ssh_server.add_server_key(host_key)
 
83
        server = StubServer()
 
84
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer, StubSFTPServer, root=self._root)
 
85
        event = threading.Event()
 
86
        ssh_server.start_server(event, server)
 
87
        event.wait(5.0)
 
88
        stop_event.wait(30.0)
 
89
 
 
90
    def setUp(self):
 
91
        TestCaseInTempDir.setUp(self)
 
92
        self._root = self.test_dir
 
93
 
 
94
        self._listener = SingleListener(self._run_server)
 
95
        self._listener.setDaemon(True)
 
96
        self._listener.start()        
 
97
        self._sftp_url = 'sftp://foo:bar@localhost:%d/' % (self._listener.port,)
 
98
        
 
99
    def tearDown(self):
 
100
        self._listener.stop()
 
101
        TestCaseInTempDir.tearDown(self)
 
102
 
 
103
        
 
104
class SFTPTransportTest (TestCaseWithSFTPServer, TestTransportMixIn):
 
105
    readonly = False
 
106
 
 
107
    def get_transport(self):
 
108
        from bzrlib.transport.sftp import SFTPTransport
 
109
        url = self._sftp_url
 
110
        return SFTPTransport(url)
 
111
 
 
112
if not paramiko_loaded:
 
113
    del SFTPTransportTest