Index: third_party/pexpect/tests/test_async.py |
diff --git a/third_party/pexpect/tests/test_async.py b/third_party/pexpect/tests/test_async.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..ce755721bb699864dae92299e8699faa8e6b0493 |
--- /dev/null |
+++ b/third_party/pexpect/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 |