python类error_perm()的实例源码

recipe-576777.py 文件源码 项目:code 作者: ActiveState 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def info(remote):
    try:
        sfmstat = remote.readlines(os.path.join(remote.root, '.sfmstat'))
    except ftplib.error_perm, err:
        if not err[0].startswith('550'):
            log(err, abort=True)
        sfmstat = None
    print
    if sfmstat:
        last_updated, mirror_path = sfmstat[0].split(None, 1)
        last_updated = datetime.datetime.fromtimestamp(float(last_updated))
        print 'Mirror of', mirror_path
        print last_updated.strftime('Last updated on %A, %d. %B %Y at %H:%M:%S')
    else:
        print 'No mirror recognized'
    print
    print 'Content of %s%s:' % (remote.host, remote.root)
    remote.ftp.dir(remote.root)
    print
download_folder_from_ftp_job.py 文件源码 项目:hoplite 作者: ni 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def _get_files_in_dir(ftp_session, source, dest, status):
    filelist = ftp_session.nlst()
    for path in filelist:
        local_path = os.path.join(dest, path)
        remote_path = source + path
        try:
            # See if it's a directory
            new_remote_path = source + path + "/"
            ftp_session.cwd(source + path + "/")
            # It is! So we should create the directory on dest
            try:
                os.mkdir(local_path)
            except OSError:
                logger.debug("Dir: {0} already exists".format(local_path))
            logger.debug("Get Folder: {0}".format(new_remote_path))
            _get_files_in_dir(ftp_session, new_remote_path, local_path, status)
        except ftplib.error_perm:
            logger.debug("Get file: {0}".format(path))
            with open(local_path, "wb") as file_handle:
                ftp_session.retrbinary(
                    "RETR " + remote_path, file_handle.write)
download_folder_from_ftp_job.py 文件源码 项目:hoplite 作者: ni 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def download_files(ftp_session, source, dest, status):
    # dest must already exist
    try:
        ftp_session.cwd(source)
    except ftplib.error_perm:
        # invalid entry (ensure input form: "/dir/folder/something/")
        msg = "Could not open the ftp directory: '{0}'".format(source)
        logger.error(msg)
        status.update({"error": msg})
        return
    if not os.path.isdir(dest):
        msg = "Local path does not exist: '{0}'".format(dest)
        logger.error(msg)
        status.update({"error": msg})
        return
    _get_files_in_dir(ftp_session, source, dest, status)
HKftp.py 文件源码 项目:Team-HK-Project 作者: GrahamMThomas 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def FtpListCommand(self, directory = '/'):
        ls = []
        if directory[-1] == '/':
            try:
                self.ftpDirectory += directory
                self.ftpDirectory.replace('//','/')
                self.ftp.retrlines('LIST %s' % self.ftpDirectory, ls.append)
            except error_perm, msg:
                errorListingMessage.text = ""
                parsedErrorMessage = (str(msg)).split(" ")[1:]
                for item in parsedErrorMessage:
                    errorListingMessage.text = errorListingMessage.text + str(item) + " "   
                listingErrorPopup.open()
        else:
            self.ftp.retrlines('LIST', ls.append)

        return self.ListParser(ls)
test_ftpfs.py 文件源码 项目:pyfilesystem2 作者: PyFilesystem 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_manager(self):
        mem_fs = open_fs('mem://')

        with self.assertRaises(errors.ResourceError):
            with ftp_errors(mem_fs, path='foo'):
                raise error_temp

        with self.assertRaises(errors.OperationFailed):
            with ftp_errors(mem_fs):
                raise error_temp

        with self.assertRaises(errors.InsufficientStorage):
            with ftp_errors(mem_fs):
                raise error_perm('552 foo')

        with self.assertRaises(errors.ResourceNotFound):
            with ftp_errors(mem_fs):
                raise error_perm('501 foo')

        with self.assertRaises(errors.PermissionDenied):
            with ftp_errors(mem_fs):
                raise error_perm('999 foo')
ftpfs.py 文件源码 项目:pyfilesystem2 作者: PyFilesystem 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def makedir(self, path, permissions=None, recreate=False):
        _path = self.validatepath(path)

        with ftp_errors(self, path=path):
            if _path == '/':
                if recreate:
                    return self.opendir(path)
                else:
                    raise errors.DirectoryExists(path)

            if not (recreate and self.isdir(path)):
                try:
                    self.ftp.mkd(_encode(_path, self.ftp.encoding))
                except error_perm as error:
                    code, _ = _parse_ftp_error(error)
                    if code == 550:
                        if self.isdir(path):
                            raise errors.DirectoryExists(path)
                        else:
                            if self.exists(path):
                                raise errors.DirectoryExists(path)
                    raise errors.ResourceNotFound(path)
        return self.opendir(path)
