Merge pull request #764 from musthafak/master

rework async unittests to rely on unittest.IsolatedAsyncioTestCase

.github/workflows/ci.yml

@@ -25,7 +25,7 @@ jobs:
     - name: Install packages
       run: |
         export PYTHONIOENCODING=UTF8
-        pip install coveralls pytest-cov ptyprocess
+        pip install -r requirements-testing.txt
 
     - name: Run tests
       run: |

requirements-testing.txt

@@ -1,4 +1,6 @@
-pytest
-pytest-cov
+aiounittest; python_version < '3.8'
 coverage
 coveralls
+ptyprocess
+pytest
+pytest-cov

tests/PexpectTestCase.py

@@ -26,8 +26,13 @@
 import sys
 import os
 
+try:
+    from unittest import IsolatedAsyncioTestCase
+except ImportError:
+    from aiounittest import AsyncTestCase as IsolatedAsyncioTestCase
 
-class PexpectTestCase(unittest.TestCase):
+
+class _PexpectTestCaseBase:
     def setUp(self):
         self.PYTHONBIN = sys.executable
         self.original_path = os.getcwd()
@@ -109,3 +114,11 @@ def assertRaisesRegex(self, excClass, pattern):
                 assert re.match(pattern, str(e))
             else:
                 raise AssertionError("%s was not raised" % excClass)
+
+
+class PexpectTestCase(_PexpectTestCaseBase, unittest.TestCase):
+    pass
+
+
+class AsyncPexpectTestCase(_PexpectTestCaseBase, IsolatedAsyncioTestCase):
+    pass

tests/test_async.py

@@ -9,89 +9,79 @@
 
 import pexpect
 from pexpect import replwrap
-from .PexpectTestCase import PexpectTestCase
 
-def run(coro):
-    return asyncio.get_event_loop().run_until_complete(coro)
+from . import PexpectTestCase
+
 
 @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)
+class AsyncTests(PexpectTestCase.AsyncPexpectTestCase):
+    async def test_simple_expect(self):
+        p = pexpect.spawn("cat")
+        p.sendline("Hello asyncio")
+        assert await p.expect(["Hello", pexpect.EOF], async_=True) == 0
+        print("Done")
+
+    async def test_timeout(self):
+        p = pexpect.spawn("cat")
         with self.assertRaises(pexpect.TIMEOUT):
-            run(coro)
+            await p.expect("foo", timeout=1, async_=True)
 
-        p = pexpect.spawn('cat')
-        coro = p.expect(['foo', pexpect.TIMEOUT], timeout=1, async_=True)
-        assert run(coro) == 1
+        p = pexpect.spawn("cat")
+        assert await p.expect(["foo", pexpect.TIMEOUT], timeout=1, async_=True) == 1
 
-    def test_eof(self):
-        p = pexpect.spawn('cat')
-        p.sendline('Hi')
-        coro = p.expect(pexpect.EOF, async_=True)
+    async def test_eof(self):
+        p = pexpect.spawn("cat")
+        p.sendline("Hi")
         p.sendeof()
-        assert run(coro) == 0
+        assert await p.expect(pexpect.EOF, async_=True) == 0
 
-        p = pexpect.spawn('cat')
+        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' % 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' % 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' % self.PYTHONBIN, encoding='utf8')
-        assert run(p.expect_exact(u'READY', async_=True)) == 0
+            await p.expect("Blah", async_=True)
+
+    async def test_expect_exact(self):
+        p = pexpect.spawn("%s list100.py" % self.PYTHONBIN)
+        assert await p.expect_exact(b"5", async_=True) == 0
+        assert await p.expect_exact(["wpeok", b"11"], async_=True) == 1
+        assert await p.expect_exact([b"foo", pexpect.EOF], async_=True) == 1
+
+    async def test_async_utf8(self):
+        p = pexpect.spawn("%s list100.py" % self.PYTHONBIN, encoding="utf8")
+        assert await p.expect_exact("5", async_=True) == 0
+        assert await p.expect_exact(["wpeok", "11"], async_=True) == 1
+        assert await p.expect_exact(["foo", pexpect.EOF], async_=True) == 1
+
+    async def test_async_and_gc(self):
+        p = pexpect.spawn("%s sleep_for.py 1" % self.PYTHONBIN, encoding="utf8")
+        assert await p.expect_exact("READY", async_=True) == 0
         gc.collect()
-        assert run(p.expect_exact(u'END', async_=True)) == 0
+        assert await p.expect_exact("END", async_=True) == 0
 
-    def test_async_and_sync(self):
-        p = pexpect.spawn('echo 1234', encoding='utf8', maxread=1)
-        assert run(p.expect_exact(u'1', async_=True)) == 0
-        assert p.expect_exact(u'2') == 0
-        assert run(p.expect_exact(u'3', async_=True)) == 0
+    async def test_async_and_sync(self):
+        p = pexpect.spawn("echo 1234", encoding="utf8", maxread=1)
+        assert await p.expect_exact("1", async_=True) == 0
+        assert p.expect_exact("2") == 0
+        assert await p.expect_exact("3", async_=True) == 0
 
-    def test_async_replwrap(self):
+    async def test_async_replwrap(self):
         bash = replwrap.bash()
-        coro = bash.run_command("time", async_=True)
-        res = run(coro)
-        assert 'real' in res, res
+        res = await bash.run_command("time", async_=True)
+        assert "real" in res, res
 
-    def test_async_replwrap_multiline(self):
+    async def test_async_replwrap_multiline(self):
         bash = replwrap.bash()
-        coro = bash.run_command("echo '1 2\n3 4'", async_=True)
-        res = run(coro)
-        self.assertEqual(res.strip().splitlines(), ['1 2', '3 4'])
+        res = await bash.run_command("echo '1 2\n3 4'", async_=True)
+        self.assertEqual(res.strip().splitlines(), ["1 2", "3 4"])
 
         # Should raise ValueError if input is incomplete
-        coro = bash.run_command("echo '5 6", async_=True)
         try:
-            run(coro)
+            await bash.run_command("echo '5 6", async_=True)
         except ValueError:
             pass
         else:
             assert False, "Didn't raise ValueError for incomplete input"
 
         # Check that the REPL was reset (SIGINT) after the incomplete input
-        coro = bash.run_command("echo '1 2\n3 4'", async_=True)
-        res = run(coro)
-        self.assertEqual(res.strip().splitlines(), ['1 2', '3 4'])
+        res = await bash.run_command("echo '1 2\n3 4'", async_=True)
+        self.assertEqual(res.strip().splitlines(), ["1 2", "3 4"])