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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
try:
import asyncio
except ImportError:
asyncio = None
import gc
import sys
import unittest
import pexpect
from pexpect import replwrap
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' % 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
gc.collect()
assert run(p.expect_exact(u'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
def test_async_replwrap(self):
bash = replwrap.bash()
coro = bash.run_command("time", async_=True)
res = run(coro)
assert 'real' in res, res
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'])
# Should raise ValueError if input is incomplete
coro = bash.run_command("echo '5 6", async_=True)
try:
run(coro)
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'])
|
