[Jython-checkins] jython: Checking and raising errors in _io module
jeff.allen
jython-checkins at python.org
Tue Dec 25 02:05:21 CET 2012
http://hg.python.org/jython/rev/044d655970b8
changeset: 6914:044d655970b8
user: Jeff Allen <ja...py at farowl.co.uk>
date: Sun Dec 23 16:49:35 2012 +0000
summary:
Checking and raising errors in _io module
Changes to _jyio.py mostly aimed at greater conformity with test.test_io
where copied _pyio code seems not to pass. Introduce _checkInitialized
mirroring CPython code. Last skip removed from test_bufio.py so we can
revert to lib-python version. test_io now scores fail/error/skip = 2/17/99.
files:
Lib/_jyio.py | 45 ++++++-
Lib/test/test_bufio.py | 80 ------------
Lib/test/test_io.py | 9 +-
src/org/python/modules/_io/PyFileIO.java | 2 +-
4 files changed, 49 insertions(+), 87 deletions(-)
diff --git a/Lib/_jyio.py b/Lib/_jyio.py
--- a/Lib/_jyio.py
+++ b/Lib/_jyio.py
@@ -156,6 +156,7 @@
"""
def __init__(self, raw):
+ self._ok = False # Jython: subclass __init__ must set when state valid
self._raw = raw
### Positioning ###
@@ -211,21 +212,31 @@
### Inquiries ###
def seekable(self):
+ self._checkInitialized() # Jython: to forbid use in an invalid state
return self.raw.seekable()
def readable(self):
+ self._checkInitialized() # Jython: to forbid use in an invalid state
return self.raw.readable()
def writable(self):
+ self._checkInitialized() # Jython: to forbid use in an invalid state
return self.raw.writable()
@property
def raw(self):
return self._raw
- @property
- def closed(self):
- return self.raw.closed
+ # Jython difference: @property closed(self) inherited from _IOBase.__closed
+
+ # Jython difference: emulate C implementation CHECK_INITIALIZED. This is for
+ # compatibility, to pass test.test_io.CTextIOWrapperTest.test_initialization.
+ def _checkInitialized(self):
+ if not self._ok:
+ if self.raw is None:
+ raise ValueError("raw stream has been detached")
+ else:
+ raise ValueError("I/O operation on uninitialized object")
@property
def name(self):
@@ -388,6 +399,7 @@
self.buffer_size = buffer_size
self._reset_read_buf()
self._read_lock = Lock()
+ self._ok = True # Jython: to enable use now in a valid state
def _reset_read_buf(self):
self._read_buf = b""
@@ -401,6 +413,7 @@
mode. If n is negative, read until EOF or until read() would
block.
"""
+ self._checkReadable() # Jython: to forbid use in an invalid state
if n is not None and n < -1:
raise ValueError("invalid number of bytes to read")
with self._read_lock:
@@ -494,6 +507,7 @@
"""Reads up to n bytes, with at most one read() system call."""
# Returns up to n bytes. If at least one byte is buffered, we
# only return buffered bytes. Otherwise, we do one raw read.
+ self._checkReadable() # Jython: to forbid use in an invalid state
if n < 0:
raise ValueError("number of bytes to read must be positive")
if n == 0:
@@ -542,8 +556,10 @@
self.buffer_size = buffer_size
self._write_buf = bytearray()
self._write_lock = Lock()
+ self._ok = True # Jython: to enable use now in a valid state
def write(self, b):
+ self._checkWritable() # Jython: to forbid use in an invalid state
if self.closed:
raise ValueError("write to closed file")
if isinstance(b, unicode):
@@ -588,6 +604,7 @@
def _flush_unlocked(self):
if self.closed:
raise ValueError("flush of closed file")
+ self._checkWritable() # Jython: to forbid use in an invalid state
while self._write_buf:
try:
n = self.raw.write(self._write_buf)
@@ -939,6 +956,7 @@
def __init__(self, buffer, encoding=None, errors=None, newline=None,
line_buffering=False):
+ self._ok = False # Jython: to forbid use in an invalid state
if newline is not None and not isinstance(newline, basestring):
raise TypeError("illegal newline type: %r" % (type(newline),))
if newline not in (None, "", "\n", "\r", "\r\n"):
@@ -977,6 +995,8 @@
self._snapshot = None # info for reconstructing decoder state
self._seekable = self._telling = self.buffer.seekable()
+ self._ok = True # Jython: to enable use now in a valid state
+
if self._seekable and self.writable():
position = self.buffer.tell()
if position != 0:
@@ -1021,15 +1041,23 @@
return self._buffer
def seekable(self):
+ self._checkInitialized() # Jython: to forbid use in an invalid state
+ self._checkClosed() # Jython: compatibility with C implementation
return self._seekable
def readable(self):
+ self._checkInitialized() # Jython: to forbid use in an invalid state
+ self._checkClosed() # Jython: compatibility with C implementation
return self.buffer.readable()
def writable(self):
+ self._checkInitialized() # Jython: to forbid use in an invalid state
+ self._checkClosed() # Jython: compatibility with C implementation
return self.buffer.writable()
def flush(self):
+ self._checkInitialized() # Jython: to forbid use in an invalid state
+ self._checkClosed() # Jython: compatibility with C implementation
self.buffer.flush()
self._telling = self._seekable
@@ -1042,6 +1070,15 @@
# Jython difference: @property closed(self) inherited from _IOBase.__closed
+ # Jython difference: emulate C implementation CHECK_INITIALIZED. This is for
+ # compatibility, to pass test.test_io.CTextIOWrapperTest.test_initialization.
+ def _checkInitialized(self):
+ if not self._ok:
+ if self.buffer is None:
+ raise ValueError("underlying buffer has been detached")
+ else:
+ raise ValueError("I/O operation on uninitialized object")
+
@property
def name(self):
return self.buffer.name
@@ -1053,6 +1090,7 @@
return self.buffer.isatty()
def write(self, s):
+ self._checkWritable() # Jython: to forbid use in an invalid state
if self.closed:
raise ValueError("write to closed file")
if not isinstance(s, unicode):
@@ -1233,6 +1271,7 @@
if self.buffer is None:
raise ValueError("buffer is already detached")
self.flush()
+ self._ok = False # Jython: to forbid use in an invalid state
buffer = self._buffer
self._buffer = None
return buffer
diff --git a/Lib/test/test_bufio.py b/Lib/test/test_bufio.py
deleted file mode 100644
--- a/Lib/test/test_bufio.py
+++ /dev/null
@@ -1,80 +0,0 @@
-import unittest
-from test import test_support as support
-
-import io # C implementation.
-import _pyio as pyio # Python implementation.
-
-# Simple test to ensure that optimizations in the IO library deliver the
-# expected results. For best testing, run this under a debug-build Python too
-# (to exercise asserts in the C code).
-
-lengths = list(range(1, 257)) + [512, 1000, 1024, 2048, 4096, 8192, 10000,
- 16384, 32768, 65536, 1000000]
-
-class BufferSizeTest(unittest.TestCase):
- def try_one(self, s):
- # Write s + "\n" + s to file, then open it and ensure that successive
- # .readline()s deliver what we wrote.
-
- # Ensure we can open TESTFN for writing.
- support.unlink(support.TESTFN)
-
- # Since C doesn't guarantee we can write/read arbitrary bytes in text
- # files, use binary mode.
- f = self.open(support.TESTFN, "wb")
- try:
- # write once with \n and once without
- f.write(s)
- f.write(b"\n")
- f.write(s)
- f.close()
- f = open(support.TESTFN, "rb")
- line = f.readline()
- self.assertEqual(line, s + b"\n")
- line = f.readline()
- self.assertEqual(line, s)
- line = f.readline()
- self.assertTrue(not line) # Must be at EOF
- f.close()
- finally:
- support.unlink(support.TESTFN)
-
- def drive_one(self, pattern):
- for length in lengths:
- # Repeat string 'pattern' as often as needed to reach total length
- # 'length'. Then call try_one with that string, a string one larger
- # than that, and a string one smaller than that. Try this with all
- # small sizes and various powers of 2, so we exercise all likely
- # stdio buffer sizes, and "off by one" errors on both sides.
- q, r = divmod(length, len(pattern))
- teststring = pattern * q + pattern[:r]
- self.assertEqual(len(teststring), length)
- self.try_one(teststring)
- self.try_one(teststring + b"x")
- self.try_one(teststring[:-1])
-
- def test_primepat(self):
- # A pattern with prime length, to avoid simple relationships with
- # stdio buffer sizes.
- self.drive_one(b"1234567890\00\01\02\03\04\05\06")
-
- def test_nullpat(self):
- self.drive_one(bytes(1000))
-
-
- at unittest.skipIf(support.is_jython, "FIXME: Not working on Jython")
-class CBufferSizeTest(BufferSizeTest):
- open = io.open
-
-class PyBufferSizeTest(BufferSizeTest):
- open = staticmethod(pyio.open)
-
-class BuiltinBufferSizeTest(BufferSizeTest):
- open = open
-
-
-def test_main():
- support.run_unittest(CBufferSizeTest, PyBufferSizeTest, BuiltinBufferSizeTest)
-
-if __name__ == "__main__":
- test_main()
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -2355,13 +2355,16 @@
class CTextIOWrapperTest(TextIOWrapperTest):
def test_initialization(self):
- r = self.BytesIO(b"\xc3\xa9\n\n")
+ # Use ascii material so decoding does not raise ValueError
+ r = self.BytesIO(b"red\nherring\n")
b = self.BufferedReader(r, 1000)
t = self.TextIOWrapper(b)
+ # Jython note: Invalid __init__ calls also leave t unreadable (in C
+ # implementation but not pure python _pyio).
self.assertRaises(TypeError, t.__init__, b, newline=42)
- self.assertRaises(ValueError, t.read)
+ self.assertRaises(ValueError, t.read) # Check t unreadable
self.assertRaises(ValueError, t.__init__, b, newline='xyzzy')
- self.assertRaises(ValueError, t.read)
+ self.assertRaises(ValueError, t.read) # Check t unreadable
@unittest.skipIf(support.is_jython, "GC nondeterministic in Jython")
def test_garbage_collection(self):
diff --git a/src/org/python/modules/_io/PyFileIO.java b/src/org/python/modules/_io/PyFileIO.java
--- a/src/org/python/modules/_io/PyFileIO.java
+++ b/src/org/python/modules/_io/PyFileIO.java
@@ -390,7 +390,7 @@
throw closedValueError();
}
if (!seekableKnown) {
- seekable = ioDelegate.seek(0, 0) >= 0;
+ seekable = ioDelegate.seek(0, 1) >= 0; // Trial seek
seekableKnown = true;
}
return seekable;
--
Repository URL: http://hg.python.org/jython
More information about the Jython-checkins
mailing list