OLD | NEW |
(Empty) | |
| 1 try: |
| 2 import asyncio |
| 3 except ImportError: |
| 4 asyncio = None |
| 5 |
| 6 import sys |
| 7 import unittest |
| 8 |
| 9 import pexpect |
| 10 from .PexpectTestCase import PexpectTestCase |
| 11 |
| 12 def run(coro): |
| 13 return asyncio.get_event_loop().run_until_complete(coro) |
| 14 |
| 15 @unittest.skipIf(asyncio is None, "Requires asyncio") |
| 16 class AsyncTests(PexpectTestCase): |
| 17 def test_simple_expect(self): |
| 18 p = pexpect.spawn('cat') |
| 19 p.sendline('Hello asyncio') |
| 20 coro = p.expect(['Hello', pexpect.EOF] , async=True) |
| 21 assert run(coro) == 0 |
| 22 print('Done') |
| 23 |
| 24 def test_timeout(self): |
| 25 p = pexpect.spawn('cat') |
| 26 coro = p.expect('foo', timeout=1, async=True) |
| 27 with self.assertRaises(pexpect.TIMEOUT): |
| 28 run(coro) |
| 29 |
| 30 p = pexpect.spawn('cat') |
| 31 coro = p.expect(['foo', pexpect.TIMEOUT], timeout=1, async=True) |
| 32 assert run(coro) == 1 |
| 33 |
| 34 def test_eof(self): |
| 35 p = pexpect.spawn('cat') |
| 36 p.sendline('Hi') |
| 37 coro = p.expect(pexpect.EOF, async=True) |
| 38 p.sendeof() |
| 39 assert run(coro) == 0 |
| 40 |
| 41 p = pexpect.spawn('cat') |
| 42 p.sendeof() |
| 43 coro = p.expect('Blah', async=True) |
| 44 with self.assertRaises(pexpect.EOF): |
| 45 run(coro) |
| 46 |
| 47 def test_expect_exact(self): |
| 48 p = pexpect.spawn('%s list100.py' % sys.executable) |
| 49 assert run(p.expect_exact(b'5', async=True)) == 0 |
| 50 assert run(p.expect_exact(['wpeok', b'11'], async=True)) == 1 |
| 51 assert run(p.expect_exact([b'foo', pexpect.EOF], async=True)) == 1 |
OLD | NEW |