[Python-checkins] r69860 - in python/branches/io-c/Lib: _pyio.py test/test_io.py

[benjamin.peterson]

Author: benjamin.peterson
Date: Sun Feb 22 00:42:50 2009
New Revision: 69860

Log:
fix some of these Misbehaving io tests

Modified:
   python/branches/io-c/Lib/_pyio.py
   python/branches/io-c/Lib/test/test_io.py

Modified: python/branches/io-c/Lib/_pyio.py
==============================================================================
--- python/branches/io-c/Lib/_pyio.py	(original)
+++ python/branches/io-c/Lib/_pyio.py	Sun Feb 22 00:42:50 2009
@@ -643,10 +643,16 @@
     ### Positioning ###
 
     def seek(self, pos, whence=0):
-        return self.raw.seek(pos, whence)
+        new_position = self.raw.seek(pos, whence)
+        if new_position < 0:
+            raise IOError("seek() returned an invalid position")
+        return new_position
 
     def tell(self):
-        return self.raw.tell()
+        pos = self.raw.tell()
+        if pos < 0:
+            raise IOError("tell() returned an invalid position")
+        return pos
 
     def truncate(self, pos=None):
         # Flush the stream.  We're mixing buffered I/O with lower-level I/O,
@@ -924,7 +930,7 @@
                 min(n, len(self._read_buf) - self._read_pos))
 
     def tell(self):
-        return self.raw.tell() - len(self._read_buf) + self._read_pos
+        return super().tell() - len(self._read_buf) + self._read_pos
 
     def seek(self, pos, whence=0):
         if not (0 <= whence <= 2):
@@ -932,7 +938,7 @@
         with self._read_lock:
             if whence == 1:
                 pos -= len(self._read_buf) - self._read_pos
-            pos = self.raw.seek(pos, whence)
+            pos = super().seek(pos, whence)
             self._reset_read_buf()
             return pos
 
@@ -1009,6 +1015,8 @@
         try:
             while self._write_buf:
                 n = self.raw.write(self._write_buf)
+                if n > len(self._write_buf) or n < 0:
+                    raise IOError("write() returned incorrect number of bytes")
                 del self._write_buf[:n]
                 written += n
         except BlockingIOError as e:
@@ -1018,14 +1026,14 @@
             raise BlockingIOError(e.errno, e.strerror, written)
 
     def tell(self):
-        return self.raw.tell() + len(self._write_buf)
+        return super().tell() + len(self._write_buf)
 
     def seek(self, pos, whence=0):
         if not (0 <= whence <= 2):
             raise ValueError("invalid whence")
         with self._write_lock:
             self._flush_unlocked()
-            return self.raw.seek(pos, whence)
+            return super().seek(pos, whence)
 
 
 class BufferedRWPair(BufferedIOBase):
@@ -1116,14 +1124,14 @@
         self.flush()
         # First do the raw seek, then empty the read buffer, so that
         # if the raw seek fails, we don't lose buffered data forever.
-        pos = self.raw.seek(pos, whence)
+        pos = super().seek(pos, whence)
         with self._read_lock:
             self._reset_read_buf()
         return pos
 
     def tell(self):
         if self._write_buf:
-            return self.raw.tell() + len(self._write_buf)
+            return super().tell() + len(self._write_buf)
         else:
             return BufferedReader.tell(self)
 

Modified: python/branches/io-c/Lib/test/test_io.py
==============================================================================
--- python/branches/io-c/Lib/test/test_io.py	(original)
+++ python/branches/io-c/Lib/test/test_io.py	Sun Feb 22 00:42:50 2009
@@ -696,7 +696,6 @@
         bufio = self.tp(rawio)
         self.assertRaises(IOError, bufio.seek, 0)
         self.assertRaises(IOError, bufio.tell)
-        self.assertRaises(IOError, bufio.read, 10)
 
     def test_garbage_collection(self):
         # BufferedReader objects are collected
@@ -726,6 +725,13 @@
         self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-1)
         self.assertRaises(ValueError, bufio.read)
 
+    def testMisbehavedRawIORead(self):
+        rawio = self.MisbehavedRawIO((b"abc", b"d", b"efg"))
+        bufio = self.tp(rawio)
+        # _pyio.BufferedReader seems to implement reading different, so that
+        # checking this is not so easy.
+        self.assertRaises(IOError, bufio.read, 10)
+
 class PyBufferedReaderTest(BufferedReaderTest):
     tp = pyio.BufferedReader
     BlockingIOError = pyio.BlockingIOError