Merge remote-tracking branch 'origin/master' into document-blocking-write

15 files changed, 795 insertions, 126 deletions

diff --git a/tests/PexpectTestCase.py b/tests/PexpectTestCase.py
index 7a9574e..307437e 100644
--- a/tests/PexpectTestCase.py
+++ b/tests/PexpectTestCase.py
@@ -22,26 +22,68 @@ from __future__ import print_function
 
 import contextlib
 import unittest
+import signal
 import sys
 import os
 
+
 class PexpectTestCase(unittest.TestCase):
     def setUp(self):
         self.PYTHONBIN = sys.executable
         self.original_path = os.getcwd()
         tests_dir = os.path.dirname(__file__)
         self.project_dir = project_dir = os.path.dirname(tests_dir)
+
+        # all tests are executed in this folder; there are many auxiliary
+        # programs in this folder executed by spawn().
         os.chdir(tests_dir)
-        os.environ['COVERAGE_PROCESS_START'] = os.path.join(project_dir, '.coveragerc')
+
+        # If the pexpect raises an exception after fork(), but before
+        # exec(), our test runner *also* forks.  We prevent this by
+        # storing our pid and asserting equality on tearDown.
+        self.pid = os.getpid()
+
+        coverage_rc = os.path.join(project_dir, '.coveragerc')
+        os.environ['COVERAGE_PROCESS_START'] = coverage_rc
         os.environ['COVERAGE_FILE'] = os.path.join(project_dir, '.coverage')
         print('\n', self.id(), end=' ')
         sys.stdout.flush()
+
+        # some build agents will ignore SIGHUP and SIGINT, which python
+        # inherits.  This causes some of the tests related to terminate()
+        # to fail.  We set them to the default handlers that they should
+        # be, and restore them back to their SIG_IGN value on tearDown.
+        #
+        # I'm not entirely convinced they need to be restored, only our
+        # test runner is affected.
+        self.restore_ignored_signals = [
+            value for value in (signal.SIGHUP, signal.SIGINT,)
+            if signal.getsignal(value) == signal.SIG_IGN]
+        if signal.SIGHUP in self.restore_ignored_signals:
+            # sighup should be set to default handler
+            signal.signal(signal.SIGHUP, signal.SIG_DFL)
+        if signal.SIGINT in self.restore_ignored_signals:
+            # SIGINT should be set to signal.default_int_handler
+            signal.signal(signal.SIGINT, signal.default_int_handler)
         unittest.TestCase.setUp(self)
 
     def tearDown(self):
-        os.chdir (self.original_path)
+        # restore original working folder
+        os.chdir(self.original_path)
+
+        if self.pid != os.getpid():
+            # The build server pattern-matches phrase 'Test runner has forked!'
+            print("Test runner has forked! This means a child process raised "
+                  "an exception before exec() in a test case, the error is "
+                  "more than likely found above this line in stderr.",
+                  file=sys.stderr)
+            exit(1)
+
+        # restore signal handlers
+        for signal_value in self.restore_ignored_signals:
+            signal.signal(signal_value, signal.SIG_IGN)
 
-    if sys.version_info < (2,7):
+    if sys.version_info < (2, 7):
         # We want to use these methods, which are new/improved in 2.7, but
         # we are still supporting 2.6 for the moment. This section can be
         # removed when we drop Python 2.6 support.
diff --git a/tests/README b/tests/README
index 295632b..ef5b613 100644
--- a/tests/README
+++ b/tests/README
@@ -1,18 +1,8 @@
 
-The best way to run these tests is from the directory above this one. Source
-the test.env environment file. This will make sure that you are using the
-correct pexpect.py file otherwise Python might try to import a different
-version if it is already installed in this environment. Then run the testall.py
-script in the tools/ directory. This script will automatically build a test
-suite from all the test scripts in the tests/ directory. This allows you to add
-new test scripts simply by dropping them in the tests/ directory. You don't
-have to register the test or do anything else to integrate it into the test
-suite.
+The best way to run these tests is from the directory above this one. Run:
 
-For example, this is the normal set of commands you would use to run all tests
-in the tests/ directory:
+    py.test
 
-    $ cd /home/user/pexpect_dev/
-    $ . test.env
-    $ ./tools/testall.py
+To run a specific test file:
 
+    py.test tests/test_constructor.py
diff --git a/tests/test_ansi.py b/tests/test_ansi.py
index 3b8d6a9..a9d445e 100755
--- a/tests/test_ansi.py
+++ b/tests/test_ansi.py
@@ -21,6 +21,9 @@ PEXPECT LICENSE
 from pexpect import ANSI
 import unittest
 from . import PexpectTestCase
+import sys
+
+PY3 = (sys.version_info[0] >= 3)
 
 write_target = 'I\'ve got a ferret sticking up my nose.                           \n' +\
 '(He\'s got a ferret sticking up his nose.)                        \n' +\
@@ -142,10 +145,81 @@ class ansiTestCase (PexpectTestCase.PexpectTestCase):
 
     def test_number_x(self):
         """Test the FSM state used to handle more than 2 numeric parameters."""
-        s = ANSI.ANSI(1, 20)
+        class TestANSI(ANSI.ANSI):
+            captured_memory = None
+            def do_sgr(self, fsm):
+                assert self.captured_memory is None
+                self.captured_memory = fsm.memory
+
+        s = TestANSI(1, 20)
         s.write('\x1b[0;1;32;45mtest')
         assert str(s) == ('test                ')
