/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/ftp_server.py

  • Committer: Ian Clatworthy
  • Date: 2007-11-23 05:52:03 UTC
  • mto: (3054.1.1 ianc-integration)
  • mto: This revision was merged to the branch mainline in revision 3055.
  • Revision ID: ian.clatworthy@internode.on.net-20071123055203-r2t2r27amhzllubr
first cut at Central development chapter

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
 
20
20
"""
21
21
 
22
22
import asyncore
23
 
import errno
24
23
import os
25
24
import select
26
25
import stat
33
32
from bzrlib import (
34
33
    tests,
35
34
    trace,
 
35
    transport,
36
36
    )
37
 
from bzrlib.tests import test_server
38
 
 
39
 
 
40
 
class test_filesystem(medusa.filesys.os_filesystem):
41
 
    """A custom filesystem wrapper to add missing functionalities."""
42
 
 
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)
46
37
 
47
38
 
48
39
class test_authorizer(object):
72
63
            and password != self.secured_password):
73
64
            return 0, 'Password invalid.', None
74
65
        else:
75
 
            return 1, 'OK.', test_filesystem(self.root)
 
66
            return 1, 'OK.', medusa.filesys.os_filesystem(self.root)
76
67
 
77
68
 
78
69
class ftp_channel(medusa.ftp_server.ftp_channel):
126
117
    def cmd_size(self, line):
127
118
        """Return the size of a file
128
119
 
129
 
        This is overloaded to help the test suite determine if the
 
120
        This is overloaded to help the test suite determine if the 
130
121
        target is a directory.
131
122
        """
132
123
        filename = line[1]
136
127
            else:
137
128
                self.respond('550 "%s" is not a file' % (filename,))
138
129
        else:
139
 
            self.respond('213 %d'
 
130
            self.respond('213 %d' 
140
131
                % (self.filesystem.stat(filename)[stat.ST_SIZE]),)
141
132
 
142
133
    def cmd_mkd(self, line):
161
152
            except:
162
153
                self.respond ('550 error creating directory.')
163
154
 
164
 
    def cmd_site(self, line):
165
 
        """Site specific commands."""
166
 
        command, args = line[1].split(' ', 1)
167
 
        if command.lower() == 'chmod':
168
 
            try:
169
 
                mode, path = args.split()
170
 
                mode = int(mode, 8)
171
 
            except ValueError:
172
 
                # We catch both malformed line and malformed mode with the same
173
 
                # ValueError.
174
 
                self.command_not_understood(' '.join(line))
175
 
                return
176
 
            try:
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
183
 
                # case
184
 
                self.command_not_authorized(' '.join(line))
185
 
        else:
186
 
            # Another site specific command was requested. We don't know that
187
 
            # one
188
 
            self.command_not_understood(' '.join(line))
189
 
 
190
155
 
191
156
class ftp_server(medusa.ftp_server.ftp_server):
192
157
    """Customize the behavior of the Medusa ftp_server.
210
175
        trace.mutter('ftp_server %s: %s', type, message)
211
176
 
212
177
 
213
 
class FTPTestServer(test_server.TestServer):
 
178
class FTPServer(transport.Server):
214
179
    """Common code for FTP server facilities."""
215
180
 
216
 
    no_unicode_support = True
217
 
 
218
181
    def __init__(self):
219
182
        self._root = None
220
183
        self._ftp_server = None
227
190
        """Calculate an ftp url to this server."""
228
191
        return 'ftp://foo:bar@localhost:%d/' % (self._port)
229
192
 
230
 
    def get_bogus_url(self):
231
 
        """Return a URL which cannot be connected to."""
232
 
        return 'ftp://127.0.0.1:1'
 
193
#    def get_bogus_url(self):
 
194
#        """Return a URL which cannot be connected to."""
 
195
#        return 'ftp://127.0.0.1:1'
233
196
 
234
197
    def log(self, message):
235
198
        """This is used by medusa.ftp_server to log connections, etc."""
236
199
        self.logs.append(message)
237
200
 
238
 
    def start_server(self, vfs_server=None):
239
 
        if not (vfs_server is None or isinstance(vfs_server,
240
 
                                                 test_server.LocalURLServer)):
241
 
            raise AssertionError(
242
 
                "FTPServer currently assumes local transport, got %s" % vfs_server)
 
201
    def setUp(self, vfs_server=None):
 
202
        from bzrlib.transport.local import LocalURLServer
 
203
        assert vfs_server is None or isinstance(vfs_server, LocalURLServer), \
 
204
            "FTPServer currently assumes local transport, got %s" % vfs_server
 
205
 
243
206
        self._root = os.getcwdu()
244
207
        self._ftp_server = ftp_server(
245
208
            authorizer=test_authorizer(root=self._root),
252
215
        # Don't let it loop forever, or handle an infinite number of requests.
253
216
        # In this case it will run for 1000s, or 10000 requests
254
217
        self._async_thread = threading.Thread(
255
 
                target=FTPTestServer._asyncore_loop_ignore_EBADF,
 
218
                target=FTPServer._asyncore_loop_ignore_EBADF,
256
219
                kwargs={'timeout':0.1, 'count':10000})
257
220
        self._async_thread.setDaemon(True)
258
221
        self._async_thread.start()
259
222
 
260
 
    def stop_server(self):
 
223
    def tearDown(self):
 
224
        """See bzrlib.transport.Server.tearDown."""
261
225
        self._ftp_server.close()
262
226
        asyncore.close_all()
263
227
        self._async_thread.join()
280
244
            if e.args[0] != errno.EBADF:
281
245
                raise
282
246
 
283
 
    def add_user(self, user, password):
284
 
        """Add a user with write access."""
285
 
        authorizer = server = self._ftp_server.authorizer
286
 
        authorizer.secured_user = user
287
 
        authorizer.secured_password = password
 
247
 
 
248
 
288
249