Python ftplib 模块,error_temp() 实例源码

我们从Python开源项目中,提取了以下27个代码示例,用于说明如何使用ftplib.error_temp()

项目:pyfilesystem2    作者:PyFilesystem    | 项目源码 | 文件源码
def close(self):
        if not self.closed:
            with self._lock:
                try:
                    if self._write_conn is not None:
                        self._write_conn.close()
                        self._write_conn = None
                    if self._read_conn is not None:
                        self._read_conn.close()
                        self._read_conn = None
                    try:
                        self.ftp.quit()
                    except error_temp:  # pragma: nocover
                        pass
                finally:
                    super(FTPFile, self).close()
项目:sdk-samples    作者:cradlepoint    | 项目源码 | 文件源码
def test_max_connections_per_ip(self):
        # Test FTPServer.max_cons_per_ip attribute
        with self.server.lock:
            self.server.server.max_cons_per_ip = 3
        self.client.quit()
        c1 = self.client_class()
        c2 = self.client_class()
        c3 = self.client_class()
        c4 = self.client_class()
        try:
            c1.connect(self.server.host, self.server.port)
            c2.connect(self.server.host, self.server.port)
            c3.connect(self.server.host, self.server.port)
            self.assertRaises(ftplib.error_temp, c4.connect, self.server.host,
                              self.server.port)
            # Make sure client has been disconnected.
            # socket.error (Windows) or EOFError (Linux) exception is
            # supposed to be raised in such a case.
            self.assertRaises((socket.error, EOFError), c4.sendcmd, 'noop')
        finally:
            for c in (c1, c2, c3, c4):
                try:
                    c.quit()
                except (socket.error, EOFError):  # already disconnected
                    c.close()
项目:sdk-samples    作者:cradlepoint    | 项目源码 | 文件源码
def test_active_conn_error(self):
        # we open a socket() but avoid to invoke accept() to
        # reproduce this error condition:
        # http://code.google.com/p/pyftpdlib/source/detail?r=905
        with contextlib.closing(socket.socket()) as sock:
            sock.bind((HOST, 0))
            port = sock.getsockname()[1]
            self.client.sock.settimeout(.1)
            try:
                resp = self.client.sendport(HOST, port)
            except ftplib.error_temp as err:
                self.assertEqual(str(err)[:3], '425')
            except (socket.timeout, getattr(ssl, "SSLError", object())):
                pass
            else:
                self.assertNotEqual(str(resp)[:3], '200')
项目:code    作者:ActiveState    | 项目源码 | 文件源码
def list(self, dir, skip_mtime=False):
        month_to_int = {'Jan': 1, 'Feb': 2, 'Mar': 3, 'Apr': 4,
            'May': 5, 'Jun': 6, 'Jul': 7, 'Aug': 8, 'Sep': 9,
            'Oct': 10, 'Nov': 11, 'Dec': 12}
        try:
            buffer = []
            self.ftp.dir('-a ', dir, buffer.append)
        except ftplib.error_temp:
            buffer = []
            self.ftp.dir(dir, buffer.append)
        dirs = []
        files = {}
        for line in buffer:
            cols = line.split(None, 8)
            name = os.path.split(cols[8])[1]
            if cols[0] == 'total' or name in ('.', '..'):
                continue
            if cols[0].startswith('d'):
                dirs.append(name)
            else:
                if skip_mtime:
                    mtime = 0
                else:
                    month = month_to_int[cols[5]]
                    day = int(cols[6])
                    if cols[7].find(':') == -1:
                        year = int(cols[7])
                        hour = minute = 0
                    else:
                        year = datetime.date.today().year
                        hour, minute = [int(s) for s in cols[7].split(':')]
                    mtime = datetime.datetime(year, month, day, hour, minute)
                    mtime = int(time.mktime(mtime.timetuple()))
                size = int(cols[4])
                files[name] = {
                    'size': size,
                    'mtime': mtime,
                    }
        return (dirs, files)
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def test_exceptions(self):
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, IOError, EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass
项目:oil    作者:oilshell    | 项目源码 | 文件源码
def test_exceptions(self):
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
项目:oil    作者:oilshell    | 项目源码 | 文件源码
def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, IOError, EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass
项目:python2-tracer    作者:extremecoders-re    | 项目源码 | 文件源码
def test_exceptions(self):
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
项目:python2-tracer    作者:extremecoders-re    | 项目源码 | 文件源码
def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, IOError, EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def test_exceptions(self):
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, IOError, EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass
项目:pyfilesystem2    作者:PyFilesystem    | 项目源码 | 文件源码
def ftp_errors(fs, path=None):
    try:
        with fs._lock:
            yield
    except socket.error:
        raise errors.RemoteConnectionError(
            msg='unable to connect to {}'.format(fs.host)
        )
    except error_temp as error:
        if path is not None:
            raise errors.ResourceError(
                path,
                msg="ftp error on resource '{}' ({})".format(path, error)
            )
        else:
            raise errors.OperationFailed(
                msg='ftp error ({})'.format(error)
            )
    except error_perm as error:
        code, message = _parse_ftp_error(error)
        if code == 552:
            raise errors.InsufficientStorage(
                path=path,
                msg=message
            )
        elif code in (501, 550):
            raise errors.ResourceNotFound(path=path)
        raise errors.PermissionDenied(
            msg=message
        )