-        assert(s.state.memory == [s, '0', '1', '32', '45'])
+        assert s.captured_memory is not None
+        assert s.captured_memory == [s, '0', '1', '32', '45']
+
+    def test_fsm_memory(self):
+        """Test the FSM stack/memory does not have numbers left on it
+        after some sequences with numbers are passed in."""
+        s = ANSI.ANSI(1, 20)
+        s.write('\x1b[0;1;2;3m\x1b[4;5;6;7q\x1b[?8h\x1b[?9ltest')
+        assert str(s) == ('test                ')
+        assert s.state.memory == [s]
+
+    def test_utf8_bytes(self):
+        """Test that when bytes are passed in containing UTF-8 encoded
+        characters, where the encoding of each character consists of
+        multiple bytes, the characters are correctly decoded.
+        Incremental decoding is also tested."""
+        s = ANSI.ANSI(2, 10, encoding='utf-8')
+        # This is the UTF-8 encoding of the UCS character "HOURGLASS"
+        # followed by the UTF-8 encoding of the UCS character
+        # "KEYBOARD".  These characters can't be encoded in cp437 or
+        # latin-1.  The "KEYBOARD" character is split into two
+        # separate writes.
+        s.write(b'\xe2\x8c\x9b')
+        s.write(b'\xe2\x8c')
+        s.write(b'\xa8')
+        if PY3:
+            assert str(s) == u'\u231b\u2328        \n          '
+        else:
+            assert unicode(s) == u'\u231b\u2328        \n          '
+            assert str(s) == b'\xe2\x8c\x9b\xe2\x8c\xa8        \n          '
+        assert s.dump() == u'\u231b\u2328                  '
+        assert s.pretty() == u'+----------+\n|\u231b\u2328        |\n|          |\n+----------+\n'
+        assert s.get_abs(1, 1) == u'\u231b'
+        assert s.get_region(1, 1, 1, 5) == [u'\u231b\u2328   ']
+
+    def test_unicode(self):
+        """Test passing in of a unicode string."""
+        s = ANSI.ANSI(2, 10, encoding="utf-8")
+        s.write(u'\u231b\u2328')
+        if PY3:
+            assert str(s) == u'\u231b\u2328        \n          '
+        else:
+            assert unicode(s) == u'\u231b\u2328        \n          '
+            assert str(s) == b'\xe2\x8c\x9b\xe2\x8c\xa8        \n          '
+        assert s.dump() == u'\u231b\u2328                  '
+        assert s.pretty() == u'+----------+\n|\u231b\u2328        |\n|          |\n+----------+\n'
+        assert s.get_abs(1, 1) == u'\u231b'
+        assert s.get_region(1, 1, 1, 5) == [u'\u231b\u2328   ']
+
+    def test_decode_error(self):
+        """Test that default handling of decode errors replaces the
+        invalid characters."""
+        s = ANSI.ANSI(2, 10, encoding="ascii")
+        s.write(b'\xff') # a non-ASCII character
+        # In unicode, the non-ASCII character is replaced with
+        # REPLACEMENT CHARACTER.
+        if PY3:
+            assert str(s) == u'\ufffd         \n          '
+        else:
+            assert unicode(s) == u'\ufffd         \n          '
+            assert str(s) == b'?         \n          '
+        assert s.dump() == u'\ufffd                   '
+        assert s.pretty() == u'+----------+\n|\ufffd         |\n|          |\n+----------+\n'
+        assert s.get_abs(1, 1) == u'\ufffd'
+        assert s.get_region(1, 1, 1, 5) == [u'\ufffd    ']
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/test_async.py b/tests/test_async.py
new file mode 100644
index 0000000..ce75572
--- /dev/null
+++ b/tests/test_async.py
@@ -0,0 +1,51 @@
+try:
+    import asyncio
+except ImportError:
+    asyncio = None
+
+import sys
+import unittest
+
+import pexpect
+from .PexpectTestCase import PexpectTestCase
+
+def run(coro):
+    return asyncio.get_event_loop().run_until_complete(coro)
+
+@unittest.skipIf(asyncio is None, "Requires asyncio")
+class AsyncTests(PexpectTestCase):
+    def test_simple_expect(self):
+        p = pexpect.spawn('cat')
+        p.sendline('Hello asyncio')
+        coro = p.expect(['Hello', pexpect.EOF] , async=True)
+        assert run(coro) == 0
+        print('Done')
+
+    def test_timeout(self):
+        p = pexpect.spawn('cat')
+        coro = p.expect('foo', timeout=1, async=True)
+        with self.assertRaises(pexpect.TIMEOUT):
+            run(coro)
+        
+        p = pexpect.spawn('cat')
+        coro = p.expect(['foo', pexpect.TIMEOUT], timeout=1, async=True)
+        assert run(coro) == 1
+
+    def test_eof(self):
+        p = pexpect.spawn('cat')
+        p.sendline('Hi')
+        coro = p.expect(pexpect.EOF, async=True)
+        p.sendeof()
+        assert run(coro) == 0
+
+        p = pexpect.spawn('cat')
+        p.sendeof()
+        coro = p.expect('Blah', async=True)
+        with self.assertRaises(pexpect.EOF):
+            run(coro)
+    
+    def test_expect_exact(self):
+        p = pexpect.spawn('%s list100.py' % sys.executable)
+        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
diff --git a/tests/test_constructor.py b/tests/test_constructor.py
index 60525a0..98c473a 100755
--- a/tests/test_constructor.py
+++ b/tests/test_constructor.py
@@ -28,11 +28,11 @@ class TestCaseConstructor(PexpectTestCase.PexpectTestCase):
         the same results for different styles of invoking __init__().
         This assumes that the root directory / is static during the test.
         '''
-        p1 = pexpect.spawn('/bin/ls -l /bin')
-        p2 = pexpect.spawn('/bin/ls' ,['-l', '/bin'])
-        p1.expect (pexpect.EOF)
-        p2.expect (pexpect.EOF)
-        assert (p1.before == p2.before)
+        p1 = pexpect.spawn('uname -m -n -p -r -s -v')
+        p2 = pexpect.spawn('uname', ['-m', '-n', '-p', '-r', '-s', '-v'])
+        p1.expect(pexpect.EOF)
+        p2.expect(pexpect.EOF)
+        assert p1.before == p2.before
 
     def test_named_parameters (self):
         '''This tests that named parameters work.
