Skip to content
Merged
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 67 additions & 52 deletions Lib/test/test_io/test_general.py
Original file line number Diff line number Diff line change
Expand Up @@ -334,6 +334,45 @@ class PyMockNonBlockWriterIO(MockNonBlockWriterIO, pyio.RawIOBase):
BlockingIOError = pyio.BlockingIOError


# Build classes which point to all the right mocks per io implementation
class CTestCase(unittest.TestCase):
io = io
is_C = True

MockRawIO = CMockRawIO
MisbehavedRawIO = CMisbehavedRawIO
MockFileIO = CMockFileIO
CloseFailureIO = CCloseFailureIO
MockNonBlockWriterIO = CMockNonBlockWriterIO
MockUnseekableIO = CMockUnseekableIO
MockRawIOWithoutRead = CMockRawIOWithoutRead
SlowFlushRawIO = CSlowFlushRawIO
MockCharPseudoDevFileIO = CMockCharPseudoDevFileIO

# Use the class as a proxy to the io module members.
def __getattr__(self, name):
return getattr(self.io, name)


class PyTestCase(unittest.TestCase):
io = pyio
is_C = False

MockRawIO = PyMockRawIO
MisbehavedRawIO = PyMisbehavedRawIO
MockFileIO = PyMockFileIO
CloseFailureIO = PyCloseFailureIO
MockNonBlockWriterIO = PyMockNonBlockWriterIO
MockUnseekableIO = PyMockUnseekableIO
MockRawIOWithoutRead = PyMockRawIOWithoutRead
SlowFlushRawIO = PySlowFlushRawIO
MockCharPseudoDevFileIO = PyMockCharPseudoDevFileIO

# Use the class as a proxy to the io module members.
def __getattr__(self, name):
return getattr(self.io, name)


class IOTest(unittest.TestCase):

def setUp(self):
Expand Down Expand Up @@ -1100,7 +1139,7 @@ def reader(file, barrier):
write_count * thread_count)


class CIOTest(IOTest):
class CIOTest(IOTest, CTestCase):

def test_IOBase_finalize(self):
# Issue #12149: segmentation fault on _PyIOBase_finalize when both a
Expand Down Expand Up @@ -1224,7 +1263,7 @@ def test_stringio_setstate(self):
obj.__setstate__(('', '', 0, {}))
self.assertEqual(obj.getvalue(), '')

class PyIOTest(IOTest):
class PyIOTest(IOTest, PyTestCase):
pass


Expand Down Expand Up @@ -1455,7 +1494,7 @@ def test_buffer_freeing(self) :
bufio.close()
self.assertEqual(sys.getsizeof(bufio), size)

class BufferedReaderTest(unittest.TestCase, CommonBufferedTests):
class BufferedReaderTest(CommonBufferedTests):
read_mode = "rb"

def test_constructor(self):
Expand Down Expand Up @@ -1790,7 +1829,7 @@ def test_seek_character_device_file(self):
self.assertEqual(buf.seek(0, io.SEEK_CUR), 0)


class CBufferedReaderTest(BufferedReaderTest, SizeofTest):
class CBufferedReaderTest(BufferedReaderTest, SizeofTest, CTestCase):
tp = io.BufferedReader

def test_initialization(self):
Expand Down Expand Up @@ -1849,11 +1888,11 @@ def test_bad_readinto_type(self):
self.assertIsInstance(cm.exception.__cause__, TypeError)


class PyBufferedReaderTest(BufferedReaderTest):
class PyBufferedReaderTest(BufferedReaderTest, PyTestCase):
tp = pyio.BufferedReader


class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
class BufferedWriterTest(CommonBufferedTests):
write_mode = "wb"

def test_constructor(self):
Expand Down Expand Up @@ -2148,8 +2187,7 @@ def test_slow_close_from_thread(self):
t.join()



class CBufferedWriterTest(BufferedWriterTest, SizeofTest):
class CBufferedWriterTest(BufferedWriterTest, SizeofTest, CTestCase):
tp = io.BufferedWriter

def test_initialization(self):
Expand Down Expand Up @@ -2189,10 +2227,10 @@ def test_args_error(self):
self.tp(self.BytesIO(), 1024, 1024, 1024)


class PyBufferedWriterTest(BufferedWriterTest):
class PyBufferedWriterTest(BufferedWriterTest, PyTestCase):
tp = pyio.BufferedWriter

class BufferedRWPairTest(unittest.TestCase):
class BufferedRWPairTest:

def test_constructor(self):
pair = self.tp(self.MockRawIO(), self.MockRawIO())
Expand Down Expand Up @@ -2221,14 +2259,14 @@ def test_constructor_max_buffer_size_removal(self):
self.tp(self.MockRawIO(), self.MockRawIO(), 8, 12)

def test_constructor_with_not_readable(self):
class NotReadable(MockRawIO):
class NotReadable(self.MockRawIO):
def readable(self):
return False

self.assertRaises(OSError, self.tp, NotReadable(), self.MockRawIO())

def test_constructor_with_not_writeable(self):
class NotWriteable(MockRawIO):
class NotWriteable(self.MockRawIO):
def writable(self):
return False

Expand Down Expand Up @@ -2374,9 +2412,9 @@ def writer_close():
writer.close = lambda: None

def test_isatty(self):
class SelectableIsAtty(MockRawIO):
class SelectableIsAtty(self.MockRawIO):
def __init__(self, isatty):
MockRawIO.__init__(self)
super().__init__()
self._isatty = isatty