项目:pefile.pypy    作者:cloudtracer    | 项目源码 | 文件源码
def test_exceptions(self):
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
项目:pefile.pypy    作者:cloudtracer    | 项目源码 | 文件源码
def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, IOError, EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def test_exceptions(self):
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, OSError, EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass
项目:ndk-python    作者:gittor    | 项目源码 | 文件源码
def test_exceptions(self):
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
项目:ndk-python    作者:gittor    | 项目源码 | 文件源码
def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, IOError, EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass
项目:sdk-samples    作者:cradlepoint    | 项目源码 | 文件源码
def test_stou_rest(self):
        # Watch for STOU preceded by REST, which makes no sense.
        self.client.sendcmd('type i')
        self.client.sendcmd('rest 10')
        self.assertRaisesRegex(ftplib.error_temp, "Can't STOU while REST",
                               self.client.sendcmd, 'stou')
项目:sdk-samples    作者:cradlepoint    | 项目源码 | 文件源码
def test_appe_rest(self):
        # Watch for APPE preceded by REST, which makes no sense.
        self.client.sendcmd('type i')
        self.client.sendcmd('rest 10')
        self.assertRaisesRegex(ftplib.error_temp, "Can't APPE while REST",
                               self.client.sendcmd, 'appe x')
项目:sdk-samples    作者:cradlepoint    | 项目源码 | 文件源码
def test_max_connections(self):
        # Test FTPServer.max_cons attribute
        with self.server.lock:
            self.server.server.max_cons = 3
        self.client.quit()
        c1 = self.client_class()
        c2 = self.client_class()
        c3 = self.client_class()
        try:
            c1.connect(self.server.host, self.server.port)
            c2.connect(self.server.host, self.server.port)
            self.assertRaises(ftplib.error_temp, c3.connect, self.server.host,
                              self.server.port)
            # with passive data channel established
            c2.quit()
            c1.login(USER, PASSWD)
            c1.makepasv()
            self.assertRaises(ftplib.error_temp, c2.connect, self.server.host,
                              self.server.port)
            # with passive data socket waiting for connection
            c1.login(USER, PASSWD)
            c1.sendcmd('pasv')
            self.assertRaises(ftplib.error_temp, c2.connect, self.server.host,
                              self.server.port)
            # with active data channel established
            c1.login(USER, PASSWD)
            with contextlib.closing(c1.makeport()):
                self.assertRaises(
                    ftplib.error_temp, c2.connect, self.server.host,
                    self.server.port)
        finally:
            for c in (c1, c2, c3):
                try:
                    c.quit()
                except (socket.error, EOFError):  # already disconnected
                    c.close()