ftpfs.py 文件源码 项目:pyfilesystem2 作者: PyFilesystem 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def removedir(self, path):
        _path = self.validatepath(path)
        if _path == '/':
            raise errors.RemoveRootError()

        with ftp_errors(self, path):
            try:
                self.ftp.rmd(_encode(_path, self.ftp.encoding))
            except error_perm as error:
                code, _ = _parse_ftp_error(error)
                if code == 550:
                    if self.isfile(path):
                        raise errors.DirectoryExpected(path)
                    if not self.isempty(path):
                        raise errors.DirectoryNotEmpty(path)
                raise  # pragma: no cover
ftpfs.py 文件源码 项目:pyfilesystem2 作者: PyFilesystem 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def getbytes(self, path):
        _path = self.validatepath(path)
        data = io.BytesIO()
        with ftp_errors(self, path):
            with self._manage_ftp() as ftp:
                try:
                    ftp.retrbinary(
                        str("RETR ") + _encode(_path, self.ftp.encoding),
                        data.write
                    )
                except error_perm as error:
                    code, _ = _parse_ftp_error(error)
                    if code == 550:
                        if self.isdir(path):
                            raise errors.FileExpected(path)
                    raise

        data_bytes = data.getvalue()
        return data_bytes
ftpsync.py 文件源码 项目:openmv-ide 作者: openmv 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def get_mtime(self, name):
        """ Return modification time of the file in ftp server.

        If the file does not exist, return None.
        If the name refers to a directory, return 0.

        To get the local modification time, use
        ``get_mtime()-clock_offset``.
        """
        filename = self.abspath(name)
        try:
            resp = self.ftp.sendcmd('MDTM ' + filename.replace('\\', '/'))
        except error_perm, msg:
            s = str(msg)
            if s.startswith('550 I can only retrieve regular files'):
                # filename is directory
                return 0
            if s.startswith('550 Can\'t check for file existence'):
                # file with filename does not exist
                return None
            raise
        assert resp[:3]=='213', `resp, filename`
        modtime = int(time.mktime(time.strptime(resp[3:].strip(), '%Y%m%d%H%M%S')))
        return modtime
ftp.py 文件源码 项目:foil 作者: portfoliome 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def ftp_walk(ftpconn: FTP, rootpath=''):
    """Recursively traverse an ftp directory to discovery directory listing."""

    current_directory = rootpath

    try:
        directories, files = directory_listing(ftpconn, current_directory)
    except ftplib.error_perm:
        return

    # Yield before recursion
    yield current_directory, directories, files

    # Recurse into sub-directories
    for name in directories:
        new_path = os.path.join(current_directory, name)

        for entry in ftp_walk(ftpconn, rootpath=new_path):
            yield entry
    else:
        return
__main__.py 文件源码 项目:Blended 作者: BlendedSiteGenerator 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def placeFiles(ftp, path):
    """Upload the built files to FTP"""
    for name in os.listdir(path):
        if name != "config.py" and name != "config.pyc" and name != "templates" and name != "content":
            localpath = os.path.join(path, name)
            if os.path.isfile(localpath):
                print("STOR", name, localpath)
                ftp.storbinary('STOR ' + name, open(localpath, 'rb'))
            elif os.path.isdir(localpath):
                print("MKD", name)

                try:
                    ftp.mkd(name)

                # ignore "directory already exists"
                except error_perm as e:
                    if not e.args[0].startswith('550'):
                        raise

                print("CWD", name)
                ftp.cwd(name)
                placeFiles(ftp, localpath)
                print("CWD", "..")
                ftp.cwd("..")
test_functional.py 文件源码 项目:sdk-samples 作者: cradlepoint 项目源码 文件源码 阅读 40 收藏 0 点赞 0 评论 0
def test_rnfr_rnto(self):
        # rename file
        tempname = os.path.basename(tempfile.mktemp(dir=HOME))
        self.client.rename(self.tempfile, tempname)
        self.client.rename(tempname, self.tempfile)
        # rename dir
        tempname = os.path.basename(tempfile.mktemp(dir=HOME))
        self.client.rename(self.tempdir, tempname)
        self.client.rename(tempname, self.tempdir)
        # rnfr/rnto over non-existing paths
        bogus = os.path.basename(tempfile.mktemp(dir=HOME))
        self.assertRaises(ftplib.error_perm, self.client.rename, bogus, '/x')
        self.assertRaises(
            ftplib.error_perm, self.client.rename, self.tempfile, u('/'))
        # rnto sent without first specifying the source
        self.assertRaises(ftplib.error_perm, self.client.sendcmd,
                          'rnto ' + self.tempfile)

        # make sure we can't rename root directory
        self.assertRaisesRegex(ftplib.error_perm,
                               "Can't rename home directory",
                               self.client.rename, '/', '/x')
test_functional.py 文件源码 项目:sdk-samples 作者: cradlepoint 项目源码 文件源码 阅读 30 收藏 0 点赞 0 评论 0
def test_unforeseen_mdtm_event(self):
        # Emulate a case where the file last modification time is prior
        # to year 1900.  This most likely will never happen unless
        # someone specifically force the last modification time of a
        # file in some way.
        # To do so we temporarily override os.path.getmtime so that it
        # returns a negative value referring to a year prior to 1900.
        # It causes time.localtime/gmtime to raise a ValueError exception
        # which is supposed to be handled by server.

        # On python 3 it seems that the trick of replacing the original
        # method with the lambda doesn't work.
        if not PY3:
            _getmtime = AbstractedFS.getmtime
            try:
                AbstractedFS.getmtime = lambda x, y: -9000000000
                self.assertRaisesRegex(
                    ftplib.error_perm,
                    "550 Can't determine file's last modification time",
                    self.client.sendcmd, 'mdtm ' + self.tempfile)
                # make sure client hasn't been disconnected
                self.client.sendcmd('noop')
            finally:
                AbstractedFS.getmtime = _getmtime
test_functional.py 文件源码 项目:sdk-samples 作者: cradlepoint 项目源码 文件源码 阅读 31 收藏 0 点赞 0 评论 0
def test_stou_orphaned_file(self):
        # Check that no orphaned file gets left behind when STOU fails.
        # Even if STOU fails the file is first created and then erased.
        # Since we can't know the name of the file the best way that
        # we have to test this case is comparing the content of the
        # directory before and after STOU has been issued.
        # Assuming that TESTFN is supposed to be a "reserved" file
        # name we shouldn't get false positives.
        safe_remove(TESTFN)
        # login as a limited user in order to make STOU fail
        self.client.login('anonymous', '@nopasswd')
        before = os.listdir(HOME)
        self.assertRaises(ftplib.error_perm, self.client.sendcmd,
                          'stou ' + TESTFN)
        after = os.listdir(HOME)
        if before != after:
            for file in after:
                self.assertFalse(file.startswith(TESTFN))
test_functional.py 文件源码 项目:sdk-samples 作者: cradlepoint 项目源码 文件源码 阅读 31 收藏 0 点赞 0 评论 0
def test_epsv(self):
        # test wrong proto
        try:
            self.client.sendcmd('epsv ' + self.other_proto)
        except ftplib.error_perm as err:
            self.assertEqual(str(err)[0:3], "522")
        else:
            self.fail("Exception not raised")

        # proto > 2
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'epsv 3')

        # test connection
        for cmd in ('EPSV', 'EPSV ' + self.proto):
            host, port = ftplib.parse229(self.client.sendcmd(cmd),
                                         self.client.sock.getpeername())
            with contextlib.closing(
                    socket.socket(self.client.af, socket.SOCK_STREAM)) as s:
                s.settimeout(TIMEOUT)
                s.connect((host, port))
                self.client.sendcmd('abor')
test_functional.py 文件源码 项目:sdk-samples 作者: cradlepoint 项目源码 文件源码 阅读 39 收藏 0 点赞 0 评论 0
def test_mlst(self):
        # utility function for extracting the line of interest
        def mlstline(cmd):
            return self.client.voidcmd(cmd).split('\n')[1]

        if self.utf8fs:
            self.assertTrue('type=dir' in
                            mlstline('mlst ' + TESTFN_UNICODE))
            self.assertTrue('/' + TESTFN_UNICODE in
                            mlstline('mlst ' + TESTFN_UNICODE))
            self.assertTrue('type=file' in
                            mlstline('mlst ' + TESTFN_UNICODE_2))
            self.assertTrue('/' + TESTFN_UNICODE_2 in
                            mlstline('mlst ' + TESTFN_UNICODE_2))
        else:
            self.assertRaises(ftplib.error_perm,
                              mlstline, 'mlst ' + TESTFN_UNICODE)

    # --- file transfer
