traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Issue #5734: BufferedRWPair was poorly tested and had several glaring bugs. 

Patch by Brian Quinlan.
This commit is contained in:
Antoine Pitrou 2009-04-19 00:09:36 +00:00
parent 561f36068f
commit cf4c749680
4 changed files with 137 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
Lib/_pyio.py
View File

@ -839,7 +839,9 @@ class BufferedReader(_BufferedIOMixin):
def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):
"""Create a new buffered reader using the given readable raw IO object.
"""
raw._checkReadable()
if not raw.readable():
raise IOError('"raw" argument must be readable.')
_BufferedIOMixin.__init__(self, raw)
if buffer_size <= 0:
raise ValueError("invalid buffer size")
@ -970,7 +972,9 @@ class BufferedWriter(_BufferedIOMixin):
def __init__(self, raw,
buffer_size=DEFAULT_BUFFER_SIZE, max_buffer_size=None):
raw._checkWritable()
if not raw.writable():
raise IOError('"raw" argument must be writable.')
_BufferedIOMixin.__init__(self, raw)
if buffer_size <= 0:
raise ValueError("invalid buffer size")
@ -1076,8 +1080,13 @@ class BufferedRWPair(BufferedIOBase):
"""
if max_buffer_size is not None:
warnings.warn("max_buffer_size is deprecated", DeprecationWarning, 2)
reader._checkReadable()
writer._checkWritable()
if not reader.readable():
raise IOError('"reader" argument must be readable.')
if not writer.writable():
raise IOError('"writer" argument must be writable.')
self.reader = BufferedReader(reader, buffer_size)
self.writer = BufferedWriter(writer, buffer_size)

103
Lib/test/test_io.py
View File

@ -1051,13 +1051,11 @@ class PyBufferedWriterTest(BufferedWriterTest):
class BufferedRWPairTest(unittest.TestCase):
def test_basic(self):
r = self.MockRawIO(())
w = self.MockRawIO()
pair = self.tp(r, w)
def test_constructor(self):
pair = self.tp(self.MockRawIO(), self.MockRawIO())
self.assertFalse(pair.closed)
def test_max_buffer_size_deprecation(self):
def test_constructor_max_buffer_size_deprecation(self):
with support.check_warnings() as w:
warnings.simplefilter("always", DeprecationWarning)
self.tp(self.MockRawIO(), self.MockRawIO(), 8, 12)
@ -1067,7 +1065,100 @@ class BufferedRWPairTest(unittest.TestCase):
self.assertEqual(str(warning.message),
"max_buffer_size is deprecated")
# XXX More Tests
def test_constructor_with_not_readable(self):
class NotReadable(MockRawIO):
def readable(self):
return False
self.assertRaises(IOError, self.tp, NotReadable(), self.MockRawIO())
def test_constructor_with_not_writeable(self):
class NotWriteable(MockRawIO):
def writable(self):
return False
self.assertRaises(IOError, self.tp, self.MockRawIO(), NotWriteable())
def test_read(self):
pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
self.assertEqual(pair.read(3), b"abc")
self.assertEqual(pair.read(1), b"d")
self.assertEqual(pair.read(), b"ef")
def test_read1(self):
# .read1() is delegated to the underlying reader object, so this test
# can be shallow.
pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
self.assertEqual(pair.read1(3), b"abc")
def test_readinto(self):
pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
data = bytearray(5)
self.assertEqual(pair.readinto(data), 5)
self.assertEqual(data, b"abcde")
def test_write(self):
w = self.MockRawIO()
pair = self.tp(self.MockRawIO(), w)
pair.write(b"abc")
pair.flush()
pair.write(b"def")
pair.flush()
self.assertEqual(w._write_stack, [b"abc", b"def"])
def test_peek(self):
pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
self.assertTrue(pair.peek(3).startswith(b"abc"))
self.assertEqual(pair.read(3), b"abc")
def test_readable(self):
pair = self.tp(self.MockRawIO(), self.MockRawIO())
self.assertTrue(pair.readable())
def test_writeable(self):
pair = self.tp(self.MockRawIO(), self.MockRawIO())
self.assertTrue(pair.writable())
def test_seekable(self):
# BufferedRWPairs are never seekable, even if their readers and writers
# are.
pair = self.tp(self.MockRawIO(), self.MockRawIO())
self.assertFalse(pair.seekable())
# .flush() is delegated to the underlying writer object and has been
# tested in the test_write method.
def test_close_and_closed(self):
pair = self.tp(self.MockRawIO(), self.MockRawIO())
self.assertFalse(pair.closed)
pair.close()
self.assertTrue(pair.closed)
def test_isatty(self):
class SelectableIsAtty(MockRawIO):
def __init__(self, isatty):
MockRawIO.__init__(self)
self._isatty = isatty
def isatty(self):
return self._isatty
pair = self.tp(SelectableIsAtty(False), SelectableIsAtty(False))
self.assertFalse(pair.isatty())
pair = self.tp(SelectableIsAtty(True), SelectableIsAtty(False))
self.assertTrue(pair.isatty())
pair = self.tp(SelectableIsAtty(False), SelectableIsAtty(True))
self.assertTrue(pair.isatty())
pair = self.tp(SelectableIsAtty(True), SelectableIsAtty(True))
self.assertTrue(pair.isatty())
class CBufferedRWPairTest(BufferedRWPairTest):
tp = io.BufferedRWPair

3
Misc/NEWS
View File

@ -72,6 +72,9 @@ Core and Builtins
Library
-------
- Issue #5734: BufferedRWPair was poorly tested and had several glaring
bugs. Patch by Brian Quinlan.
- Issue #1161031: fix readwrite select flag handling: POLLPRI now
results in a handle_expt_event call, not handle_read_event, and POLLERR
and POLLNVAL now call handle_close, not handle_expt_event. Also,

43
Modules/_io/bufferedio.c
View File

@ -1846,7 +1846,7 @@ typedef struct {
static int
BufferedRWPair_init(BufferedRWPairObject *self, PyObject *args,
PyObject *kwds)
PyObject *kwds)
{
PyObject *reader, *writer;
Py_ssize_t buffer_size = DEFAULT_BUFFER_SIZE;
@ -1865,29 +1865,18 @@ BufferedRWPair_init(BufferedRWPairObject *self, PyObject *args,
if (_PyIOBase_checkWritable(writer, Py_True) == NULL)
return -1;
args = Py_BuildValue("(n)", buffer_size);
if (args == NULL) {
Py_CLEAR(self->reader);
return -1;
}
self->reader = (BufferedObject *)PyType_GenericNew(
&PyBufferedReader_Type, args, NULL);
Py_DECREF(args);
self->reader = (BufferedObject *) PyObject_CallFunction(
(PyObject *) &PyBufferedReader_Type, "On", reader, buffer_size);
if (self->reader == NULL)
return -1;
args = Py_BuildValue("(n)", buffer_size);
if (args == NULL) {
Py_CLEAR(self->reader);
return -1;
}
self->writer = (BufferedObject *)PyType_GenericNew(
&PyBufferedWriter_Type, args, NULL);
Py_DECREF(args);
self->writer = (BufferedObject *) PyObject_CallFunction(
(PyObject *) &PyBufferedWriter_Type, "On", writer, buffer_size);
if (self->writer == NULL) {
Py_CLEAR(self->reader);
return -1;
}
return 0;
}
@ -1951,6 +1940,12 @@ BufferedRWPair_read1(BufferedRWPairObject *self, PyObject *args)
return _forward_call(self->reader, "read1", args);
}
static PyObject *
BufferedRWPair_readinto(BufferedRWPairObject *self, PyObject *args)
{
return _forward_call(self->reader, "readinto", args);
}
static PyObject *
BufferedRWPair_write(BufferedRWPairObject *self, PyObject *args)
{
@ -2000,12 +1995,17 @@ BufferedRWPair_isatty(BufferedRWPairObject *self, PyObject *args)
return _forward_call(self->reader, "isatty", args);
}
static PyObject *
BufferedRWPair_closed_get(BufferedRWPairObject *self, void *context)
{
return PyObject_GetAttr((PyObject *) self->writer, _PyIO_str_closed);
}
static PyMethodDef BufferedRWPair_methods[] = {
{"read", (PyCFunction)BufferedRWPair_read, METH_VARARGS},
{"peek", (PyCFunction)BufferedRWPair_peek, METH_VARARGS},
{"read1", (PyCFunction)BufferedRWPair_read1, METH_VARARGS},
{"readinto", (PyCFunction)Buffered_readinto, METH_VARARGS},
{"readinto", (PyCFunction)BufferedRWPair_readinto, METH_VARARGS},
{"write", (PyCFunction)BufferedRWPair_write, METH_VARARGS},
{"flush", (PyCFunction)BufferedRWPair_flush, METH_NOARGS},
@ -2019,6 +2019,11 @@ static PyMethodDef BufferedRWPair_methods[] = {
{NULL, NULL}
};
static PyGetSetDef BufferedRWPair_getset[] = {
{"closed", (getter)BufferedRWPair_closed_get, NULL, NULL},
{0}
};
PyTypeObject PyBufferedRWPair_Type = {
PyVarObject_HEAD_INIT(NULL, 0)
"_io.BufferedRWPair", /*tp_name*/
@ -2050,7 +2055,7 @@ PyTypeObject PyBufferedRWPair_Type = {
0, /* tp_iternext */
BufferedRWPair_methods, /* tp_methods */
0, /* tp_members */
0, /* tp_getset */
BufferedRWPair_getset, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */