summaryrefslogtreecommitdiff
path: root/tests/test_async.py
blob: bb945a40e5551b508eb4cb5fe7a377c5e28ec90e (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
try:
    import asyncio
except ImportError:
    asyncio = None

import pexpect
import unittest

def run(coro):
    return asyncio.get_event_loop().run_until_complete(coro)

@unittest.skipIf(asyncio is None, "Requires asyncio")
class AsyncTests(unittest.TestCase):
    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)