test_functional.py 文件源码 项目:sdk-samples 作者: cradlepoint 项目源码 文件源码 阅读 66 收藏 0 点赞 0 评论 0
def test_stor(self):
        if self.utf8fs:
            data = b'abcde12345' * 500
            os.remove(TESTFN_UNICODE_2)
            dummy = BytesIO()
            dummy.write(data)
            dummy.seek(0)
            self.client.storbinary('stor ' + TESTFN_UNICODE_2, dummy)
            dummy_recv = BytesIO()
            self.client.retrbinary('retr ' + TESTFN_UNICODE_2,
                                   dummy_recv.write)
            dummy_recv.seek(0)
            self.assertEqual(dummy_recv.read(), data)
        else:
            dummy = BytesIO()
            self.assertRaises(ftplib.error_perm, self.client.storbinary,
                              'stor ' + TESTFN_UNICODE_2, dummy)
test_functional_ssl.py 文件源码 项目:sdk-samples 作者: cradlepoint 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def test_prot(self):
        self.client.login(secure=False)
        msg = "503 PROT not allowed on insecure control connection."
        self.assertRaisesWithMsg(ftplib.error_perm, msg,
                                 self.client.sendcmd, 'prot p')
        self.client.login(secure=True)
        # secured
        self.client.prot_p()
        sock = self.client.transfercmd('list')
        with contextlib.closing(sock):
            while True:
                if not sock.recv(1024):
                    self.client.voidresp()
                    break
            self.assertTrue(isinstance(sock, ssl.SSLSocket))
            # unsecured
            self.client.prot_c()
        sock = self.client.transfercmd('list')
        with contextlib.closing(sock):
            while True:
                if not sock.recv(1024):
                    self.client.voidresp()
                    break
            self.assertFalse(isinstance(sock, ssl.SSLSocket))
jcvi.py 文件源码 项目:tredparse 作者: humanlongevity 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def ls_ftp(dir):
    from urlparse import urlparse
    from ftplib import FTP, error_perm
    o = urlparse(dir)

    ftp = FTP(o.netloc)
    ftp.login()
    ftp.cwd(o.path)

    files = []
    try:
        files = ftp.nlst()
    except error_perm, resp:
        if str(resp) == "550 No files found":
            print "no files in this directory"
        else:
            raise
    return files
ftp_class.py 文件源码 项目:Automation-Framework-for-devices 作者: tok-gogogo 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def cwd(self,path):
        #-----------------------------------------------------------------------------
        # Name:        cwd fuction
        # param:       path :cd to the path,please write absolute path
        # explain:     cd to this path
        # Author:      gongke
        #
        # Created:     2013/01/13
        #-----------------------------------------------------------------------------
        try:
            self.mftp.cwd(path)
        except ftplib.error_perm:
            #print "cannot cd to this path %s" %path
            print_mes = "cannot cd to this path  " +path 
            print print_mes
            info_public(print_mes)
            self.mftp.quit()
            return False
        #print "cd %s" %path
        print_mes = "cd  " +path 
        print print_mes
        info_public(print_mes)
        return True
ftp.py 文件源码 项目:transfert 作者: rbernand 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def open(self, flags):
        if flags in ('r',):
            self._socket = self._conn.transfercmd('RETR %s' % self.url.path)
        elif flags in ('w', 'w+'):
            try:
                self._socket = self._conn.transfercmd('STOR %s' % self.url.path)
            except ftplib.error_perm:
                raise FileExistsError(str(self.url))
ftp.py 文件源码 项目:transfert 作者: rbernand 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def size(self):
        try:
            return self._conn.size(self.url.path)
        except ftplib.error_perm as err:
            if 'not a regular file' in str(err):
                return None
            raise FileNotFoundError(self.url)
        except ValueError:
            return None
ftp.py 文件源码 项目:transfert 作者: rbernand 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def isdir(self):
        try:
            self._conn.size(self.url.path)
        except ftplib.error_perm as err:
            if "regular" in str(err):
                return True
            return False
ftp.py 文件源码 项目:transfert 作者: rbernand 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def delete(self):
        if self.isfile():
            self._conn.delete(self.url.path)
        elif self.isdir():
            try:
                self._conn.rmd(self.url.path)
            except ftplib.error_perm as err:
                raise OSError(err)
        else:
            raise FileNotFoundError(str(self.url))
