Mike Verdone's checkpoint, cleaned up.

Also implemented Neal's suggestion (add fileno() to SocketIO)
and some unrelated changes, e.g. remove Google copyright
and make BytesIO a subclass of BufferedIOBase.

1 files changed, 145 insertions, 1 deletions

diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index 0e3b03d..7384999 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -1,8 +1,43 @@
+"""Unit tests for io.py."""
+
 import unittest
 from test import test_support
 
 import io
 
+
+class MockReadIO(io.RawIOBase):
+    def __init__(self, readStack):
+        self._readStack = list(readStack)
+
+    def read(self, n=None):
+        try:
+            return self._readStack.pop(0)
+        except:
+            return io.EOF
+
+    def fileno(self):
+        return 42
+
+    def readable(self):
+        return True
+
+
+class MockWriteIO(io.RawIOBase):
+    def __init__(self):
+        self._writeStack = []
+
+    def write(self, b):
+        self._writeStack.append(b)
+        return len(b)
+
+    def writeable(self):
+        return True
+
+    def fileno(self):
+        return 42
+
+
 class IOTest(unittest.TestCase):
 
     def write_ops(self, f):
@@ -55,8 +90,117 @@ class IOTest(unittest.TestCase):
         f = io.BytesIO(data)
         self.read_ops(f)
 
+
+class BytesIOTest(unittest.TestCase):
+
+    def testInit(self):
+        buf = b"1234567890"
+        bytesIo = io.BytesIO(buf)
+
+    def testRead(self):
+        buf = b"1234567890"
+        bytesIo = io.BytesIO(buf)
+
+        self.assertEquals(buf[:1], bytesIo.read(1))
+        self.assertEquals(buf[1:5], bytesIo.read(4))
+        self.assertEquals(buf[5:], bytesIo.read(900))
+        self.assertEquals(io.EOF, bytesIo.read())
+
+    def testReadNoArgs(self):
+        buf = b"1234567890"
+        bytesIo = io.BytesIO(buf)
+
+        self.assertEquals(buf, bytesIo.read())
+        self.assertEquals(io.EOF, bytesIo.read())
+
+    def testSeek(self):
+        buf = b"1234567890"
+        bytesIo = io.BytesIO(buf)
+
+        bytesIo.read(5)
+        bytesIo.seek(0)
+        self.assertEquals(buf, bytesIo.read())
+
+        bytesIo.seek(3)
+        self.assertEquals(buf[3:], bytesIo.read())
+
+    def testTell(self):
+        buf = b"1234567890"
+        bytesIo = io.BytesIO(buf)
+
+        self.assertEquals(0, bytesIo.tell())
+        bytesIo.seek(5)
+        self.assertEquals(5, bytesIo.tell())
+        bytesIo.seek(10000)
+        self.assertEquals(10000, bytesIo.tell())
+
+
+class BufferedReaderTest(unittest.TestCase):
+
+    def testRead(self):
+        rawIo = MockReadIO((b"abc", b"d", b"efg"))
+        bufIo = io.BufferedReader(rawIo)
+
+        self.assertEquals(b"abcdef", bufIo.read(6))
+
+    def testReadToEof(self):
+        rawIo = MockReadIO((b"abc", b"d", b"efg"))
+        bufIo = io.BufferedReader(rawIo)
+
+        self.assertEquals(b"abcdefg", bufIo.read(9000))
+
+    def testReadNoArgs(self):
+        rawIo = MockReadIO((b"abc", b"d", b"efg"))
+        bufIo = io.BufferedReader(rawIo)
+
+        self.assertEquals(b"abcdefg", bufIo.read())
+
+    def testFileno(self):
+        rawIo = MockReadIO((b"abc", b"d", b"efg"))
+        bufIo = io.BufferedReader(rawIo)
+
+        self.assertEquals(42, bufIo.fileno())
+
+    def testFilenoNoFileno(self):
+        # TODO will we always have fileno() function? If so, kill
+        # this test. Else, write it.
+        pass
+
+
+class BufferedWriterTest(unittest.TestCase):
+
+    def testWrite(self):
+        # Write to the buffered IO but don't overflow the buffer.
+        writer = MockWriteIO()
+        bufIo = io.BufferedWriter(writer, 8)
+
+        bufIo.write(b"abc")
+
+        self.assertFalse(writer._writeStack)
+
+    def testWriteOverflow(self):
+        writer = MockWriteIO()
+        bufIo = io.BufferedWriter(writer, 8)
+
+        bufIo.write(b"abc")
+        bufIo.write(b"defghijkl")
+
+        self.assertEquals(b"abcdefghijkl", writer._writeStack[0])
+
+    def testFlush(self):
+        writer = MockWriteIO()
+        bufIo = io.BufferedWriter(writer, 8)
+
+        bufIo.write(b"abc")
+        bufIo.flush()
+
+        self.assertEquals(b"abc", writer._writeStack[0])
+
+# TODO. Tests for open()
+
 def test_main():
-    test_support.run_unittest(IOTest)
+    test_support.run_unittest(IOTest, BytesIOTest, BufferedReaderTest,
+                              BufferedWriterTest)
 
 if __name__ == "__main__":
     test_main()