← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/ftp_server/pyftpdlib_based.py

  • Committer: Vincent Ladeuil
  • Date: 2009-03-18 10:56:52 UTC
  • mto: (4167.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 4168.
  • Revision ID: v.ladeuil+lp@free.fr-20090318105652-qrgo4ua4nw294bg4
Final tweaks.

* bzrlib/tests/ftp_server/pyftpdlib_based.py:
(BzrConformingFS.listdir, BzrConformingFS.fs2ftp): Ensure we use
utf8 paths even if not tested.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/ftp_server/pyftpdlib_based.py
27
27
 
28
28
 
29
29
from bzrlib import (
 
30
    osutils,
30
31
    trace,
31
32
    transport,
32
33
    )
36
37
 
37
38
    def _check_permissions(self, username, perm):
38
39
        # Like base implementation but don't warn about write permissions
39
 
        # assigned to anonynous, since that's exactly our purpose.
 
40
        # assigned to anonymous, since that's exactly our purpose.
40
41
        for p in perm:
41
42
            if p not in self.read_perms + self.write_perms:
42
43
                raise ftpserver.AuthorizerError('No such permission "%s"' %p)
44
45
 
45
46
class BzrConformingFS(ftpserver.AbstractedFS):
46
47
 
47
 
    def chmod(self, path, mode):
48
 
        return os.chmod(path, mode)
49
 
 
50
48
    def listdir(self, path):
51
49
        """List the content of a directory."""
52
 
        # XXX: Something just freaks out in asyncore if given unicode strings,
53
 
        # that may need to be revisited once unicode or at least utf-8 encoded
54
 
        # paths is better handled. -- vila 20090228
55
 
        return [str(s) for s in os.listdir(path)]
 
50
        # FIXME: need tests with unicode paths
 
51
        return [osutils.safe_utf8(s) for s in os.listdir(path)]
56
52
 
57
53
    def fs2ftp(self, fspath):
58
54
        p = ftpserver.AbstractedFS.fs2ftp(self, fspath)
59
 
        # We should never send unicode strings, they are not handled properly
60
 
        # by the stack (asynchat.async_chat.initiate_send using a buffer()
61
 
        # starting with python2.6 may be the real culprit, but converting to
62
 
        # str() here fixes the problem.  that may need to be revisited once
63
 
        # unicode or at least utf-8 encoded paths is better handled. -- vila
64
 
        # 20090228
65
 
        return str(p)
 
55
        # FIXME: need tests with unicode paths
 
56
        return osutils.safe_utf8(p)
 
57
 
66
58
 
67
59
class BZRConformingFTPHandler(ftpserver.FTPHandler):
68
60
 
103
95
            # ValueError.
104
96
            self.respond("500 'SITE CHMOD %s': command not understood."
105
97
                         % line)
106
 
            self.log('FAIL SITE CHMOD MKD ' % line)
 
98
            self.log('FAIL SITE CHMOD ' % line)
107
99
            return
108
100
        ftp_path = self.fs.fs2ftp(path)
109
101
        try:
124
116
    auth_needed=True, arg_needed=True, check_path=False,
125
117
    help='Syntax: SITE CHMOD <SP>  octal_mode_bits file-name (chmod file)',
126
118
    )
127
 
ftpserver.proto_cmds['PASS'] = ftpserver._CommandProperty(
128
 
    perm=None,
129
 
    auth_needed=False,
130
 
    # An empty password is valid, hence the arg is neither mandatory not
131
 
    # forbidden
132
 
    arg_needed=None,
133
 
    check_path=False,
134
 
    help='Syntax: PASS [<SP> password] (set user password).',
135
 
    )
 
119
# An empty password is valid, hence the arg is neither mandatory not forbidden
 
120
ftpserver.proto_cmds['PASS'].arg_needed = None
136
121
 
137
122
 
138
123
class ftp_server(ftpserver.FTPServer):
165
150
        return 'ftp://127.0.0.1:1/'
166
151
 
167
152
    def log(self, message):
168
 
        """This is used by medusa.ftp_server to log connections, etc."""
 
153
        """This is used by ftp_server to log connections, etc."""
169
154
        self.logs.append(message)
170
155
 
171
156
    def setUp(self, vfs_server=None):
190
175
        ftpserver.logerror = self.log
191
176
 
192
177
        self._port = self._ftp_server.socket.getsockname()[1]
193
 
        # Don't let it loop forever, or handle an infinite number of requests.
194
 
        # In this case it will run for 1000s, or 10000 requests
195
178
        self._ftpd_starting = threading.Lock()
196
179
        self._ftpd_starting.acquire() # So it can be released by the server
197
180
        self._ftpd_thread = threading.Thread(