← 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: Patch Queue Manager
  • Date: 2012-03-15 21:35:53 UTC
  • mfrom: (6507.1.1 trunk)
  • Revision ID: pqm@pqm.ubuntu.com-20120315213553-pj1g3cjj19nstvns
(vila) Merge 2.5 branch including fix for bug #956027 (Vincent
 Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/ftp_server/pyftpdlib_based.py
96
96
        else:
97
97
            ftpserver.FTPHandler.ftp_NLST(self, path)
98
98
 
99
 
    def ftp_SITE_CHMOD(self, line):
100
 
        try:
101
 
            mode, path = line.split(None, 1)
102
 
            mode = int(mode, 8)
103
 
        except ValueError:
104
 
            # We catch both malformed line and malformed mode with the same
105
 
            # ValueError.
106
 
            self.respond("500 'SITE CHMOD %s': command not understood."
107
 
                         % line)
108
 
            self.log('FAIL SITE CHMOD ' % line)
109
 
            return
110
 
        ftp_path = self.fs.fs2ftp(path)
111
 
        try:
112
 
            self.run_as_current_user(self.fs.chmod, self.fs.ftp2fs(path), mode)
113
 
        except OSError, err:
114
 
            why = ftpserver._strerror(err)
115
 
            self.log('FAIL SITE CHMOD 0%03o "%s". %s.' % (mode, ftp_path, why))
116
 
            self.respond('550 %s.' % why)
117
 
        else:
118
 
            self.log('OK SITE CHMOD 0%03o "%s".' % (mode, ftp_path))
119
 
            self.respond('200 SITE CHMOD succesful.')
120
 
 
121
 
    if pyftplib_version >= (0, 6, 0):
122
 
        def log_cmd(self, cmd, arg, respcode, respstr):
123
 
            # base class version choke on unicode, the alternative is to just
124
 
            # provide an empty implementation and relies on the client to do
125
 
            # the logging for debugging purposes. Not worth the trouble so far
126
 
            # -- vila 20110607
127
 
            if cmd in ("DELE", "RMD", "RNFR", "RNTO", "MKD"):
128
 
                line = '"%s" %s' % (' '.join([cmd, unicode(arg)]).strip(),
129
 
                                    respcode)
130
 
                self.log(line)
131
 
 
132
 
 
133
 
if pyftplib_version < (0, 6,0):
134
 
    # pyftpdlib says to define SITE commands by declaring ftp_SITE_<CMD>
135
 
    # methods, but fails to recognize them.
136
 
    ftpserver.proto_cmds['SITE CHMOD'] = ftpserver._CommandProperty(
137
 
        # Best fit choice even if not exactly right (can be d, f or m too)
138
 
        perm='w',
139
 
        auth_needed=True, arg_needed=True, check_path=False,
140
 
        help='Syntax: SITE CHMOD <SP>  octal_mode_bits file-name (chmod file)',
141
 
        )
142
 
    # An empty password is valid, hence the arg is neither mandatory not
143
 
    # forbidden
144
 
    ftpserver.proto_cmds['PASS'].arg_needed = None
145
 
else:
146
 
    # Same rationale as above (the password should be optional), but the hole
147
 
    # in pyftplib-0.6.0 is narrower
148
 
    ftpserver.proto_cmds['PASS']['arg'] = None
149
 
 
 
99
    def log_cmd(self, cmd, arg, respcode, respstr):
 
100
        # base class version choke on unicode, the alternative is to just
 
101
        # provide an empty implementation and relies on the client to do
 
102
        # the logging for debugging purposes. Not worth the trouble so far
 
103
        # -- vila 20110607
 
104
        if cmd in ("DELE", "RMD", "RNFR", "RNTO", "MKD"):
 
105
            line = '"%s" %s' % (' '.join([cmd, unicode(arg)]).strip(), respcode)
 
106
            self.log(line)
 
107
 
 
108
 
 
109
# An empty password is valid, hence the arg is neither mandatory nor forbidden
 
110
ftpserver.proto_cmds['PASS']['arg'] = None
150
111
 
151
112
class ftp_server(ftpserver.FTPServer):
152
113
 
191
152
 
192
153
        address = ('localhost', 0) # bind to a random port
193
154
        authorizer = AnonymousWithWriteAccessAuthorizer()
194
 
        authorizer.add_anonymous(self._root, perm='elradfmw')
 
155
        authorizer.add_anonymous(self._root, perm='elradfmwM')
195
156
        self._ftp_server = ftp_server(address, BzrConformingFTPHandler,
196
157
                                      authorizer)
197
158
        # This is hacky as hell, will not work if we need two servers working
227
188
                             % (self._ftpd_thread.ident,))
228
189
 
229
190
    def _run_server(self):
230
 
        """Run the server until stop_server is called, shut it down properly then.
 
191
        """Run the server until stop_server is called.
 
192
 
 
193
        Shut it down properly then.
231
194
        """
232
195
        self._ftpd_running = True
233
196
        self._ftpd_starting.release()
242
205
    def add_user(self, user, password):
243
206
        """Add a user with write access."""
244
207
        self._ftp_server.authorizer.add_user(user, password, self._root,
245
 
                                             perm='elradfmw')
 
208
                                             perm='elradfmwM')