mirror of
https://github.com/python/cpython.git
synced 2025-12-04 00:30:19 +00:00
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.
This commit is contained in:
parent
a4f9fc6494
commit
68bbcd2a71
2 changed files with 277 additions and 13 deletions
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue