python类S_IROTH的实例源码

memdump.py 文件源码 项目:nitro 作者: KVM-VMI 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def main(args):
    vm_name = args['<vm_name>']
    # get domain from libvirt
    con = libvirt.open('qemu:///system')
    domain = con.lookupByName(vm_name)

    path = os.path.join(os.getcwd(), '{}.raw'.format(vm_name))
    with open(path, 'w') as f:
        # chmod to be r/w by everyone
        os.chmod(path, stat.S_IRUSR | stat.S_IWUSR |
                                stat.S_IRGRP | stat.S_IWGRP |
                                stat.S_IROTH | stat.S_IWOTH)
        # take a ram dump
        flags = libvirt.VIR_DUMP_MEMORY_ONLY
        dumpformat = libvirt.VIR_DOMAIN_CORE_DUMP_FORMAT_RAW
        domain.coreDumpWithFormat(path, dumpformat, flags)
test_file_loader.py 文件源码 项目:zippy 作者: securesystemslab 项目源码 文件源码 阅读 30 收藏 0 点赞 0 评论 0
def test_read_only_bytecode(self):
        # When bytecode is read-only but should be rewritten, fail silently.
        with source_util.create_modules('_temp') as mapping:
            # Create bytecode that will need to be re-created.
            py_compile.compile(mapping['_temp'])
            bytecode_path = imp.cache_from_source(mapping['_temp'])
            with open(bytecode_path, 'r+b') as bytecode_file:
                bytecode_file.seek(0)
                bytecode_file.write(b'\x00\x00\x00\x00')
            # Make the bytecode read-only.
            os.chmod(bytecode_path,
                        stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
            try:
                # Should not raise IOError!
                self.import_(mapping['_temp'], '_temp')
            finally:
                # Make writable for eventual clean-up.
                os.chmod(bytecode_path, stat.S_IWUSR)
test_import.py 文件源码 项目:zippy 作者: securesystemslab 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_execute_bit_not_copied(self):
        # Issue 6070: under posix .pyc files got their execute bit set if
        # the .py file had the execute bit set, but they aren't executable.
        with temp_umask(0o022):
            sys.path.insert(0, os.curdir)
            try:
                fname = TESTFN + os.extsep + "py"
                open(fname, 'w').close()
                os.chmod(fname, (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH |
                                 stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH))
                fn = imp.cache_from_source(fname)
                unlink(fn)
                __import__(TESTFN)
                if not os.path.exists(fn):
                    self.fail("__import__ did not result in creation of "
                              "either a .pyc or .pyo file")
                s = os.stat(fn)
                self.assertEqual(stat.S_IMODE(s.st_mode),
                                 stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
            finally:
                del sys.path[0]
                remove_files(TESTFN)
                unload(TESTFN)
test_import.py 文件源码 项目:oil 作者: oilshell 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def test_execute_bit_not_copied(self):
        # Issue 6070: under posix .pyc files got their execute bit set if
        # the .py file had the execute bit set, but they aren't executable.
        oldmask = os.umask(022)
        sys.path.insert(0, os.curdir)
        try:
            fname = TESTFN + os.extsep + "py"
            f = open(fname, 'w').close()
            os.chmod(fname, (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH |
                             stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH))
            __import__(TESTFN)
            fn = fname + 'c'
            if not os.path.exists(fn):
                fn = fname + 'o'
                if not os.path.exists(fn):
                    self.fail("__import__ did not result in creation of "
                              "either a .pyc or .pyo file")
            s = os.stat(fn)
            self.assertEqual(stat.S_IMODE(s.st_mode),
                             stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
        finally:
            os.umask(oldmask)
            remove_files(TESTFN)
            unload(TESTFN)
            del sys.path[0]
test_import.py 文件源码 项目:python2-tracer 作者: extremecoders-re 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_execute_bit_not_copied(self):
        # Issue 6070: under posix .pyc files got their execute bit set if
        # the .py file had the execute bit set, but they aren't executable.
        oldmask = os.umask(022)
        sys.path.insert(0, os.curdir)
        try:
            fname = TESTFN + os.extsep + "py"
            f = open(fname, 'w').close()
            os.chmod(fname, (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH |
                             stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH))
            __import__(TESTFN)
            fn = fname + 'c'
            if not os.path.exists(fn):
                fn = fname + 'o'
                if not os.path.exists(fn):
                    self.fail("__import__ did not result in creation of "
                              "either a .pyc or .pyo file")
            s = os.stat(fn)
            self.assertEqual(stat.S_IMODE(s.st_mode),
                             stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
        finally:
            os.umask(oldmask)
            remove_files(TESTFN)
            unload(TESTFN)
            del sys.path[0]
makeExecutable.py 文件源码 项目:emulator.tools.retroarch 作者: JoKeRzBoX 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def makeFilesExecutable():
    scriptPath = xbmc.translatePath(xbmcaddon.Addon(id = 'emulator.tools.retroarch').getAddonInfo('path'))
    scriptPath = os.path.join(scriptPath, 'bin')
    file1 = os.path.join(scriptPath, 'retroarch.sh')
    file2 = os.path.join(scriptPath, 'retroarch.start')
    file3 = os.path.join(scriptPath, 'retroarch')

    try:
        os.chmod(file1, stat.S_IRWXU|stat.S_IRWXG|stat.S_IROTH|stat.S_IXOTH)
        os.chmod(file2, stat.S_IRWXU|stat.S_IRWXG|stat.S_IROTH|stat.S_IXOTH)
        os.chmod(file3, stat.S_IRWXU|stat.S_IRWXG|stat.S_IROTH|stat.S_IXOTH)
        d = xbmcgui.Dialog()
        d.ok('RetroArch', 'File permissions applied', 'scripts should now be executable')        
    except:
        d = xbmcgui.Dialog()
        d.ok('RetroArch', 'Failed to apply permissions', 'Please try again later or do it manualy via ssh')
test_import.py 文件源码 项目:polyloader 作者: elfsternberg 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def test_execute_bit_not_copied(self):
        # Issue 6070: under posix .pyc files got their execute bit set if
        # the .py file had the execute bit set, but they aren't executable.
        oldmask = os.umask(0o22)
        sys.path.insert(0, os.curdir)
        try:
            fname = TESTFN + os.extsep + "py"
            f = open(fname, 'w').close()
            os.chmod(fname, (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH |
                             stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH))
            __import__(TESTFN)
            fn = fname + 'c'
            if not os.path.exists(fn):
                fn = fname + 'o'
                if not os.path.exists(fn):
                    self.fail("__import__ did not result in creation of "
                              "either a .pyc or .pyo file")
            s = os.stat(fn)
            assert(s.st_mode & (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH))
            assert(not (s.st_mode & (stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)))
        finally:
            os.umask(oldmask)
            clean_tmpfiles(fname)
            unload(TESTFN)
            del sys.path[0]
test_file_loader.py 文件源码 项目:web_ctp 作者: molebot 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def test_read_only_bytecode(self):
        # When bytecode is read-only but should be rewritten, fail silently.
        with source_util.create_modules('_temp') as mapping:
            # Create bytecode that will need to be re-created.
            py_compile.compile(mapping['_temp'])
            bytecode_path = imp.cache_from_source(mapping['_temp'])
            with open(bytecode_path, 'r+b') as bytecode_file:
                bytecode_file.seek(0)
                bytecode_file.write(b'\x00\x00\x00\x00')
            # Make the bytecode read-only.
            os.chmod(bytecode_path,
                        stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
            try:
                # Should not raise IOError!
                self.import_(mapping['_temp'], '_temp')
            finally:
                # Make writable for eventual clean-up.
                os.chmod(bytecode_path, stat.S_IWUSR)
find_circ.py 文件源码 项目:find_circ2 作者: rajewsky-lab 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def store_index(self,ipath):
        self.logger.info("# indexed_fasta.store_index('%s')" % ipath)

        # write to tmp-file first and in the end rename in order to have this atomic 
        # otherwise parallel building of the same index may screw it up.

        import tempfile
        tmp = tempfile.NamedTemporaryFile(mode="w",dir = os.path.dirname(ipath),delete=False)
        for chrom in sorted(self.chrom_stats.keys()):
            ofs,ldata,skip,skipchar,size = self.chrom_stats[chrom]
            tmp.write("%s\t%d\t%d\t%d\t%r\t%d\n" % (chrom,ofs,ldata,skip,skipchar,size))

        # make sure everything is on disk
        os.fsync(tmp)
        tmp.close()

        # make it accessible to everyone
        import stat
        os.chmod(tmp.name, stat.S_IROTH | stat.S_IRGRP | stat.S_IRUSR)

        # this is atomic on POSIX as we have created tmp in the same directory, 
        # therefore same filesystem
        os.rename(tmp.name,ipath)
find_circ.py 文件源码 项目:find_circ 作者: rajewsky-lab 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def store_index(self,ipath):
        debug("# indexed_fasta.store_index('%s')" % ipath)

        # write to tmp-file first and in the end rename in order to have this atomic 
        # otherwise parallel building of the same index may screw it up.

        import tempfile
        tmp = tempfile.NamedTemporaryFile(mode="w",dir = os.path.dirname(ipath),delete=False)
        for chrom in sorted(self.chrom_stats.keys()):
            ofs,ldata,skip,skipchar,size = self.chrom_stats[chrom]
            tmp.write("%s\t%d\t%d\t%d\t%r\t%d\n" % (chrom,ofs,ldata,skip,skipchar,size))

        # make sure everything is on disk
        os.fsync(tmp)
        tmp.close()

        # make it accessible to everyone
        import stat
        os.chmod(tmp.name, stat.S_IROTH | stat.S_IRGRP | stat.S_IRUSR)

        # this is atomic on POSIX as we have created tmp in the same directory, 
        # therefore same filesystem
        os.rename(tmp.name,ipath)
build.py 文件源码 项目:ClockworkVMs 作者: csd-dev-tools 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def tearDown(self):
        '''
        Disconnect ramdisk, unloading data to pre-build location.
        '''
        self.mbl.chownR(self.keyuser, self.tmphome + "/src")

        # chmod so it's readable by everyone, writable by the group
        self.mbl.chmodR(stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH |
                        stat.S_IWGRP, self.tmphome + "/src", "append")
        self.libc.sync()
        self.libc.sync()
        # Copy back to pseudo-build directory
        call([self.RSYNC, "-aqp", self.tmphome + "/src/MacBuild/dmgs", self.buildHome])
        call([self.RSYNC, "-aqp", self.tmphome + "/src/", self.buildHome + "/builtSrc"])
        self.libc.sync()
        self.libc.sync()

        os.chdir(self.buildHome)
        self._exit(self.ramdisk, self.luggage, 0)
eAE.py 文件源码 项目:deepsleepnet 作者: akaraspt 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def _create_eae_zipfile(self, zip_file_name, main_file_path, data_files=None):

        to_zip = []
        if data_files is None:
            data_files = []

        # Handle main script
        to_zip.append(main_file_path)

        # Prepare the zip file
        zip_path = "/tmp/" + zip_file_name
        zipf = zipfile.ZipFile(zip_path, mode='w', compression=zipfile.ZIP_DEFLATED, allowZip64=True)
        for f in to_zip:
            zipf.write(f)
        zipf.close()

        # Handle other files & dirs
        for f in data_files:
            zipCommand = "zip -r -u -0 " + zip_path + " " + f
            call([zipCommand], shell=True)

        # Chmod 666 the zip file so it can be accessed
        os.chmod(zip_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IWGRP | stat.S_IROTH | stat.S_IWOTH)

        return zip_path
utils_test.py 文件源码 项目:treadmill 作者: Morgan-Stanley 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def test_create_script_perms(self):
        """this tests the create_script function (permissions).
        """
        script_file = os.path.join(self.root, 'script')
        # Test non-default mode (+x)
        mode = (stat.S_IRUSR |
                stat.S_IRGRP |
                stat.S_IROTH)

        utils.create_script(
            script_file,
            's6.run',
            mode=mode,
            user='testproid',
            home='home',
            shell='shell',
            _alias={
                's6_setuidgid': '/test/s6-setuidgid',
            }
        )

        self.assertEqual(utils.os.stat(script_file).st_mode, 33060)
__init__.py 文件源码 项目:treadmill 作者: Morgan-Stanley 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def save_app(manifest, container_dir, app_json=STATE_JSON):
    """Saves app manifest and freezes to object."""
    # Save the manifest with allocated vip and ports in the state
    state_file = os.path.join(container_dir, app_json)
    fs.write_safe(
        state_file,
        lambda f: json.dump(manifest, f)
    )
    # chmod for the file to be world readable.
    if os.name == 'posix':
        os.chmod(
            state_file,
            stat.S_IWUSR | stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH
        )

    # Freeze the app data into a namedtuple object
    return utils.to_obj(manifest)
test_import.py 文件源码 项目:pefile.pypy 作者: cloudtracer 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def test_execute_bit_not_copied(self):
        # Issue 6070: under posix .pyc files got their execute bit set if
        # the .py file had the execute bit set, but they aren't executable.
        oldmask = os.umask(022)
        sys.path.insert(0, os.curdir)
        try:
            fname = TESTFN + os.extsep + "py"
            f = open(fname, 'w').close()
            os.chmod(fname, (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH |
                             stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH))
            __import__(TESTFN)
            fn = fname + 'c'
            if not os.path.exists(fn):
                fn = fname + 'o'
                if not os.path.exists(fn):
                    self.fail("__import__ did not result in creation of "
                              "either a .pyc or .pyo file")
            s = os.stat(fn)
            self.assertEqual(stat.S_IMODE(s.st_mode),
                             stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
        finally:
            os.umask(oldmask)
            remove_files(TESTFN)
            unload(TESTFN)
            del sys.path[0]
test_file_loader.py 文件源码 项目:ouroboros 作者: pybee 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def test_read_only_bytecode(self):
        # When bytecode is read-only but should be rewritten, fail silently.
        with source_util.create_modules('_temp') as mapping:
            # Create bytecode that will need to be re-created.
            py_compile.compile(mapping['_temp'])
            bytecode_path = self.util.cache_from_source(mapping['_temp'])
            with open(bytecode_path, 'r+b') as bytecode_file:
                bytecode_file.seek(0)
                bytecode_file.write(b'\x00\x00\x00\x00')
            # Make the bytecode read-only.
            os.chmod(bytecode_path,
                        stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
            try:
                # Should not raise OSError!
                self.import_(mapping['_temp'], '_temp')
            finally:
                # Make writable for eventual clean-up.
                os.chmod(bytecode_path, stat.S_IWUSR)
test_import.py 文件源码 项目:ndk-python 作者: gittor 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def test_execute_bit_not_copied(self):
        # Issue 6070: under posix .pyc files got their execute bit set if
        # the .py file had the execute bit set, but they aren't executable.
        oldmask = os.umask(022)
        sys.path.insert(0, os.curdir)
        try:
            fname = TESTFN + os.extsep + "py"
            f = open(fname, 'w').close()
            os.chmod(fname, (stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH |
                             stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH))
            __import__(TESTFN)
            fn = fname + 'c'
            if not os.path.exists(fn):
                fn = fname + 'o'
                if not os.path.exists(fn):
                    self.fail("__import__ did not result in creation of "
                              "either a .pyc or .pyo file")
            s = os.stat(fn)
            self.assertEqual(stat.S_IMODE(s.st_mode),
                             stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
        finally:
            os.umask(oldmask)
            remove_files(TESTFN)
            unload(TESTFN)
            del sys.path[0]
formatter.py 文件源码 项目:spark 作者: chenguolin 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def _octal_to_perm(octal):
    perms = list("-" * 9)
    if octal & stat.S_IRUSR:
        perms[0] = "r"
    if octal & stat.S_IWUSR:
        perms[1] = "w"
    if octal & stat.S_IXUSR:
        perms[2] = "x"
    if octal & stat.S_IRGRP:
        perms[3] = "r"
    if octal & stat.S_IWGRP:
        perms[4] = "w"
    if octal & stat.S_IXGRP:
        perms[5] = "x"
    if octal & stat.S_IROTH:
        perms[6] = "r"
    if octal & stat.S_IWOTH:
        perms[7] = "w"
    if octal & stat.S_IXOTH:
        perms[8] = "x"
    return "".join(perms)
test_file_loader.py 文件源码 项目:kbe_server 作者: xiaohaoppy 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def test_read_only_bytecode(self):
        # When bytecode is read-only but should be rewritten, fail silently.
        with source_util.create_modules('_temp') as mapping:
            # Create bytecode that will need to be re-created.
            py_compile.compile(mapping['_temp'])
            bytecode_path = self.util.cache_from_source(mapping['_temp'])
            with open(bytecode_path, 'r+b') as bytecode_file:
                bytecode_file.seek(0)
                bytecode_file.write(b'\x00\x00\x00\x00')
            # Make the bytecode read-only.
            os.chmod(bytecode_path,
                        stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
            try:
                # Should not raise OSError!
                self.import_(mapping['_temp'], '_temp')
            finally:
                # Make writable for eventual clean-up.
                os.chmod(bytecode_path, stat.S_IWUSR)
generator_utils.py 文件源码 项目:tensor2tensor 作者: tensorflow 项目源码 文件源码 阅读 33 收藏 0 点赞 0 评论 0
def gunzip_file(gz_path, new_path):
  """Unzips from gz_path into new_path.

  Args:
    gz_path: path to the zipped file.
    new_path: path to where the file will be unzipped.
  """
  if tf.gfile.Exists(new_path):
    tf.logging.info("File %s already exists, skipping unpacking" % new_path)
    return
  tf.logging.info("Unpacking %s to %s" % (gz_path, new_path))
  # We may be unpacking into a newly created directory, add write mode.
  mode = stat.S_IRWXU or stat.S_IXGRP or stat.S_IRGRP or stat.S_IROTH
  os.chmod(os.path.dirname(new_path), mode)
  with gzip.open(gz_path, "rb") as gz_file:
    with tf.gfile.GFile(new_path, mode="wb") as new_file:
      for line in gz_file:
        new_file.write(line)
util.py 文件源码 项目:OnlineSchemaChange 作者: facebookincubator 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
def is_file_readable(filepath):
    """
    Check if the file given is readable to the user we are currently running
    at
    """
    uid = os.getuid()
    euid = os.geteuid()
    gid = os.getgid()
    egid = os.getegid()

    # This is probably true most of the time, so just let os.access()
    # handle it.  Avoids potential bugs in the rest of this function.
    if uid == euid and gid == egid:
        return os.access(filepath, os.R_OK)

    st = os.stat(filepath)

    if st.st_uid == euid:
        return st.st_mode & stat.S_IRUSR != 0

    groups = os.getgroups()
    if st.st_gid == egid or st.st_gid in groups:
        return st.st_mode & stat.S_IRGRP != 0

    return st.st_mode & stat.S_IROTH != 0
tasks.py 文件源码 项目:SoS 作者: vatlab 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def kill_task(task):
    status = check_task(task)
    if status == 'pending':
        return 'cancelled'
    # remove job file as well
    job_file =  os.path.join(os.path.expanduser('~'), '.sos', 'tasks', task + '.sh')
    if os.path.isfile(job_file):
        try:
            os.remove(job_file)
        except Exception:
            pass
    if status != 'running':
        return status
    # job is running
    pulse_file =  os.path.join(os.path.expanduser('~'), '.sos', 'tasks', task + '.pulse')
    from stat import S_IREAD, S_IRGRP, S_IROTH
    os.chmod(pulse_file, S_IREAD|S_IRGRP|S_IROTH)
    return 'killed'
recipe-577848.py 文件源码 项目:code 作者: ActiveState 项目源码 文件源码 阅读 34 收藏 0 点赞 0 评论 0
def set_executable(top):
    error_count=0
    def set_exec(name):
        mode = os.stat(name).st_mode
        new_mode = mode
        if new_mode & stat.S_IRUSR:
            new_mode = new_mode | stat.S_IXUSR
        if new_mode & stat.S_IRGRP:
            new_mode = new_mode | stat.S_IXGRP
        if new_mode & stat.S_IROTH:
            new_mode = new_mode | stat.S_IXOTH
        if (mode != new_mode):
            print "Setting exec for '%s' (mode %o => %o)" % (name, mode, new_mode)
            os.chmod(name, new_mode)
    def unset_exec(name):
        mode = os.stat(name).st_mode
        new_mode = mode & ~(stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
        if (mode != new_mode):
            print "Unsetting exec for '%s' (mode %o => %o)" % (name, mode, new_mode)
            os.chmod(name, new_mode)
    for root, dirs, files in os.walk(top):
        for name in files:
            complete_name = os.path.join(root, name)
            if os.path.islink(complete_name): continue
            try: 
                f = open(complete_name, 'r')
                header = f.read(4)
                f.close()
                if header[0:2] == '#!' or header[1:4] == 'ELF':
                    set_exec(complete_name)
                else:
                    unset_exec(complete_name)
            except Exception as e:
                print "%s: %s" %  (complete_name, e.__str__())
                error_count += 1
        for name in dirs:
            complete_name = os.path.join(root, name)
            set_exec(complete_name)
    return error_count
_broker.py 文件源码 项目:ave 作者: sonyxperiadev 项目源码 文件源码 阅读 40 收藏 0 点赞 0 评论 0
def begin_handover(self, fdtx_dir='/tmp/ave'):
        # call this function on the broker that is going to be replaced.
        # steps:
        # stop listening for new connections from clients, stop the notifier,
        # disconnect all shares, disable further allocation (a session will
        # receive a Restarting exception if it manages to allocate at precisely
        # this moment and is expected to try again once or twice), create a
        # UNIX domain socket to transfer file descriptors (in a separate step),
        # serialize the state of the local allocator and all live sessions
        # (PIDs and RPC keys). finally return serialized data and the path to
        # UNIX domain socket.
        # first of all check that fdtx_dir is writable. otherwise the socket
        # will not be created in it and everything fails
        if os.path.exists(fdtx_dir) and os.path.isdir(fdtx_dir):
            if not os.access(fdtx_dir, os.R_OK | os.X_OK | os.W_OK):
                raise Exception('directory not writable: %s' % fdtx_dir)
        self.stop_listening()
        self.stop_sharing()
        self.drop_all_shares()
        self.stop_listers()
        self.allocating = False
        self.fdtx = FdTx(None)
        uds_path = self.fdtx.listen(fdtx_dir, 'handover-%s' % rand_authkey())
        # make sure the caller will be able to interact with the new socket by
        # making it world readable and world writable
        mode = (stat.S_IRUSR  # owner has read permission
             |  stat.S_IWUSR  # owner has write permission
             |  stat.S_IRGRP  # group has read permission
             |  stat.S_IWGRP  # group has write permission
             |  stat.S_IROTH  # others have read permission
             |  stat.S_IWOTH) # others have write permission
        os.chmod(uds_path, mode)
        return self.serialize(), self.config, uds_path
_broker.py 文件源码 项目:ave 作者: sonyxperiadev 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def begin_handover(self, fdtx_dir='/tmp/ave'):
        # call this function on the broker that is going to be replaced.
        # steps:
        # stop listening for new connections from clients, stop the notifier,
        # disconnect all shares, disable further allocation (a session will
        # receive a Restarting exception if it manages to allocate at precisely
        # this moment and is expected to try again once or twice), create a
        # UNIX domain socket to transfer file descriptors (in a separate step),
        # serialize the state of the local allocator and all live sessions
        # (PIDs and RPC keys). finally return serialized data and the path to
        # UNIX domain socket.
        # first of all check that fdtx_dir is writable. otherwise the socket
        # will not be created in it and everything fails
        if os.path.exists(fdtx_dir) and os.path.isdir(fdtx_dir):
            if not os.access(fdtx_dir, os.R_OK | os.X_OK | os.W_OK):
                raise Exception('directory not writable: %s' % fdtx_dir)
        self.stop_listening()
        self.stop_sharing()
        self.drop_all_shares()
        self.stop_listers()
        self.allocating = False
        self.fdtx = FdTx(None)
        uds_path = self.fdtx.listen(fdtx_dir, 'handover-%s' % rand_authkey())
        # make sure the caller will be able to interact with the new socket by
        # making it world readable and world writable
        mode = (stat.S_IRUSR  # owner has read permission
             |  stat.S_IWUSR  # owner has write permission
             |  stat.S_IRGRP  # group has read permission
             |  stat.S_IWGRP  # group has write permission
             |  stat.S_IROTH  # others have read permission
             |  stat.S_IWOTH) # others have write permission
        os.chmod(uds_path, mode)
        return self.serialize(), self.config, uds_path
asserts.py 文件源码 项目:Taigabot 作者: FrozenPigs 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def assert_mode_644(mode):
    """Verify given mode is 644"""
    assert (mode & stat.S_IROTH) and (mode & stat.S_IRGRP)
    assert (mode & stat.S_IWUSR) and (mode & stat.S_IRUSR) and not (mode & stat.S_IXUSR)
asserts.py 文件源码 项目:Taigabot 作者: FrozenPigs 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def assert_mode_755(mode):
    """Verify given mode is 755"""
    assert (mode & stat.S_IROTH) and (mode & stat.S_IRGRP) and (mode & stat.S_IXOTH) and (mode & stat.S_IXGRP)
    assert (mode & stat.S_IWUSR) and (mode & stat.S_IRUSR) and (mode & stat.S_IXUSR)
cdrom.py 文件源码 项目:nitro 作者: KVM-VMI 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def __init__(self):
        # create cdrom dir
        self.cdrom_dir_tmp = TemporaryDirectory()
        self.tmp_dir = TemporaryDirectory()
        self.cdrom_iso_tmp = None
        # give qemu permission to execute and read in this directory
        os.chmod(self.tmp_dir.name, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR |
                                    stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH)
        self.cdrom_dir = self.cdrom_dir_tmp.name
        self.cdrom_iso_tmp = None
commands.py 文件源码 项目:argosd 作者: danielkoster 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def run(self):
        """Runs the default installer and sets rights on log directory."""
        install.run(self)

        # Make argosd user owned of log directory
        uid = getpwnam('argosd').pw_uid
        gid = getpwnam('argosd').pw_gid
        os.chown(settings.LOG_PATH, uid, gid)

        # User has full access, others can read
        mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR | \
            stat.S_IRGRP | stat.S_IROTH
        os.chmod(settings.LOG_PATH, mode)
find.py 文件源码 项目:DevOps 作者: YoLoveLife 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
def statinfo(st):
    return {
        'mode'     : "%04o" % stat.S_IMODE(st.st_mode),
        'isdir'    : stat.S_ISDIR(st.st_mode),
        'ischr'    : stat.S_ISCHR(st.st_mode),
        'isblk'    : stat.S_ISBLK(st.st_mode),
        'isreg'    : stat.S_ISREG(st.st_mode),
        'isfifo'   : stat.S_ISFIFO(st.st_mode),
        'islnk'    : stat.S_ISLNK(st.st_mode),
        'issock'   : stat.S_ISSOCK(st.st_mode),
        'uid'      : st.st_uid,
        'gid'      : st.st_gid,
        'size'     : st.st_size,
        'inode'    : st.st_ino,
        'dev'      : st.st_dev,
        'nlink'    : st.st_nlink,
        'atime'    : st.st_atime,
        'mtime'    : st.st_mtime,
        'ctime'    : st.st_ctime,
        'wusr'     : bool(st.st_mode & stat.S_IWUSR),
        'rusr'     : bool(st.st_mode & stat.S_IRUSR),
        'xusr'     : bool(st.st_mode & stat.S_IXUSR),
        'wgrp'     : bool(st.st_mode & stat.S_IWGRP),
        'rgrp'     : bool(st.st_mode & stat.S_IRGRP),
        'xgrp'     : bool(st.st_mode & stat.S_IXGRP),
        'woth'     : bool(st.st_mode & stat.S_IWOTH),
        'roth'     : bool(st.st_mode & stat.S_IROTH),
        'xoth'     : bool(st.st_mode & stat.S_IXOTH),
        'isuid'    : bool(st.st_mode & stat.S_ISUID),
        'isgid'    : bool(st.st_mode & stat.S_ISGID),
    }


问题


面经


文章

微信
公众号

扫码关注公众号