我们从Python开源项目中，提取了以下38个代码示例，用于说明如何使用pexpect.spawnu()。
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
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 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 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 __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)
