diff options
author | dluyer <53582923+dluyer@users.noreply.github.com> | 2020-01-09 14:27:40 -0800 |
---|---|---|
committer | GitHub <noreply@github.com> | 2020-01-09 14:27:40 -0800 |
commit | e62022859afaa1d5e435d75fe4ea5491c1323dd9 (patch) | |
tree | 1c010f3e67ceeaa2bf1923a8bb3a4890d2f4da90 | |
parent | 7ab86626d1e88a7f844865c598658121e2f68a42 (diff) | |
parent | e99f21c595af125a72d8fbb5bc562917ba24eff9 (diff) | |
download | pexpect-git-e62022859afaa1d5e435d75fe4ea5491c1323dd9.tar.gz |
Merge pull request #3 from dluyer/master
Merge updates from master
-rw-r--r-- | README.rst | 2 | ||||
-rwxr-xr-x | tests/deprecated_test_filedescriptor.py (renamed from tests/depricated_test_filedescriptor.py) | 0 | ||||
-rwxr-xr-x | tests/deprecated_test_run_out_of_pty.py (renamed from tests/test_run_out_of_pty.py) | 0 | ||||
-rwxr-xr-x | tests/test_ansi.py | 16 | ||||
-rw-r--r-- | tests/test_async.py | 6 | ||||
-rwxr-xr-x | tests/test_ctrl_chars.py | 13 | ||||
-rwxr-xr-x | tests/test_expect.py | 2 | ||||
-rwxr-xr-x | tests/test_interact.py | 2 | ||||
-rwxr-xr-x | tests/test_isalive.py | 4 | ||||
-rwxr-xr-x | tests/test_misc.py | 4 | ||||
-rw-r--r-- | tests/test_pxssh.py | 10 | ||||
-rwxr-xr-x | tests/test_run.py | 2 | ||||
-rw-r--r-- | tests/test_socket.py | 32 | ||||
-rw-r--r-- | tests/test_which.py | 7 |
14 files changed, 68 insertions, 32 deletions
@@ -13,7 +13,7 @@ Don Libes' Expect. Pexpect allows your script to spawn a child application and control it as if a human were typing commands. Pexpect can be used for automating interactive applications such as ssh, ftp, -passwd, telnet, etc. It can be used to a automate setup scripts for duplicating +passwd, telnet, etc. It can be used to automate setup scripts for duplicating software package installations on different servers. It can be used for automated software testing. Pexpect is in the spirit of Don Libes' Expect, but Pexpect is pure Python. diff --git a/tests/depricated_test_filedescriptor.py b/tests/deprecated_test_filedescriptor.py index 6b0ef3e..6b0ef3e 100755 --- a/tests/depricated_test_filedescriptor.py +++ b/tests/deprecated_test_filedescriptor.py diff --git a/tests/test_run_out_of_pty.py b/tests/deprecated_test_run_out_of_pty.py index 3090147..3090147 100755 --- a/tests/test_run_out_of_pty.py +++ b/tests/deprecated_test_run_out_of_pty.py diff --git a/tests/test_ansi.py b/tests/test_ansi.py index a9d445e..3d73fe8 100755 --- a/tests/test_ansi.py +++ b/tests/test_ansi.py @@ -21,7 +21,10 @@ PEXPECT LICENSE from pexpect import ANSI import unittest from . import PexpectTestCase +import os +import shutil import sys +import tempfile PY3 = (sys.version_info[0] >= 3) @@ -120,8 +123,17 @@ class ansiTestCase (PexpectTestCase.PexpectTestCase): s = ANSI.ANSI (24,80) with open('torturet.vt') as f: sample_text = f.read() - for c in sample_text: - s.process (c) + # This causes ANSI.py's DoLog to write in the cwd. Make sure we're in a + # writeable directory. + d = tempfile.mkdtemp() + old_cwd = os.getcwd() + os.chdir(d) + try: + for c in sample_text: + s.process (c) + finally: + os.chdir(old_cwd) + shutil.rmtree(d) assert s.pretty() == torture_target, 'processed: \n' + s.pretty() + '\nexpected:\n' + torture_target def test_tetris (self): diff --git a/tests/test_async.py b/tests/test_async.py index 991890c..466d56f 100644 --- a/tests/test_async.py +++ b/tests/test_async.py @@ -47,19 +47,19 @@ class AsyncTests(PexpectTestCase): run(coro) def test_expect_exact(self): - p = pexpect.spawn('%s list100.py' % sys.executable) + p = pexpect.spawn('%s list100.py' % self.PYTHONBIN) assert run(p.expect_exact(b'5', async_=True)) == 0 assert run(p.expect_exact(['wpeok', b'11'], async_=True)) == 1 assert run(p.expect_exact([b'foo', pexpect.EOF], async_=True)) == 1 def test_async_utf8(self): - p = pexpect.spawn('%s list100.py' % sys.executable, encoding='utf8') + p = pexpect.spawn('%s list100.py' % self.PYTHONBIN, encoding='utf8') assert run(p.expect_exact(u'5', async_=True)) == 0 assert run(p.expect_exact([u'wpeok', u'11'], async_=True)) == 1 assert run(p.expect_exact([u'foo', pexpect.EOF], async_=True)) == 1 def test_async_and_gc(self): - p = pexpect.spawn('%s sleep_for.py 1' % sys.executable, encoding='utf8') + p = pexpect.spawn('%s sleep_for.py 1' % self.PYTHONBIN, encoding='utf8') assert run(p.expect_exact(u'READY', async_=True)) == 0 gc.collect() assert run(p.expect_exact(u'END', async_=True)) == 0 diff --git a/tests/test_ctrl_chars.py b/tests/test_ctrl_chars.py index 10d03db..032027c 100755 --- a/tests/test_ctrl_chars.py +++ b/tests/test_ctrl_chars.py @@ -36,11 +36,14 @@ else: byte = chr class TestCtrlChars(PexpectTestCase.PexpectTestCase): + def setUp(self): + super(TestCtrlChars, self).setUp() + self.getch_cmd = self.PYTHONBIN + ' getch.py' def test_control_chars(self): '''This tests that we can send all 256 8-bit characters to a child process.''' - child = pexpect.spawn('python getch.py', echo=False, timeout=5) + child = pexpect.spawn(self.getch_cmd, echo=False, timeout=5) child.expect('READY') for i in range(1, 256): child.send(byte(i)) @@ -54,7 +57,7 @@ class TestCtrlChars(PexpectTestCase.PexpectTestCase): assert child.exitstatus == 0 def test_sendintr (self): - child = pexpect.spawn('python getch.py', echo=False, timeout=5) + child = pexpect.spawn(self.getch_cmd, echo=False, timeout=5) child.expect('READY') child.sendintr() child.expect(str(ord(ptyprocess._INTR)) + '<STOP>') @@ -66,7 +69,7 @@ class TestCtrlChars(PexpectTestCase.PexpectTestCase): assert child.exitstatus == 0 def test_sendeof(self): - child = pexpect.spawn('python getch.py', echo=False, timeout=5) + child = pexpect.spawn(self.getch_cmd, echo=False, timeout=5) child.expect('READY') child.sendeof() child.expect(str(ord(ptyprocess._EOF)) + '<STOP>') @@ -80,14 +83,14 @@ class TestCtrlChars(PexpectTestCase.PexpectTestCase): def test_bad_sendcontrol_chars (self): '''This tests that sendcontrol will return 0 for an unknown char. ''' - child = pexpect.spawn('python getch.py', echo=False, timeout=5) + child = pexpect.spawn(self.getch_cmd, echo=False, timeout=5) child.expect('READY') assert 0 == child.sendcontrol('1') def test_sendcontrol(self): '''This tests that we can send all special control codes by name. ''' - child = pexpect.spawn('python getch.py', echo=False, timeout=5) + child = pexpect.spawn(self.getch_cmd, echo=False, timeout=5) child.expect('READY') for ctrl in 'abcdefghijklmnopqrstuvwxyz': assert child.sendcontrol(ctrl) == 1 diff --git a/tests/test_expect.py b/tests/test_expect.py index 2c74744..c62231a 100755 --- a/tests/test_expect.py +++ b/tests/test_expect.py @@ -38,7 +38,7 @@ from .utils import no_coverage_env FILTER=''.join([(len(repr(chr(x)))==3) and chr(x) or '.' for x in range(256)]) def hex_dump(src, length=16): result=[] - for i in xrange(0, len(src), length): + for i in range(0, len(src), length): s = src[i:i+length] hexa = ' '.join(["%02X"%ord(x) for x in s]) printable = s.translate(FILTER) diff --git a/tests/test_interact.py b/tests/test_interact.py index 865353b..4afbd18 100755 --- a/tests/test_interact.py +++ b/tests/test_interact.py @@ -41,7 +41,7 @@ class InteractTestCase (PexpectTestCase.PexpectTestCase): else: env['PYTHONPATH'] = self.project_dir - self.interact_py = ('{sys.executable} interact.py'.format(sys=sys)) + self.interact_py = ('{self.PYTHONBIN} interact.py'.format(self=self)) def test_interact_escape(self): " Ensure `escape_character' value exits interactive mode. " diff --git a/tests/test_isalive.py b/tests/test_isalive.py index cd79d09..5e3021e 100755 --- a/tests/test_isalive.py +++ b/tests/test_isalive.py @@ -57,7 +57,7 @@ class IsAliveTestCase(PexpectTestCase.PexpectTestCase): '''Test calling wait with a process terminated by a signal.''' if not hasattr(signal, 'SIGALRM'): return 'SKIP' - p = pexpect.spawn(sys.executable, ['alarm_die.py']) + p = pexpect.spawn(self.PYTHONBIN, ['alarm_die.py']) p.wait() assert p.exitstatus is None self.assertEqual(p.signalstatus, signal.SIGALRM) @@ -99,7 +99,7 @@ class IsAliveTestCase(PexpectTestCase.PexpectTestCase): assert not p.isalive() def test_forced_terminate(self): - p = pexpect.spawn(sys.executable, ['needs_kill.py']) + p = pexpect.spawn(self.PYTHONBIN, ['needs_kill.py']) p.expect('READY') assert p.terminate(force=True) == True p.expect(pexpect.EOF) diff --git a/tests/test_misc.py b/tests/test_misc.py index 6052b6a..7784759 100755 --- a/tests/test_misc.py +++ b/tests/test_misc.py @@ -171,7 +171,7 @@ class TestCaseMisc(PexpectTestCase.PexpectTestCase): def test_with(self): "spawn can be used as a context manager" - with pexpect.spawn(sys.executable + ' echo_w_prompt.py') as p: + with pexpect.spawn(self.PYTHONBIN + ' echo_w_prompt.py') as p: p.expect('<in >') p.sendline(b'alpha') p.expect(b'<out>alpha') @@ -187,7 +187,7 @@ class TestCaseMisc(PexpectTestCase.PexpectTestCase): def test_sighup(self): " validate argument `ignore_sighup=True` and `ignore_sighup=False`. " - getch = sys.executable + ' getch.py' + getch = self.PYTHONBIN + ' getch.py' child = pexpect.spawn(getch, ignore_sighup=True) child.expect('READY') child.kill(signal.SIGHUP) diff --git a/tests/test_pxssh.py b/tests/test_pxssh.py index 0d49b23..c6ec4e2 100644 --- a/tests/test_pxssh.py +++ b/tests/test_pxssh.py @@ -1,18 +1,24 @@ #!/usr/bin/env python import os +import shutil import tempfile import unittest from pexpect import pxssh +from .PexpectTestCase import PexpectTestCase -class SSHTestBase(unittest.TestCase): +class SSHTestBase(PexpectTestCase): def setUp(self): + super(SSHTestBase, self).setUp() + self.tempdir = tempfile.mkdtemp() self.orig_path = os.environ.get('PATH') + os.symlink(self.PYTHONBIN, os.path.join(self.tempdir, 'python')) fakessh_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'fakessh')) - os.environ['PATH'] = fakessh_dir + \ + os.environ['PATH'] = self.tempdir + os.pathsep + fakessh_dir + \ ((os.pathsep + self.orig_path) if self.orig_path else '') def tearDown(self): + shutil.rmtree(self.tempdir) if self.orig_path: os.environ['PATH'] = self.orig_path else: diff --git a/tests/test_run.py b/tests/test_run.py index 1b3c92f..401ddc6 100755 --- a/tests/test_run.py +++ b/tests/test_run.py @@ -181,7 +181,7 @@ class RunUnicodeFuncTestCase(RunFuncTestCase): else: return True # Stop the child process - output = pexpect.runu(sys.executable + ' echo_w_prompt.py', + output = pexpect.runu(self.PYTHONBIN + ' echo_w_prompt.py', env={'PYTHONIOENCODING': 'utf-8'}, events={pattern: callback}) assert isinstance(output, unicode_type), type(output) diff --git a/tests/test_socket.py b/tests/test_socket.py index 21648f4..a8c8595 100644 --- a/tests/test_socket.py +++ b/tests/test_socket.py @@ -39,7 +39,17 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): def setUp(self): print(self.id()) PexpectTestCase.PexpectTestCase.setUp(self) + self.af = socket.AF_INET self.host = '127.0.0.1' + try: + socket.socket(socket.AF_INET, socket.SOCK_STREAM) + except socket.error: + try: + socket.socket(socket.AF_INET6, socket.SOCK_STREAM) + self.af = socket.AF_INET6 + self.host = '::1' + except socket.error: + pass self.port = 49152 + 10000 self.motd = b"""\ ------------------------------------------------------------------------------ @@ -92,7 +102,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): def socket_server(self, server_up): sock = None try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((self.host, self.port)) sock.listen(5) @@ -126,7 +136,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): def socket_fn(self, timed_out, all_read): result = 0 try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock, timeout=10) # Get all data from server @@ -140,7 +150,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): exit(result) def test_socket(self): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock.fileno(), timeout=10) session.expect(self.prompt1) @@ -154,7 +164,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): self.assertEqual(session.before, b'') def test_socket_with_write(self): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock.fileno(), timeout=10) session.expect(self.prompt1) @@ -177,7 +187,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): def test_timeout(self): with self.assertRaises(pexpect.TIMEOUT): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock, timeout=10) session.expect(b'Bogus response') @@ -211,7 +221,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): self.assertEqual(test_proc.exitcode, errno.ETIMEDOUT) def test_maxread(self): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock.fileno(), timeout=10) session.maxread = 1100 @@ -226,7 +236,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): self.assertEqual(session.before, b'') def test_fd_isalive(self): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock.fileno(), timeout=10) assert session.isalive() @@ -234,7 +244,7 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): assert not session.isalive(), "Should not be alive after close()" def test_fd_isalive_poll(self): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock.fileno(), timeout=10, use_poll=True) assert session.isalive() @@ -242,21 +252,21 @@ class ExpectTestCase(PexpectTestCase.PexpectTestCase): assert not session.isalive(), "Should not be alive after close()" def test_fd_isatty(self): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock.fileno(), timeout=10) assert not session.isatty() session.close() def test_fd_isatty_poll(self): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock.fileno(), timeout=10, use_poll=True) assert not session.isatty() session.close() def test_fileobj(self): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock = socket.socket(self.af, socket.SOCK_STREAM) sock.connect((self.host, self.port)) session = fdpexpect.fdspawn(sock, timeout=10) # Should get the fileno from the socket session.expect(self.prompt1) diff --git a/tests/test_which.py b/tests/test_which.py index 15a8944..e1b19a8 100644 --- a/tests/test_which.py +++ b/tests/test_which.py @@ -2,6 +2,7 @@ import subprocess import tempfile import shutil +import sys import errno import os @@ -41,8 +42,12 @@ class TestCaseWhich(PexpectTestCase.PexpectTestCase): " which() finds an executable in $PATH and returns its abspath. " bin_dir = tempfile.mkdtemp() + if sys.getfilesystemencoding() in ('ascii', 'ANSI_X3.4-1968'): + prefix = 'ascii-' + else: + prefix = u'ǝpoɔıun-' temp_obj = tempfile.NamedTemporaryFile( - suffix=u'.sh', prefix=u'ǝpoɔıun-', + suffix=u'.sh', prefix=prefix, dir=bin_dir, delete=False) bin_path = temp_obj.name fname = os.path.basename(temp_obj.name) |