| """General tests for the io module. |
| |
| New tests should go in more specific modules; see test_io/__init__.py |
| """ |
| |
| import abc |
| import array |
| import errno |
| import os |
| import pickle |
| import sys |
| import textwrap |
| import threading |
| import unittest |
| import warnings |
| import weakref |
| from test import support |
| from test.support.script_helper import ( |
| assert_python_ok, assert_python_failure, run_python_until_end) |
| from test.support import ( |
| import_helper, is_apple, os_helper, threading_helper, warnings_helper, |
| ) |
| from test.support.os_helper import FakePath |
| from .utils import byteslike, CTestCase, PyTestCase |
| |
| import io # C implementation of io |
| import _pyio as pyio # Python implementation of io |
| |
| |
| class IOTest: |
| |
| def setUp(self): |
| os_helper.unlink(os_helper.TESTFN) |
| |
| def tearDown(self): |
| os_helper.unlink(os_helper.TESTFN) |
| |
| def write_ops(self, f): |
| self.assertEqual(f.write(b"blah."), 5) |
| f.truncate(0) |
| self.assertEqual(f.tell(), 5) |
| f.seek(0) |
| |
| self.assertEqual(f.write(b"blah."), 5) |
| self.assertEqual(f.seek(0), 0) |
| self.assertEqual(f.write(b"Hello."), 6) |
| self.assertEqual(f.tell(), 6) |
| self.assertEqual(f.seek(-1, 1), 5) |
| self.assertEqual(f.tell(), 5) |
| buffer = bytearray(b" world\n\n\n") |
| self.assertEqual(f.write(buffer), 9) |
| buffer[:] = b"*" * 9 # Overwrite our copy of the data |
| self.assertEqual(f.seek(0), 0) |
| self.assertEqual(f.write(b"h"), 1) |
| self.assertEqual(f.seek(-1, 2), 13) |
| self.assertEqual(f.tell(), 13) |
| |
| self.assertEqual(f.truncate(12), 12) |
| self.assertEqual(f.tell(), 13) |
| self.assertRaises(TypeError, f.seek, 0.0) |
| |
| def read_ops(self, f, buffered=False): |
| data = f.read(5) |
| self.assertEqual(data, b"hello") |
| data = byteslike(data) |
| self.assertEqual(f.readinto(data), 5) |
| self.assertEqual(bytes(data), b" worl") |
| data = bytearray(5) |
| self.assertEqual(f.readinto(data), 2) |
| self.assertEqual(len(data), 5) |
| self.assertEqual(data[:2], b"d\n") |
| self.assertEqual(f.seek(0), 0) |
| self.assertEqual(f.read(20), b"hello world\n") |
| self.assertEqual(f.read(1), b"") |
| self.assertEqual(f.readinto(byteslike(b"x")), 0) |
| self.assertEqual(f.seek(-6, 2), 6) |
| self.assertEqual(f.read(5), b"world") |
| self.assertEqual(f.read(0), b"") |
| self.assertEqual(f.readinto(byteslike()), 0) |
| self.assertEqual(f.seek(-6, 1), 5) |
| self.assertEqual(f.read(5), b" worl") |
| self.assertEqual(f.tell(), 10) |
| self.assertRaises(TypeError, f.seek, 0.0) |
| if buffered: |
| f.seek(0) |
| self.assertEqual(f.read(), b"hello world\n") |
| f.seek(6) |
| self.assertEqual(f.read(), b"world\n") |
| self.assertEqual(f.read(), b"") |
| f.seek(0) |
| data = byteslike(5) |
| self.assertEqual(f.readinto1(data), 5) |
| self.assertEqual(bytes(data), b"hello") |
| |
| LARGE = 2**31 |
| |
| def large_file_ops(self, f): |
| assert f.readable() |
| assert f.writable() |
| try: |
| self.assertEqual(f.seek(self.LARGE), self.LARGE) |
| except (OverflowError, ValueError): |
| self.skipTest("no largefile support") |
| self.assertEqual(f.tell(), self.LARGE) |
| self.assertEqual(f.write(b"xxx"), 3) |
| self.assertEqual(f.tell(), self.LARGE + 3) |
| self.assertEqual(f.seek(-1, 1), self.LARGE + 2) |
| self.assertEqual(f.truncate(), self.LARGE + 2) |
| self.assertEqual(f.tell(), self.LARGE + 2) |
| self.assertEqual(f.seek(0, 2), self.LARGE + 2) |
| self.assertEqual(f.truncate(self.LARGE + 1), self.LARGE + 1) |
| self.assertEqual(f.tell(), self.LARGE + 2) |
| self.assertEqual(f.seek(0, 2), self.LARGE + 1) |
| self.assertEqual(f.seek(-1, 2), self.LARGE) |
| self.assertEqual(f.read(2), b"x") |
| |
| def test_invalid_operations(self): |
| # Try writing on a file opened in read mode and vice-versa. |
| exc = self.UnsupportedOperation |
| with self.open(os_helper.TESTFN, "w", encoding="utf-8") as fp: |
| self.assertRaises(exc, fp.read) |
| self.assertRaises(exc, fp.readline) |
| with self.open(os_helper.TESTFN, "wb") as fp: |
| self.assertRaises(exc, fp.read) |
| self.assertRaises(exc, fp.readline) |
| with self.open(os_helper.TESTFN, "wb", buffering=0) as fp: |
| self.assertRaises(exc, fp.read) |
| self.assertRaises(exc, fp.readall) |
| self.assertRaises(exc, fp.readline) |
| with self.open(os_helper.TESTFN, "rb", buffering=0) as fp: |
| self.assertRaises(exc, fp.write, b"blah") |
| self.assertRaises(exc, fp.writelines, [b"blah\n"]) |
| with self.open(os_helper.TESTFN, "rb") as fp: |
| self.assertRaises(exc, fp.write, b"blah") |
| self.assertRaises(exc, fp.writelines, [b"blah\n"]) |
| with self.open(os_helper.TESTFN, "r", encoding="utf-8") as fp: |
| self.assertRaises(exc, fp.write, "blah") |
| self.assertRaises(exc, fp.writelines, ["blah\n"]) |
| # Non-zero seeking from current or end pos |
| self.assertRaises(exc, fp.seek, 1, self.SEEK_CUR) |
| self.assertRaises(exc, fp.seek, -1, self.SEEK_END) |
| |
| @support.cpython_only |
| def test_startup_optimization(self): |
| # gh-132952: Test that `io` is not imported at startup and that the |
| # __module__ of UnsupportedOperation is set to "io". |
| assert_python_ok("-S", "-c", textwrap.dedent( |
| """ |
| import sys |
| assert "io" not in sys.modules |
| try: |
| sys.stdin.truncate() |
| except Exception as e: |
| typ = type(e) |
| assert typ.__module__ == "io", (typ, typ.__module__) |
| assert typ.__name__ == "UnsupportedOperation", (typ, typ.__name__) |
| else: |
| raise AssertionError("Expected UnsupportedOperation") |
| """ |
| )) |
| |
| @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") |
| def test_optional_abilities(self): |
| # Test for OSError when optional APIs are not supported |
| # The purpose of this test is to try fileno(), reading, writing and |
| # seeking operations with various objects that indicate they do not |
| # support these operations. |
| |
| def pipe_reader(): |
| [r, w] = os.pipe() |
| os.close(w) # So that read() is harmless |
| return self.FileIO(r, "r") |
| |
| def pipe_writer(): |
| [r, w] = os.pipe() |
| self.addCleanup(os.close, r) |
| # Guarantee that we can write into the pipe without blocking |
| thread = threading.Thread(target=os.read, args=(r, 100)) |
| thread.start() |
| self.addCleanup(thread.join) |
| return self.FileIO(w, "w") |
| |
| def buffered_reader(): |
| return self.BufferedReader(self.MockUnseekableIO()) |
| |
| def buffered_writer(): |
| return self.BufferedWriter(self.MockUnseekableIO()) |
| |
| def buffered_random(): |
| return self.BufferedRandom(self.BytesIO()) |
| |
| def buffered_rw_pair(): |
| return self.BufferedRWPair(self.MockUnseekableIO(), |
| self.MockUnseekableIO()) |
| |
| def text_reader(): |
| class UnseekableReader(self.MockUnseekableIO): |
| writable = self.BufferedIOBase.writable |
| write = self.BufferedIOBase.write |
| return self.TextIOWrapper(UnseekableReader(), "ascii") |
| |
| def text_writer(): |
| class UnseekableWriter(self.MockUnseekableIO): |
| readable = self.BufferedIOBase.readable |
| read = self.BufferedIOBase.read |
| return self.TextIOWrapper(UnseekableWriter(), "ascii") |
| |
| tests = ( |
| (pipe_reader, "fr"), (pipe_writer, "fw"), |
| (buffered_reader, "r"), (buffered_writer, "w"), |
| (buffered_random, "rws"), (buffered_rw_pair, "rw"), |
| (text_reader, "r"), (text_writer, "w"), |
| (self.BytesIO, "rws"), (self.StringIO, "rws"), |
| ) |
| |
| def do_test(test, obj, abilities): |
| readable = "r" in abilities |
| self.assertEqual(obj.readable(), readable) |
| writable = "w" in abilities |
| self.assertEqual(obj.writable(), writable) |
| |
| if isinstance(obj, self.TextIOBase): |
| data = "3" |
| elif isinstance(obj, (self.BufferedIOBase, self.RawIOBase)): |
| data = b"3" |
| else: |
| self.fail("Unknown base class") |
| |
| if "f" in abilities: |
| obj.fileno() |
| else: |
| self.assertRaises(OSError, obj.fileno) |
| |
| if readable: |
| obj.read(1) |
| obj.read() |
| else: |
| self.assertRaises(OSError, obj.read, 1) |
| self.assertRaises(OSError, obj.read) |
| |
| if writable: |
| obj.write(data) |
| else: |
| self.assertRaises(OSError, obj.write, data) |
| |
| if sys.platform.startswith("win") and test in ( |
| pipe_reader, pipe_writer): |
| # Pipes seem to appear as seekable on Windows |
| return |
| seekable = "s" in abilities |
| self.assertEqual(obj.seekable(), seekable) |
| |
| if seekable: |
| obj.tell() |
| obj.seek(0) |
| else: |
| self.assertRaises(OSError, obj.tell) |
| self.assertRaises(OSError, obj.seek, 0) |
| |
| if writable and seekable: |
| obj.truncate() |
| obj.truncate(0) |
| else: |
| self.assertRaises(OSError, obj.truncate) |
| self.assertRaises(OSError, obj.truncate, 0) |
| |
| for [test, abilities] in tests: |
| with self.subTest(test): |
| if test == pipe_writer and not threading_helper.can_start_thread: |
| self.skipTest("Need threads") |
| with test() as obj: |
| do_test(test, obj, abilities) |
| |
| |
| def test_open_handles_NUL_chars(self): |
| fn_with_NUL = 'foo\0bar' |
| self.assertRaises(ValueError, self.open, fn_with_NUL, 'w', encoding="utf-8") |
| |
| bytes_fn = bytes(fn_with_NUL, 'ascii') |
| with warnings.catch_warnings(): |
| warnings.simplefilter("ignore", DeprecationWarning) |
| self.assertRaises(ValueError, self.open, bytes_fn, 'w', encoding="utf-8") |
| |
| def test_raw_file_io(self): |
| with self.open(os_helper.TESTFN, "wb", buffering=0) as f: |
| self.assertEqual(f.readable(), False) |
| self.assertEqual(f.writable(), True) |
| self.assertEqual(f.seekable(), True) |
| self.write_ops(f) |
| with self.open(os_helper.TESTFN, "rb", buffering=0) as f: |
| self.assertEqual(f.readable(), True) |
| self.assertEqual(f.writable(), False) |
| self.assertEqual(f.seekable(), True) |
| self.read_ops(f) |
| |
| def test_buffered_file_io(self): |
| with self.open(os_helper.TESTFN, "wb") as f: |
| self.assertEqual(f.readable(), False) |
| self.assertEqual(f.writable(), True) |
| self.assertEqual(f.seekable(), True) |
| self.write_ops(f) |
| with self.open(os_helper.TESTFN, "rb") as f: |
| self.assertEqual(f.readable(), True) |
| self.assertEqual(f.writable(), False) |
| self.assertEqual(f.seekable(), True) |
| self.read_ops(f, True) |
| |
| def test_readline(self): |
| with self.open(os_helper.TESTFN, "wb") as f: |
| f.write(b"abc\ndef\nxyzzy\nfoo\x00bar\nanother line") |
| with self.open(os_helper.TESTFN, "rb") as f: |
| self.assertEqual(f.readline(), b"abc\n") |
| self.assertEqual(f.readline(10), b"def\n") |
| self.assertEqual(f.readline(2), b"xy") |
| self.assertEqual(f.readline(4), b"zzy\n") |
| self.assertEqual(f.readline(), b"foo\x00bar\n") |
| self.assertEqual(f.readline(None), b"another line") |
| self.assertRaises(TypeError, f.readline, 5.3) |
| with self.open(os_helper.TESTFN, "r", encoding="utf-8") as f: |
| self.assertRaises(TypeError, f.readline, 5.3) |
| |
| def test_readline_nonsizeable(self): |
| # Issue #30061 |
| # Crash when readline() returns an object without __len__ |
| class R(self.IOBase): |
| def readline(self): |
| return None |
| self.assertRaises((TypeError, StopIteration), next, R()) |
| |
| def test_next_nonsizeable(self): |
| # Issue #30061 |
| # Crash when __next__() returns an object without __len__ |
| class R(self.IOBase): |
| def __next__(self): |
| return None |
| self.assertRaises(TypeError, R().readlines, 1) |
| |
| def test_raw_bytes_io(self): |
| f = self.BytesIO() |
| self.write_ops(f) |
| data = f.getvalue() |
| self.assertEqual(data, b"hello world\n") |
| f = self.BytesIO(data) |
| self.read_ops(f, True) |
| |
| def test_large_file_ops(self): |
| # On Windows and Apple platforms this test consumes large resources; It |
| # takes a long time to build the >2 GiB file and takes >2 GiB of disk |
| # space therefore the resource must be enabled to run this test. |
| if sys.platform[:3] == 'win' or is_apple: |
| support.requires( |
| 'largefile', |
| 'test requires %s bytes and a long time to run' % self.LARGE) |
| with self.open(os_helper.TESTFN, "w+b", 0) as f: |
| self.large_file_ops(f) |
| with self.open(os_helper.TESTFN, "w+b") as f: |
| self.large_file_ops(f) |
| |
| def test_with_open(self): |
| for bufsize in (0, 100): |
| with self.open(os_helper.TESTFN, "wb", bufsize) as f: |
| f.write(b"xxx") |
| self.assertEqual(f.closed, True) |
| try: |
| with self.open(os_helper.TESTFN, "wb", bufsize) as f: |
| 1/0 |
| except ZeroDivisionError: |
| self.assertEqual(f.closed, True) |
| else: |
| self.fail("1/0 didn't raise an exception") |
| |
| # issue 5008 |
| def test_append_mode_tell(self): |
| with self.open(os_helper.TESTFN, "wb") as f: |
| f.write(b"xxx") |
| with self.open(os_helper.TESTFN, "ab", buffering=0) as f: |
| self.assertEqual(f.tell(), 3) |
| with self.open(os_helper.TESTFN, "ab") as f: |
| self.assertEqual(f.tell(), 3) |
| with self.open(os_helper.TESTFN, "a", encoding="utf-8") as f: |
| self.assertGreater(f.tell(), 0) |
| |
| def test_destructor(self): |
| record = [] |
| class MyFileIO(self.FileIO): |
| def __del__(self): |
| record.append(1) |
| try: |
| f = super().__del__ |
| except AttributeError: |
| pass |
| else: |
| f() |
| def close(self): |
| record.append(2) |
| super().close() |
| def flush(self): |
| record.append(3) |
| super().flush() |
| with warnings_helper.check_warnings(('', ResourceWarning)): |
| f = MyFileIO(os_helper.TESTFN, "wb") |
| f.write(b"xxx") |
| del f |
| support.gc_collect() |
| self.assertEqual(record, [1, 2, 3]) |
| with self.open(os_helper.TESTFN, "rb") as f: |
| self.assertEqual(f.read(), b"xxx") |
| |
| def _check_base_destructor(self, base): |
| record = [] |
| class MyIO(base): |
| def __init__(self): |
| # This exercises the availability of attributes on object |
| # destruction. |
| # (in the C version, close() is called by the tp_dealloc |
| # function, not by __del__) |
| self.on_del = 1 |
| self.on_close = 2 |
| self.on_flush = 3 |
| def __del__(self): |
| record.append(self.on_del) |
| try: |
| f = super().__del__ |
| except AttributeError: |
| pass |
| else: |
| f() |
| def close(self): |
| record.append(self.on_close) |
| super().close() |
| def flush(self): |
| record.append(self.on_flush) |
| super().flush() |
| f = MyIO() |
| del f |
| support.gc_collect() |
| self.assertEqual(record, [1, 2, 3]) |
| |
| def test_IOBase_destructor(self): |
| self._check_base_destructor(self.IOBase) |
| |
| def test_RawIOBase_destructor(self): |
| self._check_base_destructor(self.RawIOBase) |
| |
| def test_BufferedIOBase_destructor(self): |
| self._check_base_destructor(self.BufferedIOBase) |
| |
| def test_TextIOBase_destructor(self): |
| self._check_base_destructor(self.TextIOBase) |
| |
| def test_close_flushes(self): |
| with self.open(os_helper.TESTFN, "wb") as f: |
| f.write(b"xxx") |
| with self.open(os_helper.TESTFN, "rb") as f: |
| self.assertEqual(f.read(), b"xxx") |
| |
| def test_array_writes(self): |
| a = array.array('i', range(10)) |
| n = len(a.tobytes()) |
| def check(f): |
| with f: |
| self.assertEqual(f.write(a), n) |
| f.writelines((a,)) |
| check(self.BytesIO()) |
| check(self.FileIO(os_helper.TESTFN, "w")) |
| check(self.BufferedWriter(self.MockRawIO())) |
| check(self.BufferedRandom(self.MockRawIO())) |
| check(self.BufferedRWPair(self.MockRawIO(), self.MockRawIO())) |
| |
| def test_closefd(self): |
| self.assertRaises(ValueError, self.open, os_helper.TESTFN, 'w', |
| encoding="utf-8", closefd=False) |
| |
| def test_read_closed(self): |
| with self.open(os_helper.TESTFN, "w", encoding="utf-8") as f: |
| f.write("egg\n") |
| with self.open(os_helper.TESTFN, "r", encoding="utf-8") as f: |
| file = self.open(f.fileno(), "r", encoding="utf-8", closefd=False) |
| self.assertEqual(file.read(), "egg\n") |
| file.seek(0) |
| file.close() |
| self.assertRaises(ValueError, file.read) |
| with self.open(os_helper.TESTFN, "rb") as f: |
| file = self.open(f.fileno(), "rb", closefd=False) |
| self.assertEqual(file.read()[:3], b"egg") |
| file.close() |
| self.assertRaises(ValueError, file.readinto, bytearray(1)) |
| |
| def test_no_closefd_with_filename(self): |
| # can't use closefd in combination with a file name |
| self.assertRaises(ValueError, self.open, os_helper.TESTFN, "r", |
| encoding="utf-8", closefd=False) |
| |
| def test_closefd_attr(self): |
| with self.open(os_helper.TESTFN, "wb") as f: |
| f.write(b"egg\n") |
| with self.open(os_helper.TESTFN, "r", encoding="utf-8") as f: |
| self.assertEqual(f.buffer.raw.closefd, True) |
| file = self.open(f.fileno(), "r", encoding="utf-8", closefd=False) |
| self.assertEqual(file.buffer.raw.closefd, False) |
| |
| def test_garbage_collection(self): |
| # FileIO objects are collected, and collecting them flushes |
| # all data to disk. |
| # |
| # Note that using warnings_helper.check_warnings() will keep the |
| # file alive due to the `source` argument to warn(). So, use |
| # catch_warnings() instead. |
| with warnings.catch_warnings(): |
| warnings.simplefilter("ignore", ResourceWarning) |
| f = self.FileIO(os_helper.TESTFN, "wb") |
| f.write(b"abcxxx") |
| f.f = f |
| wr = weakref.ref(f) |
| del f |
| support.gc_collect() |
| self.assertIsNone(wr(), wr) |
| with self.open(os_helper.TESTFN, "rb") as f: |
| self.assertEqual(f.read(), b"abcxxx") |
| |
| def test_unbounded_file(self): |
| # Issue #1174606: reading from an unbounded stream such as /dev/zero. |
| zero = "/dev/zero" |
| if not os.path.exists(zero): |
| self.skipTest("{0} does not exist".format(zero)) |
| if sys.maxsize > 0x7FFFFFFF: |
| self.skipTest("test can only run in a 32-bit address space") |
| if support.real_max_memuse < support._2G: |
| self.skipTest("test requires at least 2 GiB of memory") |
| with self.open(zero, "rb", buffering=0) as f: |
| self.assertRaises(OverflowError, f.read) |
| with self.open(zero, "rb") as f: |
| self.assertRaises(OverflowError, f.read) |
| with self.open(zero, "r") as f: |
| self.assertRaises(OverflowError, f.read) |
| |
| def check_flush_error_on_close(self, *args, **kwargs): |
| # Test that the file is closed despite failed flush |
| # and that flush() is called before file closed. |
| f = self.open(*args, **kwargs) |
| closed = [] |
| def bad_flush(): |
| closed[:] = [f.closed] |
| raise OSError() |
| f.flush = bad_flush |
| self.assertRaises(OSError, f.close) # exception not swallowed |
| self.assertTrue(f.closed) |
| self.assertTrue(closed) # flush() called |
| self.assertFalse(closed[0]) # flush() called before file closed |
| f.flush = lambda: None # break reference loop |
| |
| def test_flush_error_on_close(self): |
| # raw file |
| # Issue #5700: io.FileIO calls flush() after file closed |
| self.check_flush_error_on_close(os_helper.TESTFN, 'wb', buffering=0) |
| fd = os.open(os_helper.TESTFN, os.O_WRONLY|os.O_CREAT) |
| self.check_flush_error_on_close(fd, 'wb', buffering=0) |
| fd = os.open(os_helper.TESTFN, os.O_WRONLY|os.O_CREAT) |
| self.check_flush_error_on_close(fd, 'wb', buffering=0, closefd=False) |
| os.close(fd) |
| # buffered io |
| self.check_flush_error_on_close(os_helper.TESTFN, 'wb') |
| fd = os.open(os_helper.TESTFN, os.O_WRONLY|os.O_CREAT) |
| self.check_flush_error_on_close(fd, 'wb') |
| fd = os.open(os_helper.TESTFN, os.O_WRONLY|os.O_CREAT) |
| self.check_flush_error_on_close(fd, 'wb', closefd=False) |
| os.close(fd) |
| # text io |
| self.check_flush_error_on_close(os_helper.TESTFN, 'w', encoding="utf-8") |
| fd = os.open(os_helper.TESTFN, os.O_WRONLY|os.O_CREAT) |
| self.check_flush_error_on_close(fd, 'w', encoding="utf-8") |
| fd = os.open(os_helper.TESTFN, os.O_WRONLY|os.O_CREAT) |
| self.check_flush_error_on_close(fd, 'w', encoding="utf-8", closefd=False) |
| os.close(fd) |
| |
| def test_multi_close(self): |
| f = self.open(os_helper.TESTFN, "wb", buffering=0) |
| f.close() |
| f.close() |
| f.close() |
| self.assertRaises(ValueError, f.flush) |
| |
| def test_RawIOBase_read(self): |
| # Exercise the default limited RawIOBase.read(n) implementation (which |
| # calls readinto() internally). |
| rawio = self.MockRawIOWithoutRead((b"abc", b"d", None, b"efg", None)) |
| self.assertEqual(rawio.read(2), b"ab") |
| self.assertEqual(rawio.read(2), b"c") |
| self.assertEqual(rawio.read(2), b"d") |
| self.assertEqual(rawio.read(2), None) |
| self.assertEqual(rawio.read(2), b"ef") |
| self.assertEqual(rawio.read(2), b"g") |
| self.assertEqual(rawio.read(2), None) |
| self.assertEqual(rawio.read(2), b"") |
| |
| def test_RawIOBase_read_bounds_checking(self): |
| # Make sure a `.readinto` call which returns a value outside |
| # (0, len(buffer)) raises. |
| class Misbehaved(self.io.RawIOBase): |
| def __init__(self, readinto_return) -> None: |
| self._readinto_return = readinto_return |
| def readinto(self, b): |
| return self._readinto_return |
| |
| with self.assertRaises(ValueError) as cm: |
| Misbehaved(2).read(1) |
| self.assertEqual(str(cm.exception), "readinto returned 2 outside buffer size 1") |
| for bad_size in (2147483647, sys.maxsize, -1, -1000): |
| with self.assertRaises(ValueError): |
| Misbehaved(bad_size).read() |
| |
| def test_RawIOBase_read_gh60107(self): |
| # gh-60107: Ensure a "Raw I/O" which keeps a reference to the |
| # mutable memory doesn't allow making a mutable bytes. |
| class RawIOKeepsReference(self.MockRawIOWithoutRead): |
| def __init__(self, *args, **kwargs): |
| self.buf = None |
| super().__init__(*args, **kwargs) |
| |
| def readinto(self, buf): |
| # buf is the bytearray so keeping a reference to it doesn't keep |
| # the memory alive; a memoryview does. |
| self.buf = memoryview(buf) |
| buf[0:4] = self._read_stack.pop() |
| return 3 |
| |
| with self.assertRaises(BufferError): |
| rawio = RawIOKeepsReference([b"1234"]) |
| rawio.read(4) |
| |
| def test_types_have_dict(self): |
| test = ( |
| self.IOBase(), |
| self.RawIOBase(), |
| self.TextIOBase(), |
| self.StringIO(), |
| self.BytesIO() |
| ) |
| for obj in test: |
| self.assertHasAttr(obj, "__dict__") |
| |
| def test_opener(self): |
| with self.open(os_helper.TESTFN, "w", encoding="utf-8") as f: |
| f.write("egg\n") |
| fd = os.open(os_helper.TESTFN, os.O_RDONLY) |
| def opener(path, flags): |
| return fd |
| with self.open("non-existent", "r", encoding="utf-8", opener=opener) as f: |
| self.assertEqual(f.read(), "egg\n") |
| |
| def test_bad_opener_negative_1(self): |
| # Issue #27066. |
| def badopener(fname, flags): |
| return -1 |
| with self.assertRaises(ValueError) as cm: |
| self.open('non-existent', 'r', opener=badopener) |
| self.assertEqual(str(cm.exception), 'opener returned -1') |
| |
| def test_bad_opener_other_negative(self): |
| # Issue #27066. |
| def badopener(fname, flags): |
| return -2 |
| with self.assertRaises(ValueError) as cm: |
| self.open('non-existent', 'r', opener=badopener) |
| self.assertEqual(str(cm.exception), 'opener returned -2') |
| |
| def test_opener_invalid_fd(self): |
| # Check that OSError is raised with error code EBADF if the |
| # opener returns an invalid file descriptor (see gh-82212). |
| fd = os_helper.make_bad_fd() |
| with self.assertRaises(OSError) as cm: |
| self.open('foo', opener=lambda name, flags: fd) |
| self.assertEqual(cm.exception.errno, errno.EBADF) |
| |
| def test_fileio_closefd(self): |
| # Issue #4841 |
| with self.open(__file__, 'rb') as f1, \ |
| self.open(__file__, 'rb') as f2: |
| fileio = self.FileIO(f1.fileno(), closefd=False) |
| # .__init__() must not close f1 |
| fileio.__init__(f2.fileno(), closefd=False) |
| f1.readline() |
| # .close() must not close f2 |
| fileio.close() |
| f2.readline() |
| |
| def test_nonbuffered_textio(self): |
| with warnings_helper.check_no_resource_warning(self): |
| with self.assertRaises(ValueError): |
| self.open(os_helper.TESTFN, 'w', encoding="utf-8", buffering=0) |
| |
| def test_invalid_newline(self): |
| with warnings_helper.check_no_resource_warning(self): |
| with self.assertRaises(ValueError): |
| self.open(os_helper.TESTFN, 'w', encoding="utf-8", newline='invalid') |
| |
| def test_buffered_readinto_mixin(self): |
| # Test the implementation provided by BufferedIOBase |
| class Stream(self.BufferedIOBase): |
| def read(self, size): |
| return b"12345" |
| read1 = read |
| stream = Stream() |
| for method in ("readinto", "readinto1"): |
| with self.subTest(method): |
| buffer = byteslike(5) |
| self.assertEqual(getattr(stream, method)(buffer), 5) |
| self.assertEqual(bytes(buffer), b"12345") |
| |
| def test_fspath_support(self): |
| def check_path_succeeds(path): |
| with self.open(path, "w", encoding="utf-8") as f: |
| f.write("egg\n") |
| |
| with self.open(path, "r", encoding="utf-8") as f: |
| self.assertEqual(f.read(), "egg\n") |
| |
| check_path_succeeds(FakePath(os_helper.TESTFN)) |
| check_path_succeeds(FakePath(os.fsencode(os_helper.TESTFN))) |
| |
| with self.open(os_helper.TESTFN, "w", encoding="utf-8") as f: |
| bad_path = FakePath(f.fileno()) |
| with self.assertRaises(TypeError): |
| self.open(bad_path, 'w', encoding="utf-8") |
| |
| bad_path = FakePath(None) |
| with self.assertRaises(TypeError): |
| self.open(bad_path, 'w', encoding="utf-8") |
| |
| bad_path = FakePath(FloatingPointError) |
| with self.assertRaises(FloatingPointError): |
| self.open(bad_path, 'w', encoding="utf-8") |
| |
| # ensure that refcounting is correct with some error conditions |
| with self.assertRaisesRegex(ValueError, 'read/write/append mode'): |
| self.open(FakePath(os_helper.TESTFN), 'rwxa', encoding="utf-8") |
| |
| def test_RawIOBase_readall(self): |
| # Exercise the default unlimited RawIOBase.read() and readall() |
| # implementations. |
| rawio = self.MockRawIOWithoutRead((b"abc", b"d", b"efg")) |
| self.assertEqual(rawio.read(), b"abcdefg") |
| rawio = self.MockRawIOWithoutRead((b"abc", b"d", b"efg")) |
| self.assertEqual(rawio.readall(), b"abcdefg") |
| |
| def test_BufferedIOBase_readinto(self): |
| # Exercise the default BufferedIOBase.readinto() and readinto1() |
| # implementations (which call read() or read1() internally). |
| class Reader(self.BufferedIOBase): |
| def __init__(self, avail): |
| self.avail = avail |
| def read(self, size): |
| result = self.avail[:size] |
| self.avail = self.avail[size:] |
| return result |
| def read1(self, size): |
| """Returns no more than 5 bytes at once""" |
| return self.read(min(size, 5)) |
| tests = ( |
| # (test method, total data available, read buffer size, expected |
| # read size) |
| ("readinto", 10, 5, 5), |
| ("readinto", 10, 6, 6), # More than read1() can return |
| ("readinto", 5, 6, 5), # Buffer larger than total available |
| ("readinto", 6, 7, 6), |
| ("readinto", 10, 0, 0), # Empty buffer |
| ("readinto1", 10, 5, 5), # Result limited to single read1() call |
| ("readinto1", 10, 6, 5), # Buffer larger than read1() can return |
| ("readinto1", 5, 6, 5), # Buffer larger than total available |
| ("readinto1", 6, 7, 5), |
| ("readinto1", 10, 0, 0), # Empty buffer |
| ) |
| UNUSED_BYTE = 0x81 |
| for test in tests: |
| with self.subTest(test): |
| method, avail, request, result = test |
| reader = Reader(bytes(range(avail))) |
| buffer = bytearray((UNUSED_BYTE,) * request) |
| method = getattr(reader, method) |
| self.assertEqual(method(buffer), result) |
| self.assertEqual(len(buffer), request) |
| self.assertSequenceEqual(buffer[:result], range(result)) |
| unused = (UNUSED_BYTE,) * (request - result) |
| self.assertSequenceEqual(buffer[result:], unused) |
| self.assertEqual(len(reader.avail), avail - result) |
| |
| def test_close_assert(self): |
| class R(self.IOBase): |
| def __setattr__(self, name, value): |
| pass |
| def flush(self): |
| raise OSError() |
| f = R() |
| # This would cause an assertion failure. |
| self.assertRaises(OSError, f.close) |
| |
| # Silence destructor error |
| R.flush = lambda self: None |
| |
| @threading_helper.requires_working_threading() |
| def test_write_readline_races(self): |
| # gh-134908: Concurrent iteration over a file caused races |
| thread_count = 2 |
| write_count = 100 |
| read_count = 100 |
| |
| def writer(file, barrier): |
| barrier.wait() |
| for _ in range(write_count): |
| file.write("x") |
| |
| def reader(file, barrier): |
| barrier.wait() |
| for _ in range(read_count): |
| for line in file: |
| self.assertEqual(line, "") |
| |
| with self.open(os_helper.TESTFN, "w+") as f: |
| barrier = threading.Barrier(thread_count + 1) |
| reader = threading.Thread(target=reader, args=(f, barrier)) |
| writers = [threading.Thread(target=writer, args=(f, barrier)) |
| for _ in range(thread_count)] |
| with threading_helper.catch_threading_exception() as cm: |
| with threading_helper.start_threads(writers + [reader]): |
| pass |
| self.assertIsNone(cm.exc_type) |
| |
| self.assertEqual(os.stat(os_helper.TESTFN).st_size, |
| write_count * thread_count) |
| |
| |
| class CIOTest(IOTest, CTestCase): |
| |
| def test_IOBase_finalize(self): |
| # Issue #12149: segmentation fault on _PyIOBase_finalize when both a |
| # class which inherits IOBase and an object of this class are caught |
| # in a reference cycle and close() is already in the method cache. |
| class MyIO(self.IOBase): |
| def close(self): |
| pass |
| |
| # create an instance to populate the method cache |
| MyIO() |
| obj = MyIO() |
| obj.obj = obj |
| wr = weakref.ref(obj) |
| del MyIO |
| del obj |
| support.gc_collect() |
| self.assertIsNone(wr(), wr) |
| |
| @support.cpython_only |
| class TestIOCTypes(unittest.TestCase): |
| def setUp(self): |
| _io = import_helper.import_module("_io") |
| self.types = [ |
| _io.BufferedRWPair, |
| _io.BufferedRandom, |
| _io.BufferedReader, |
| _io.BufferedWriter, |
| _io.BytesIO, |
| _io.FileIO, |
| _io.IncrementalNewlineDecoder, |
| _io.StringIO, |
| _io.TextIOWrapper, |
| _io._BufferedIOBase, |
| _io._BytesIOBuffer, |
| _io._IOBase, |
| _io._RawIOBase, |
| _io._TextIOBase, |
| ] |
| if sys.platform == "win32": |
| self.types.append(_io._WindowsConsoleIO) |
| self._io = _io |
| |
| def test_immutable_types(self): |
| for tp in self.types: |
| with self.subTest(tp=tp): |
| with self.assertRaisesRegex(TypeError, "immutable"): |
| tp.foo = "bar" |
| |
| def test_class_hierarchy(self): |
| def check_subs(types, base): |
| for tp in types: |
| with self.subTest(tp=tp, base=base): |
| self.assertIsSubclass(tp, base) |
| |
| def recursive_check(d): |
| for k, v in d.items(): |
| if isinstance(v, dict): |
| recursive_check(v) |
| elif isinstance(v, set): |
| check_subs(v, k) |
| else: |
| self.fail("corrupt test dataset") |
| |
| _io = self._io |
| hierarchy = { |
| _io._IOBase: { |
| _io._BufferedIOBase: { |
| _io.BufferedRWPair, |
| _io.BufferedRandom, |
| _io.BufferedReader, |
| _io.BufferedWriter, |
| _io.BytesIO, |
| }, |
| _io._RawIOBase: { |
| _io.FileIO, |
| }, |
| _io._TextIOBase: { |
| _io.StringIO, |
| _io.TextIOWrapper, |
| }, |
| }, |
| } |
| if sys.platform == "win32": |
| hierarchy[_io._IOBase][_io._RawIOBase].add(_io._WindowsConsoleIO) |
| |
| recursive_check(hierarchy) |
| |
| def test_subclassing(self): |
| _io = self._io |
| dataset = {k: True for k in self.types} |
| dataset[_io._BytesIOBuffer] = False |
| |
| for tp, is_basetype in dataset.items(): |
| with self.subTest(tp=tp, is_basetype=is_basetype): |
| name = f"{tp.__name__}_subclass" |
| bases = (tp,) |
| if is_basetype: |
| _ = type(name, bases, {}) |
| else: |
| msg = "not an acceptable base type" |
| with self.assertRaisesRegex(TypeError, msg): |
| _ = type(name, bases, {}) |
| |
| def test_disallow_instantiation(self): |
| _io = self._io |
| support.check_disallow_instantiation(self, _io._BytesIOBuffer) |
| |
| def test_stringio_setstate(self): |
| # gh-127182: Calling __setstate__() with invalid arguments must not crash |
| obj = self._io.StringIO() |
| with self.assertRaisesRegex( |
| TypeError, |
| 'initial_value must be str or None, not int', |
| ): |
| obj.__setstate__((1, '', 0, {})) |
| |
| obj.__setstate__((None, '', 0, {})) # should not crash |
| self.assertEqual(obj.getvalue(), '') |
| |
| obj.__setstate__(('', '', 0, {})) |
| self.assertEqual(obj.getvalue(), '') |
| |
| class PyIOTest(IOTest, PyTestCase): |
| pass |
| |
| |
| @support.cpython_only |
| class APIMismatchTest(unittest.TestCase): |
| |
| def test_RawIOBase_io_in_pyio_match(self): |
| """Test that pyio RawIOBase class has all c RawIOBase methods""" |
| mismatch = support.detect_api_mismatch(pyio.RawIOBase, io.RawIOBase, |
| ignore=('__weakref__', '__static_attributes__')) |
| self.assertEqual(mismatch, set(), msg='Python RawIOBase does not have all C RawIOBase methods') |
| |
| def test_RawIOBase_pyio_in_io_match(self): |
| """Test that c RawIOBase class has all pyio RawIOBase methods""" |
| mismatch = support.detect_api_mismatch(io.RawIOBase, pyio.RawIOBase) |
| self.assertEqual(mismatch, set(), msg='C RawIOBase does not have all Python RawIOBase methods') |
| |
| |
| # XXX Tests for open() |
| |
| class MiscIOTest: |
| |
| # for test__all__, actual values are set in subclasses |
| name_of_module = None |
| extra_exported = () |
| not_exported = () |
| |
| def tearDown(self): |
| os_helper.unlink(os_helper.TESTFN) |
| |
| def test___all__(self): |
| support.check__all__(self, self.io, self.name_of_module, |
| extra=self.extra_exported, |
| not_exported=self.not_exported) |
| |
| def test_attributes(self): |
| f = self.open(os_helper.TESTFN, "wb", buffering=0) |
| self.assertEqual(f.mode, "wb") |
| f.close() |
| |
| f = self.open(os_helper.TESTFN, "w+", encoding="utf-8") |
| self.assertEqual(f.mode, "w+") |
| self.assertEqual(f.buffer.mode, "wb+") |
| self.assertEqual(f.buffer.raw.mode, "wb+") |
| |
| g = self.open(f.fileno(), "wb", closefd=False) |
| self.assertEqual(g.mode, "wb") |
| self.assertEqual(g.raw.mode, "wb") |
| self.assertEqual(g.name, f.fileno()) |
| self.assertEqual(g.raw.name, f.fileno()) |
| f.close() |
| g.close() |
| |
| def test_removed_u_mode(self): |
| # bpo-37330: The "U" mode has been removed in Python 3.11 |
| for mode in ("U", "rU", "r+U"): |
| with self.assertRaises(ValueError) as cm: |
| self.open(os_helper.TESTFN, mode) |
| self.assertIn('invalid mode', str(cm.exception)) |
| |
| @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") |
| def test_open_pipe_with_append(self): |
| # bpo-27805: Ignore ESPIPE from lseek() in open(). |
| r, w = os.pipe() |
| self.addCleanup(os.close, r) |
| f = self.open(w, 'a', encoding="utf-8") |
| self.addCleanup(f.close) |
| # Check that the file is marked non-seekable. On Windows, however, lseek |
| # somehow succeeds on pipes. |
| if sys.platform != 'win32': |
| self.assertFalse(f.seekable()) |
| |
| def test_io_after_close(self): |
| for kwargs in [ |
| {"mode": "w"}, |
| {"mode": "wb"}, |
| {"mode": "w", "buffering": 1}, |
| {"mode": "w", "buffering": 2}, |
| {"mode": "wb", "buffering": 0}, |
| {"mode": "r"}, |
| {"mode": "rb"}, |
| {"mode": "r", "buffering": 1}, |
| {"mode": "r", "buffering": 2}, |
| {"mode": "rb", "buffering": 0}, |
| {"mode": "w+"}, |
| {"mode": "w+b"}, |
| {"mode": "w+", "buffering": 1}, |
| {"mode": "w+", "buffering": 2}, |
| {"mode": "w+b", "buffering": 0}, |
| ]: |
| if "b" not in kwargs["mode"]: |
| kwargs["encoding"] = "utf-8" |
| f = self.open(os_helper.TESTFN, **kwargs) |
| f.close() |
| self.assertRaises(ValueError, f.flush) |
| self.assertRaises(ValueError, f.fileno) |
| self.assertRaises(ValueError, f.isatty) |
| self.assertRaises(ValueError, f.__iter__) |
| if hasattr(f, "peek"): |
| self.assertRaises(ValueError, f.peek, 1) |
| self.assertRaises(ValueError, f.read) |
| if hasattr(f, "read1"): |
| self.assertRaises(ValueError, f.read1, 1024) |
| self.assertRaises(ValueError, f.read1) |
| if hasattr(f, "readall"): |
| self.assertRaises(ValueError, f.readall) |
| if hasattr(f, "readinto"): |
| self.assertRaises(ValueError, f.readinto, bytearray(1024)) |
| if hasattr(f, "readinto1"): |
| self.assertRaises(ValueError, f.readinto1, bytearray(1024)) |
| self.assertRaises(ValueError, f.readline) |
| self.assertRaises(ValueError, f.readlines) |
| self.assertRaises(ValueError, f.readlines, 1) |
| self.assertRaises(ValueError, f.seek, 0) |
| self.assertRaises(ValueError, f.tell) |
| self.assertRaises(ValueError, f.truncate) |
| self.assertRaises(ValueError, f.write, |
| b"" if "b" in kwargs['mode'] else "") |
| self.assertRaises(ValueError, f.writelines, []) |
| self.assertRaises(ValueError, next, f) |
| |
| def test_blockingioerror(self): |
| # Various BlockingIOError issues |
| class C(str): |
| pass |
| c = C("") |
| b = self.BlockingIOError(1, c) |
| c.b = b |
| b.c = c |
| wr = weakref.ref(c) |
| del c, b |
| support.gc_collect() |
| self.assertIsNone(wr(), wr) |
| |
| def test_abcs(self): |
| # Test the visible base classes are ABCs. |
| self.assertIsInstance(self.IOBase, abc.ABCMeta) |
| self.assertIsInstance(self.RawIOBase, abc.ABCMeta) |
| self.assertIsInstance(self.BufferedIOBase, abc.ABCMeta) |
| self.assertIsInstance(self.TextIOBase, abc.ABCMeta) |
| |
| def _check_abc_inheritance(self, abcmodule): |
| with self.open(os_helper.TESTFN, "wb", buffering=0) as f: |
| self.assertIsInstance(f, abcmodule.IOBase) |
| self.assertIsInstance(f, abcmodule.RawIOBase) |
| self.assertNotIsInstance(f, abcmodule.BufferedIOBase) |
| self.assertNotIsInstance(f, abcmodule.TextIOBase) |
| with self.open(os_helper.TESTFN, "wb") as f: |
| self.assertIsInstance(f, abcmodule.IOBase) |
| self.assertNotIsInstance(f, abcmodule.RawIOBase) |
| self.assertIsInstance(f, abcmodule.BufferedIOBase) |
| self.assertNotIsInstance(f, abcmodule.TextIOBase) |
| with self.open(os_helper.TESTFN, "w", encoding="utf-8") as f: |
| self.assertIsInstance(f, abcmodule.IOBase) |
| self.assertNotIsInstance(f, abcmodule.RawIOBase) |
| self.assertNotIsInstance(f, abcmodule.BufferedIOBase) |
| self.assertIsInstance(f, abcmodule.TextIOBase) |
| |
| def test_abc_inheritance(self): |
| # Test implementations inherit from their respective ABCs |
| self._check_abc_inheritance(self) |
| |
| def test_abc_inheritance_official(self): |
| # Test implementations inherit from the official ABCs of the |
| # baseline "io" module. |
| self._check_abc_inheritance(io) |
| |
| def _check_warn_on_dealloc(self, *args, **kwargs): |
| f = self.open(*args, **kwargs) |
| r = repr(f) |
| with self.assertWarns(ResourceWarning) as cm: |
| f = None |
| support.gc_collect() |
| self.assertIn(r, str(cm.warning.args[0])) |
| |
| def test_warn_on_dealloc(self): |
| self._check_warn_on_dealloc(os_helper.TESTFN, "wb", buffering=0) |
| self._check_warn_on_dealloc(os_helper.TESTFN, "wb") |
| self._check_warn_on_dealloc(os_helper.TESTFN, "w", encoding="utf-8") |
| |
| def _check_warn_on_dealloc_fd(self, *args, **kwargs): |
| fds = [] |
| def cleanup_fds(): |
| for fd in fds: |
| try: |
| os.close(fd) |
| except OSError as e: |
| if e.errno != errno.EBADF: |
| raise |
| self.addCleanup(cleanup_fds) |
| r, w = os.pipe() |
| fds += r, w |
| self._check_warn_on_dealloc(r, *args, **kwargs) |
| # When using closefd=False, there's no warning |
| r, w = os.pipe() |
| fds += r, w |
| with warnings_helper.check_no_resource_warning(self): |
| self.open(r, *args, closefd=False, **kwargs) |
| |
| @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") |
| def test_warn_on_dealloc_fd(self): |
| self._check_warn_on_dealloc_fd("rb", buffering=0) |
| self._check_warn_on_dealloc_fd("rb") |
| self._check_warn_on_dealloc_fd("r", encoding="utf-8") |
| |
| |
| def test_pickling(self): |
| # Pickling file objects is forbidden |
| msg = "cannot pickle" |
| for kwargs in [ |
| {"mode": "w"}, |
| {"mode": "wb"}, |
| {"mode": "wb", "buffering": 0}, |
| {"mode": "r"}, |
| {"mode": "rb"}, |
| {"mode": "rb", "buffering": 0}, |
| {"mode": "w+"}, |
| {"mode": "w+b"}, |
| {"mode": "w+b", "buffering": 0}, |
| ]: |
| if "b" not in kwargs["mode"]: |
| kwargs["encoding"] = "utf-8" |
| for protocol in range(pickle.HIGHEST_PROTOCOL + 1): |
| with self.subTest(protocol=protocol, kwargs=kwargs): |
| with self.open(os_helper.TESTFN, **kwargs) as f: |
| with self.assertRaisesRegex(TypeError, msg): |
| pickle.dumps(f, protocol) |
| |
| @unittest.skipIf(support.is_emscripten, "Emscripten corrupts memory when writing to nonblocking fd") |
| def test_nonblock_pipe_write_bigbuf(self): |
| self._test_nonblock_pipe_write(16*1024) |
| |
| @unittest.skipIf(support.is_emscripten, "Emscripten corrupts memory when writing to nonblocking fd") |
| def test_nonblock_pipe_write_smallbuf(self): |
| self._test_nonblock_pipe_write(1024) |
| |
| @unittest.skipUnless(hasattr(os, 'set_blocking'), |
| 'os.set_blocking() required for this test') |
| @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") |
| def _test_nonblock_pipe_write(self, bufsize): |
| sent = [] |
| received = [] |
| r, w = os.pipe() |
| os.set_blocking(r, False) |
| os.set_blocking(w, False) |
| |
| # To exercise all code paths in the C implementation we need |
| # to play with buffer sizes. For instance, if we choose a |
| # buffer size less than or equal to _PIPE_BUF (4096 on Linux) |
| # then we will never get a partial write of the buffer. |
| rf = self.open(r, mode='rb', closefd=True, buffering=bufsize) |
| wf = self.open(w, mode='wb', closefd=True, buffering=bufsize) |
| |
| with rf, wf: |
| for N in 9999, 73, 7574: |
| try: |
| i = 0 |
| while True: |
| msg = bytes([i % 26 + 97]) * N |
| sent.append(msg) |
| wf.write(msg) |
| i += 1 |
| |
| except self.BlockingIOError as e: |
| self.assertEqual(e.args[0], errno.EAGAIN) |
| self.assertEqual(e.args[2], e.characters_written) |
| sent[-1] = sent[-1][:e.characters_written] |
| received.append(rf.read()) |
| msg = b'BLOCKED' |
| wf.write(msg) |
| sent.append(msg) |
| |
| while True: |
| try: |
| wf.flush() |
| break |
| except self.BlockingIOError as e: |
| self.assertEqual(e.args[0], errno.EAGAIN) |
| self.assertEqual(e.args[2], e.characters_written) |
| self.assertEqual(e.characters_written, 0) |
| received.append(rf.read()) |
| |
| received += iter(rf.read, None) |
| |
| sent, received = b''.join(sent), b''.join(received) |
| self.assertEqual(sent, received) |
| self.assertTrue(wf.closed) |
| self.assertTrue(rf.closed) |
| |
| def test_create_fail(self): |
| # 'x' mode fails if file is existing |
| with self.open(os_helper.TESTFN, 'w', encoding="utf-8"): |
| pass |
| self.assertRaises(FileExistsError, self.open, os_helper.TESTFN, 'x', encoding="utf-8") |
| |
| def test_create_writes(self): |
| # 'x' mode opens for writing |
| with self.open(os_helper.TESTFN, 'xb') as f: |
| f.write(b"spam") |
| with self.open(os_helper.TESTFN, 'rb') as f: |
| self.assertEqual(b"spam", f.read()) |
| |
| def test_open_allargs(self): |
| # there used to be a buffer overflow in the parser for rawmode |
| self.assertRaises(ValueError, self.open, os_helper.TESTFN, 'rwax+', encoding="utf-8") |
| |
| def test_check_encoding_errors(self): |
| # bpo-37388: open() and TextIOWrapper must check encoding and errors |
| # arguments in dev mode |
| mod = self.io.__name__ |
| filename = __file__ |
| invalid = 'Boom, Shaka Laka, Boom!' |
| code = textwrap.dedent(f''' |
| import sys |
| from {mod} import open, TextIOWrapper |
| |
| try: |
| open({filename!r}, encoding={invalid!r}) |
| except LookupError: |
| pass |
| else: |
| sys.exit(21) |
| |
| try: |
| open({filename!r}, errors={invalid!r}) |
| except LookupError: |
| pass |
| else: |
| sys.exit(22) |
| |
| fp = open({filename!r}, "rb") |
| with fp: |
| try: |
| TextIOWrapper(fp, encoding={invalid!r}) |
| except LookupError: |
| pass |
| else: |
| sys.exit(23) |
| |
| try: |
| TextIOWrapper(fp, errors={invalid!r}) |
| except LookupError: |
| pass |
| else: |
| sys.exit(24) |
| |
| sys.exit(10) |
| ''') |
| proc = assert_python_failure('-X', 'dev', '-c', code) |
| self.assertEqual(proc.rc, 10, proc) |
| |
| def test_check_encoding_warning(self): |
| # PEP 597: Raise warning when encoding is not specified |
| # and sys.flags.warn_default_encoding is set. |
| mod = self.io.__name__ |
| filename = __file__ |
| code = textwrap.dedent(f'''\ |
| import sys |
| from {mod} import open, TextIOWrapper |
| import pathlib |
| |
| with open({filename!r}) as f: # line 5 |
| pass |
| |
| pathlib.Path({filename!r}).read_text() # line 8 |
| ''') |
| proc = assert_python_ok('-X', 'warn_default_encoding', '-c', code) |
| warnings = proc.err.splitlines() |
| self.assertEqual(len(warnings), 2) |
| self.assertStartsWith(warnings[0], b"<string>:5: EncodingWarning: ") |
| self.assertStartsWith(warnings[1], b"<string>:8: EncodingWarning: ") |
| |
| def test_text_encoding(self): |
| # PEP 597, bpo-47000. io.text_encoding() returns "locale" or "utf-8" |
| # based on sys.flags.utf8_mode |
| code = "import io; print(io.text_encoding(None))" |
| |
| proc = assert_python_ok('-X', 'utf8=0', '-c', code) |
| self.assertEqual(b"locale", proc.out.strip()) |
| |
| proc = assert_python_ok('-X', 'utf8=1', '-c', code) |
| self.assertEqual(b"utf-8", proc.out.strip()) |
| |
| |
| class CMiscIOTest(MiscIOTest, CTestCase): |
| name_of_module = "io", "_io" |
| extra_exported = "BlockingIOError", |
| |
| def test_readinto_buffer_overflow(self): |
| # Issue #18025 |
| class BadReader(self.io.BufferedIOBase): |
| def read(self, n=-1): |
| return b'x' * 10**6 |
| bufio = BadReader() |
| b = bytearray(2) |
| self.assertRaises(ValueError, bufio.readinto, b) |
| |
| def check_daemon_threads_shutdown_deadlock(self, stream_name): |
| # Issue #23309: deadlocks at shutdown should be avoided when a |
| # daemon thread and the main thread both write to a file. |
| code = """if 1: |
| import sys |
| import time |
| import threading |
| from test.support import SuppressCrashReport |
| |
| file = sys.{stream_name} |
| |
| def run(): |
| while True: |
| file.write('.') |
| file.flush() |
| |
| crash = SuppressCrashReport() |
| crash.__enter__() |
| # don't call __exit__(): the crash occurs at Python shutdown |
| |
| thread = threading.Thread(target=run) |
| thread.daemon = True |
| thread.start() |
| |
| time.sleep(0.5) |
| file.write('!') |
| file.flush() |
| """.format_map(locals()) |
| res, _ = run_python_until_end("-c", code) |
| err = res.err.decode() |
| if res.rc != 0: |
| # Failure: should be a fatal error |
| pattern = (r"Fatal Python error: _enter_buffered_busy: " |
| r"could not acquire lock " |
| r"for <(_io\.)?BufferedWriter name='<{stream_name}>'> " |
| r"at interpreter shutdown, possibly due to " |
| r"daemon threads".format_map(locals())) |
| self.assertRegex(err, pattern) |
| else: |
| self.assertFalse(err.strip('.!')) |
| |
| @threading_helper.requires_working_threading() |
| @support.requires_resource('walltime') |
| def test_daemon_threads_shutdown_stdout_deadlock(self): |
| self.check_daemon_threads_shutdown_deadlock('stdout') |
| |
| @threading_helper.requires_working_threading() |
| @support.requires_resource('walltime') |
| def test_daemon_threads_shutdown_stderr_deadlock(self): |
| self.check_daemon_threads_shutdown_deadlock('stderr') |
| |
| |
| class PyMiscIOTest(MiscIOTest, PyTestCase): |
| name_of_module = "_pyio", "io" |
| extra_exported = "BlockingIOError", "open_code", |
| not_exported = "valid_seek_flags", |
| |
| |
| class ProtocolsTest(unittest.TestCase): |
| class MyReader: |
| def read(self, sz=-1): |
| return b"" |
| |
| class MyWriter: |
| def write(self, b: bytes): |
| pass |
| |
| def test_reader_subclass(self): |
| self.assertIsSubclass(self.MyReader, io.Reader) |
| self.assertNotIsSubclass(str, io.Reader) |
| |
| def test_writer_subclass(self): |
| self.assertIsSubclass(self.MyWriter, io.Writer) |
| self.assertNotIsSubclass(str, io.Writer) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
