本文整理汇总了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;未经允许,请勿转载。 |
请发表评论