diff -r eb1b93fd68f5 Lib/test/test_io.py --- a/Lib/test/test_io.py Thu May 12 02:07:00 2011 +0200 +++ b/Lib/test/test_io.py Sat May 28 14:42:56 2011 -0400 @@ -801,6 +801,69 @@ self.assertEqual(bufio.readinto(b), 1) self.assertEqual(b, b"cb") + def test_prefetch(self): + rawio = self.MockRawIO((b'abc', b'd', b'ef', None)) + bufio = self.tp(rawio) + b = bytearray(2) + # Invalid args + self.assertRaises(TypeError, bufio.prefetch, None, 0, 0) + self.assertRaises(TypeError, bufio.prefetch, 0, None, 0) + self.assertRaises(TypeError, bufio.prefetch, 0, 0, None) + self.assertRaises(ValueError, bufio.prefetch, b, 0, 4) + self.assertRaises(ValueError, bufio.prefetch, b, -1, 0) + self.assertRaises(ValueError, bufio.prefetch, b, 0, -1) + + self.assertEqual(bufio.prefetch(b), 2) + self.assertEqual(b, b'ab') + self.assertEqual(bufio.prefetch(b, 0, 0), 2) + self.assertEqual(b, b'ab') + self.assertEqual(bufio.prefetch(b, 0, 2), 2) + self.assertEqual(b, b'ab') + self.assertEqual(rawio._reads, 1) + self.assertEqual(bufio.prefetch(b, 1, 2), 2) + self.assertEqual(b, b'de') + self.assertEqual(rawio._reads, 3) + self.assertEqual(bufio.prefetch(b, 0, 2), 1) + self.assertEqual(b, b'fe') + self.assertEqual(rawio._reads, 4) + self.assertEqual(bufio.prefetch(b, 1, 1), 0) + self.assertEqual(b, b'fe') + self.assertEqual(rawio._reads, 5) + # EOF during skip + b = bytearray(2) + bufio.seek(0) + self.assertEqual(bufio.prefetch(b, 8, 2), 0) + self.assertEqual(b, b'\x00\x00') + + def test_unseekable_prefetch(self): + rawio = self.MockUnseekableIO(b'abc' * 10) + bufio = self.tp(rawio) + b = bytearray(3) + self.assertEqual(bufio.prefetch(b, 0, 0), 3) + self.assertEqual(b, b'abc') + self.assertEqual(bufio.prefetch(b, 1, 2), 3) + self.assertEqual(b, b'bca') + self.assertEqual(bufio.prefetch(b, 24, 0), 3) + self.assertEqual(b, b'abc') + self.assertEqual(bufio.prefetch(b, 2, 3), 1) + self.assertEqual(b, b'cbc') + # large skips > bufsize + bufsize = 8 * 1024 + data = b'abcxyz' * bufsize + bufio = self.tp(self.MockUnseekableIO(data)) + b = bytearray(9) + self.assertEqual(bufio.prefetch(b, len(data) - 9, 2), 9) + self.assertEqual(b, b'xyzabcxyz') + self.assertEqual(bufio.prefetch(b, 1, 7), 6) + self.assertEqual(b, b'abcxyzxyz') + # EOF during skip + b = bytearray(2) + rawio = self.MockUnseekableIO(b'abc') + bufio = self.tp(rawio) + self.assertEqual(bufio.prefetch(b, 8, 2), 0) + self.assertEqual(b, b'\x00\x00') + + def test_readlines(self): def bufio(): rawio = self.MockRawIO((b"abc\n", b"d\n", b"ef")) @@ -1185,7 +1248,6 @@ DeprecationWarning)): self.tp(self.MockRawIO(), 8, 12) - class CBufferedWriterTest(BufferedWriterTest): tp = io.BufferedWriter @@ -1510,6 +1572,7 @@ # You can't construct a BufferedRandom over a non-seekable stream. test_unseekable = None + test_unseekable_prefetch = None class CBufferedRandomTest(BufferedRandomTest): tp = io.BufferedRandom