inherit IOStream from TextIOBase

ensures all expected APIs are present
and raise appropriate errors for undefined methods,
rather than AttributeError.

ipykernel/iostream.py

@@ -11,7 +11,7 @@
 import sys
 import threading
 import warnings
-from io import StringIO, UnsupportedOperation
+from io import StringIO, UnsupportedOperation, TextIOBase
 
 import zmq
 from zmq.eventloop.ioloop import IOLoop
@@ -249,23 +249,22 @@ def send_multipart(self, *args, **kwargs):
         return self.io_thread.send_multipart(*args, **kwargs)
 
 
-class OutStream(object):
+class OutStream(TextIOBase):
     """A file like object that publishes the stream to a 0MQ PUB socket.
     
     Output is handed off to an IO Thread
     """
 
     # The time interval between automatic flushes, in seconds.
     flush_interval = 0.2
-    topic=None
+    topic = None
+    encoding = 'UTF-8'
 
     def __init__(self, session, pub_thread, name, pipe=None):
         if pipe is not None:
             warnings.warn("pipe argument to OutStream is deprecated and ignored",
                 DeprecationWarning)
-        self.encoding = 'UTF-8'
         # This is necessary for compatibility with Python built-in streams
-        self.errors = None
         self.session = session
         if not isinstance(pub_thread, IOPubThread):
             # Backward-compat: given socket, not thread. Wrap in a thread.
@@ -339,24 +338,6 @@ def _flush(self):
             content = {u'name':self.name, u'text':data}
             self.session.send(self.pub_thread, u'stream', content=content,
                 parent=self.parent_header, ident=self.topic)
-
-    def isatty(self):
-        return False
-
-    def __next__(self):
-        raise IOError('Read not supported on a write only stream.')
-
-    if not py3compat.PY3:
-        next = __next__
-
-    def read(self, size=-1):
-        raise IOError('Read not supported on a write only stream.')
-
-    def readline(self, size=-1):
-        raise IOError('Read not supported on a write only stream.')
-
-    def fileno(self):
-        raise UnsupportedOperation("IOStream has no fileno.")
 
     def write(self, string):
         if self.pub_thread is None:

ipykernel/tests/test_io.py

@@ -0,0 +1,42 @@
+"""Test IO capturing functionality"""
+
+import io
+
+import zmq
+
+from jupyter_client.session import Session
+from ipykernel.iostream import IOPubThread, OutStream
+
+import nose.tools as nt
+
+def test_io_api():
+    """Test that wrapped stdout has the same API as a normal TextIO object"""
+    session = Session()
+    ctx = zmq.Context()
+    pub = ctx.socket(zmq.PUB)
+    thread = IOPubThread(pub)
+    thread.start()
+
+    stream = OutStream(session, thread, 'stdout')
+
+    # cleanup unused zmq objects before we start testing
+    thread.stop()
+    thread.close()
+    ctx.term()
+
+    assert stream.errors is None
+    assert not stream.isatty()
+    with nt.assert_raises(io.UnsupportedOperation):
+        stream.detach()
+    with nt.assert_raises(io.UnsupportedOperation):
+        next(stream)
+    with nt.assert_raises(io.UnsupportedOperation):
+        stream.read()
+    with nt.assert_raises(io.UnsupportedOperation):
+        stream.readline()
+    with nt.assert_raises(io.UnsupportedOperation):
+        stream.seek()
+    with nt.assert_raises(io.UnsupportedOperation):
+        stream.tell()
+
+