• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python os.write函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中os.write函数的典型用法代码示例。如果您正苦于以下问题:Python write函数的具体用法?Python write怎么用?Python write使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了write函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_runcmd_redirects_stdin_from_file

 def test_runcmd_redirects_stdin_from_file(self):
     fd, filename = tempfile.mkstemp()
     os.write(fd, 'foobar')
     os.lseek(fd, 0, os.SEEK_SET)
     self.assertEqual(cliapp.runcmd_unchecked(['cat'], stdin=fd),
                      (0, 'foobar', ''))
     os.close(fd)
开发者ID:lawer,项目名称:fog_client_linux,代码行数:7,代码来源:runcmd_tests.py


示例2: child

def child(pipeout):
	zzz = 0
	while True:
		time.sleep(1 if zzz == 0 else zzz)
		msg = ('Spam %03d' % zzz).encode()
		os.write(pipeout, msg)
		zzz = (zzz+1) % 5
开发者ID:bjshan,项目名称:pp4e,代码行数:7,代码来源:pipe-thread.py


示例3: save_data

def save_data(data, output_filename):
    """
    Save data to file.

    If the file already exists, the function will not save the data and return
    1

    Parameters
    ----------
    data : str
         String containing the data you wish to write out to a file

    output_filename : str
         Path (full or relative) to the file you will save the data into.

    Returns
    -------
    out : int
        Return 0 if the data was saved successfully. Return 1 if the file 
        already exists.

    Hint
    ----
    Check out the os module for determining whether a file exists already.

    """
    if not os.path.exists(output_filename): # output_file doesn't currently exist
        
        fd = os.open(output_filename,os.O_WRONLY|os.O_CREAT)
        os.write(fd,data)
        os.close(fd)
        return 0
    else:
        return 1
开发者ID:benjaminleroy,项目名称:lab6_exercise,代码行数:34,代码来源:data_acquisition.py


示例4: create_signed_cert

 def create_signed_cert(self, ou, san="IP:127.0.0.1,IP:::1,DNS:localhost"):
     print_ok("generating {0}.key, {0}.crt, {0}.p12".format(ou))
     fd, openssl_config = mkstemp(dir='.')
     os.write(fd, "extendedKeyUsage=clientAuth,serverAuth\n".encode('utf-8'))
     os.write(fd, "subjectAltName = {0}".format(san).encode('utf-8'))
     call("openssl genrsa -out {0}.key 1024".format(ou),
          shell=True, stderr=FNULL)
     call(
         "openssl req -new -key {0}.key -out {0}.csr -subj /C=US/ST=CA/O=ghostunnel/OU={0}".format(ou),
         shell=True,
         stderr=FNULL)
     call("chmod 600 {0}.key".format(ou), shell=True)
     call(
         "openssl x509 -req -in {0}.csr -CA {1}.crt -CAkey {1}.key -CAcreateserial -out {0}_temp.crt -days 5 -extfile {2}".format(
             ou,
             self.name,
             openssl_config),
         shell=True,
         stderr=FNULL)
     call(
         "openssl pkcs12 -export -out {0}_temp.p12 -in {0}_temp.crt -inkey {0}.key -password pass:".format(ou),
         shell=True)
     os.rename("{0}_temp.crt".format(ou), "{0}.crt".format(ou))
     os.rename("{0}_temp.p12".format(ou), "{0}.p12".format(ou))
     os.close(fd)
     os.remove(openssl_config)
     self.leaf_certs.append(ou)
开发者ID:security-geeks,项目名称:ghostunnel,代码行数:27,代码来源:common.py


示例5: _write

