diff options
author | Musthafa Kariyaparambil <musthafakdml@gmail.com> | 2023-08-28 16:02:29 +0200 |
---|---|---|
committer | Musthafa Kariyaparambil <musthafakdml@gmail.com> | 2023-08-28 16:02:29 +0200 |
commit | c59c991ebac3f4d77d9860896e9a337fc68c6ce9 (patch) | |
tree | 0f4ed476f6edbee0bc089111473238eaa0bb5292 | |
parent | 2532721644781543ca660e52d48a35bd93872fc1 (diff) | |
download | pexpect-c59c991ebac3f4d77d9860896e9a337fc68c6ce9.tar.gz |
rework async unittests to rely on unittest.IsolatedAsyncioTestCase
-rw-r--r-- | tests/PexpectTestCase.py | 10 | ||||
-rw-r--r-- | tests/test_async.py | 112 |
2 files changed, 60 insertions, 62 deletions
diff --git a/tests/PexpectTestCase.py b/tests/PexpectTestCase.py index a762d8f..eb58de1 100644 --- a/tests/PexpectTestCase.py +++ b/tests/PexpectTestCase.py @@ -27,7 +27,7 @@ import sys import os -class PexpectTestCase(unittest.TestCase): +class _PexpectTestCaseBase: def setUp(self): self.PYTHONBIN = sys.executable self.original_path = os.getcwd() @@ -109,3 +109,11 @@ class PexpectTestCase(unittest.TestCase): assert re.match(pattern, str(e)) else: raise AssertionError("%s was not raised" % excClass) + + +class PexpectTestCase(_PexpectTestCaseBase, unittest.TestCase): + pass + + +class AsyncPexpectTestCase(_PexpectTestCaseBase, unittest.IsolatedAsyncioTestCase): + pass diff --git a/tests/test_async.py b/tests/test_async.py index 466d56f..68c5fc2 100644 --- a/tests/test_async.py +++ b/tests/test_async.py @@ -9,89 +9,79 @@ import unittest 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"]) |