Support for SEEK_HOLE/SEEK_DATA

Lib/_pyio.py

 """
 Python implementation of the io module.
 """
 
 import os
 import abc
 import codecs
 import warnings
 import errno
 # Import _thread instead of threading to reduce startup cost
 try:
     from _thread import allocate_lock as Lock
 except ImportError:
     from _dummy_thread import allocate_lock as Lock
 
 import io
 from io import (__all__, SEEK_SET, SEEK_CUR, SEEK_END)
+
+valid_seek_flags = {0, 1, 2}  # Hardwired values
+if hasattr(os, 'SEEK_HOLE') :
+    valid_seek_flags.add(os.SEEK_HOLE)
+    valid_seek_flags.add(os.SEEK_DATA)
 
 # open() uses st_blksize whenever we can
 DEFAULT_BUFFER_SIZE = 8 * 1024  # bytes
 
 # NOTE: Base classes defined here are registered with the "official" ABCs
 # defined in io.py. We don't use real inheritance though, because we don't
 # want to inherit the C implementations.
 
 # Rebind for compatibility
 BlockingIOError = BlockingIOError
@@ skipping 1007 matching lines @@
             return b""
         with self._read_lock:
             self._peek_unlocked(1)
             return self._read_unlocked(
                 min(n, len(self._read_buf) - self._read_pos))
 
     def tell(self):
         return _BufferedIOMixin.tell(self) - len(self._read_buf) + self._read_pos
 
     def seek(self, pos, whence=0):
+        if whence not in valid_seek_flags:
+            raise ValueError("invalid whence value")
         with self._read_lock:
             if whence == 1:
                 pos -= len(self._read_buf) - self._read_pos
             pos = _BufferedIOMixin.seek(self, pos, whence)
             self._reset_read_buf()
             return pos
 
 class BufferedWriter(_BufferedIOMixin):
 
     """A buffer for a writeable sequential RawIO object.
@@ skipping 75 matching lines @@
                     errno.EAGAIN,
                     "write could not complete without blocking", 0)
             if n > len(self._write_buf) or n < 0:
                 raise IOError("write() returned incorrect number of bytes")
             del self._write_buf[:n]
 
     def tell(self):
         return _BufferedIOMixin.tell(self) + len(self._write_buf)
 
     def seek(self, pos, whence=0):
+        if whence not in valid_seek_flags:
+            raise ValueError("invalid whence value")
         with self._write_lock:
             self._flush_unlocked()
             return _BufferedIOMixin.seek(self, pos, whence)
 
 
 class BufferedRWPair(BufferedIOBase):
 
     """A buffered reader and writer object together.
 
     A buffered reader object and buffered writer object put together to
@@ skipping 75 matching lines @@
 
     _warning_stack_offset = 3
 
     def __init__(self, raw,
                  buffer_size=DEFAULT_BUFFER_SIZE, max_buffer_size=None):
         raw._checkSeekable()
         BufferedReader.__init__(self, raw, buffer_size)
         BufferedWriter.__init__(self, raw, buffer_size, max_buffer_size)
 
     def seek(self, pos, whence=0):
+        if whence not in valid_seek_flags:
+            raise ValueError("invalid whence value")
         self.flush()
         if self._read_buf:
             # Undo read ahead.
             with self._read_lock:
                 self.raw.seek(self._read_pos - len(self._read_buf), 1)
         # 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)
         with self._read_lock:
             self._reset_read_buf()
@@ skipping 814 matching lines @@
     def errors(self):
         return None
 
     @property
     def encoding(self):
         return None
 
     def detach(self):
         # This doesn't make sense on StringIO.
         self._unsupported("detach")