def _write(root, path, buf, offset, fh):
    f_path = full_path(root, path)
    vnfs_ops = VNFSOperations(root)
    file_name = vnfs_ops.vnfs_get_file_name(f_path)
    #if file_name == "action":
    if file_name in special_files and special_files[file_name]+'_write' in globals():
        try:
            nf_config = get_nf_config(vnfs_ops, f_path)
            # call the custom write function
            logger.info('Writing to ' + file_name  + ' in ' + 
                nf_config['nf_instance_name'] + '@' + nf_config['host'])
            ret_str = globals()[special_files[file_name]+'_write'](vnfs_ops._hypervisor, 
                nf_config, buf.rstrip("\n"))
        except errors.HypervisorError, ex:
            logger.debug('raised OSErro ' + str(ex.errno))
            raise OSError(ex.errno, os.strerror(ex.errno))
        logger.info('Successfully wrote ' + file_name + 
            ' in ' + nf_config['nf_instance_name'] + '@' + nf_config['host'])

        #if buf.rstrip("\n") == "activate":
        #    try:
        #        vnfs_ops.vnfs_deploy_nf(nf_path)
        #    except errors.VNFCreateError:
        #        #raise OSError(errno.EBUSY, os.strerror(errno.EBUSY))
        #        raise OSError(747, 'Cannot create VNF')
        #elif buf.rstrip("\n") == "stop":
        #    vnfs_ops.vnfs_stop_vnf(nf_path)
        #elif buf.rstrip("\n") == "start":
        #    vnfs_ops.vnfs_start_vnf(nf_path)
        #elif buf.rstrip("\n") == "destroy":
        #    vnfs_ops.vnfs_destroy_vnf(nf_path)
        os.lseek(fh, offset, os.SEEK_SET)
        os.write(fh, buf.rstrip("\n"))
        return len(buf)
开发者ID:williamsalles,项目名称:nf.io,代码行数:34,代码来源:nginx.py


示例6: run_once

    def run_once(self):
        super(simple_base, self).run_once()
        self.loginfo("Starting background docker command, timeout %s seconds", self.config["docker_timeout"])

        attach_in_pipe_r, attach_in_pipe_w = os.pipe()
        self.sub_stuff["file_desc"].append(attach_in_pipe_r)
        self.sub_stuff["file_desc"].append(attach_in_pipe_w)

        self.sub_stuff["subargs_a"].append(self.sub_stuff["rand_name"])

        dkrcmd = AsyncDockerCmd(self, "attach", self.sub_stuff["subargs_a"], verbose=True)
        # Runs in background
        self.sub_stuff["cmd_attach"] = dkrcmd
        self.sub_stuff["cmdresult_attach"] = dkrcmd.execute(attach_in_pipe_r)
        self.wait_interactive_cmd()
        self.logdebug("Before input should be ignored: %s", dkrcmd.cmdresult)

        # This input should be ignored.
        os.write(self.sub_stuff["run_in_pipe_w"], self.config["interactive_cmd_run"] + "\n")

        self.logdebug("Before input should be passed: %s", dkrcmd.cmdresult)
        # This input should be passed to container.
        os.write(attach_in_pipe_w, self.config["interactive_cmd_attach"] + "\n")

        self.wait_interactive_cmd()
        self.logdebug("After input was passsed: %s", dkrcmd.cmdresult)
开发者ID:lsm5,项目名称:autotest-docker,代码行数:26,代码来源:attach.py


示例7: Run

  def Run(self, args):
    """Initializes the driver."""
    self.SyncTransactionLog()

    # This will raise if the signature is bad.
    args.driver.Verify(config_lib.CONFIG["Client.driver_signing_public_key"])

    if args.force_reload:
      try:
        self.UninstallDriver(None, args.driver_name, delete_file=False)
      except Exception as e:  # pylint: disable=broad-except
        logging.debug("Error uninstalling driver: %s", e)

    path_handle, path_name = tempfile.mkstemp(suffix=".sys")
    try:
      # TODO(user): Ensure we have lock here, no races
      logging.info("Writing driver to %s", path_name)

      # Note permissions default to global read, user only write.
      try:
        os.write(path_handle, args.driver.data)
      finally:
        os.close(path_handle)

      self.InstallDriver(path_name, args.driver_name, args.driver_display_name)

    finally:
      os.unlink(path_name)
