def recv_ClientInit(self, block):
# start reward proxy.
self._log_info('Starting reward proxy server')
self.reward_proxy = pexpect.spawnu(self.factory.reward_proxy_bin,
logfile=sys.stdout,
timeout=None)
# wait on reward proxy to be up.
self._log_info('Waiting for reward proxy server')
self.reward_proxy.expect('\[RewardProxyServer\]')
self.reward_proxy_thread = threading.Thread(target=lambda: self.reward_proxy.expect(pexpect.EOF))
self.reward_proxy_thread.start()
self._log_info('Reward proxy server is up %s', self.reward_proxy.before)
super(DualProxyServer, self).recv_ClientInit(block)
self.logfile_dir = self.log_manager.logfile_dir
python类spawnu()的实例源码
def launch_process(self, command):
"""* What you can do
- It starts jumanpp process and keep it.
"""
# type: (text_type)->None
if not self.option is None:
command_plus_option = self.unix_command + " " + self.option
else:
command_plus_option = self.unix_command
if six.PY3:
if shutil.which(command) is None:
raise Exception("No command at {}".format(command))
else:
self.process_analyzer = pexpect.spawnu(command_plus_option)
self.process_id = self.process_analyzer.pid
else:
doc_command_string = "echo '' | {}".format(command)
command_check = os.system(doc_command_string)
if not command_check == 0:
raise Exception("No command at {}".format(command))
else:
self.process_analyzer = pexpect.spawnu(command_plus_option)
self.process_id = self.process_analyzer.pid
def run_interactive_task(cmd, logger, msg):
"""Run a task interactively and log when started.
Run given task using ``pexpect.spawn``. Log the command used.
Performs neither validation of the process - if the process
successfully started or is still running - nor killing of the
process. The user must do both.
:param cmd: Exact command to be executed
:param logger: Logger to write details to
:param msg: Message to be shown to user
:returns: ``pexpect.child`` object
"""
logger.info(msg)
logger.debug('%s%s', CMD_PREFIX, cmd)
child = pexpect.spawnu(cmd)
if settings.getValue('VERBOSITY') == 'debug':
child.logfile_read = sys.stdout
return child
def __init__(self, cmd_or_spawn, orig_prompt, prompt_change,
new_prompt=PEXPECT_PROMPT,
continuation_prompt=PEXPECT_CONTINUATION_PROMPT):
if isinstance(cmd_or_spawn, str):
self.child = pexpect.spawnu(cmd_or_spawn, echo=False)
else:
self.child = cmd_or_spawn
if self.child.echo:
# Existing spawn instance has echo enabled, disable it
# to prevent our input from being repeated to output.
self.child.setecho(False)
self.child.waitnoecho()
if prompt_change is None:
self.prompt = orig_prompt
else:
self.set_prompt(orig_prompt,
prompt_change.format(new_prompt, continuation_prompt))
self.prompt = new_prompt
self.continuation_prompt = continuation_prompt
self._expect_prompt()
def __init__(self, cmd_or_spawn, orig_prompt, prompt_change,
new_prompt=PEXPECT_PROMPT,
continuation_prompt=PEXPECT_CONTINUATION_PROMPT):
if isinstance(cmd_or_spawn, str):
self.child = pexpect.spawnu(cmd_or_spawn, echo=False)
else:
self.child = cmd_or_spawn
if self.child.echo:
# Existing spawn instance has echo enabled, disable it
# to prevent our input from being repeated to output.
self.child.setecho(False)
self.child.waitnoecho()
if prompt_change is None:
self.prompt = orig_prompt
else:
self.set_prompt(orig_prompt,
prompt_change.format(new_prompt, continuation_prompt))
self.prompt = new_prompt
self.continuation_prompt = continuation_prompt
self._expect_prompt()
def login(self):
"""Run the process using pexpect.
"""
self.cli = pexpect.spawnu('scm login')
def spawnu(*args, **kwargs):
"""Deprecated: pass encoding to spawn() instead."""
kwargs.setdefault('encoding', 'utf-8')
return spawn(*args, **kwargs)
def get_shell(self):
"""Gets or creates the shell in which to run commands for the
supplied demo
"""
if self._shell == None:
child = pexpect.spawnu('/bin/bash', env=self.demo.env.get(), echo=False, timeout=None)
ps1 = PEXPECT_PROMPT[:5] + u'\[\]' + PEXPECT_PROMPT[5:]
ps2 = PEXPECT_CONTINUATION_PROMPT[:5] + u'\[\]' + PEXPECT_CONTINUATION_PROMPT[5:]
prompt_change = u"PS1='{0}' PS2='{1}' PROMPT_COMMAND=''".format(ps1, ps2)
self._shell = pexpect.replwrap.REPLWrapper(child, u'\$', prompt_change)
return self._shell
def spawnu(*args, **kwargs):
"""Deprecated: pass encoding to spawn() instead."""
kwargs.setdefault('encoding', 'utf-8')
return spawn(*args, **kwargs)
def restart_process(self):
""""""
# type: ()->None
if not self.option is None:
command_plus_option = self.unix_command + " " + self.option
else:
command_plus_option = self.unix_command
self.process_analyzer.kill(sig=9)
self.process_analyzer = pexpect.spawnu(command_plus_option)
self.process_id = self.process_analyzer.pid
def initialize(self, environment):
self.env = environment
try:
self.server = ptyprocess.PtyProcessUnicode.spawn(['/usr/bin/tclsh', self.env['runtime']['settingsManager'].getSetting('speech', 'serverPath')])
#self.server = pexpect.spawnu('tclsh ' + self.env['runtime']['settingsManager'].getSetting('speech', 'serverPath'))
self._isInitialized = True
except Exception as e:
self.env['runtime']['debug'].writeDebugOut('speechDriver:initialize:' + str(e),debug.debugLevel.ERROR)
print(e)
def step_run_cli(self):
self.cli = pexpect.spawnu('kube-shell')
def test_existing_spawn(self):
child = pexpect.spawnu("bash", timeout=5, echo=False)
repl = replwrap.REPLWrapper(child, re.compile('[$#]'),
"PS1='{0}' PS2='{1}' "
"PROMPT_COMMAND=''")
res = repl.run_command("echo $HOME")
assert res.startswith('/'), res
def test_no_change_prompt(self):
if platform.python_implementation() == 'PyPy':
raise unittest.SkipTest("This test fails on PyPy because of REPL differences")
child = pexpect.spawnu('python', echo=False, timeout=5)
# prompt_change=None should mean no prompt change
py = replwrap.REPLWrapper(child, replwrap.u(">>> "), prompt_change=None,
continuation_prompt=replwrap.u("... "))
assert py.prompt == ">>> "
res = py.run_command("for a in range(3): print(a)\n")
assert res.strip().splitlines() == ['0', '1', '2']
def test_interact_spawnu_eof(self):
" Ensure subprocess receives unicode, EOF, and exit. "
p = pexpect.spawnu(self.interact_ucs_py, timeout=5, env=self.env)
p.expect('<in >')
p.sendline('?lpha')
p.sendline('?eta')
p.expect('<out>?lpha')
p.expect('<out>?eta')
p.sendeof()
p.expect_exact('<eof>')
p.expect_exact('Escaped interact')
p.expect(pexpect.EOF)
assert not p.isalive()
assert p.exitstatus == 0
def test_expect_basic (self):
p = pexpect.spawnu('cat')
p.sendline('Hello')
p.sendline('there')
p.sendline('Mr. þython') # þ is more like th than p, but never mind
p.expect('Hello')
p.expect('there')
p.expect('Mr. þython')
p.sendeof ()
p.expect (pexpect.EOF)
def test_expect_setecho_toggle(self):
'''This tests that echo may be toggled off.
'''
p = pexpect.spawnu('cat', timeout=5)
try:
self._expect_echo_toggle_off(p)
except IOError:
if sys.platform.lower().startswith('sunos'):
if hasattr(unittest, 'SkipTest'):
raise unittest.SkipTest("Not supported on this platform.")
return 'skip'
raise
self._expect_echo_toggle_on(p)
def test_expect_echo_exact (self):
'''Like test_expect_echo(), but using expect_exact().
'''
p = pexpect.spawnu('cat', timeout=5)
p.expect = p.expect_exact
self._expect_echo(p)
def test_expect_setecho_toggle_exact(self):
p = pexpect.spawnu('cat', timeout=5)
p.expect = p.expect_exact
try:
self._expect_echo_toggle_off(p)
except IOError:
if sys.platform.lower().startswith('sunos'):
if hasattr(unittest, 'SkipTest'):
raise unittest.SkipTest("Not supported on this platform.")
return 'skip'
raise
self._expect_echo_toggle_on(p)
def test_log_unicode(self):
msg = "abc?÷"
filename_send = tempfile.mktemp()
filename_read = tempfile.mktemp()
p = pexpect.spawnu('cat')
if platform.python_version_tuple() < ('3', '0', '0'):
import codecs
def open(fname, mode, **kwargs):
if 'newline' in kwargs:
del kwargs['newline']
return codecs.open(fname, mode, **kwargs)
else:
import io
open = io.open
p.logfile_send = open(filename_send, 'w', encoding='utf-8')
p.logfile_read = open(filename_read, 'w', encoding='utf-8')
p.sendline(msg)
p.sendeof()
p.expect(pexpect.EOF)
p.close()
p.logfile_send.close()
p.logfile_read.close()
# ensure the 'send' log is correct,
with open(filename_send, 'r', encoding='utf-8') as f:
self.assertEqual(f.read(), msg + '\n\x04')
# ensure the 'read' log is correct,
with open(filename_read, 'r', encoding='utf-8', newline='') as f:
output = f.read().replace(_CAT_EOF, '')
self.assertEqual(output, (msg + '\r\n')*2 )
def test_spawn_utf8_incomplete(self):
# This test case ensures correct incremental decoding, which
# otherwise fails when the stream inspected by os.read()
# does not align exactly at a utf-8 multibyte boundry:
# UnicodeDecodeError: 'utf8' codec can't decode byte 0xe2 in
# position 0: unexpected end of data
p = pexpect.spawnu('cat', maxread=1)
p.sendline('????????')
p.sendeof()
p.expect('????????')
def main():
p = pexpect.spawnu(sys.executable + ' echo_w_prompt.py',
env=no_coverage_env())
p.interact()
print("Escaped interact")
def spawnu(*args, **kwargs):
"""Deprecated: pass encoding to spawn() instead."""
kwargs.setdefault('encoding', 'utf-8')
return spawn(*args, **kwargs)
def __init__(self, cmd_or_spawn, orig_prompt, prompt_change,
new_prompt=PEXPECT_PROMPT,
continuation_prompt=PEXPECT_CONTINUATION_PROMPT,
extra_init_cmd=None):
if isinstance(cmd_or_spawn, str):
self.child = pexpect.spawnu(cmd_or_spawn, echo=False)
else:
self.child = cmd_or_spawn
if self.child.echo:
# Existing spawn instance has echo enabled, disable it
# to prevent our input from being repeated to output.
self.child.setecho(False)
self.child.waitnoecho()
if prompt_change is None:
self.prompt = orig_prompt
else:
self.set_prompt(orig_prompt,
prompt_change.format(new_prompt, continuation_prompt))
self.prompt = new_prompt
self.continuation_prompt = continuation_prompt
self._expect_prompt()
if extra_init_cmd is not None:
self.run_command(extra_init_cmd)
def runu(command, timeout=-1, withexitstatus=False, events=None,
extra_args=None, logfile=None, cwd=None, env=None, **kwargs):
"""This offers the same interface as :func:`run`, but using unicode.
Like :class:`spawnu`, you can pass ``encoding`` and ``errors`` parameters,
which will be used for both input and output.
"""
return _run(command, timeout=timeout, withexitstatus=withexitstatus,
events=events, extra_args=extra_args, logfile=logfile, cwd=cwd,
env=env, _spawn=spawnu, **kwargs)
def __init__(self, *args, **kwargs):
self.encoding = kwargs.pop('encoding', 'utf-8')
self.errors = kwargs.pop('errors', 'strict')
self._decoder = codecs.getincrementaldecoder(self.encoding)(errors=self.errors)
super(spawnu, self).__init__(*args, **kwargs)
def create_term(self, rows, cols, cwd=None):
"""Create a new virtual terminal."""
pid = hashlib.md5(str(time.time()).encode('utf-8')).hexdigest()[0:6]
if WINDOWS:
tty = pty.PTY(cols, rows)
tty.spawn(self.cmd, cwd=cwd)
else:
tty = pty.spawnu(self.cmd, cwd=cwd)
tty.setwinsize(rows, cols)
self.consoles[pid] = {'tty': tty, 'read': None}
raise tornado.gen.Return(pid)
def spawnu(*args, **kwargs):
"""Deprecated: pass encoding to spawn() instead."""
kwargs.setdefault('encoding', 'utf-8')
return spawn(*args, **kwargs)
def test_existing_spawn(self):
child = pexpect.spawnu("bash", timeout=5, echo=False)
repl = replwrap.REPLWrapper(child, re.compile('[$#]'),
"PS1='{0}' PS2='{1}' "
"PROMPT_COMMAND=''")
res = repl.run_command("echo $HOME")
assert res.startswith('/'), res
def test_no_change_prompt(self):
if platform.python_implementation() == 'PyPy':
raise unittest.SkipTest("This test fails on PyPy because of REPL differences")
child = pexpect.spawnu('python', echo=False, timeout=5)
# prompt_change=None should mean no prompt change
py = replwrap.REPLWrapper(child, replwrap.u(">>> "), prompt_change=None,
continuation_prompt=replwrap.u("... "))
assert py.prompt == ">>> "
res = py.run_command("for a in range(3): print(a)\n")
assert res.strip().splitlines() == ['0', '1', '2']
