← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/ftp_server.py

  • Committer: Jelmer Vernooij
  • Date: 2008-07-08 15:58:12 UTC
  • mfrom: (3353.1.4 paramiko-1.7.2-compat)
  • mto: This revision was merged to the branch mainline in revision 3531.
  • Revision ID: jelmer@samba.org-20080708155812-gacwrgs7d7n48a0d
Merge bashprompt script that prints the bzr branch nick.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/ftp_server.py
1
 
# Copyright (C) 2007-2010 Canonical Ltd
 
1
# Copyright (C) 2007 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
"""
17
17
FTP test server.
18
18
 
33
33
from bzrlib import (
34
34
    tests,
35
35
    trace,
 
36
    transport,
36
37
    )
37
 
from bzrlib.tests import test_server
38
38
 
39
39
 
40
40
class test_filesystem(medusa.filesys.os_filesystem):
126
126
    def cmd_size(self, line):
127
127
        """Return the size of a file
128
128
 
129
 
        This is overloaded to help the test suite determine if the
 
129
        This is overloaded to help the test suite determine if the 
130
130
        target is a directory.
131
131
        """
132
132
        filename = line[1]
136
136
            else:
137
137
                self.respond('550 "%s" is not a file' % (filename,))
138
138
        else:
139
 
            self.respond('213 %d'
 
139
            self.respond('213 %d' 
140
140
                % (self.filesystem.stat(filename)[stat.ST_SIZE]),)
141
141
 
142
142
    def cmd_mkd(self, line):
210
210
        trace.mutter('ftp_server %s: %s', type, message)
211
211
 
212
212
 
213
 
class FTPTestServer(test_server.TestServer):
 
213
class FTPServer(transport.Server):
214
214
    """Common code for FTP server facilities."""
215
215
 
216
 
    no_unicode_support = True
217
 
 
218
216
    def __init__(self):
219
217
        self._root = None
220
218
        self._ftp_server = None
235
233
        """This is used by medusa.ftp_server to log connections, etc."""
236
234
        self.logs.append(message)
237
235
 
238
 
    def start_server(self, vfs_server=None):
239
 
        if not (vfs_server is None or isinstance(vfs_server,
240
 
                                                 test_server.LocalURLServer)):
 
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)):
241
239
            raise AssertionError(
242
240
                "FTPServer currently assumes local transport, got %s" % vfs_server)
243
241
        self._root = os.getcwdu()
252
250
        # Don't let it loop forever, or handle an infinite number of requests.
253
251
        # In this case it will run for 1000s, or 10000 requests
254
252
        self._async_thread = threading.Thread(
255
 
                target=FTPTestServer._asyncore_loop_ignore_EBADF,
 
253
                target=FTPServer._asyncore_loop_ignore_EBADF,
256
254
                kwargs={'timeout':0.1, 'count':10000})
257
 
        if 'threads' in tests.selftest_debug_flags:
258
 
            sys.stderr.write('Thread started: %s\n'
259
 
                             % (self._async_thread.ident,))
260
255
        self._async_thread.setDaemon(True)
261
256
        self._async_thread.start()
262
257
 
263
 
    def stop_server(self):
 
258
    def tearDown(self):
 
259
        """See bzrlib.transport.Server.tearDown."""
264
260
        self._ftp_server.close()
265
261
        asyncore.close_all()
266
262
        self._async_thread.join()
267
 
        if 'threads' in tests.selftest_debug_flags:
268
 
            sys.stderr.write('Thread  joined: %s\n'
269
 
                             % (self._async_thread.ident,))
270
263
 
271
264
    @staticmethod
272
265
    def _asyncore_loop_ignore_EBADF(*args, **kwargs):
286
279
            if e.args[0] != errno.EBADF:
287
280
                raise
288
281
 
289
 
    def add_user(self, user, password):
290
 
        """Add a user with write access."""
291
 
        authorizer = server = self._ftp_server.authorizer
292
 
        authorizer.secured_user = user
293
 
        authorizer.secured_password = password
 
282
 
 
283
 
294
284