diff --git a/tests/test_ctrl_chars.py b/tests/test_ctrl_chars.py
index 9c7b869..10d03db 100755
--- a/tests/test_ctrl_chars.py
+++ b/tests/test_ctrl_chars.py
@@ -26,6 +26,9 @@ from . import PexpectTestCase
 import time
 import sys
 
+from ptyprocess import ptyprocess
+ptyprocess._make_eof_intr()
+
 if sys.version_info[0] >= 3:
     def byte(i):
         return bytes([i])
@@ -54,7 +57,7 @@ class TestCtrlChars(PexpectTestCase.PexpectTestCase):
         child = pexpect.spawn('python getch.py', echo=False, timeout=5)
         child.expect('READY')
         child.sendintr()
-        child.expect(str(child._INTR) + '<STOP>')
+        child.expect(str(ord(ptyprocess._INTR)) + '<STOP>')
 
         child.send(byte(0))
         child.expect('0<STOP>')
@@ -66,7 +69,7 @@ class TestCtrlChars(PexpectTestCase.PexpectTestCase):
         child = pexpect.spawn('python getch.py', echo=False, timeout=5)
         child.expect('READY')
         child.sendeof()
-        child.expect(str(child._EOF) + '<STOP>')
+        child.expect(str(ord(ptyprocess._EOF)) + '<STOP>')
 
         child.send(byte(0))
         child.expect('0<STOP>')
