import sys import threading import unittest from test import support from test.support import import_helper from test.support import threading_helper _testlimitedcapi = import_helper.import_module('_testlimitedcapi') _testcapi = import_helper.import_module('_testcapi') from _testcapi import PY_SSIZE_T_MIN, PY_SSIZE_T_MAX NULL = None class BytesSubclass(bytes): pass class BytesLike: def __init__(self, value): self.value = value def __bytes__(self): return self.value class CAPITest(unittest.TestCase): def test_check(self): # Test PyBytes_Check() check = _testlimitedcapi.bytes_check self.assertTrue(check(b'abc')) self.assertTrue(check(b'')) self.assertFalse(check('abc')) self.assertFalse(check(bytearray(b'abc'))) self.assertTrue(check(BytesSubclass(b'abc'))) self.assertFalse(check(BytesLike(b'abc'))) self.assertFalse(check(3)) self.assertFalse(check([])) self.assertFalse(check(object())) # CRASHES check(NULL) def test_checkexact(self): # Test PyBytes_CheckExact() check = _testlimitedcapi.bytes_checkexact self.assertTrue(check(b'abc')) self.assertTrue(check(b'')) self.assertFalse(check('abc')) self.assertFalse(check(bytearray(b'abc'))) self.assertFalse(check(BytesSubclass(b'abc'))) self.assertFalse(check(BytesLike(b'abc'))) self.assertFalse(check(3)) self.assertFalse(check([])) self.assertFalse(check(object())) # CRASHES check(NULL) def test_fromstringandsize(self): # Test PyBytes_FromStringAndSize() fromstringandsize = _testlimitedcapi.bytes_fromstringandsize self.assertEqual(fromstringandsize(b'abc'), b'abc') self.assertEqual(fromstringandsize(b'abc', 2), b'ab') self.assertEqual(fromstringandsize(b'abc\0def'), b'abc\0def') # Return singleton for 0 or 1 bytes self.assertIs(fromstringandsize(b'', 0), b'') self.assertIs(fromstringandsize(NULL, 0), b'') self.assertIs(fromstringandsize(b'a', 1), b'a') self.assertEqual(len(fromstringandsize(NULL, 3)), 3) self.assertRaises((MemoryError, OverflowError), fromstringandsize, NULL, PY_SSIZE_T_MAX) self.assertRaises(SystemError, fromstringandsize, b'abc', -1) self.assertRaises(SystemError, fromstringandsize, b'abc', PY_SSIZE_T_MIN) self.assertRaises(SystemError, fromstringandsize, NULL, -1) self.assertRaises(SystemError, fromstringandsize, NULL, PY_SSIZE_T_MIN) def test_fromstring(self): # Test PyBytes_FromString() fromstring = _testlimitedcapi.bytes_fromstring self.assertEqual(fromstring(b'abc\0def'), b'abc') # Return singleton for 0 or 1 bytes self.assertIs(fromstring(b''), b'') self.assertIs(fromstring(b'a'), b'a') # CRASHES fromstring(NULL) def test_fromobject(self): # Test PyBytes_FromObject() fromobject = _testlimitedcapi.bytes_fromobject self.assertEqual(fromobject(b''), b'') self.assertEqual(fromobject(b'abc'), b'abc') self.assertEqual(fromobject(bytearray(b'abc')), b'abc') self.assertEqual(fromobject(BytesSubclass(b'abc')), b'abc') self.assertEqual(fromobject([97, 98, 99]), b'abc') self.assertRaises(TypeError, fromobject, 3) self.assertRaises(TypeError, fromobject, BytesLike(b'abc')) self.assertRaises(TypeError, fromobject, 'abc') self.assertRaises(TypeError, fromobject, object()) self.assertRaises(SystemError, fromobject, NULL) def test_size(self): # Test PyBytes_Size() size = _testlimitedcapi.bytes_size self.assertEqual(size(b''), 0) self.assertEqual(size(b'abc'), 3) self.assertEqual(size(BytesSubclass(b'abc')), 3) self.assertRaises(TypeError, size, bytearray(b'abc')) self.assertRaises(TypeError, size, 'abc') self.assertRaises(TypeError, size, object()) # CRASHES size(NULL) def test_asstring(self): """Test PyBytes_AsString()""" asstring = _testlimitedcapi.bytes_asstring self.assertEqual(asstring(b'abc', 4), b'abc\0') self.assertEqual(asstring(b'abc\0def', 8), b'abc\0def\0') self.assertEqual(asstring(b'', 1), b'\0') self.assertRaises(TypeError, asstring, 'abc', 0) self.assertRaises(TypeError, asstring, object(), 0) # CRASHES asstring(NULL, 0) def test_asstringandsize(self): """Test PyBytes_AsStringAndSize()""" asstringandsize = _testlimitedcapi.bytes_asstringandsize asstringandsize_null = _testlimitedcapi.bytes_asstringandsize_null self.assertEqual(asstringandsize(b'abc', 4), (b'abc\0', 3)) self.assertEqual(asstringandsize(b'abc\0def', 8), (b'abc\0def\0', 7)) self.assertEqual(asstringandsize(b'', 1), (b'\0', 0)) self.assertEqual(asstringandsize_null(b'abc', 4), b'abc\0') self.assertRaises(ValueError, asstringandsize_null, b'abc\0def', 8) self.assertRaises(TypeError, asstringandsize, 'abc', 0) self.assertRaises(TypeError, asstringandsize_null, 'abc', 0) self.assertRaises(TypeError, asstringandsize, object(), 0) self.assertRaises(TypeError, asstringandsize_null, object(), 0) # CRASHES asstringandsize(NULL, 0) # CRASHES asstringandsize_null(NULL, 0) def test_repr(self): # Test PyBytes_Repr() bytes_repr = _testlimitedcapi.bytes_repr self.assertEqual(bytes_repr(b'', 0), r"""b''""") self.assertEqual(bytes_repr(b'''abc''', 0), r"""b'abc'""") self.assertEqual(bytes_repr(b'''abc''', 1), r"""b'abc'""") self.assertEqual(bytes_repr(b'''a'b"c"d''', 0), r"""b'a\'b"c"d'""") self.assertEqual(bytes_repr(b'''a'b"c"d''', 1), r"""b'a\'b"c"d'""") self.assertEqual(bytes_repr(b'''a'b"c''', 0), r"""b'a\'b"c'""") self.assertEqual(bytes_repr(b'''a'b"c''', 1), r"""b'a\'b"c'""") self.assertEqual(bytes_repr(b'''a'b'c"d''', 0), r"""b'a\'b\'c"d'""") self.assertEqual(bytes_repr(b'''a'b'c"d''', 1), r"""b'a\'b\'c"d'""") self.assertEqual(bytes_repr(b'''a'b'c'd''', 0), r"""b'a\'b\'c\'d'""") self.assertEqual(bytes_repr(b'''a'b'c'd''', 1), r'''b"a'b'c'd"''') self.assertEqual(bytes_repr(BytesSubclass(b'abc'), 0), r"""b'abc'""") # UDEFINED bytes_repr(object(), 0) # CRASHES bytes_repr(NULL, 0) def test_concat(self, concat=None): """Test PyBytes_Concat()""" if concat is None: concat = _testlimitedcapi.bytes_concat self.assertEqual(concat(b'abc', b'def'), b'abcdef') self.assertEqual(concat(b'a\0b', b'c\0d'), b'a\0bc\0d') self.assertEqual(concat(bytearray(b'abc'), b'def'), b'abcdef') self.assertEqual(concat(b'abc', bytearray(b'def')), b'abcdef') self.assertEqual(concat(bytearray(b'abc'), b''), b'abc') self.assertEqual(concat(b'', bytearray(b'def')), b'def') self.assertEqual(concat(memoryview(b'xabcy')[1:4], b'def'), b'abcdef') self.assertEqual(concat(b'abc', memoryview(b'xdefy')[1:4]), b'abcdef') self.assertEqual(concat(b'', b''), b'') self.assertEqual(concat(b'abc', b'def', True), b'abcdef') self.assertEqual(concat(b'abc', bytearray(b'def'), True), b'abcdef') # Check that it does not change the singleton self.assertEqual(concat(bytes(), b'def', True), b'def') self.assertEqual(len(bytes()), 0) self.assertRaises(TypeError, concat, memoryview(b'axbycz')[::2], b'def') self.assertRaises(TypeError, concat, b'abc', memoryview(b'dxeyfz')[::2]) self.assertRaises(TypeError, concat, b'abc', 'def') self.assertRaises(TypeError, concat, 'abc', b'def') self.assertRaises(TypeError, concat, 'abc', 'def') self.assertRaises(TypeError, concat, [], b'def') self.assertRaises(TypeError, concat, b'abc', []) self.assertRaises(TypeError, concat, [], []) self.assertEqual(concat(NULL, b'def'), NULL) self.assertEqual(concat(b'abc', NULL), NULL) self.assertEqual(concat(NULL, object()), NULL) self.assertEqual(concat(object(), NULL), NULL) def test_concatanddel(self): """Test PyBytes_ConcatAndDel()""" self.test_concat(_testlimitedcapi.bytes_concatanddel) def test_decodeescape(self): """Test PyBytes_DecodeEscape()""" decodeescape = _testlimitedcapi.bytes_decodeescape self.assertEqual(decodeescape(b''), b'') self.assertEqual(decodeescape(b'abc'), b'abc') self.assertEqual(decodeescape(br'\t\n\r\x0b\x0c\x00\\\'\"'), b'''\t\n\r\v\f\0\\'"''') self.assertEqual(decodeescape(b'\t\n\r\x0b\x0c\x00'), b'\t\n\r\v\f\0') self.assertEqual(decodeescape(br'\xa1\xa2'), b'\xa1\xa2') self.assertEqual(decodeescape(br'\2\24\241'), b'\x02\x14\xa1') self.assertEqual(decodeescape(b'\xa1\xa2'), b'\xa1\xa2') with self.assertWarns(DeprecationWarning): self.assertEqual(decodeescape(br'\u4f60'), br'\u4f60') with self.assertWarns(DeprecationWarning): self.assertEqual(decodeescape(br'\z'), br'\z') with self.assertWarns(DeprecationWarning): self.assertEqual(decodeescape(br'\541'), b'a') for b in b'\\', br'\x', br'\xa', br'\xz', br'\xaz': self.assertRaises(ValueError, decodeescape, b) self.assertRaises(ValueError, decodeescape, b, 'strict') self.assertEqual(decodeescape(br'x\xa', 'replace'), b'x?') self.assertEqual(decodeescape(br'x\xay', 'replace'), b'x?y') self.assertEqual(decodeescape(br'x\xa\xy', 'replace'), b'x??y') self.assertEqual(decodeescape(br'x\xa\xy', 'ignore'), b'xy') self.assertRaises(ValueError, decodeescape, b'\\', 'spam') self.assertRaises(OverflowError, decodeescape, b'abc', NULL, PY_SSIZE_T_MAX) self.assertRaises(OverflowError, decodeescape, NULL, NULL, PY_SSIZE_T_MAX) # INVALID decodeescape(NULL) # CRASHES decodeescape(b'abc', NULL, -1) # CRASHES decodeescape(NULL, NULL, 1) def test_resize(self): """Test _PyBytes_Resize()""" _resize = _testcapi.bytes_resize def assert_is_fresh_copy(result, refcnt, is_new_obj): self.assertEqual(refcnt, 1) self.assertTrue(is_new_obj) self.assertFalse(sys._is_immortal(result)) def resize(obj, size, new, compute_hash=False): old_size = len(obj) result, refcnt, is_new_obj = _resize(obj, size, new, compute_hash=compute_hash) if size == old_size: # Return the same object unchanged self.assertFalse(is_new_obj) elif old_size == 0: assert_is_fresh_copy(result, refcnt, is_new_obj) elif size == 0: # check that the result is the empty bytes string singleton self.assertEqual(result, b'') self.assertTrue(sys._is_immortal(result)) self.assertTrue(is_new_obj) elif (not new) or compute_hash: # gh-156995: Make sure that the result is a fresh object. # Previously, _PyBytes_Resize(&obj, 1) returned a singleton # if _PyObject_IsUniquelyReferenced() is false. assert_is_fresh_copy(result, refcnt, is_new_obj) else: # An in-place resize can return the same memory address, or # not. 'is_new_obj' cannot be tested. self.assertEqual(refcnt, 1) self.assertFalse(sys._is_immortal(result)) return result for new in True, False: for compute_hash in True, False: with self.subTest(new=new, compute_hash=compute_hash): self.assertEqual(resize(b'abc', 0, new, compute_hash), b'') self.assertEqual(resize(b'abc', 1, new, compute_hash), b'a') self.assertEqual(resize(b'abc', 2, new, compute_hash), b'ab') self.assertEqual(resize(b'abc', 3, new, compute_hash), b'abc') b = resize(b'abc', 4, new, compute_hash) self.assertEqual(len(b), 4) self.assertEqual(b[:3], b'abc') self.assertEqual(resize(b'a', 0, new, compute_hash), b'') self.assertEqual(resize(b'a', 1, new, compute_hash), b'a') b = resize(b'a', 2, new, compute_hash) self.assertEqual(len(b), 2) self.assertEqual(b[:1], b'a') self.assertEqual(resize(b'', 0, new, compute_hash), b'') self.assertEqual(len(resize(b'', 1, new, compute_hash)), 1) self.assertEqual(len(resize(b'', 2, new, compute_hash)), 2) self.assertRaises(SystemError, resize, b'abc', -1, False) self.assertRaises(SystemError, resize, bytearray(b'abc'), 3, False) # CRASHES resize(NULL, 0, False) # CRASHES resize(NULL, 3, False) def test_join(self): """Test PyBytes_Join()""" bytes_join = _testcapi.bytes_join self.assertEqual(bytes_join(b'', []), b'') self.assertEqual(bytes_join(b'sep', []), b'') self.assertEqual(bytes_join(b'', [b'a', b'b', b'c']), b'abc') self.assertEqual(bytes_join(b'-', [b'a', b'b', b'c']), b'a-b-c') self.assertEqual(bytes_join(b' - ', [b'a', b'b', b'c']), b'a - b - c') self.assertEqual(bytes_join(b'-', [bytearray(b'abc'), memoryview(b'def')]), b'abc-def') self.assertEqual(bytes_join(b'-', iter([b'a', b'b', b'c'])), b'a-b-c') # invalid 'sep' argument with self.assertRaises(TypeError): bytes_join(bytearray(b'sep'), []) with self.assertRaises(TypeError): bytes_join(memoryview(b'sep'), []) with self.assertRaises(TypeError): bytes_join('', []) # empty Unicode string with self.assertRaises(TypeError): bytes_join('unicode', []) with self.assertRaises(TypeError): bytes_join(123, []) with self.assertRaises(SystemError): self.assertEqual(bytes_join(NULL, [b'a', b'b', b'c']), b'abc') # invalid 'iterable' argument with self.assertRaises(TypeError): bytes_join(b'', [b'bytes', 'unicode']) with self.assertRaises(TypeError): bytes_join(b'', [b'bytes', 123]) with self.assertRaises(TypeError): bytes_join(b'', 123) with self.assertRaises(SystemError): bytes_join(b'', NULL) class BaseWriterTest: RESULT_TYPE = NotImplementedError SMALL_BUFFER = 11 # bytes assert SMALL_BUFFER < _testcapi.PyBytesWriter_small_buffer LARGE_BUFFER = _testcapi.PyBytesWriter_small_buffer + 17 # bytes NEW_BYTE = b'\xff' def create_writer(self, alloc=0, string=b''): raise NotImplementedError def bytes_equal(self, result, expected): # Similar to assertEqual(), but check also that the result type # is RESULT_TYPE. self.assertEqual(result, expected) self.assertEqual(type(result), self.RESULT_TYPE) def test_create(self): # Test PyBytesWriter_Create() writer = self.create_writer() self.assertEqual(writer.get_size(), 0) self.bytes_equal(writer.finish(), b'') writer = self.create_writer(3) writer.write(0, b'abc') self.assertEqual(writer.get_size(), 3) self.bytes_equal(writer.finish(), b'abc') @unittest.skipUnless(support.Py_DEBUG, 'need Py_DEBUG') def test_get_data(self): # Test PyBytesWriter_GetData() writer = self.create_writer(6) NEW_BYTE = self.NEW_BYTE self.assertEqual(writer.get_data(), NEW_BYTE * 6) writer.write(0, b'abc') self.assertEqual(writer.get_data(), b'abc' + NEW_BYTE * 3) writer.write(3, b'123') self.assertEqual(writer.get_data(), b'abc123') # Switch from small buffer to large buffer small, large = self.SMALL_BUFFER, self.LARGE_BUFFER writer = self.create_writer(small) self.assertEqual(writer.get_data(), NEW_BYTE * small) writer.write(0, b's' * small) self.assertEqual(writer.get_data(), b's' * small) writer.resize(large) self.assertEqual(writer.get_data(), b's' * small + NEW_BYTE * (large - small)) writer.write(small, b'L' * (large - small)) self.assertEqual(writer.get_data(), b's' * small + b'L' * (large - small)) # Resize large buffer small, large = self.SMALL_BUFFER, self.LARGE_BUFFER writer = self.create_writer(large) self.assertEqual(writer.get_data(), NEW_BYTE * large) writer.write(0, b'L' * large) self.assertEqual(writer.get_data(), b'L' * large) writer.resize(large + 10) self.assertEqual(writer.get_data(), b'L' * large + NEW_BYTE * 10) writer.write(large, b'#' * 10) self.assertEqual(writer.get_data(), b'L' * large + b'#' * 10) def test_finish_with_size(self): # Test PyBytesWriter_FinishWithSize() writer = self.create_writer(10) writer.write(0, b'abc123') self.assertEqual(writer.get_size(), 10) self.bytes_equal(writer.finish_with_size(3), b'abc') # Error if the size is negative writer = self.create_writer(3, ) writer.write(0, b'abc') with self.assertRaises(ValueError): writer.finish_with_size(-3) # Error if the requested size is larger than the allocated size writer = self.create_writer(3) writer.write(0, b'abc') with self.assertRaises(ValueError): writer.finish_with_size(4) def test_write_bytes(self): # Test PyBytesWriter_WriteBytes() writer = self.create_writer() writer.write_bytes(b'Hello World!', -1) self.bytes_equal(writer.finish(), b'Hello World!') writer = self.create_writer() writer.write_bytes(b'Hello ', -1) writer.write_bytes(b'World! ', 6) self.bytes_equal(writer.finish(), b'Hello World!') def test_resize(self): # Test PyBytesWriter_Resize() writer = self.create_writer() writer.resize(len(b'hello')) writer.write(0, b'hello') self.bytes_equal(writer.finish(), b'hello') writer = self.create_writer() writer.resize(0) # noop writer.resize(len(b'number')) writer.write(0, b'number') writer.resize(len(b'number=')) writer.write(len(b'number'), b'=') writer.resize(len(b'number=123'), ) writer.write(len(b'number='), b'123') writer.resize(len(b'number=123')) # noop self.bytes_equal(writer.finish(), b'number=123') # Switch from small buffer to large buffer writer = self.create_writer() small, large = self.SMALL_BUFFER, self.LARGE_BUFFER writer.resize(small) writer.write(0, b's' * small) writer.resize(large) writer.write(small, b'L' * (large - small)) self.bytes_equal(writer.finish(), b's' * small + b'L' * (large - small)) # Make sure that it's possible to write after a resize to zero # when a bytes/bytearray object is allocated. writer = self.create_writer() writer.resize(self.LARGE_BUFFER) writer.resize(0) writer.write_bytes(b'abc', 3) self.bytes_equal(writer.finish(), b'abc') # invalid size for size in (self.SMALL_BUFFER, self.LARGE_BUFFER): with self.subTest(size=size): writer = self.create_writer() writer.write_bytes(b'x' * size, -1) with self.assertRaisesRegex(ValueError, 'size must be >= 0'): writer.resize(-1) with self.assertRaises((MemoryError, OverflowError)): writer.resize(_testcapi.PY_SSIZE_T_MAX) self.bytes_equal(writer.finish(), b'x' * size) @support.nomemtest def test_resize_error(self): # Test PyBytesWriter_Resize() error init = b'x' * self.LARGE_BUFFER writer = self.create_writer(len(init)) writer.write(0, init) size = len(init) + 100 with self.assertRaises(MemoryError): with support.inject_memory_error_cm(): writer.resize(size) suffix = b'still working' writer.write_bytes(suffix, -1) self.bytes_equal(writer.finish(), init + suffix) # Note: PyBytesWriter_Resize() leaves the buffer unchanged (no resize) # if the new size is smaller than the allocated size def test_grow(self): # Test PyBytesWriter_Grow() writer = self.create_writer(0) writer.grow(len(b'number=123')) writer.write(0, b'number=123') self.bytes_equal(writer.finish(), b'number=123') writer = self.create_writer() writer.grow(0) # noop writer.grow(len(b'number')) writer.write(0, b'number') writer.grow(len(b'=')) writer.write(len(b'number'), b'=') writer.grow(len(b'123'), ) writer.write(len(b'number='), b'123') writer.grow(0) # noop self.bytes_equal(writer.finish(), b'number=123') # Switch from small buffer to large buffer writer = self.create_writer() small, large = self.SMALL_BUFFER, self.LARGE_BUFFER writer.grow(small) writer.write(0, b's' * small) writer.grow(large - small) writer.write(small, b'L' * (large - small)) self.bytes_equal(writer.finish(), b's' * small + b'L' * (large - small)) # invalid size for size in (self.SMALL_BUFFER, self.LARGE_BUFFER): with self.subTest(size=size): writer = self.create_writer() writer.write_bytes(b'x' * size, -1) with self.assertRaisesRegex(ValueError, 'invalid size'): writer.grow(-size - 1) with self.assertRaises(MemoryError): writer.grow(_testcapi.PY_SSIZE_T_MAX) self.bytes_equal(writer.finish(), b'x' * size) @support.nomemtest def test_grow_error(self): # Test PyBytesWriter_Grow() error init = b'x' * self.LARGE_BUFFER writer = self.create_writer(len(init)) writer.write(0, init) with self.assertRaises(MemoryError): with support.inject_memory_error_cm(): writer.grow(100) suffix = b'still working' writer.write_bytes(suffix, -1) self.bytes_equal(writer.finish(), init + suffix) # Note: PyBytesWriter_Grow() leaves the buffer unchanged (no resize) # if grow is negative. def test_format_i(self): # Test PyBytesWriter_Format() writer = self.create_writer() writer.format_i(b'x=%i', 123456) self.bytes_equal(writer.finish(), b'x=123456') writer = self.create_writer() writer.format_i(b'x=%i, ', 123) writer.format_i(b'y=%d', 456) self.bytes_equal(writer.finish(), b'x=123, y=456') # Test invalid format: "%t" is invalid, stop the parser and copy # remaining bytes writer = self.create_writer() writer.write_bytes(b'prefix ', -1) writer.format_i(b'y=%._ t %i', 0) writer.write_bytes(b'.', 1) self.bytes_equal(writer.finish(), b'prefix y=%._ t %i.') def test_format_s(self): # Test PyBytesWriter_Format() writer = self.create_writer() writer.format_s(b's=%s', b'Hello World') self.bytes_equal(writer.finish(), b's=Hello World') # Test precision writer = self.create_writer() writer.format_s(b'%.6s', b'Monty Python') writer.format_s(b'%.20s', b'3.0') self.bytes_equal(writer.finish(), b'Monty 3.0') @support.nomemtest def test_format_s_memory_error(self): writer = self.create_writer() s = b'x' * self.LARGE_BUFFER with self.assertRaises(MemoryError): _testcapi.call_with_nomemory(0, 0, writer.format_s, b's=%s', s) writer.write_bytes(b'after', -1) self.bytes_equal(writer.finish(), b'after') @threading_helper.requires_working_threading() def test_thread(self): # PyBytesWriter can be used by multiple threads: it's up to the caller # to implement a lock to prevent concurrent accesses. writer = self.create_writer(0) size = None data = None def thread_func(writer, LARGE_BUFFER): nonlocal size, data # create a bytes object for the buffer writer.write_bytes(b'x' * LARGE_BUFFER, LARGE_BUFFER) # so we can check a write with a bytes object writer.write_bytes(b'yz', 2) writer.format_i(b'i=%i', 5) writer.resize(10) data = writer.get_data() size = writer.get_size() thread = threading.Thread(target=thread_func, args=(writer, self.LARGE_BUFFER)) thread.start() threading_helper.join_thread(thread) self.assertEqual(size, 10) self.assertEqual(data, b'x' * 10) self.bytes_equal(writer.finish(), b'x' * 10) class BytesWriterTest(BaseWriterTest, unittest.TestCase): RESULT_TYPE = bytes def create_writer(self, size=0): # Test PyBytesWriter_Create() return _testcapi.PyBytesWriter(size, 0) # Only PyBytesWriter_Create() returns singletons def test_singletons(self): empty = b'' singletons = {ch: bytes((ch,)) for ch in range(256)} writer = self.create_writer() self.assertIs(writer.finish(), empty) # Large buffer writer = self.create_writer() unused_text = b'x' * self.LARGE_BUFFER writer.write_bytes(unused_text, len(unused_text)) self.assertIs(writer.finish_with_size(0), empty) # Large buffer with resize writer = self.create_writer() unused_text = b'x' * self.LARGE_BUFFER writer.write_bytes(unused_text, len(unused_text)) writer.resize(0) self.assertIs(writer.finish(), empty) for ch in range(256): byte = bytes((ch,)) writer = self.create_writer() writer.write_bytes(byte, len(byte)) self.assertIs(writer.finish(), singletons[ch]) def test_example_abc(self): self.assertEqual(_testcapi.byteswriter_abc(), b'abc') def test_example_resize(self): self.assertEqual(_testcapi.byteswriter_resize(), b'Hello World') def test_example_highlevel(self): self.assertEqual(_testcapi.byteswriter_highlevel(), b'Hello World!') class ByteArrayWriterTest(BaseWriterTest, unittest.TestCase): RESULT_TYPE = bytearray def create_writer(self, size=0): # Test private _PyBytesWriter_CreateByteArray() return _testcapi.PyBytesWriter(size, 1) if __name__ == "__main__": unittest.main()