开发者ID:kartikeyap,项目名称:grr,代码行数:28,代码来源:windows.py


示例8: cosimTimeZero

 def cosimTimeZero():
     wt = int(os.environ['MYHDL_TO_PIPE'])
     rf = int(os.environ['MYHDL_FROM_PIPE'])
     buf = "TO 01 "
     for s, w in zip(fromSignames, fromSizes):
         buf += "%s %s " % (s, w)
     os.write(wt, to_bytes(buf))
开发者ID:jmgc,项目名称:myhdl-numeric,代码行数:7,代码来源:test_Cosimulation.py


示例9: cosimToSignalVals

 def cosimToSignalVals():
     wt = int(os.environ['MYHDL_TO_PIPE'])
     rf = int(os.environ['MYHDL_FROM_PIPE'])
     buf = "FROM 00 "
     for s, w in zip(fromSignames, fromSizes):
         buf += "%s %s " % (s, w)
     os.write(wt, to_bytes(buf))
     os.read(rf, MAXLINE)
     buf = "TO 00 "
     for s, w in zip(toSignames, toSizes):
         buf += "%s %s " % (s, w)
     os.write(wt, to_bytes(buf))
     os.read(rf, MAXLINE)
     os.write(wt, b"START")
     os.read(rf, MAXLINE)
     buf = "0 "
     for s, v in zip(toSignames, toVals):
         buf += s
         buf += " "
         buf += hex(v)[2:]
         buf += " "
     os.write(wt, to_bytes(buf))
     os.read(rf, MAXLINE)
     buf = "0 "
     for s, v in zip(toSignames, toXVals):
         buf += s
         buf += " "
         buf += v
         buf += " "
     os.write(wt, to_bytes(buf))
开发者ID:jmgc,项目名称:myhdl-numeric,代码行数:30,代码来源:test_Cosimulation.py


