← 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: Canonical.com Patch Queue Manager
  • Date: 2009-03-18 02:14:31 UTC
  • mfrom: (4158.2.1 integration)
  • Revision ID: pqm@pqm.ubuntu.com-20090318021431-md1n8o3542wwsvai
(mbp) merge back 1.13final to trunk

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):
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