项目:sdk-samples    作者:cradlepoint    | 项目源码 | 文件源码
def test_on_incomplete_file_received(self):
        _file = []

        class TestHandler(FTPHandler):

            def on_incomplete_file_received(self, file):
                _file.append(file)

        self._setUp(TestHandler)
        data = b'abcde12345' * 100000
        self.dummyfile.write(data)
        self.dummyfile.seek(0)

        with contextlib.closing(
                self.client.transfercmd('stor ' + TESTFN)) as conn:
            bytes_sent = 0
            while True:
                chunk = self.dummyfile.read(BUFSIZE)
                conn.sendall(chunk)
                bytes_sent += len(chunk)
                # stop transfer while it isn't finished yet
                if bytes_sent >= INTERRUPTED_TRANSF_SIZE or not chunk:
                    self.client.putcmd('abor')
                    break
        self.assertRaises(ftplib.error_temp, self.client.getresp)  # 426
        self.client.quit()  # prevent race conditions
        call_until(lambda: _file, "ret == [os.path.abspath(TESTFN)]")
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def test_exceptions(self):
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, OSError, EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass
项目:FTPwalker    作者:Bohdan-Khomtchouk    | 项目源码 | 文件源码
def main_run(self, args):
        """
        .. py:attribute:: main_run()
        Run threads by passing a leading directory to `traverse_branch` function.

           :param args: a tuple contain root and another tuple contains base and
           leadings. The root is the path of parent directory (assigned to a process)
           base is a tuple contain the path of sub-directory and file names that are
           associated with.
           :type args: iterable
           :rtype: None

        """
        if self.resume:
            root, leadings = args
            base = ['/']
        else:
            root, (base, leadings) = args
        print ('---' * 5, datetime.now(), '{}'.format(root), '---' * 5)
        try:
            # base, leadings = self.find_leading(root)
            # print("base and leadings for {} --> {}, {}".format(root, base, leadings))
            if not self.resume:
                leadings = [(ospath.join('/', root, i.strip('/')), root) for i in leadings]
            if leadings:
                pool = ThreadPool()
                pool.map(self.traverse_branch, leadings)
                pool.close()
                pool.join()
            else:
                self.all_path.put(base[0])
        except (ftplib.error_temp, ftplib.error_perm, socket.gaierror) as exp:
            print(exp)
项目:eemeter    作者:openeemeter    | 项目源码 | 文件源码
def _retreive_file_lines(self, filename_format, station, year):
        string = BytesIO()

        if self.ftp is None:
            self.ftp = self._get_ftp_connection()

        for station_id in self._get_potential_station_ids(station):
            filename = filename_format.format(station=station_id, year=year)
            try:
                self.ftp.retrbinary('RETR {}'.format(filename), string.write)
            except (IOError, ftplib.error_perm) as e1:
                logger.warn(
                    "Failed FTP RETR for station {}: {}."
                    " Not attempting reconnect."
                    .format(station_id, e1)
                )
            except (ftplib.error_temp, EOFError) as e2:
                # Bad connection. attempt to reconnect.
                logger.warn(
                    "Failed FTP RETR for station {}: {}."
                    " Attempting reconnect."
                    .format(station_id, e2)
                )
                self.ftp.close()
                self.ftp = self._get_ftp_connection()
                try:
                    self.ftp.retrbinary('RETR {}'.format(filename),
                                        string.write)
                except (IOError, ftplib.error_perm) as e3:
                    logger.warn(
                        "Failed FTP RETR for station {}: {}."
                        " Trying another station id."
                        .format(station_id, e3)
                    )
                else:
                    break
            else:
                break

        logger.info(
            'Successfully retrieved ftp://ftp.ncdc.noaa.gov{}'
            .format(filename)
        )

        string.seek(0)
        f = gzip.GzipFile(fileobj=string)
        lines = f.readlines()
        string.close()
        return lines