ftp.py 文件源码 项目:transfert 作者: rbernand 项目源码 文件源码 阅读 32 收藏 0 点赞 0 评论 0
def mkdir(self, name=None):
        try:
            if name is None:
                if self.isdir():
                    return self
                self._conn.mkd(self.url.path)
                return self
            else:
                if self.join(name).isdir():
                    return self.join(name)
                self._conn.mkd(name)
                return self.join(name)
        except ftplib.error_perm as err:
            raise TransfertPermissionError(err)
urllib.py 文件源码 项目:kinect-2-libras 作者: inessadl 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def retrfile(self, file, type):
        import ftplib
        self.endtransfer()
        if type in ('d', 'D'): cmd = 'TYPE A'; isdir = 1
        else: cmd = 'TYPE ' + type; isdir = 0
        try:
            self.ftp.voidcmd(cmd)
        except ftplib.all_errors:
            self.init()
            self.ftp.voidcmd(cmd)
        conn = None
        if file and not isdir:
            # Try to retrieve as a file
            try:
                cmd = 'RETR ' + file
                conn = self.ftp.ntransfercmd(cmd)
            except ftplib.error_perm, reason:
                if str(reason)[:3] != '550':
                    raise IOError, ('ftp error', reason), sys.exc_info()[2]
        if not conn:
            # Set transfer mode to ASCII!
            self.ftp.voidcmd('TYPE A')
            # Try a directory listing. Verify that directory exists.
            if file:
                pwd = self.ftp.pwd()
                try:
                    try:
                        self.ftp.cwd(file)
                    except ftplib.error_perm, reason:
                        raise IOError, ('ftp error', reason), sys.exc_info()[2]
                finally:
                    self.ftp.cwd(pwd)
                cmd = 'LIST ' + file
            else:
                cmd = 'LIST'
            conn = self.ftp.ntransfercmd(cmd)
        self.busy = 1
        # Pass back both a suitably decorated object and a retrieval length
        return (addclosehook(conn[0].makefile('rb'),
                             self.endtransfer), conn[1])
dump.py 文件源码 项目:mygene.info 作者: biothings 项目源码 文件源码 阅读 35 收藏 0 点赞 0 评论 0
def get_ftpfile_lastmodified(self, file_path):
        '''return lastmodified for a given file on ftp server.'''
        try:
            response = self.client.sendcmd('MDTM ' + file_path)
        except error_perm as e:
            self.logger.debug("Skip %s: %s" % (file_path,e))
            return None
        code, lastmodified = response.split()
        # an example: 'last-modified': '20121128150000'
        lastmodified = int(time.mktime(datetime.strptime(lastmodified, '%Y%m%d%H%M%S').timetuple()))
        return lastmodified
ftp.py 文件源码 项目:core-python 作者: yidao620c 项目源码 文件源码 阅读 30 收藏 0 点赞 0 评论 0
def main():
    try:
        f = ftplib.FTP(HOST)
    except (socket.error, socket.gaierror) as e:
        print('ERROR: connot reach "%s"' % HOST)
        exit(1)
    print('*** Connected to host "%s"' % HOST)

    try:
        f.login()
    except ftplib.error_perm:
        print('ERROR: connot login anonymously')
        f.quit()
        exit(1)
    print('*** Logged in as "anonymous"')

    try:
        f.cwd(DIRN)
    except ftplib.error_perm:
        print('ERROR: cannot cd to "%s"' % DIRN)
        f.quit()
        exit(1)
    print('*** Changed to "%s" folder' % DIRN)

    try:
        locFile = open(FILE, 'wb')
        f.retrbinary('RETR %s' % FILE, locFile.write)
    except ftplib.error_perm:
        print('ERROR: cannot read file "%s"' % FILE)
        os.unlink(FILE)
    else:
        print('*** Downloaded "%s" to CWD' % FILE)
    finally:
        locFile.close()
    f.quit()
    return
ftpscout.py 文件源码 项目:ftpscout 作者: RubenRocha 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def get_directory_listing(ftp_con):
    try:
        files = ftp_con.nlst()
        if "." in files: files.remove(".")
        if ".." in files: files.remove("..")

        rnd_files = random.sample(set(files), 3)

        return "{} files listed - sample of files: {}".format(len(files), rnd_files)
    except ftplib.error_perm as resp:
        if "550" in resp:
            return "No files"
    finally:
        con.quit()
        con.close()
test_ftplib.py 文件源码 项目:zippy 作者: securesystemslab 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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')


问题


面经


文章

微信
公众号

扫码关注公众号