diff --git a/tests/test_expect.py b/tests/test_expect.py
index 8ccb9c5..3f4c9d8 100755
--- a/tests/test_expect.py
+++ b/tests/test_expect.py
@@ -18,11 +18,13 @@ PEXPECT LICENSE
     OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 
 '''
+import multiprocessing
 import unittest
 import subprocess
 import time
 import signal
 import sys
+import os
 
 import pexpect
 from . import PexpectTestCase
@@ -542,7 +544,40 @@ class ExpectTestCase (PexpectTestCase.PexpectTestCase):
         signal.alarm(1)
         p1.expect('END')
 
+    def test_stdin_closed(self):
+        '''
+        Ensure pexpect continues to operate even when stdin is closed
+        '''
+        class Closed_stdin_proc(multiprocessing.Process):
+            def run(self):
+                sys.__stdin__.close()
+                cat = pexpect.spawn('cat')
+                cat.sendeof()
+                cat.expect(pexpect.EOF)
+
+        proc = Closed_stdin_proc()
+        proc.start()
+        proc.join()
+        assert proc.exitcode == 0
+
+    def test_stdin_stdout_closed(self):
+        '''
+        Ensure pexpect continues to operate even when stdin and stdout is closed
+        '''
+        class Closed_stdin_stdout_proc(multiprocessing.Process):
+            def run(self):
+                sys.__stdin__.close()
+                sys.__stdout__.close()
+                cat = pexpect.spawn('cat')
+                cat.sendeof()
+                cat.expect(pexpect.EOF)
+
+        proc = Closed_stdin_stdout_proc()
+        proc.start()
+        proc.join()
+        assert proc.exitcode == 0
+
 if __name__ == '__main__':
     unittest.main()
 
-suite = unittest.makeSuite(ExpectTestCase,'test')
+suite = unittest.makeSuite(ExpectTestCase, 'test')
diff --git a/tests/test_interact.py b/tests/test_interact.py
index 06fc44a..e635cb0 100755
--- a/tests/test_interact.py
+++ b/tests/test_interact.py
@@ -66,9 +66,12 @@ class InteractTestCase (PexpectTestCase.PexpectTestCase):
         p.expect(b'<out>alpha')
         p.expect(b'<out>beta')
         p.sendeof()
-        p.expect_exact('<eof>')
-        p.expect_exact('Escaped interact')
-        p.expect(pexpect.EOF)
+        # strangely, on travis-ci, sendeof() terminates the subprocess,
+        # it doesn't receive ^D, just immediately throws EOF.
+        idx = p.expect_exact(['<eof>', pexpect.EOF])
+        if idx == 0:
+            p.expect_exact('Escaped interact')
+            p.expect(pexpect.EOF)
         assert not p.isalive()
         assert p.exitstatus == 0
 
@@ -81,9 +84,12 @@ class InteractTestCase (PexpectTestCase.PexpectTestCase):
         p.expect('<out>ɑlpha')
         p.expect('<out>Βeta')
         p.sendeof()
-        p.expect_exact('<eof>')
-        p.expect_exact('Escaped interact')
-        p.expect(pexpect.EOF)
+        # strangely, on travis-ci, sendeof() terminates the subprocess,
+        # it doesn't receive ^D, just immediately throws EOF.
+        idx = p.expect_exact(['<eof>', pexpect.EOF])
+        if idx == 0:
+            p.expect_exact('Escaped interact')
+            p.expect(pexpect.EOF)
         assert not p.isalive()
         assert p.exitstatus == 0
 
diff --git a/tests/test_maxcanon.py b/tests/test_maxcanon.py
new file mode 100644
index 0000000..772a3b7
--- /dev/null
+++ b/tests/test_maxcanon.py
@@ -0,0 +1,176 @@
+""" Module for canonical-mode tests. """
+# std imports
+import sys
+import os
+
+# local
+import pexpect
+from . import PexpectTestCase
+
+# 3rd-party
+import pytest
+
+
+class TestCaseCanon(PexpectTestCase.PexpectTestCase):
+    """
+    Test expected Canonical mode behavior (limited input line length).
+
+    All systems use the value of MAX_CANON which can be found using
+    fpathconf(3) value PC_MAX_CANON -- with the exception of Linux
+    and FreeBSD.
+
+    Linux, though defining a value of 255, actually honors the value
+    of 4096 from linux kernel include file tty.h definition
+    N_TTY_BUF_SIZE.
+
+    Linux also does not honor IMAXBEL. termios(3) states, "Linux does not
+    implement this bit, and acts as if it is always set." Although these
+    tests ensure it is enabled, this is a non-op for Linux.
+
+    FreeBSD supports neither, and instead uses a fraction (1/5) of the tty
+    speed which is always 9600.  Therefor, the maximum limited input line
+    length is 9600 / 5 = 1920.
+
+    These tests only ensure the correctness of the behavior described by
+    the sendline() docstring. pexpect is not particularly involved in
+    these scenarios, though if we wish to expose some kind of interface
+    to tty.setraw, for example, these tests may be re-purposed as such.
+
+    Lastly, portions of these tests are skipped on Travis-CI. It produces
+    unexpected behavior not reproduced on Debian/GNU Linux.
+    """
+
+    def setUp(self):
+        super(TestCaseCanon, self).setUp()
+
+        self.echo = False
+        if sys.platform.lower().startswith('linux'):
+            # linux is 4096, N_TTY_BUF_SIZE.
+            self.max_input = 4096
+            self.echo = True
+        elif sys.platform.lower().startswith('sunos'):
+            # SunOS allows PC_MAX_CANON + 1; see
+            # https://bitbucket.org/illumos/illumos-gate/src/d07a59219ab7fd2a7f39eb47c46cf083c88e932f/usr/src/uts/common/io/ldterm.c?at=default#cl-1888
+            self.max_input = os.fpathconf(0, 'PC_MAX_CANON') + 1
+        elif sys.platform.lower().startswith('freebsd'):
+            # http://lists.freebsd.org/pipermail/freebsd-stable/2009-October/052318.html
+            self.max_input = 9600 / 5
+        else:
+            # All others (probably) limit exactly at PC_MAX_CANON
+            self.max_input = os.fpathconf(0, 'PC_MAX_CANON')
+
+    @pytest.mark.skipif(
+        sys.platform.lower().startswith('freebsd'),
+        reason='os.write to BLOCK indefinitely on FreeBSD in this case'
+    )
+    def test_under_max_canon(self):
+        " BEL is not sent by terminal driver at maximum bytes - 1. "
+        # given,
+        child = pexpect.spawn('bash', echo=self.echo, timeout=5)
+        child.sendline('echo READY')
+        child.sendline('stty icanon imaxbel')
+        child.sendline('echo BEGIN; cat')
+
+        # some systems BEL on (maximum - 1), not able to receive CR,
+        # even though all characters up until then were received, they
+        # simply cannot be transmitted, as CR is part of the transmission.
+        send_bytes = self.max_input - 1
+
+        # exercise,
+        child.sendline('_' * send_bytes)
+
+        # fast forward beyond 'cat' command, as ^G can be found as part of
+        # set-xterm-title sequence of $PROMPT_COMMAND or $PS1.
+        child.expect_exact('BEGIN')
+
+        # verify, all input is found in echo output,
+        child.expect_exact('_' * send_bytes)
+
+        # BEL is not found,
+        with self.assertRaises(pexpect.TIMEOUT):
+            child.expect_exact('\a', timeout=1)
+
+        # cleanup,
+        child.sendeof()           # exit cat(1)
+        child.sendline('exit 0')  # exit bash(1)
+        child.expect(pexpect.EOF)
+        assert not child.isalive()
+        assert child.exitstatus == 0
+
+    @pytest.mark.skipif(
+        sys.platform.lower().startswith('freebsd'),
+        reason='os.write to BLOCK indefinitely on FreeBSD in this case'
+    )
+    def test_beyond_max_icanon(self):
+        " a single BEL is sent when maximum bytes is reached. "
+        # given,
+        child = pexpect.spawn('bash', echo=self.echo, timeout=5)
+        child.sendline('stty icanon imaxbel erase ^H')
+        child.sendline('cat')
+        send_bytes = self.max_input
+
+        # exercise,
+        child.sendline('_' * send_bytes)
+        child.expect_exact('\a')
+
+        # exercise, we must now backspace to send CR.
+        child.sendcontrol('h')
+        child.sendline()
+
+        if os.environ.get('TRAVIS', None) == 'true':
+            # Travis-CI has intermittent behavior here, possibly
+            # because the master process is itself, a PTY?
+            return
+
+        # verify the length of (maximum - 1) received by cat(1),
+        # which has written it back out,
+        child.expect_exact('_' * (send_bytes - 1))
+        # and not a byte more.
+        with self.assertRaises(pexpect.TIMEOUT):
+            child.expect_exact('_', timeout=1)
+
+        # cleanup,
+        child.sendeof()           # exit cat(1)
+        child.sendline('exit 0')  # exit bash(1)
+        child.expect_exact(pexpect.EOF)
+        assert not child.isalive()
+        assert child.exitstatus == 0
+
+    @pytest.mark.skipif(
+        sys.platform.lower().startswith('freebsd'),
+        reason='os.write to BLOCK indefinitely on FreeBSD in this case'
+    )
+    def test_max_no_icanon(self):
+        " may exceed maximum input bytes if canonical mode is disabled. "
+        # given,
+        child = pexpect.spawn('bash', echo=self.echo, timeout=5)
+        child.sendline('stty -icanon imaxbel')
+        child.sendline('echo BEGIN; cat')
+        send_bytes = self.max_input + 11
+
+        # exercise,
+        child.sendline('_' * send_bytes)
+
+        # fast forward beyond 'cat' command, as ^G can be found as part of
+        # set-xterm-title sequence of $PROMPT_COMMAND or $PS1.
+        child.expect_exact('BEGIN')
+
+        if os.environ.get('TRAVIS', None) == 'true':
+            # Travis-CI has intermittent behavior here, possibly
+            # because the master process is itself, a PTY?
+            return
+
+        # BEL is *not* found,
+        with self.assertRaises(pexpect.TIMEOUT):
+            child.expect_exact('\a', timeout=1)
+
+        # verify, all input is found in output,
+        child.expect_exact('_' * send_bytes)
+
+        # cleanup,
+        child.sendcontrol('c')    # exit cat(1) (eof wont work in -icanon)
+        child.sendcontrol('c')
+        child.sendline('exit 0')  # exit bash(1)
+        child.expect(pexpect.EOF)
+        assert not child.isalive()
+        assert child.exitstatus == 0
diff --git a/tests/test_misc.py b/tests/test_misc.py
index d9205e4..e439240 100755
--- a/tests/test_misc.py
+++ b/tests/test_misc.py
@@ -178,6 +178,16 @@ class TestCaseMisc(PexpectTestCase.PexpectTestCase):
         with self.assertRaises(pexpect.EOF):
             child.expect('the unexpected')
 
+    def test_with(self):
+        "spawn can be used as a context manager"
+        with pexpect.spawn(sys.executable + ' echo_w_prompt.py') as p:
+            p.expect('<in >')
+            p.sendline(b'alpha')
+            p.expect(b'<out>alpha')
+            assert p.isalive()
+        
+        assert not p.isalive()
+
     def test_terminate(self):
         " test force terminate always succeeds (SIGKILL). "
         child = pexpect.spawn('cat')
@@ -186,41 +196,24 @@ class TestCaseMisc(PexpectTestCase.PexpectTestCase):
 
     def test_sighup(self):
         " validate argument `ignore_sighup=True` and `ignore_sighup=False`. "
-        # If a parent process sets an Ignore handler for SIGHUP (as on Fedora's
-        # build machines), this test breaks. We temporarily restore the default
-        # handler, so the child process will quit. However, we can't simply
-        # replace any installed handler, because getsignal returns None for
-        # handlers not set in Python code, so we wouldn't be able to restore
-        # them.
-        if signal.getsignal(signal.SIGHUP) == signal.SIG_IGN:
-            signal.signal(signal.SIGHUP, signal.SIG_DFL)
-            restore_sig_ign = True
-        else:
-            restore_sig_ign = False
-
         getch = sys.executable + ' getch.py'
-        try:
-            child = pexpect.spawn(getch, ignore_sighup=True)
-            child.expect('READY')
-            child.kill(signal.SIGHUP)
-            for _ in range(10):
-                if not child.isalive():
-                    self.fail('Child process should not have exited.')
-                time.sleep(0.1)
-
-            child = pexpect.spawn(getch, ignore_sighup=False)
-            child.expect('READY')
-            child.kill(signal.SIGHUP)
-            for _ in range(10):
-                if not child.isalive():
-                    break
-                time.sleep(0.1)
-            else:
-                self.fail('Child process should have exited.')
-
-        finally:
-            if restore_sig_ign:
-                signal.signal(signal.SIGHUP, signal.SIG_IGN)
+        child = pexpect.spawn(getch, ignore_sighup=True)
+        child.expect('READY')
+        child.kill(signal.SIGHUP)
+        for _ in range(10):
+            if not child.isalive():
+                self.fail('Child process should not have exited.')
+            time.sleep(0.1)
+
+        child = pexpect.spawn(getch, ignore_sighup=False)
+        child.expect('READY')
+        child.kill(signal.SIGHUP)
+        for _ in range(10):
+            if not child.isalive():
+                break
+            time.sleep(0.1)
+        else:
+            self.fail('Child process should have exited.')
 
     def test_bad_child_pid(self):
         " assert bad condition error in isalive(). "
@@ -228,7 +221,7 @@ class TestCaseMisc(PexpectTestCase.PexpectTestCase):
         child = pexpect.spawn('cat')
         child.terminate(force=1)
         # Force an invalid state to test isalive
-        child.terminated = 0
+        child.ptyproc.terminated = 0
         try:
             with self.assertRaisesRegexp(pexpect.ExceptionPexpect,
                                          ".*" + expect_errmsg):
@@ -361,9 +354,9 @@ class TestCaseMisc(PexpectTestCase.PexpectTestCase):
         " test forced self.__fork_pty() and __pty_make_controlling_tty "
         # given,
         class spawn_ourptyfork(pexpect.spawn):
-            def _spawn(self, command, args=[]):
+            def _spawn(self, command, args=[], preexec_fn=None):
                 self.use_native_pty_fork = False
-                pexpect.spawn._spawn(self, command, args)
+                pexpect.spawn._spawn(self, command, args, preexec_fn)
 
         # exercise,
         p = spawn_ourptyfork('cat', echo=False)
diff --git a/tests/test_replwrap.py b/tests/test_replwrap.py
index 14f7c39..28c7599 100644
--- a/tests/test_replwrap.py
+++ b/tests/test_replwrap.py
@@ -26,7 +26,7 @@ class REPLWrapTestCase(unittest.TestCase):
         assert 'real' in res, res
 
         # PAGER should be set to cat, otherwise man hangs
-        res = bash.run_command('man sleep', timeout=2)
+        res = bash.run_command('man sleep', timeout=5)
         assert 'SLEEP' in res, res
 
     def test_multiline(self):
diff --git a/tests/test_repr.py b/tests/test_repr.py
new file mode 100644
index 0000000..ce618d4
--- /dev/null
+++ b/tests/test_repr.py
@@ -0,0 +1,26 @@
+""" Test __str__ methods. """
+import pexpect
+
+from . import PexpectTestCase
+
+
+class TestCaseMisc(PexpectTestCase.PexpectTestCase):
+
+    def test_str_spawnu(self):
+        """ Exercise spawnu.__str__() """
+        # given,
+        p = pexpect.spawnu('cat')
+        # exercise,
+        value = str(p)
+        # verify
+        assert isinstance(value, str)
+
+    def test_str_spawn(self):
+        """ Exercise spawn.__str__() """
+        # given,
+        p = pexpect.spawn('cat')
+        # exercise,
+        value = str(p)
+        # verify
+        assert isinstance(value, str)
+
diff --git a/tests/test_run.py b/tests/test_run.py
index 814b70a..1b3c92f 100755
--- a/tests/test_run.py
+++ b/tests/test_run.py
@@ -22,72 +22,170 @@ PEXPECT LICENSE
 import pexpect
 import unittest
 import subprocess
+import tempfile
 import sys
+import os
 from . import PexpectTestCase
 
-# TODO Many of these test cases blindly assume that sequential
-# TODO listing of the /bin directory will yield the same results.
-# 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"]>3:
+
+def timeout_callback(values):
+    if values["event_count"] > 3:
         return 1
     return 0
 
+
+def function_events_callback(values):
+    try:
+        previous_echoed = (values["child_result_list"][-1]
+                           .decode().split("\n")[-2].strip())
+        if previous_echoed.endswith("stage-1"):
+            return "echo stage-2\n"
+        elif previous_echoed.endswith("stage-2"):
+            return "echo stage-3\n"
+        elif previous_echoed.endswith("stage-3"):
+            return "exit\n"
+        else:
+            raise Exception("Unexpected output {0}".format(previous_echoed))
+    except IndexError:
+        return "echo stage-1\n"
+
+
 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):
+    def setUp(self):
+        fd, self.rcfile = tempfile.mkstemp()
+        os.write(fd, b'PS1=GO: \n')
+        os.close(fd)
+        super(RunFuncTestCase, self).setUp()
+
+    def tearDown(self):
+        os.unlink(self.rcfile)
+        super(RunFuncTestCase, self).tearDown()
+
+    def test_run_exit(self):
         (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) = self.runfunc('ls -l /bin', withexitstatus=1)
+    def test_run(self):
+        the_old_way = subprocess.Popen(
+            args=['uname', '-m', '-n'],
+            stdout=subprocess.PIPE
+        ).communicate()[0].rstrip()
+
+        (the_new_way, exitstatus) = self.runfunc(
+            'uname -m -n', 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...
-        self.runfunc("cat", timeout=1, events={pexpect.TIMEOUT:timeout_callback})
+    def test_run_callback(self):
+        # TODO it seems like this test could block forever if run fails...
+        events = {pexpect.TIMEOUT: timeout_callback}
+        self.runfunc("cat", timeout=1, events=events)
 
-    def test_run_bad_exitstatus (self):
-        (the_new_way, exitstatus) = self.runfunc('ls -l /najoeufhdnzkxjd',
-                                                    withexitstatus=1)
+    def test_run_bad_exitstatus(self):
+        (the_new_way, exitstatus) = self.runfunc(
+            'ls -l /najoeufhdnzkxjd', withexitstatus=1)
         assert exitstatus != 0
 
+    def test_run_event_as_string(self):
+        events = [
+            # second match on 'abc', echo 'def'
+            ('abc\r\n.*GO:', 'echo "def"\n'),
+            # final match on 'def': exit
+            ('def\r\n.*GO:', 'exit\n'),
+            # first match on 'GO:' prompt, echo 'abc'
+            ('GO:', 'echo "abc"\n')
+        ]
+
+        (data, exitstatus) = pexpect.run(
+            'bash --rcfile {0}'.format(self.rcfile),
+            withexitstatus=True,
+            events=events,
+            timeout=10)
+        assert exitstatus == 0
+
+    def test_run_event_as_function(self):
+        events = [
+            ('GO:', function_events_callback)
+        ]
+
+        (data, exitstatus) = pexpect.run(
+            'bash --rcfile {0}'.format(self.rcfile),
+            withexitstatus=True,
+            events=events,
+            timeout=10)
+        assert exitstatus == 0
+
+    def test_run_event_as_method(self):
+        events = [
+            ('GO:', self._method_events_callback)
+        ]
+
+        (data, exitstatus) = pexpect.run(
+            'bash --rcfile {0}'.format(self.rcfile),
+            withexitstatus=True,
+            events=events,
+            timeout=10)
+        assert exitstatus == 0
+
+    def test_run_event_typeerror(self):
+        events = [('GO:', -1)]
+        with self.assertRaises(TypeError):
+            pexpect.run('bash --rcfile {0}'.format(self.rcfile),
+                        withexitstatus=True,
+                        events=events,
+                        timeout=10)
+
+    def _method_events_callback(self, values):
+        try:
+            previous_echoed = (values["child_result_list"][-1].decode()
+                               .split("\n")[-2].strip())
+            if previous_echoed.endswith("foo1"):
+                return "echo foo2\n"
+            elif previous_echoed.endswith("foo2"):
+                return "echo foo3\n"
+            elif previous_echoed.endswith("foo3"):
+                return "exit\n"
+            else:
+                raise Exception("Unexpected output {0!r}"
+                                .format(previous_echoed))
+        except IndexError:
+            return "echo foo1\n"
+
+
 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)   # þ
+            char = chr(254)   # þ
             pattern = '<in >'
         else:
-            c = unichr(254)  # analysis:ignore
+            char = unichr(254)  # analysis:ignore
             pattern = '<in >'.decode('ascii')
 
-        def callback(d):
-            if d['event_count'] == 0:
-                return c + '\n'
+        def callback(values):
+            if values['event_count'] == 0:
+                return char + '\n'
             else:
                 return True  # Stop the child process
 
         output = pexpect.runu(sys.executable + ' echo_w_prompt.py',
-                              env={'PYTHONIOENCODING':'utf-8'},
-                              events={pattern:callback})
+                              env={'PYTHONIOENCODING': 'utf-8'},
+                              events={pattern: callback})
         assert isinstance(output, unicode_type), type(output)
-        assert '<out>'+c in output, output
+        assert ('<out>' + char) in output, output
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/test_screen.py b/tests/test_screen.py
index 3f0736b..2429e57 100755
--- a/tests/test_screen.py
+++ b/tests/test_screen.py
@@ -19,10 +19,14 @@ PEXPECT LICENSE
 
 '''
 
+import sys
+
 from pexpect import screen
 import unittest
 from . import PexpectTestCase
 
+PY3 = (sys.version_info[0] >= 3)
+
 fill1_target='XXXXXXXXXX\n' + \
 'XOOOOOOOOX\n' + \
 'XO::::::OX\n' + \
@@ -76,6 +80,17 @@ insert_target = 'ZXZZZZZZXZ\n' +\
 'ZZ/2.4.6ZZ'
 get_region_target = ['......', '.\\/...', './\\...', '......']
 
+unicode_box_unicode_result = u'\u2554\u2557\n\u255A\u255D'
+unicode_box_pretty_result = u'''\
++--+
+|\u2554\u2557|
+|\u255A\u255D|
++--+
+'''
+unicode_box_ascii_bytes_result = b'??\n??'
+unicode_box_cp437_bytes_result = b'\xc9\xbb\n\xc8\xbc'
+unicode_box_utf8_bytes_result = b'\xe2\x95\x94\xe2\x95\x97\n\xe2\x95\x9a\xe2\x95\x9d'
+
 class screenTestCase (PexpectTestCase.PexpectTestCase):
     def make_screen_with_put (self):
         s = screen.screen(10,10)
@@ -168,20 +183,101 @@ class screenTestCase (PexpectTestCase.PexpectTestCase):
         s.insert_abs (10,9,'Z')
         s.insert_abs (10,9,'Z')
         assert str(s) == insert_target
- #   def test_write (self):
- #       s = screen.screen (6,65)
- #       s.fill('.')
- #       s.cursor_home()
- #       for c in write_text:
- #           s.write (c)
- #       print str(s)
- #       assert str(s) == write_target
- #   def test_tetris (self):
- #       s = screen.screen (24,80)
- #       tetris_text = open ('tetris.data').read()
- #       for c in tetris_text:
- #           s.write (c)
- #       assert str(s) == tetris_target
+
+    def make_screen_with_box_unicode(self, *args, **kwargs):
+        '''Creates a screen containing a box drawn using double-line
+        line drawing characters. The characters are fed in as
+        unicode. '''
+        s = screen.screen (2,2,*args,**kwargs)
+        s.put_abs (1,1,u'\u2554')
+        s.put_abs (1,2,u'\u2557')
+        s.put_abs (2,1,u'\u255A')
+        s.put_abs (2,2,u'\u255D')
+        return s
+
+    def make_screen_with_box_cp437(self, *args, **kwargs):
+        '''Creates a screen containing a box drawn using double-line
+        line drawing characters. The characters are fed in as
+        CP437. '''
+        s = screen.screen (2,2,*args,**kwargs)
+        s.put_abs (1,1,b'\xc9')
+        s.put_abs (1,2,b'\xbb')
+        s.put_abs (2,1,b'\xc8')
+        s.put_abs (2,2,b'\xbc')
+        return s
+
+    def make_screen_with_box_utf8(self, *args, **kwargs):
+        '''Creates a screen containing a box drawn using double-line
+        line drawing characters. The characters are fed in as
+        UTF-8. '''
+        s = screen.screen (2,2,*args,**kwargs)
+        s.put_abs (1,1,b'\xe2\x95\x94')
+        s.put_abs (1,2,b'\xe2\x95\x97')
+        s.put_abs (2,1,b'\xe2\x95\x9a')
+        s.put_abs (2,2,b'\xe2\x95\x9d')
+        return s
+
+    def test_unicode_ascii (self):
+        # With the default encoding set to ASCII, we should still be
+        # able to feed in unicode strings and get them back out:
+        s = self.make_screen_with_box_unicode('ascii')
+        if PY3:
+            assert str(s) == unicode_box_unicode_result
+        else:
+            assert unicode(s) == unicode_box_unicode_result
+            # And we should still get something for Python 2 str(), though
+            # it might not be very useful
+            str(s)
+
+        assert s.pretty() == unicode_box_pretty_result
+
+    def test_decoding_errors(self):
+        # With strict error handling, it should reject bytes it can't decode
+        with self.assertRaises(UnicodeDecodeError):
+            self.make_screen_with_box_cp437('ascii', 'strict')
+
+        # replace should turn them into unicode replacement characters, U+FFFD
+        s = self.make_screen_with_box_cp437('ascii', 'replace')
+        expected = u'\ufffd\ufffd\n\ufffd\ufffd'
+        if PY3:
+            assert str(s) == expected
+        else:
+            assert unicode(s) == expected
+
+    def test_unicode_cp437 (self):
+        # Verify decoding from and re-encoding to CP437.
+        s = self.make_screen_with_box_cp437('cp437','strict')
+        if PY3:
+            assert str(s) == unicode_box_unicode_result
+        else:
+            assert unicode(s) == unicode_box_unicode_result
+            assert str(s) == unicode_box_cp437_bytes_result
+        assert s.pretty() == unicode_box_pretty_result
+
+    def test_unicode_utf8 (self):
+        # Verify decoding from and re-encoding to UTF-8.
+        s = self.make_screen_with_box_utf8('utf-8','strict')
+        if PY3:
+            assert str(s) == unicode_box_unicode_result
+        else:
+            assert unicode(s) == unicode_box_unicode_result
+            assert str(s) == unicode_box_utf8_bytes_result
+        assert s.pretty() == unicode_box_pretty_result
+
+    def test_no_bytes(self):
+        s = screen.screen(2, 2, encoding=None)
+        s.put_abs(1, 1, u'A')
+        s.put_abs(2, 2, u'D')
+
+        with self.assertRaises(TypeError):
+            s.put_abs(1, 2, b'B')
+
+        if PY3:
+            assert str(s) == u'A \n D'
+        else:
+            assert unicode(s) == u'A \n D'
+            # This will still work if it's limited to ascii
+            assert str(s) == b'A \n D'
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/test_which.py b/tests/test_which.py
index 83575fb..bda3333 100644
--- a/tests/test_which.py
+++ b/tests/test_which.py
@@ -1,9 +1,14 @@
+import subprocess
 import tempfile
