mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-01 18:51:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 21

asyncio: Minimal pty support in UNIX read pipe, by Jonathan Slenders.

Browse source
This commit is contained in:
Guido van Rossum 2014-01-10 13:30:04 -08:00
parent 4835f17c24
commit 02757ea7e9
2 changed files with 47 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

42
Lib/test/test_asyncio/test_events.py
View file

@ -132,6 +132,8 @@ class MyReadPipeProto(protocols.Protocol):
self.state.append('EOF')
def connection_lost(self, exc):
if 'EOF' not in self.state:
self.state.append('EOF') # It is okay if EOF is missed.
assert self.state == ['INITIAL', 'CONNECTED', 'EOF'], self.state
self.state.append('CLOSED')
if self.done:
@ -953,6 +955,46 @@ class EventLoopTestsMixin:
# extra info is available
self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
@unittest.skipUnless(sys.platform != 'win32',
"Don't support pipes for Windows")
def test_read_pty_output(self):
proto = None
def factory():
nonlocal proto
proto = MyReadPipeProto(loop=self.loop)
return proto
master, slave = os.openpty()
master_read_obj = io.open(master, 'rb', 0)
@tasks.coroutine
def connect():
t, p = yield from self.loop.connect_read_pipe(factory,
master_read_obj)
self.assertIs(p, proto)
self.assertIs(t, proto.transport)
self.assertEqual(['INITIAL', 'CONNECTED'], proto.state)
self.assertEqual(0, proto.nbytes)
self.loop.run_until_complete(connect())
os.write(slave, b'1')
test_utils.run_until(self.loop, lambda: proto.nbytes)
self.assertEqual(1, proto.nbytes)
os.write(slave, b'2345')
test_utils.run_until(self.loop, lambda: proto.nbytes >= 5)
self.assertEqual(['INITIAL', 'CONNECTED'], proto.state)
self.assertEqual(5, proto.nbytes)
os.close(slave)
self.loop.run_until_complete(proto.done)
self.assertEqual(
['INITIAL', 'CONNECTED', 'EOF', 'CLOSED'], proto.state)
# extra info is available
self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
@unittest.skipUnless(sys.platform != 'win32',
"Don't support pipes for Windows")
def test_write_pipe(self):