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