示例10: publish_from_path

 def publish_from_path(self, path, content=None):
     """
     Gets filename and content for a path, attempts to create directory if 
     necessary, writes to file.
     
     """
     fn, directory, fngz = self.get_filename_from_path(path)
     
     if not content:
         content = self.get_content_from_path(path)
     
     if not os.path.exists(directory):
         try:
             os.makedirs(directory)
         except:
             raise StaticGeneratorException('Could not create the directory: %s' % directory)
     
     try:
         f, tmpname = tempfile.mkstemp(dir=directory)
         os.write(f, content)
         os.close(f)
         os.chmod(tmpname, stat.S_IREAD | stat.S_IWRITE | stat.S_IWUSR | stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
         os.rename(tmpname, fn)
         try:
             tmpnamegz = '%s.gz' % tmpname
             f = gzip.open(tmpnamegz, 'wb')
             f.write(content)
             f.close()
             os.chmod(tmpnamegz, stat.S_IREAD | stat.S_IWRITE | stat.S_IWUSR | stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
             os.rename(tmpnamegz, fngz)
         except:
             raise StaticGeneratorException('Could not create the file: %s' % fngz)
     except:
         raise StaticGeneratorException('Could not create the file: %s' % fn)
开发者ID:andriijas,项目名称:staticgenerator,代码行数:34,代码来源:__init__.py


示例11: testContiunePollingOnSubscriberError

    def testContiunePollingOnSubscriberError(self):

        def cleanUp(_):
            w.stop()

        def fileDeleted(_):
            d = defer.Deferred()
            d.addCallback(cleanUp)
            reactor.callLater(0, d.callback, _)
            return d

        def fileChanged(_):
            os.remove(tempFile)
            raise Exception("This exception shouldn't be raisen")

        fd, tempFile = tempfile.mkstemp()
        w = watcher.FilesWatcher([tempFile], 0.001)
        w.subscribe(fileChanged = fileChanged)
        d = defer.Deferred()
        d.addCallback(fileDeleted)
        w.subscribe(fileDeleted = d.callback)
        w.start()
        os.write(fd, "test")
        os.close(fd)
        return d
开发者ID:offlinehacker,项目名称:flumotion,代码行数:25,代码来源:test_component_base_watcher.py


示例12: _exec

  def _exec(self, data):
    """Starts the registered function as a second process

    This will fork and start the registered function on a second process.
    This shouldn't block on anything. It merely starts then returns.

    Args:
      data: The data that should be written to stdin on the sub process.

    Returns:
      Nothing.
    """
    logging.warning('Executing process: %s' % self._description)
    try:
      p = MinimalSubprocess(self._description, data, timeout=self._timeout)
      p.fork_exec(self._run_func, self._uid, self._gid)
      self._lock.acquire()
      self._processes.append(p)
      self._lock.release()
      if default_ping_fd is not None:
        os.write(default_ping_fd, '\0')
    except UnknownUserError:
      logging.error('%s: Unable to find user %s', self._description,
                    self._uid)
    except UnknownGroupError:
      logging.error('%s: Unable to find group %s', self._description,
                    self._gid)
开发者ID:bdacode,项目名称:twitcher,代码行数:27,代码来源:core.py


示例13: set_

def set_(package, question, type, value, *extra):
    '''
    Set answers to debconf questions for a package.

    CLI Example:

    .. code-block:: bash

        salt '*' debconf.set <package> <question> <type> <value> [<value> ...]
    '''

    if extra:
        value = ' '.join((value,) + tuple(extra))

    fd_, fname = salt.utils.mkstemp(prefix="salt-", close_fd=False)

    line = "{0} {1} {2} {3}".format(package, question, type, value)
    os.write(fd_, line)
    os.close(fd_)

    _set_file(fname)

    os.unlink(fname)

    return True
开发者ID:DavideyLee,项目名称:salt,代码行数:25,代码来源:debconfmod.py


示例14: do_edit

def do_edit(given_cl, current_cl, cl_file_path):
  if given_cl.is_unspecified():
    # Show an editor if CL not specified on the command-line
    tmp_fd, tmp_path = tempfile.mkstemp(prefix='appspot-', suffix='.txt')
    os.write(tmp_fd, editable_change(current_cl))
    os.close(tmp_fd)

    retcode = subprocess.call(
        '%s %s' % (os.getenv('VISUAL', os.getenv('EDITOR', 'vi')),
                   commands.mkarg(tmp_path)),
        shell=True)
    try:
      if retcode < 0:
        raise Exception('editor closed with signal %s' % -retcode)
      elif retcode > 0:
        raise Exception('editor exited with error value %s' % retcode)
      edited_cl = parse_change(open(tmp_path).read())
    finally:
      os.remove(tmp_path)
    if edited_cl.is_unspecified():
      print >>sys.stderr, 'cancelled edit'
      return
    edited_cl.merge_into(current_cl)
  else:
    given_cl.merge_into(current_cl)
  out = open(cl_file_path, 'w')
  out.write(editable_change(current_cl))
  out.close()
开发者ID:minrk,项目名称:google-caja-mirror,代码行数:28,代码来源:appspot.py


示例15: rlogout

def rlogout(tty, timeout=TIMEOUT):
    """
    End remote session.
    """

    os.write(tty, "exit" + os.linesep);
    expect(tty, ["closed."], timeout=timeout);
开发者ID:joe42,项目名称:ClusterBot,代码行数:7,代码来源:ssh.py


示例16: __init__

 def __init__(self, data):
     fd, fname = tempfile.mkstemp()
     gzd = GzipFile(mode='r', fileobj=StringIO(b64decode(data)))
     os.write(fd, gzd.read())
     os.close(fd)
     gzd.close()
     self.name = fname
开发者ID:ynsta,项目名称:fwrtss,代码行数:7,代码来源:fwrtss-plot.py


示例17: write_to_tempfile

def write_to_tempfile(content, path=None, suffix="", prefix="tmp"):
    """Create temporary file or use existing file.

    This util is needed for creating temporary file with
    specified content, suffix and prefix. If path is not None,
    it will be used for writing content. If the path doesn't
    exist it'll be created.

    :param content: content for temporary file.
    :param path: same as parameter 'dir' for mkstemp
    :param suffix: same as parameter 'suffix' for mkstemp
    :param prefix: same as parameter 'prefix' for mkstemp

    For example: it can be used in database tests for creating
    configuration files.
    """
    if path:
        ensure_tree(path)

    (fd, path) = tempfile.mkstemp(suffix=suffix, dir=path, prefix=prefix)
    try:
        os.write(fd, content)
    finally:
        os.close(fd)
    return path
开发者ID:vmthunder,项目名称:virtman,代码行数:25,代码来源:fileutils.py


示例18: _call

    def _call(self, request, async_object):
        """Ensure there's an active connection and put the request in
        the queue if there is.

        Returns False if the call short circuits due to AUTH_FAILED,
        CLOSED, EXPIRED_SESSION or CONNECTING state.

        """

        if self._state == KeeperState.AUTH_FAILED:
            async_object.set_exception(AuthFailedError())
            return False
        elif self._state == KeeperState.CLOSED:
            async_object.set_exception(ConnectionClosedError(
                "Connection has been closed"))
            return False
        elif self._state in (KeeperState.EXPIRED_SESSION,
                             KeeperState.CONNECTING):
            async_object.set_exception(SessionExpiredError())
            return False

        self._queue.append((request, async_object))

        # wake the connection, guarding against a race with close()
        write_pipe = self._connection._write_pipe
        if write_pipe is None:
            async_object.set_exception(ConnectionClosedError(
                "Connection has been closed"))
        try:
            os.write(write_pipe, b'\0')
        except:
            async_object.set_exception(ConnectionClosedError(
                "Connection has been closed"))
开发者ID:adam-ho,项目名称:misc,代码行数:33,代码来源:client.py


示例19: _create_cookie

    def _create_cookie(self, timefunc=time.time):
        
        lockfd = self._get_lock()

        cookies = self._get_cookies(timefunc)

        cookie_id = 1
        for tpl in cookies:
            if int(tpl[0]) >= cookie_id:
                cookie_id = int(tpl[0]) + 1

        cookie = hexlify(os.urandom(24))
        
        cookies.append( (str(cookie_id), str(int(timefunc())), cookie) )

        for c in cookies:
            os.write(lockfd, ' '.join(c).encode('ascii') + b'\n')

        os.close(lockfd)
        if os.geteuid() == 0:
            os.chown(self.lock_file, self.uid, self.gid)

        os.rename(self.lock_file, self.cookie_file)

        self.cookieId = cookie_id
        self.cookie   = cookie
开发者ID:Ed-von-Schleck,项目名称:gruvi,代码行数:26,代码来源:authentication.py


示例20: start_gdb

def start_gdb(gdb_path, gdb_commands, gdb_flags=None):
    """Start gdb in the background and block until it finishes.

    Args:
        gdb_path: Path of the gdb binary.
        gdb_commands: Contents of GDB script to run.
        gdb_flags: List of flags to append to gdb command.
    """

    # Windows disallows opening the file while it's open for writing.
    gdb_script_fd, gdb_script_path = tempfile.mkstemp()
    os.write(gdb_script_fd, gdb_commands)
    os.close(gdb_script_fd)
    gdb_args = [gdb_path, "-x", gdb_script_path] + (gdb_flags or [])

    kwargs = {}
    if sys.platform.startswith("win"):
        kwargs["creationflags"] = subprocess.CREATE_NEW_CONSOLE

    gdb_process = subprocess.Popen(gdb_args, **kwargs)
    while gdb_process.returncode is None:
        try:
            gdb_process.communicate()
        except KeyboardInterrupt:
            pass

    os.unlink(gdb_script_path)
开发者ID:xeronith,项目名称:platform_development,代码行数:27,代码来源:__init__.py



注:本文中的os.write函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python environ.copy函数代码示例发布时间:2022-05-25
下一篇:
Python os.walk函数代码示例发布时间:2022-05-25
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap