FreeBSD /bin/sh test workaround, simple "call and response" test using bc(1)

ptyprocess/ptyprocess.py

@@ -28,8 +28,6 @@
     _platform.startswith('solaris') or
     _platform.startswith('sunos'))
 
-_is_irix = _platform.startswith('irix')
-
 if _is_solaris:
     use_native_pty_fork = False
     from . import _fork_pty

tests/test_echo.py

@@ -1,5 +1,6 @@
 import time
 import unittest
+from ptyprocess.ptyprocess import _is_solaris
 from ptyprocess import PtyProcess
 
 class PtyEchoTestCase(unittest.TestCase):
@@ -12,6 +13,7 @@ def _read_until_eof(self, proc):
             except EOFError:
                 return
 
+    @unittest.skipIf(_is_solaris, "waitnoecho cannot be called on this platform.")
     def test_waitnoecho_forever(self):
         """Ensure waitnoecho() with no timeout will return when echo=False."""
         cat = PtyProcess.spawn(['cat'], echo=False)
@@ -22,6 +24,7 @@ def test_waitnoecho_forever(self):
         self._read_until_eof(cat)
         assert cat.wait() == 0
 
+    @unittest.skipIf(_is_solaris, "waitnoecho cannot be called on this platform.")
     def test_waitnoecho_timeout(self):
         """Ensure waitnoecho() with timeout will return when using stty to unset echo."""
         cat = PtyProcess.spawn(['cat'], echo=True)

tests/test_spawn.py

@@ -1,44 +1,113 @@
 import os
 import time
+import select
 import unittest
+from ptyprocess.ptyprocess import which
 from ptyprocess import PtyProcess, PtyProcessUnicode
 
 class PtyTestCase(unittest.TestCase):
+    def setUp(self):
+        self.cmd = u'echo $ENV_KEY; exit 0\n'
+        self.env = os.environ.copy()
+        self.env_key = u'ENV_KEY'
+        self.env_value = u'env_value'
+        self.env[self.env_key] = self.env_value
+
+    def _canread(self, fd, timeout=1):
+        return fd in select.select([fd], [], [], timeout)[0]
+
+    def _spawn_sh(self, ptyp, cmd, outp, env_value):
+        # given,
+        p = ptyp.spawn(['sh'], env=self.env)
+        p.write(cmd)
+
+        # exercise,
+        while True:
+            try:
+                outp += p.read()
+            except EOFError:
+                break
+
+        # verify, input is echo to output
+        assert cmd.strip() in outp
+
+        # result of echo $ENV_KEY in output
+        assert env_value in outp
+
+        # exit succesfully (exit 0)
+        assert p.wait() == 0
+
+
     def test_spawn_sh(self):
-        env = os.environ.copy()
-        env['FOO'] = 'rebar'
-        p = PtyProcess.spawn(['sh'], env=env)
-        p.read()
-        p.write(b'echo $FOO\n')
-        time.sleep(0.1)
-        response = p.read()
-        assert b'rebar' in response
-
-        p.sendeof()
-        p.readline()
-
-        with self.assertRaises(EOFError):
-            p.read()
-
-    def test_spawn_unicode_sh(self):
-        env = os.environ.copy()
-        env['FOO'] = 'rebar'
-        p = PtyProcessUnicode.spawn(['sh'], env=env)
-        p.read()
-        p.write(u'echo $FOO\n')
-        time.sleep(0.1)
-        response = p.read()
-        assert u'rebar' in response
-
-        p.sendeof()
-        p.readline()
-
-        with self.assertRaises(EOFError):
-            p.read()
+        outp = b''
+        self._spawn_sh(PtyProcess, self.cmd.encode('ascii'),
+                       outp, self.env_value.encode('ascii'))
+
+    def test_spawn_sh_unicode(self):
+        outp = u''
+        self._spawn_sh(PtyProcessUnicode, self.cmd,
+                       outp, self.env_value)
 
     def test_quick_spawn(self):
         """Spawn a very short-lived process."""
         # so far only reproducable on Solaris 11, spawning a process
         # that exits very quickly raised an exception at 'inst.setwinsize',
         # because the pty file descriptor was quickly lost after exec().
-        PtyProcess.spawn(['/bin/true'])
+        PtyProcess.spawn(['true'])
+
+    def _interactive_repl_unicode(self, echo):
+        """Test Call and response with echo ON/OFF."""
+        # given,
+        bc = PtyProcessUnicode.spawn(['bc'], echo=echo)
+        given_input = u'2+2+2+2+2+2+2+2+2+2+2+2+2+2+2+2+2+2+2+2\n'
+        expected_output = u'40'
+
+        # gnu-bc will display a long FSF banner on startup,
+        # whereas bsd-bc (on FreeBSD, Solaris) display no
+        # banner at all.  To ensure we've read up to our
+        # current prompt, read until the response of '2^16' is found.
+        time.sleep(1)
+
+        bc.write(u'2^16\n')
+        outp = u''
+        while self._canread(bc.fd):
+            outp += bc.read()
+        assert u'65536' in outp
+
+        # exercise,
+        bc.write(given_input)
+
+        while self._canread(bc.fd, timeout=2):
+            outp += bc.read()
+
+        # with echo ON, we should see our input.
+        #
+        # note: we cannot assert the reverse: on Solaris, FreeBSD,
+        # and OSX, our input is echoed to output even with echo=False,
+        # something to do with the non-gnu version of bc(1), perhaps.
+        if echo:
+            assert given_input.strip() in outp
+
+        # we should most certainly see the result output.
+        assert expected_output in outp
+
+        # exercise sending EOF
+        bc.sendeof()
+
+        # validate EOF on read
+        while True:
+            try:
+                bc.read()
+            except EOFError:
+                break
+
+        # validate exit status,
+        assert bc.wait() == 0
+
+    @unittest.skipIf(which('bc') is None, "bc(1) not found on this server.")
+    def test_interactive_repl_unicode_noecho(self):
+        self._interactive_repl_unicode(echo=False)
+
+    @unittest.skipIf(which('bc') is None, "bc(1) not found on this server.")
+    def test_interactive_repl_unicode_echo(self):
+        self._interactive_repl_unicode(echo=True)

tests/test_wait.py

@@ -2,32 +2,31 @@
 import time
 import unittest
 from ptyprocess import PtyProcess
-from ptyprocess.ptyprocess import which
 
 
 class TestWaitAfterTermination(unittest.TestCase):
     """Various test cases for PtyProcess.wait()"""
 
     def test_wait_true_shortproc(self):
         """Ensure correct (True) wait status for short-lived processes."""
-        child = PtyProcess.spawn([which('true')])
+        child = PtyProcess.spawn(['true'])
         # Wait so we're reasonable sure /bin/true has terminated
         time.sleep(0.2)
         self.assertEqual(child.wait(), 0)
 
     def test_wait_false_shortproc(self):
         """Ensure correct (False) wait status for short-lived processes."""
-        child = PtyProcess.spawn([which('false')])
+        child = PtyProcess.spawn(['false'])
         # Wait so we're reasonable sure /bin/false has terminated
         time.sleep(0.2)
-        self.assertEqual(child.wait(), 1)
+        self.assertNotEqual(child.wait(), 0)
 
     def test_wait_twice_longproc(self):
         """Ensure correct wait status when called twice."""
         # previous versions of ptyprocess raises PtyProcessError when
         # wait was called more than once with "Cannot wait for dead child
         # process.".  No longer true since v0.5.
-        child = PtyProcess.spawn([which('sleep'), '1'])
+        child = PtyProcess.spawn(['sleep', '1'])
         # this call to wait() will block for 1s
         for count in range(2):
             self.assertEqual(child.wait(), 0, count)