From ace88cb65f641e935c78a2019e75db744744bba1 Mon Sep 17 00:00:00 2001 From: Thomas Kluyver Date: Wed, 2 Oct 2013 10:47:37 -0700 Subject: Add runu function (unicode interface to run()) --- doc/api/pexpect.rst | 4 ++++ pexpect/__init__.py | 26 ++++++++++++++++++++++---- tests/test_run.py | 53 ++++++++++++++++++++++++++++++++++++++++++----------- 3 files changed, 68 insertions(+), 15 deletions(-) diff --git a/doc/api/pexpect.rst b/doc/api/pexpect.rst index 93e38c3..565f0ef 100644 --- a/doc/api/pexpect.rst +++ b/doc/api/pexpect.rst @@ -77,6 +77,8 @@ However, for a proper unicode API to a subprocess, use this subclass: .. autoclass:: spawnu :show-inheritance: +There is also a :func:`runu` function, the unicode counterpart to :func:`run`. + .. note:: Unicode handling with pexpect works the same way on Python 2 and 3, despite @@ -90,6 +92,8 @@ run function .. autofunction:: run +.. autofunction:: runu + Exceptions ---------- diff --git a/pexpect/__init__.py b/pexpect/__init__.py index 28824ff..733ef75 100644 --- a/pexpect/__init__.py +++ b/pexpect/__init__.py @@ -224,13 +224,31 @@ def run(command, timeout=-1, withexitstatus=False, events=None, the next event. A callback may also return a string which will be sent to the child. 'extra_args' is not used by directly run(). It provides a way to pass data to a callback function through run() through the locals - dictionary passed to a callback. ''' + dictionary passed to a callback. + ''' + return _run(command, timeout=timeout, withexitstatus=withexitstatus, + events=events, extra_args=extra_args, logfile=logfile, cwd=cwd, + env=env, _spawn=spawn) + +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 _run(command, timeout, withexitstatus, events, extra_args, logfile, cwd, + env, _spawn, **kwargs): if timeout == -1: - child = spawn(command, maxread=2000, logfile=logfile, cwd=cwd, env=env) + child = _spawn(command, maxread=2000, logfile=logfile, cwd=cwd, env=env, + **kwargs) else: - child = spawn(command, timeout=timeout, maxread=2000, logfile=logfile, - cwd=cwd, env=env) + child = _spawn(command, timeout=timeout, maxread=2000, logfile=logfile, + cwd=cwd, env=env, **kwargs) if events is not None: patterns = list(events.keys()) responses = list(events.values()) diff --git a/tests/test_run.py b/tests/test_run.py index 3887e1e..ce470bb 100755 --- a/tests/test_run.py +++ b/tests/test_run.py @@ -1,4 +1,5 @@ #!/usr/bin/env python +# encoding: utf-8 ''' PEXPECT LICENSE @@ -21,6 +22,7 @@ PEXPECT LICENSE import pexpect import unittest import subprocess +import sys import PexpectTestCase # TODO Many of these test cases blindly assume that sequential @@ -28,34 +30,63 @@ import PexpectTestCase # TODO This may not always be true, but seems adequate for testing for now. # TODO I should fix this at some point. +unicode_type = str if pexpect.PY3 else unicode + def timeout_callback (d): # print d["event_count"], - if d["event_count"]>5: + if d["event_count"]>3: return 1 return 0 -class ExpectTestCase(PexpectTestCase.PexpectTestCase): +class RunFuncTestCase(PexpectTestCase.PexpectTestCase): + runfunc = staticmethod(pexpect.run) + cr = b'\r' + empty = b'' + prep_subprocess_out = staticmethod(lambda x: x) + def test_run_exit (self): - (data, exitstatus) = pexpect.run ('python exit1.py', withexitstatus=1) + (data, exitstatus) = self.runfunc('python exit1.py', withexitstatus=1) assert exitstatus == 1, "Exit status of 'python exit1.py' should be 1." def test_run (self): the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'], stdout=subprocess.PIPE).communicate()[0].rstrip() - (the_new_way, exitstatus) = pexpect.run ('ls -l /bin', withexitstatus=1) - the_new_way = the_new_way.replace(b'\r',b'').rstrip() - self.assertEqual(the_old_way, the_new_way) + (the_new_way, exitstatus) = self.runfunc('ls -l /bin', withexitstatus=1) + the_new_way = the_new_way.replace(self.cr, self.empty).rstrip() + self.assertEqual(self.prep_subprocess_out(the_old_way), the_new_way) self.assertEqual(exitstatus, 0) def test_run_callback (self): # TODO it seems like this test could block forever if run fails... - pexpect.run("cat", timeout=1, events={pexpect.TIMEOUT:timeout_callback}) + self.runfunc("cat", timeout=1, events={pexpect.TIMEOUT:timeout_callback}) def test_run_bad_exitstatus (self): - (the_new_way, exitstatus) = pexpect.run ('ls -l /najoeufhdnzkxjd', withexitstatus=1) + (the_new_way, exitstatus) = self.runfunc('ls -l /najoeufhdnzkxjd', + withexitstatus=1) assert exitstatus != 0 -if __name__ == '__main__': - unittest.main() +class RunUnicodeFuncTestCase(RunFuncTestCase): + runfunc = staticmethod(pexpect.runu) + cr = b'\r'.decode('ascii') + empty = b''.decode('ascii') + prep_subprocess_out = staticmethod(lambda x: x.decode('utf-8', 'replace')) + def test_run_unicode(self): + if pexpect.PY3: + c = chr(254) # รพ + pattern = '' + else: + c = unichr(254) # analysis:ignore + pattern = ''.decode('ascii') -suite = unittest.makeSuite(ExpectTestCase,'test') + def callback(d): + if d['event_count'] == 0: + return c + '\n' + else: + return True # Stop the child process + output = pexpect.runu(sys.executable + ' echo_w_prompt.py', + events={pattern:callback}) + assert isinstance(output, unicode_type), type(output) + assert ''+c in output, output + +if __name__ == '__main__': + unittest.main() -- cgit v1.2.1