| import sys | |
| import textwrap | |
| import threading | |
| import unittest | |
| from test import support | |
| from test.support import import_helper | |
| from test.support import threading_helper | |
| from test.support.script_helper import assert_python_failure | |
| _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) | |
| @unittest.skipUnless(support.Py_DEBUG, 'need debug build (Py_DEBUG)') | |
| def test_detect_overflow(self): | |
| # Test detection of buffer overflow | |
| size = 123 # bytes | |
| overflow = 1 # bytes | |
| code = textwrap.dedent(f''' | |
| from test.support import SuppressCrashReport | |
| import _testcapi | |
| size = {size} | |
| overflow = {overflow} | |
| with SuppressCrashReport(): | |
| # Trigger a buffer overflow in a new bytes | |
| ba = _testcapi.bytes_overflow(size, overflow) | |
| ba = None | |
| ''') | |
| proc = assert_python_failure('-c', code) | |
| self.assertIn(b'Buffer overflow detected in bytes object', proc.err) | |
| self.assertIn(f'at position {size}'.encode(), proc.err) | |
| def get_data_canary(writer): | |
| size = writer.get_size() + 1 | |
| return writer.get_data(size) | |
| 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' | |
| CANARY_BYTE = b'\xdd' | |
| 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 | |
| CANARY_BYTE = self.CANARY_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! <truncated>', 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) | |
| @unittest.skipUnless(support.Py_DEBUG, 'need debug build') | |
| def test_resize_canary(self): | |
| CANARY_BYTE = self.CANARY_BYTE | |
| for size in (self.SMALL_BUFFER, self.LARGE_BUFFER): | |
| with self.subTest(size=size): | |
| # Truncate the last byte | |
| data = b'x' * size | |
| writer = self.create_writer(size) | |
| writer.write(0, data) | |
| self.assertEqual(get_data_canary(writer), data + CANARY_BYTE) | |
| writer.resize(size - 1) | |
| self.assertEqual(get_data_canary(writer), data[:-1] + CANARY_BYTE) | |
| self.bytes_equal(writer.finish(), data[:-1]) | |
| # Make the buffer empty | |
| writer = self.create_writer(size) | |
| writer.write(0, data) | |
| writer.resize(0) | |
| self.assertEqual(get_data_canary(writer), CANARY_BYTE) | |
| self.bytes_equal(writer.finish(), b'') | |
| @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) | |
| @unittest.skipUnless(support.Py_DEBUG, 'need debug build') | |
| def test_grow_canary(self): | |
| CANARY_BYTE = self.CANARY_BYTE | |
| for size in (self.SMALL_BUFFER, self.LARGE_BUFFER): | |
| with self.subTest(size=size): | |
| # Truncate the last byte | |
| data = b'x' * size | |
| writer = self.create_writer(size) | |
| writer.write(0, data) | |
| self.assertEqual(get_data_canary(writer), data + CANARY_BYTE) | |
| writer.grow(-1) | |
| self.assertEqual(get_data_canary(writer), data[:-1] + CANARY_BYTE) | |
| self.bytes_equal(writer.finish(), data[:-1]) | |
| # Make the buffer empty | |
| writer = self.create_writer(size) | |
| writer.write(0, data) | |
| writer.grow(-size) | |
| self.assertEqual(writer.get_data(), b'') | |
| self.bytes_equal(writer.finish(), b'') | |
| @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') | |
| # Make sure that %i allocates the right number of bytes | |
| n_str = '1234567' | |
| for i in range(1, len(n_str) + 1): | |
| n = int(n_str[:i]) | |
| writer = self.create_writer(0) | |
| writer.format_i(b'n=%i', n) | |
| self.bytes_equal(writer.finish(), f'n={n}'.encode()) | |
| # 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') | |
| # Make sure that %s allocates the right number of bytes | |
| text = b'1234567' | |
| for i in range(1, len(text) + 1): | |
| s = text[:i] | |
| writer = self.create_writer(0) | |
| writer.format_s(b's=%s', s) | |
| self.bytes_equal(writer.finish(), b's=' + s) | |
| @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') | |
| @unittest.skipUnless(support.Py_DEBUG, 'need a Python debug build') | |
| def test_canary_byte(self): | |
| small_buffer = _testcapi.PyBytesWriter_small_buffer | |
| large_size = small_buffer * 10 | |
| use_bytearray = (self.RESULT_TYPE == bytearray) | |
| # Test small buffer and large buffer | |
| for size in (0, self.SMALL_BUFFER, self.LARGE_BUFFER): | |
| for operation in ( | |
| 'writer.get_data()', | |
| 'writer.get_size()', | |
| f'writer.resize({size} * 2)', | |
| f'writer.grow({size})', | |
| 'writer.discard()', | |
| 'writer.finish()', | |
| ): | |
| with self.subTest(size=size, operation=operation): | |
| code = textwrap.dedent(f""" | |
| from test.support import SuppressCrashReport | |
| import os | |
| import _testcapi | |
| size = {size} | |
| # Add an extra '#' byte to trigger a buffer overflow | |
| data = b'x' * size + b'#' | |
| use_bytearray = {use_bytearray} | |
| writer = _testcapi.PyBytesWriter(size, use_bytearray) | |
| with SuppressCrashReport(): | |
| writer.write(0, data, check=False) | |
| try: | |
| {operation} | |
| except: | |
| # Ignore all exceptions | |
| pass | |
| # If we reached this line, the operation didn't | |
| # detect the overflow. Exit immediatetly without | |
| # calling the writer destructor since it can detect | |
| # the overflow. | |
| os._exit(0) | |
| """) | |
| proc = assert_python_failure('-c', code) | |
| self.assertIn(b'Buffer overflow detected in PyBytesWriter', | |
| proc.err) | |
| self.assertIn(f'at position {size}'.encode(), | |
| proc.err) | |
| @unittest.skipUnless(support.Py_DEBUG, 'need debug build') | |
| def test_get_data_canary(self): | |
| # Test PyBytesWriter_GetData() | |
| NEW_BYTE = self.NEW_BYTE | |
| CANARY_BYTE = self.CANARY_BYTE | |
| writer = self.create_writer(6) | |
| self.assertEqual(get_data_canary(writer), | |
| NEW_BYTE * 6 + CANARY_BYTE) | |
| writer.write(0, b'abc') | |
| self.assertEqual(get_data_canary(writer), | |
| b'abc' + NEW_BYTE * 3 + CANARY_BYTE) | |
| writer.write(3, b'123') | |
| self.assertEqual(get_data_canary(writer), | |
| b'abc123' + CANARY_BYTE) | |
| @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]) | |
| # Large buffer | |
| writer = self.create_writer() | |
| unused_text = b'x' * self.LARGE_BUFFER | |
| writer.write_bytes(byte + unused_text, | |
| len(byte) + len(unused_text)) | |
| self.assertIs(writer.finish_with_size(1), 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() |