diff options
Diffstat (limited to 'Lib/test/test_io.py')
| -rw-r--r-- | Lib/test/test_io.py | 116 |
1 files changed, 65 insertions, 51 deletions
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py index 9b8920211e..c1ea6f245f 100644 --- a/Lib/test/test_io.py +++ b/Lib/test/test_io.py @@ -165,7 +165,7 @@ class CloseFailureIO(MockRawIO): def close(self): if not self.closed: self.closed = 1 - raise IOError + raise OSError class CCloseFailureIO(CloseFailureIO, io.RawIOBase): pass @@ -601,9 +601,9 @@ class IOTest(unittest.TestCase): def test_flush_error_on_close(self): f = self.open(support.TESTFN, "wb", buffering=0) def bad_flush(): - raise IOError() + raise OSError() f.flush = bad_flush - self.assertRaises(IOError, f.close) # exception not swallowed + self.assertRaises(OSError, f.close) # exception not swallowed self.assertTrue(f.closed) def test_multi_close(self): @@ -762,7 +762,7 @@ class CommonBufferedTests: if s: # The destructor *may* have printed an unraisable error, check it self.assertEqual(len(s.splitlines()), 1) - self.assertTrue(s.startswith("Exception IOError: "), s) + self.assertTrue(s.startswith("Exception OSError: "), s) self.assertTrue(s.endswith(" ignored"), s) def test_repr(self): @@ -778,22 +778,22 @@ class CommonBufferedTests: def test_flush_error_on_close(self): raw = self.MockRawIO() def bad_flush(): - raise IOError() + raise OSError() raw.flush = bad_flush b = self.tp(raw) - self.assertRaises(IOError, b.close) # exception not swallowed + self.assertRaises(OSError, b.close) # exception not swallowed self.assertTrue(b.closed) def test_close_error_on_close(self): raw = self.MockRawIO() def bad_flush(): - raise IOError('flush') + raise OSError('flush') def bad_close(): - raise IOError('close') + raise OSError('close') raw.close = bad_close b = self.tp(raw) b.flush = bad_flush - with self.assertRaises(IOError) as err: # exception not swallowed + with self.assertRaises(OSError) as err: # exception not swallowed b.close() self.assertEqual(err.exception.args, ('close',)) self.assertEqual(err.exception.__context__.args, ('flush',)) @@ -833,6 +833,14 @@ class SizeofTest: bufio = self.tp(rawio, buffer_size=bufsize2) self.assertEqual(sys.getsizeof(bufio), size + bufsize2) + @support.cpython_only + def test_buffer_freeing(self) : + bufsize = 4096 + rawio = self.MockRawIO() + bufio = self.tp(rawio, buffer_size=bufsize) + size = sys.getsizeof(bufio) - bufsize + bufio.close() + self.assertEqual(sys.getsizeof(bufio), size) class BufferedReaderTest(unittest.TestCase, CommonBufferedTests): read_mode = "rb" @@ -1007,8 +1015,8 @@ class BufferedReaderTest(unittest.TestCase, CommonBufferedTests): def test_misbehaved_io(self): rawio = self.MisbehavedRawIO((b"abc", b"d", b"efg")) bufio = self.tp(rawio) - self.assertRaises(IOError, bufio.seek, 0) - self.assertRaises(IOError, bufio.tell) + self.assertRaises(OSError, bufio.seek, 0) + self.assertRaises(OSError, bufio.tell) def test_no_extraneous_read(self): # Issue #9550; when the raw IO object has satisfied the read request, @@ -1059,17 +1067,18 @@ class CBufferedReaderTest(BufferedReaderTest, SizeofTest): bufio = self.tp(rawio) # _pyio.BufferedReader seems to implement reading different, so that # checking this is not so easy. - self.assertRaises(IOError, bufio.read, 10) + self.assertRaises(OSError, bufio.read, 10) def test_garbage_collection(self): # C BufferedReader objects are collected. # The Python version has __del__, so it ends into gc.garbage instead - rawio = self.FileIO(support.TESTFN, "w+b") - f = self.tp(rawio) - f.f = f - wr = weakref.ref(f) - del f - support.gc_collect() + with support.check_warnings(('', ResourceWarning)): + rawio = self.FileIO(support.TESTFN, "w+b") + f = self.tp(rawio) + f.f = f + wr = weakref.ref(f) + del f + support.gc_collect() self.assertTrue(wr() is None, wr) def test_args_error(self): @@ -1312,9 +1321,9 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests): def test_misbehaved_io(self): rawio = self.MisbehavedRawIO() bufio = self.tp(rawio, 5) - self.assertRaises(IOError, bufio.seek, 0) - self.assertRaises(IOError, bufio.tell) - self.assertRaises(IOError, bufio.write, b"abcdef") + self.assertRaises(OSError, bufio.seek, 0) + self.assertRaises(OSError, bufio.tell) + self.assertRaises(OSError, bufio.write, b"abcdef") def test_max_buffer_size_removal(self): with self.assertRaises(TypeError): @@ -1323,11 +1332,11 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests): def test_write_error_on_close(self): raw = self.MockRawIO() def bad_write(b): - raise IOError() + raise OSError() raw.write = bad_write b = self.tp(raw) b.write(b'spam') - self.assertRaises(IOError, b.close) # exception not swallowed + self.assertRaises(OSError, b.close) # exception not swallowed self.assertTrue(b.closed) @@ -1358,13 +1367,14 @@ class CBufferedWriterTest(BufferedWriterTest, SizeofTest): # C BufferedWriter objects are collected, and collecting them flushes # all data to disk. # The Python version has __del__, so it ends into gc.garbage instead - rawio = self.FileIO(support.TESTFN, "w+b") - f = self.tp(rawio) - f.write(b"123xxx") - f.x = f - wr = weakref.ref(f) - del f - support.gc_collect() + with support.check_warnings(('', ResourceWarning)): + rawio = self.FileIO(support.TESTFN, "w+b") + f = self.tp(rawio) + f.write(b"123xxx") + f.x = f + wr = weakref.ref(f) + del f + support.gc_collect() self.assertTrue(wr() is None, wr) with self.open(support.TESTFN, "rb") as f: self.assertEqual(f.read(), b"123xxx") @@ -1397,14 +1407,14 @@ class BufferedRWPairTest(unittest.TestCase): def readable(self): return False - self.assertRaises(IOError, self.tp, NotReadable(), self.MockRawIO()) + self.assertRaises(OSError, 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()) + self.assertRaises(OSError, self.tp, self.MockRawIO(), NotWriteable()) def test_read(self): pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO()) @@ -2165,7 +2175,7 @@ class TextIOWrapperTest(unittest.TestCase): if s: # The destructor *may* have printed an unraisable error, check it self.assertEqual(len(s.splitlines()), 1) - self.assertTrue(s.startswith("Exception IOError: "), s) + self.assertTrue(s.startswith("Exception OSError: "), s) self.assertTrue(s.endswith(" ignored"), s) # Systematic tests of the text I/O API @@ -2237,7 +2247,7 @@ class TextIOWrapperTest(unittest.TestCase): f.seek(0) for line in f: self.assertEqual(line, "\xff\n") - self.assertRaises(IOError, f.tell) + self.assertRaises(OSError, f.tell) self.assertEqual(f.tell(), p2) f.close() @@ -2341,7 +2351,7 @@ class TextIOWrapperTest(unittest.TestCase): def readable(self): return False txt = self.TextIOWrapper(UnReadable()) - self.assertRaises(IOError, txt.read) + self.assertRaises(OSError, txt.read) def test_read_one_by_one(self): txt = self.TextIOWrapper(self.BytesIO(b"AA\r\nBB")) @@ -2516,9 +2526,9 @@ class TextIOWrapperTest(unittest.TestCase): def test_flush_error_on_close(self): txt = self.TextIOWrapper(self.BytesIO(self.testdata), encoding="ascii") def bad_flush(): - raise IOError() + raise OSError() txt.flush = bad_flush - self.assertRaises(IOError, txt.close) # exception not swallowed + self.assertRaises(OSError, txt.close) # exception not swallowed self.assertTrue(txt.closed) def test_multi_close(self): @@ -2599,14 +2609,15 @@ class CTextIOWrapperTest(TextIOWrapperTest): # C TextIOWrapper objects are collected, and collecting them flushes # all data to disk. # The Python version has __del__, so it ends in gc.garbage instead. - rawio = io.FileIO(support.TESTFN, "wb") - b = self.BufferedWriter(rawio) - t = self.TextIOWrapper(b, encoding="ascii") - t.write("456def") - t.x = t - wr = weakref.ref(t) - del t - support.gc_collect() + with support.check_warnings(('', ResourceWarning)): + rawio = io.FileIO(support.TESTFN, "wb") + b = self.BufferedWriter(rawio) + t = self.TextIOWrapper(b, encoding="ascii") + t.write("456def") + t.x = t + wr = weakref.ref(t) + del t + support.gc_collect() self.assertTrue(wr() is None, wr) with self.open(support.TESTFN, "rb") as f: self.assertEqual(f.read(), b"456def") @@ -2896,7 +2907,7 @@ class MiscIOTest(unittest.TestCase): for fd in fds: try: os.close(fd) - except EnvironmentError as e: + except OSError as e: if e.errno != errno.EBADF: raise self.addCleanup(cleanup_fds) @@ -3062,15 +3073,18 @@ class SignalsTest(unittest.TestCase): try: wio = self.io.open(w, **fdopen_kwargs) t.start() - signal.alarm(1) # Fill the pipe enough that the write will be blocking. # It will be interrupted by the timer armed above. Since the # other thread has read one byte, the low-level write will # return with a successful (partial) result rather than an EINTR. # The buffered IO layer must check for pending signal # handlers, which in this case will invoke alarm_interrupt(). - self.assertRaises(ZeroDivisionError, - wio.write, item * (support.PIPE_MAX_SIZE // len(item) + 1)) + signal.alarm(1) + try: + self.assertRaises(ZeroDivisionError, + wio.write, item * (support.PIPE_MAX_SIZE // len(item) + 1)) + finally: + signal.alarm(0) t.join() # We got one byte, get another one and check that it isn't a # repeat of the first one. @@ -3084,7 +3098,7 @@ class SignalsTest(unittest.TestCase): # buffer, and block again. try: wio.close() - except IOError as e: + except OSError as e: if e.errno != errno.EBADF: raise @@ -3212,7 +3226,7 @@ class SignalsTest(unittest.TestCase): # buffer, and could block (in case of failure). try: wio.close() - except IOError as e: + except OSError as e: if e.errno != errno.EBADF: raise |