provides additional utility functions and classes that allow the use of a
wider range of codecs when working with binary files:
-.. function:: open(filename, mode='r', encoding=None, errors='strict', buffering=1)
+.. function:: open(filename, mode='r', encoding=None, errors='strict', buffering=-1)
Open an encoded file using the given *mode* and return an instance of
:class:`StreamReaderWriter`, providing transparent encoding/decoding.
*errors* may be given to define the error handling. It defaults to ``'strict'``
which causes a :exc:`ValueError` to be raised in case an encoding error occurs.
- *buffering* has the same meaning as for the built-in :func:`open` function. It
- defaults to line buffered.
+ *buffering* has the same meaning as for the built-in :func:`open` function.
+ It defaults to -1 which means that the default buffer size will be used.
.. function:: EncodedFile(file, data_encoding, file_encoding=None, errors='strict')
raise ValueError("binary mode doesn't take an errors argument")
if binary and newline is not None:
raise ValueError("binary mode doesn't take a newline argument")
+ if binary and buffering == 1:
+ import warnings
+ warnings.warn("line buffering (buffering=1) isn't supported in binary "
+ "mode, the default buffer size will be used",
+ RuntimeWarning, 2)
raw = FileIO(file,
(creating and "x" or "") +
(reading and "r" or "") +
### Shortcuts
-def open(filename, mode='r', encoding=None, errors='strict', buffering=1):
+def open(filename, mode='r', encoding=None, errors='strict', buffering=-1):
""" Open an encoded file using the given mode and return
a wrapped version providing transparent encoding/decoding.
encoding error occurs.
buffering has the same meaning as for the builtin open() API.
- It defaults to line buffered.
+ It defaults to -1 which means that the default buffer size will
+ be used.
The returned wrapped file object provides an extra attribute
.encoding which allows querying the used encoding. This
self._closed_child_pipe_fds = False
+ if self.text_mode:
+ if bufsize == 1:
+ line_buffering = True
+ # Use the default buffer size for the underlying binary streams
+ # since they don't support line buffering.
+ bufsize = -1
+ else:
+ line_buffering = False
+
try:
if p2cwrite != -1:
self.stdin = io.open(p2cwrite, 'wb', bufsize)
if self.text_mode:
self.stdin = io.TextIOWrapper(self.stdin, write_through=True,
- line_buffering=(bufsize == 1),
+ line_buffering=line_buffering,
encoding=encoding, errors=errors)
if c2pread != -1:
self.stdout = io.open(c2pread, 'rb', bufsize)
# threads
"threading_setup", "threading_cleanup", "reap_threads", "start_threads",
# miscellaneous
- "check_warnings", "check_no_resource_warning", "EnvironmentVarGuard",
+ "check_warnings", "check_no_resource_warning", "check_no_warnings",
+ "EnvironmentVarGuard",
"run_with_locale", "swap_item",
"swap_attr", "Matcher", "set_memlimit", "SuppressCrashReport", "sortdict",
"run_with_tz", "PGO", "missing_compiler_executable", "fd_count",
return _filterwarnings(filters, quiet)
+@contextlib.contextmanager
+def check_no_warnings(testcase, message='', category=Warning, force_gc=False):
+ """Context manager to check that no warnings are emitted.
+
+ This context manager enables a given warning within its scope
+ and checks that no warnings are emitted even with that warning
+ enabled.
+
+ If force_gc is True, a garbage collection is attempted before checking
+ for warnings. This may help to catch warnings emitted when objects
+ are deleted, such as ResourceWarning.
+
+ Other keyword arguments are passed to warnings.filterwarnings().
+ """
+ with warnings.catch_warnings(record=True) as warns:
+ warnings.filterwarnings('always',
+ message=message,
+ category=category)
+ yield
+ if force_gc:
+ gc_collect()
+ testcase.assertEqual(warns, [])
+
+
@contextlib.contextmanager
def check_no_resource_warning(testcase):
"""Context manager to check that no ResourceWarning is emitted.
You must remove the object which may emit ResourceWarning before
the end of the context manager.
"""
- with warnings.catch_warnings(record=True) as warns:
- warnings.filterwarnings('always', category=ResourceWarning)
+ with check_no_warnings(testcase, category=ResourceWarning, force_gc=True):
yield
- gc_collect()
- testcase.assertEqual(warns, [])
class CleanImport(object):
@contextlib.contextmanager
def interactive_python(self, separate_stderr=False):
if separate_stderr:
- p = spawn_python('-i', bufsize=1, stderr=subprocess.PIPE)
+ p = spawn_python('-i', stderr=subprocess.PIPE)
stderr = p.stderr
else:
- p = spawn_python('-i', bufsize=1, stderr=subprocess.STDOUT)
+ p = spawn_python('-i', stderr=subprocess.STDOUT)
stderr = p.stdout
try:
# Drain stderr until prompt
f.close()
self.fail("no error for invalid mode: %s" % bad_mode)
+ def _checkBufferSize(self, s):
+ try:
+ f = self.open(TESTFN, 'wb', s)
+ f.write(str(s).encode("ascii"))
+ f.close()
+ f.close()
+ f = self.open(TESTFN, 'rb', s)
+ d = int(f.read().decode("ascii"))
+ f.close()
+ f.close()
+ except OSError as msg:
+ self.fail('error setting buffer size %d: %s' % (s, str(msg)))
+ self.assertEqual(d, s)
+
def testSetBufferSize(self):
# make sure that explicitly setting the buffer size doesn't cause
# misbehaviour especially with repeated close() calls
- for s in (-1, 0, 1, 512):
- try:
- f = self.open(TESTFN, 'wb', s)
- f.write(str(s).encode("ascii"))
- f.close()
- f.close()
- f = self.open(TESTFN, 'rb', s)
- d = int(f.read().decode("ascii"))
- f.close()
- f.close()
- except OSError as msg:
- self.fail('error setting buffer size %d: %s' % (s, str(msg)))
- self.assertEqual(d, s)
+ for s in (-1, 0, 512):
+ with support.check_no_warnings(self,
+ message='line buffering',
+ category=RuntimeWarning):
+ self._checkBufferSize(s)
+
+ # test that attempts to use line buffering in binary mode cause
+ # a warning
+ with self.assertWarnsRegex(RuntimeWarning, 'line buffering'):
+ self._checkBufferSize(1)
def testTruncateOnWindows(self):
# SF bug <http://www.python.org/sf/801631>
self.large_file_ops(f)
def test_with_open(self):
- for bufsize in (0, 1, 100):
+ for bufsize in (0, 100):
f = None
with self.open(support.TESTFN, "wb", bufsize) as f:
f.write(b"xxx")
# line is not flushed in binary mode with bufsize=1.
# we should get empty response
line = b'line' + os.linesep.encode() # assume ascii-based locale
- self._test_bufsize_equal_one(line, b'', universal_newlines=False)
+ with self.assertWarnsRegex(RuntimeWarning, 'line buffering'):
+ self._test_bufsize_equal_one(line, b'', universal_newlines=False)
def test_leaking_fds_on_error(self):
# see bug #5179: Popen leaks file descriptors to PIPEs if
--- /dev/null
+Warn that line buffering is not supported if :func:`open` is called with
+binary mode and ``buffering=1``.
goto error;
}
+ if (binary && buffering == 1) {
+ if (PyErr_WarnEx(PyExc_RuntimeWarning,
+ "line buffering (buffering=1) isn't supported in "
+ "binary mode, the default buffer size will be used",
+ 1) < 0) {
+ goto error;
+ }
+ }
+
/* Create the Raw file stream */
{
PyObject *RawIO_class = (PyObject *)&PyFileIO_Type;
projects / python / commitdiff