+import shutil
+import errno
 import os
 
 import pexpect
 from . import PexpectTestCase
 
+import pytest
+
 
 class TestCaseWhich(PexpectTestCase.PexpectTestCase):
     " Tests for pexpect.which(). "
@@ -162,27 +167,101 @@ class TestCaseWhich(PexpectTestCase.PexpectTestCase):
         try:
             # setup
             os.environ['PATH'] = bin_dir
-            with open(bin_path, 'w') as fp:
-                fp.write('#!/bin/sh\necho hello, world\n')
-            for should_match, mode in ((False, 0o000),
-                                       (True,  0o005),
-                                       (True,  0o050),
-                                       (True,  0o500),
-                                       (False, 0o004),
-                                       (False, 0o040),
-                                       (False, 0o400)):
+
+            # an interpreted script requires the ability to read,
+            # whereas a binary program requires only to be executable.
+            #
+            # to gain access to a binary program, we make a copy of
+            # the existing system program echo(1).
+            bin_echo = None
+            for pth in ('/bin/echo', '/usr/bin/echo'):
+                if os.path.exists(pth):
+                    bin_echo = pth
+                    break
+            bin_which = None
+            for pth in ('/bin/which', '/usr/bin/which'):
+                if os.path.exists(pth):
+                    bin_which = pth
+                    break
+            if not bin_echo or not bin_which:
+                pytest.skip('needs `echo` and `which` binaries')
+            shutil.copy(bin_echo, bin_path)
+            isroot = os.getuid() == 0
+            for should_match, mode in (
+                # note that although the file may have matching 'group' or
+                # 'other' executable permissions, it is *not* executable
+                # because the current uid is the owner of the file -- which
+                # takes precedence
+                (False,  0o000),   # ----------, no
+                (isroot, 0o001),   # ---------x, no
+                (isroot, 0o010),   # ------x---, no
+                (True,   0o100),   # ---x------, yes
+                (False,  0o002),   # --------w-, no
+                (False,  0o020),   # -----w----, no
+                (False,  0o200),   # --w-------, no
+                (isroot, 0o003),   # --------wx, no
+                (isroot, 0o030),   # -----wx---, no
+                (True,   0o300),   # --wx------, yes
+                (False,  0o004),   # -------r--, no
+                (False,  0o040),   # ----r-----, no
+                (False,  0o400),   # -r--------, no
+                (isroot, 0o005),   # -------r-x, no
+                (isroot, 0o050),   # ----r-x---, no
+                (True,   0o500),   # -r-x------, yes
+                (False,  0o006),   # -------rw-, no
+                (False,  0o060),   # ----rw----, no
+                (False,  0o600),   # -rw-------, no
+                (isroot, 0o007),   # -------rwx, no
+                (isroot, 0o070),   # ----rwx---, no
+                (True,   0o700),   # -rwx------, yes
+                (isroot, 0o4001),  # ---S-----x, no
+                (isroot, 0o4010),  # ---S--x---, no
+                (True,   0o4100),  # ---s------, yes
+                (isroot, 0o4003),  # ---S----wx, no
+                (isroot, 0o4030),  # ---S-wx---, no
+                (True,   0o4300),  # --ws------, yes
+                (isroot, 0o2001),  # ------S--x, no
+                (isroot, 0o2010),  # ------s---, no
+                (True,   0o2100),  # ---x--S---, yes
+
+            ):
+                mode_str = '{0:0>4o}'.format(mode)
+
+                # given file mode,
                 os.chmod(bin_path, mode)
 
