with self.assertRaisesRegex(TypeError, "BufferedWriter"):
self.tp(self.BytesIO(), 1024, 1024, 1024)
+ def test_non_boolean_closed_attr(self):
+ # gh-140650: check TypeError is raised
+ class MockRawIOWithoutClosed(self.MockRawIO):
+ closed = NotImplemented
+
+ bufio = self.tp(MockRawIOWithoutClosed())
+ self.assertRaises(TypeError, bufio.write, b"")
+ self.assertRaises(TypeError, bufio.flush)
+ self.assertRaises(TypeError, bufio.close)
+
+ def test_closed_attr_raises(self):
+ class MockRawIOClosedRaises(self.MockRawIO):
+ @property
+ def closed(self):
+ raise ValueError("test")
+
+ bufio = self.tp(MockRawIOClosedRaises())
+ self.assertRaisesRegex(ValueError, "test", bufio.write, b"")
+ self.assertRaisesRegex(ValueError, "test", bufio.flush)
+ self.assertRaisesRegex(ValueError, "test", bufio.close)
+
class PyBufferedWriterTest(BufferedWriterTest, PyTestCase):
tp = pyio.BufferedWriter
}
#define IS_CLOSED(self) \
- (!self->buffer || \
+ (!self->buffer ? 1 : \
(self->fast_closed_checks \
? _PyFileIO_closed(self->raw) \
: buffered_closed(self)))
#define CHECK_CLOSED(self, error_msg) \
- if (IS_CLOSED(self) && (Py_SAFE_DOWNCAST(READAHEAD(self), Py_off_t, Py_ssize_t) == 0)) { \
- PyErr_SetString(PyExc_ValueError, error_msg); \
- return NULL; \
- } \
+ do { \
+ int _closed = IS_CLOSED(self); \
+ if (_closed < 0) { \
+ return NULL; \
+ } \
+ if (_closed && \
+ (Py_SAFE_DOWNCAST(READAHEAD(self), Py_off_t, Py_ssize_t) == 0)) \
+ { \
+ PyErr_SetString(PyExc_ValueError, error_msg); \
+ return NULL; \
+ } \
+ } while (0);
#define VALID_READ_BUFFER(self) \
(self->readable && self->read_end != -1)
PyObject *res = NULL;
Py_ssize_t written, avail, remaining;
Py_off_t offset;
+ int r;
CHECK_INITIALIZED(self)
/* Issue #31976: Check for closed file after acquiring the lock. Another
thread could be holding the lock while closing the file. */
- if (IS_CLOSED(self)) {
+ r = IS_CLOSED(self);
+ if (r < 0) {
+ goto error;
+ }
+ if (r > 0) {
PyErr_SetString(PyExc_ValueError, "write to closed file");
goto error;
}