1
# Copyright (C) 2007 Canonical Ltd
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.
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.
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
19
Based on medusa: http://www.amk.ca/python/code/medusa.html
31
import medusa.ftp_server
40
class test_filesystem(medusa.filesys.os_filesystem):
41
"""A custom filesystem wrapper to add missing functionalities."""
43
def chmod(self, path, mode):
44
p = self.normalize(self.path_module.join (self.wd, path))
45
return os.chmod(self.translate(p), mode)
48
class test_authorizer(object):
49
"""A custom Authorizer object for running the test suite.
51
The reason we cannot use dummy_authorizer, is because it sets the
52
channel to readonly, which we don't always want to do.
55
def __init__(self, root):
57
# If secured_user is set secured_password will be checked
58
self.secured_user = None
59
self.secured_password = None
61
def authorize(self, channel, username, password):
62
"""Return (success, reply_string, filesystem)"""
63
channel.persona = -1, -1
64
if username == 'anonymous':
69
# Check secured_user if set
70
if (self.secured_user is not None
71
and username == self.secured_user
72
and password != self.secured_password):
73
return 0, 'Password invalid.', None
75
return 1, 'OK.', test_filesystem(self.root)
78
class ftp_channel(medusa.ftp_server.ftp_channel):
79
"""Customized ftp channel"""
81
def log(self, message):
82
"""Redirect logging requests."""
83
trace.mutter('ftp_channel: %s', message)
85
def log_info(self, message, type='info'):
86
"""Redirect logging requests."""
87
trace.mutter('ftp_channel %s: %s', type, message)
89
def cmd_rnfr(self, line):
90
"""Prepare for renaming a file."""
91
self._renaming = line[1]
92
self.respond('350 Ready for RNTO')
93
# TODO: jam 20060516 in testing, the ftp server seems to
94
# check that the file already exists, or it sends
95
# 550 RNFR command failed
97
def cmd_rnto(self, line):
98
"""Rename a file based on the target given.
100
rnto must be called after calling rnfr.
102
if not self._renaming:
103
self.respond('503 RNFR required first.')
104
pfrom = self.filesystem.translate(self._renaming)
105
self._renaming = None
106
pto = self.filesystem.translate(line[1])
107
if os.path.exists(pto):
108
self.respond('550 RNTO failed: file exists')
111
os.rename(pfrom, pto)
112
except (IOError, OSError), e:
113
# TODO: jam 20060516 return custom responses based on
114
# why the command failed
115
# (bialix 20070418) str(e) on Python 2.5 @ Windows
116
# sometimes don't provide expected error message;
117
# so we obtain such message via os.strerror()
118
self.respond('550 RNTO failed: %s' % os.strerror(e.errno))
120
self.respond('550 RNTO failed')
121
# For a test server, we will go ahead and just die
124
self.respond('250 Rename successful.')
126
def cmd_size(self, line):
127
"""Return the size of a file
129
This is overloaded to help the test suite determine if the
130
target is a directory.
133
if not self.filesystem.isfile(filename):
134
if self.filesystem.isdir(filename):
135
self.respond('550 "%s" is a directory' % (filename,))
137
self.respond('550 "%s" is not a file' % (filename,))
139
self.respond('213 %d'
140
% (self.filesystem.stat(filename)[stat.ST_SIZE]),)
142
def cmd_mkd(self, line):
143
"""Create a directory.
145
Overloaded because default implementation does not distinguish
146
*why* it cannot make a directory.
149
self.command_not_understood(''.join(line))
153
self.filesystem.mkdir (path)
154
self.respond ('257 MKD command successful.')
155
except (IOError, OSError), e:
156
# (bialix 20070418) str(e) on Python 2.5 @ Windows
157
# sometimes don't provide expected error message;
158
# so we obtain such message via os.strerror()
159
self.respond ('550 error creating directory: %s' %
160
os.strerror(e.errno))
162
self.respond ('550 error creating directory.')
164
def cmd_site(self, line):
165
"""Site specific commands."""
166
command, args = line[1].split(' ', 1)
167
if command.lower() == 'chmod':
169
mode, path = args.split()
172
# We catch both malformed line and malformed mode with the same
174
self.command_not_understood(' '.join(line))
177
# Yes path and mode are reversed
178
self.filesystem.chmod(path, mode)
179
self.respond('200 SITE CHMOD command successful')
180
except AttributeError:
181
# The chmod method is not available in read-only and will raise
182
# AttributeError since a different filesystem is used in that
184
self.command_not_authorized(' '.join(line))
186
# Another site specific command was requested. We don't know that
188
self.command_not_understood(' '.join(line))
191
class ftp_server(medusa.ftp_server.ftp_server):
192
"""Customize the behavior of the Medusa ftp_server.
194
There are a few warts on the ftp_server, based on how it expects
198
ftp_channel_class = ftp_channel
200
def __init__(self, *args, **kwargs):
201
trace.mutter('Initializing ftp_server: %r, %r', args, kwargs)
202
medusa.ftp_server.ftp_server.__init__(self, *args, **kwargs)
204
def log(self, message):
205
"""Redirect logging requests."""
206
trace.mutter('ftp_server: %s', message)
208
def log_info(self, message, type='info'):
209
"""Override the asyncore.log_info so we don't stipple the screen."""
210
trace.mutter('ftp_server %s: %s', type, message)
213
class FTPServer(transport.Server):
214
"""Common code for FTP server facilities."""
218
self._ftp_server = None
220
self._async_thread = None
225
"""Calculate an ftp url to this server."""
226
return 'ftp://foo:bar@localhost:%d/' % (self._port)
228
def get_bogus_url(self):
229
"""Return a URL which cannot be connected to."""
230
return 'ftp://127.0.0.1:1'
232
def log(self, message):
233
"""This is used by medusa.ftp_server to log connections, etc."""
234
self.logs.append(message)
236
def setUp(self, vfs_server=None):
237
from bzrlib.transport.local import LocalURLServer
238
if not (vfs_server is None or isinstance(vfs_server, LocalURLServer)):
239
raise AssertionError(
240
"FTPServer currently assumes local transport, got %s" % vfs_server)
241
self._root = os.getcwdu()
242
self._ftp_server = ftp_server(
243
authorizer=test_authorizer(root=self._root),
245
port=0, # bind to a random port
247
logger_object=self # Use FTPServer.log() for messages
249
self._port = self._ftp_server.getsockname()[1]
250
# Don't let it loop forever, or handle an infinite number of requests.
251
# In this case it will run for 1000s, or 10000 requests
252
self._async_thread = threading.Thread(
253
target=FTPServer._asyncore_loop_ignore_EBADF,
254
kwargs={'timeout':0.1, 'count':10000})
255
self._async_thread.setDaemon(True)
256
self._async_thread.start()
259
"""See bzrlib.transport.Server.tearDown."""
260
self._ftp_server.close()
262
self._async_thread.join()
265
def _asyncore_loop_ignore_EBADF(*args, **kwargs):
266
"""Ignore EBADF during server shutdown.
268
We close the socket to get the server to shutdown, but this causes
269
select.select() to raise EBADF.
272
asyncore.loop(*args, **kwargs)
273
# FIXME: If we reach that point, we should raise an exception
274
# explaining that the 'count' parameter in setUp is too low or
275
# testers may wonder why their test just sits there waiting for a
276
# server that is already dead. Note that if the tester waits too
277
# long under pdb the server will also die.
278
except select.error, e:
279
if e.args[0] != errno.EBADF:
added
removed
bzrlib/tests/ftp_server.py