-                if not should_match:
-                    # should not be found because it is not executable
-                    assert pexpect.which(fname) is None
-                else:
-                    # should match full path
-                    assert pexpect.which(fname) == bin_path
+                # exercise whether we may execute
+                can_execute = True
+                try:
+                    subprocess.Popen(fname).wait() == 0
+                except OSError as err:
+                    if err.errno != errno.EACCES:
+                        raise
+                    # permission denied
+                    can_execute = False
+
+                assert should_match == can_execute, (
+                    should_match, can_execute, mode_str)
+
+                # exercise whether which(1) would match
+                proc = subprocess.Popen((bin_which, fname),
+                                        env={'PATH': bin_dir},
+                                        stdout=subprocess.PIPE)
+                bin_which_match = bool(not proc.wait())
+                assert should_match == bin_which_match, (
+                    should_match, bin_which_match, mode_str)
+
+                # finally, exercise pexpect's which(1) matches
+                # the same.
+                pexpect_match = bool(pexpect.which(fname))
+
+                assert should_match == pexpect_match == bin_which_match, (
+                    should_match, pexpect_match, bin_which_match, mode_str)
 
         finally:
             # restore,
             os.environ['PATH'] = save_path
+
             # destroy scratch files and folders,
             if os.path.exists(bin_path):
                 os.unlink(bin_path)