def isatty(self):
Expand All @@ -2400,10 +2438,10 @@ def test_weakref_clearing(self):
brw = None
ref = None # Shouldn't segfault.

class CBufferedRWPairTest(BufferedRWPairTest):
class CBufferedRWPairTest(BufferedRWPairTest, CTestCase):
tp = io.BufferedRWPair

class PyBufferedRWPairTest(BufferedRWPairTest):
class PyBufferedRWPairTest(BufferedRWPairTest, PyTestCase):
tp = pyio.BufferedRWPair


Expand Down Expand Up @@ -2662,7 +2700,7 @@ def test_interleaved_readline_write(self):
test_truncate_on_read_only = None


class CBufferedRandomTest(BufferedRandomTest, SizeofTest):
class CBufferedRandomTest(BufferedRandomTest, SizeofTest, CTestCase):
tp = io.BufferedRandom

def test_garbage_collection(self):
Expand All @@ -2675,7 +2713,7 @@ def test_args_error(self):
self.tp(self.BytesIO(), 1024, 1024, 1024)


class PyBufferedRandomTest(BufferedRandomTest):
class PyBufferedRandomTest(BufferedRandomTest, PyTestCase):
tp = pyio.BufferedRandom


Expand Down Expand Up @@ -4075,8 +4113,7 @@ def _to_memoryview(buf):
return memoryview(arr)


class CTextIOWrapperTest(TextIOWrapperTest):
io = io
class CTextIOWrapperTest(TextIOWrapperTest, CTestCase):
shutdown_error = "LookupError: unknown encoding: ascii"

def test_initialization(self):
Expand Down Expand Up @@ -4176,8 +4213,7 @@ def write(self, data):
buf._write_stack)


class PyTextIOWrapperTest(TextIOWrapperTest):
io = pyio
class PyTextIOWrapperTest(TextIOWrapperTest, PyTestCase):
shutdown_error = "LookupError: unknown encoding: ascii"


Expand Down Expand Up @@ -4299,6 +4335,8 @@ def test_translate(self):
self.assertEqual(decoder.decode(b"\r\r\n"), "\r\r\n")

class CIncrementalNewlineDecoderTest(IncrementalNewlineDecoderTest):
IncrementalNewlineDecoder = io.IncrementalNewlineDecoder

@support.cpython_only
def test_uninitialized(self):
uninitialized = self.IncrementalNewlineDecoder.__new__(
Expand All @@ -4310,7 +4348,7 @@ def test_uninitialized(self):


class PyIncrementalNewlineDecoderTest(IncrementalNewlineDecoderTest):
pass
IncrementalNewlineDecoder = pyio.IncrementalNewlineDecoder


# XXX Tests for open()
Expand Down Expand Up @@ -4679,8 +4717,7 @@ def test_text_encoding(self):
self.assertEqual(b"utf-8", proc.out.strip())


class CMiscIOTest(MiscIOTest):
io = io
class CMiscIOTest(MiscIOTest, CTestCase):
name_of_module = "io", "_io"
extra_exported = "BlockingIOError",

Expand Down Expand Up @@ -4745,8 +4782,7 @@ def test_daemon_threads_shutdown_stderr_deadlock(self):
self.check_daemon_threads_shutdown_deadlock('stderr')


class PyMiscIOTest(MiscIOTest):
io = pyio
class PyMiscIOTest(MiscIOTest, PyTestCase):
name_of_module = "_pyio", "io"
extra_exported = "BlockingIOError", "open_code",
not_exported = "valid_seek_flags",
Expand Down Expand Up @@ -5007,11 +5043,11 @@ def test_interrupted_write_retry_text(self):
self.check_interrupted_write_retry("x", mode="w", encoding="latin1")


class CSignalsTest(SignalsTest):
io = io
class CSignalsTest(SignalsTest, CTestCase):
pass

class PySignalsTest(SignalsTest):
io = pyio
class PySignalsTest(SignalsTest, PyTestCase):
pass

# Handling reentrancy issues would slow down _pyio even more, so the
# tests are disabled.
Expand Down Expand Up @@ -5050,27 +5086,6 @@ def load_tests(loader, tests, pattern):
CSignalsTest, PySignalsTest, TestIOCTypes,
)

# Put the namespaces of the IO module we are testing and some useful mock
# classes in the __dict__ of each test.
mocks = (MockRawIO, MisbehavedRawIO, MockFileIO, CloseFailureIO,
MockNonBlockWriterIO, MockUnseekableIO, MockRawIOWithoutRead,
SlowFlushRawIO, MockCharPseudoDevFileIO)
all_members = io.__all__
c_io_ns = {name : getattr(io, name) for name in all_members}
py_io_ns = {name : getattr(pyio, name) for name in all_members}
globs = globals()
c_io_ns.update((x.__name__, globs["C" + x.__name__]) for x in mocks)
py_io_ns.update((x.__name__, globs["Py" + x.__name__]) for x in mocks)
for test in tests:
if test.__name__.startswith("C"):
for name, obj in c_io_ns.items():
setattr(test, name, obj)
test.is_C = True
elif test.__name__.startswith("Py"):
for name, obj in py_io_ns.items():
setattr(test, name, obj)
test.is_C = False

suite = loader.suiteClass()
for test in tests:
suite.addTest(loader.loadTestsFromTestCase(test))
Expand Down
Loading