[Python-checkins] cpython: asyncio: Minimal pty support in UNIX read pipe, by Jonathan Slenders.
guido.van.rossum
python-checkins at python.org
Fri Jan 10 22:30:40 CET 2014
http://hg.python.org/cpython/rev/8a41abdb190a
changeset: 88401:8a41abdb190a
user: Guido van Rossum <guido at python.org>
date: Fri Jan 10 13:30:04 2014 -0800
summary:
asyncio: Minimal pty support in UNIX read pipe, by Jonathan Slenders.
files:
Lib/asyncio/unix_events.py | 7 +-
Lib/test/test_asyncio/test_events.py | 42 ++++++++++++++++
2 files changed, 47 insertions(+), 2 deletions(-)
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -190,7 +190,9 @@
self._pipe = pipe
self._fileno = pipe.fileno()
mode = os.fstat(self._fileno).st_mode
- if not (stat.S_ISFIFO(mode) or stat.S_ISSOCK(mode)):
+ if not (stat.S_ISFIFO(mode) or
+ stat.S_ISSOCK(mode) or
+ stat.S_ISCHR(mode)):
raise ValueError("Pipe transport is for pipes/sockets only.")
_set_nonblocking(self._fileno)
self._protocol = protocol
@@ -228,7 +230,8 @@
def _fatal_error(self, exc):
# should be called by exception handler only
- logger.exception('Fatal error for %s', self)
+ if not (isinstance(exc, OSError) and exc.errno == errno.EIO):
+ logger.exception('Fatal error for %s', self)
self._close(exc)
def _close(self, exc):
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -132,6 +132,8 @@
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:
@@ -955,6 +957,46 @@
@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):
proto = None
transport = None
--
Repository URL: http://hg.python.org/cpython
More information about the Python-checkins
mailing list