Patch by Brian Quinlan.
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")
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")
"""
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)
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)
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
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,
static int
BufferedRWPair_init(BufferedRWPairObject *self, PyObject *args,
- PyObject *kwds)
+ PyObject *kwds)
{
PyObject *reader, *writer;
Py_ssize_t buffer_size = DEFAULT_BUFFER_SIZE;
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;
}
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)
{
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},
{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*/
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 */
projects / python / commitdiff