Align bytes and bytearray operations with Python 3.15 (#9165)

Correct quote escaping, replacement count keyword binding, signed-minimum hex grouping, and clear storage/export handling. Migrate canonical bytes coverage through fail-closed preservation of the existing RustPython customizations, and retire only two freshly verified formatting guards.

Assisted-by: OpenAI Codex:model version unavailable

Lib/test/test_bytes.py

@@ -553,10 +553,10 @@ class BaseBytesTest:
         self.assertEqual(three_bytes.hex('*', -2), 'b901*ef')
         self.assertEqual(three_bytes.hex(sep=':', bytes_per_sep=2), 'b9:01ef')
         self.assertEqual(three_bytes.hex(sep='*', bytes_per_sep=-2), 'b901*ef')
-        for bytes_per_sep in 3, -3, 2**31-1, -(2**31-1):
+        for bytes_per_sep in 3, -3, sys.maxsize, -sys.maxsize:
             with self.subTest(bytes_per_sep=bytes_per_sep):
                 self.assertEqual(three_bytes.hex(':', bytes_per_sep), 'b901ef')
-        for bytes_per_sep in 2**31, -2**31, 2**1000, -2**1000:
+        for bytes_per_sep in sys.maxsize+1, -sys.maxsize-1, 2**1000, -2**1000:
             with self.subTest(bytes_per_sep=bytes_per_sep):
                 try:
                     self.assertEqual(three_bytes.hex(':', bytes_per_sep), 'b901ef')
@@ -586,6 +586,37 @@ class BaseBytesTest:
         self.assertEqual(six_bytes.hex(':', -6), '0306090c0f12')
         self.assertEqual(six_bytes.hex(' ', -95), '0306090c0f12')
 
+    def test_hex_simd_boundaries(self):
+        # Test lengths around the SIMD threshold (16 bytes).
+        # SIMD processes 16 bytes at a time; smaller inputs use scalar code.
+        for length in (14, 15, 16, 17, 31, 32, 33, 64, 65):
+            data = self.type2test(bytes(range(length)))
+            expected = ''.join(f'{b:02x}' for b in range(length))
+            with self.subTest(length=length):
+                self.assertEqual(data.hex(), expected)
+
+    def test_hex_nibble_boundaries(self):
+        # Test the nibble value boundary at 9/10 (where '9' becomes 'a').
+        # SIMD uses signed comparison for efficiency; verify correctness
+        # at this boundary for various nibble combinations.
+        boundary_bytes = self.type2test(bytes([
+            0x09,  # both nibbles: 0, 9
+            0x0a,  # both nibbles: 0, 10
+            0x90,  # both nibbles: 9, 0
+            0x99,  # both nibbles: 9, 9 (max all-digit)
+            0x9a,  # both nibbles: 9, 10
+            0xa0,  # both nibbles: 10, 0
+            0xa9,  # both nibbles: 10, 9
+            0xaa,  # both nibbles: 10, 10 (min all-letter)
+            0x00,  # min value
+            0xff,  # max value
+        ]))
+        self.assertEqual(boundary_bytes.hex(), '090a90999aa0a9aa00ff')
+
+        # Repeat with 16+ bytes to exercise SIMD path
+        simd_boundary = self.type2test(boundary_bytes * 2)
+        self.assertEqual(simd_boundary.hex(), '090a90999aa0a9aa00ff' * 2)
+
     def test_join(self):
         self.assertEqual(self.type2test(b"").join([]), b"")
         self.assertEqual(self.type2test(b"").join([b""]), b"")
@@ -616,6 +647,32 @@ class BaseBytesTest:
         with self.assertRaises(TypeError):
             dot_join([memoryview(b"ab"), "cd", b"ef"])
 
+    def test_join_concurrent_buffer_mutation(self):
+        # __buffer__() can release the GIL, letting another thread concurrently
+        # mutate the joined sequence (simulated here by mutating in __buffer__).
+        # See: https://github.com/python/cpython/issues/151295
+        def make_seq(mutate):
+            # Item is only referenced from the list slot, so mutate() frees it.
+            class Item:
+                def __buffer__(self, flags):
+                    mutate(seq)
+                    return memoryview(b'x')
+            seq = [b'a', Item(), b'c']
+            return seq
+
+        for sep in (self.type2test(b''), self.type2test(b'::')):
+            with self.subTest(sep=sep):
+                # Changing the list length is reported as a RuntimeError.
+                seq = make_seq(lambda seq: seq.clear())
+                self.assertRaises(RuntimeError, sep.join, seq)
+
+                # The list length is unchanged, so the size-change recheck
+                # cannot fire: only keeping the item alive avoids the crash.
+                def replace(seq):
+                    seq[1] = b'z'
+                seq = make_seq(replace)
+                self.assertEqual(sep.join(seq), sep.join([b'a', b'x', b'c']))
+
     def test_count(self):
         b = self.type2test(b'mississippi')
         i = 105
@@ -794,16 +851,16 @@ class BaseBytesTest:
         pi = PseudoFloat(3.1415)
 
         exceptions_params = [
-            ('%x format: an integer is required, not float', b'%x', 3.14),
-            ('%X format: an integer is required, not float', b'%X', 2.11),
-            ('%o format: an integer is required, not float', b'%o', 1.79),
-            ('%x format: an integer is required, not PseudoFloat', b'%x', pi),
-            ('%x format: an integer is required, not complex', b'%x', 3j),
-            ('%X format: an integer is required, not complex', b'%X', 2j),
-            ('%o format: an integer is required, not complex', b'%o', 1j),
-            ('%u format: a real number is required, not complex', b'%u', 3j),
-            ('%i format: a real number is required, not complex', b'%i', 2j),
-            ('%d format: a real number is required, not complex', b'%d', 2j),
+            ('%x requires an integer, not float', b'%x', 3.14),
+            ('%X requires an integer, not float', b'%X', 2.11),
+            ('%o requires an integer, not float', b'%o', 1.79),
+            (r'%x requires an integer, not .*\.PseudoFloat', b'%x', pi),
+            ('%x requires an integer, not complex', b'%x', 3j),
+            ('%X requires an integer, not complex', b'%X', 2j),
+            ('%o requires an integer, not complex', b'%o', 1j),
+            ('%u requires a real number, not complex', b'%u', 3j),
+            ('%i requires a real number, not complex', b'%i', 2j),
+            ('%d requires a real number, not complex', b'%d', 2j),
             (
                 r'%c requires an integer in range\(256\)'
                 r' or a single byte, not .*\.PseudoFloat',
@@ -812,7 +869,7 @@ class BaseBytesTest:
         ]
 
         for msg, format_bytes, value in exceptions_params:
-            with self.assertRaisesRegex(TypeError, msg):
+            with self.assertRaisesRegex(TypeError, 'format argument: ' + msg):
                 operator.mod(format_bytes, value)
 
     def test_memory_leak_gh_140939(self):
@@ -849,6 +906,13 @@ class BaseBytesTest:
         self.assertEqual(b.replace(b'i', b'a'), b'massassappa')
         self.assertEqual(b.replace(b'ss', b'x'), b'mixixippi')
 
+    def test_replace_count_keyword(self):
+        b = self.type2test(b'aa')
+        self.assertEqual(b.replace(b'a', b'b', count=0), b'aa')
+        self.assertEqual(b.replace(b'a', b'b', count=1), b'ba')
+        self.assertEqual(b.replace(b'a', b'b', count=2), b'bb')
+        self.assertEqual(b.replace(b'a', b'b', count=3), b'bb')
+
     def test_replace_int_error(self):
         self.assertRaises(TypeError, self.type2test(b'a b').replace, 32, b'')
 
@@ -1337,10 +1401,6 @@ class BytesTest(BaseBytesTest, unittest.TestCase):
         self.assertNotEqual(id(s), id(1 * s))
         self.assertNotEqual(id(s), id(s * 2))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
-    def test_mod(self):
-        return super().test_mod()
-
 
 class ByteArrayTest(BaseBytesTest, unittest.TestCase):
     type2test = bytearray
@@ -1431,6 +1491,16 @@ class ByteArrayTest(BaseBytesTest, unittest.TestCase):
         b.append(ord('p'))
         self.assertEqual(b, b'p')
 
+        # Cleared object should be empty.
+        b = bytearray(b'abc')
+        b.clear()
+        self.assertEqual(b.__alloc__(), 0)
+        base_size = sys.getsizeof(bytearray())
+        self.assertEqual(sys.getsizeof(b), base_size)
+        c = b.copy()
+        self.assertEqual(c.__alloc__(), 0)
+        self.assertEqual(sys.getsizeof(c), base_size)
+
     def test_copy(self):
         b = bytearray(b'abc')
         bb = b.copy()
@@ -1493,6 +1563,124 @@ class ByteArrayTest(BaseBytesTest, unittest.TestCase):
         self.assertRaises(MemoryError, bytearray().resize, sys.maxsize)
         self.assertRaises(MemoryError, bytearray(1000).resize, sys.maxsize)
 
+    def test_take_bytes(self):
+        ba = bytearray(b'ab')
+        self.assertEqual(ba.take_bytes(), b'ab')
+        self.assertEqual(len(ba), 0)
+        self.assertEqual(ba, bytearray(b''))
+        self.assertEqual(ba.__alloc__(), 0)
+        base_size = sys.getsizeof(bytearray())
+        self.assertEqual(sys.getsizeof(ba), base_size)
+
+        # Positive and negative slicing.
+        ba = bytearray(b'abcdef')
+        self.assertEqual(ba.take_bytes(1), b'a')
+        self.assertEqual(ba, bytearray(b'bcdef'))
+        self.assertEqual(len(ba), 5)
+        self.assertEqual(ba.take_bytes(-5), b'')
+        self.assertEqual(ba, bytearray(b'bcdef'))
+        self.assertEqual(len(ba), 5)
+        self.assertEqual(ba.take_bytes(-3), b'bc')
+        self.assertEqual(ba, bytearray(b'def'))
+        self.assertEqual(len(ba), 3)
+        self.assertEqual(ba.take_bytes(3), b'def')
+        self.assertEqual(ba, bytearray(b''))
+        self.assertEqual(len(ba), 0)
+
+        # Take nothing from emptiness.
+        self.assertEqual(ba.take_bytes(0), b'')
+        self.assertEqual(ba.take_bytes(), b'')
+        self.assertEqual(ba.take_bytes(None), b'')
+
+        # Out of bounds, bad take value.
+        self.assertRaises(IndexError, ba.take_bytes, -1)
+        self.assertRaises(TypeError, ba.take_bytes, 3.14)
+        ba = bytearray(b'abcdef')
+        self.assertRaises(IndexError, ba.take_bytes, 7)
+
+        # Offset between physical and logical start (ob_bytes != ob_start).
+        ba = bytearray(b'abcde')
+        del ba[:2]
+        self.assertEqual(ba, bytearray(b'cde'))
+        self.assertEqual(ba.take_bytes(), b'cde')
+
+        # Overallocation at end.
+        ba = bytearray(b'abcde')
+        del ba[-2:]
+        self.assertEqual(ba, bytearray(b'abc'))
+        self.assertEqual(ba.take_bytes(), b'abc')
+        ba = bytearray(b'abcde')
+        ba.resize(4)
+        self.assertEqual(ba.take_bytes(), b'abcd')
+
+        # Take of a bytearray with references should fail.
+        ba = bytearray(b'abc')
+        with memoryview(ba) as mv:
+            self.assertRaises(BufferError, ba.take_bytes)
+        self.assertEqual(ba.take_bytes(), b'abc')
+
+        # Leaving one byte must not adopt the shared single-byte bytes object
+        # as the buffer.
+        ba = bytearray(b'abc')
+        self.assertEqual(ba.take_bytes(2), b'ab')
+        ba[0] = ord('A')
+        self.assertEqual(ba, bytearray(b'A'))
+        self.assertEqual(ord(b'c'), ord('c'))
+
+    @support.cpython_only  # tests an implementation detail
+    def test_take_bytes_optimization(self):
+        # Validate optimization around taking lots of little chunks out of a
+        # much bigger buffer. Save work by only copying a little rather than
+        # moving a lot.
+        ba = bytearray(b'abcdef' + b'0' * 1000)
+        start_alloc = ba.__alloc__()
+
+        # Take two bytes at a time, checking alloc doesn't change.
+        self.assertEqual(ba.take_bytes(2), b'ab')
+        self.assertEqual(ba.__alloc__(), start_alloc)
+        self.assertEqual(len(ba), 4 + 1000)
+        self.assertEqual(ba.take_bytes(2), b'cd')
+        self.assertEqual(ba.__alloc__(), start_alloc)
+        self.assertEqual(len(ba), 2 + 1000)
+        self.assertEqual(ba.take_bytes(2), b'ef')
+        self.assertEqual(ba.__alloc__(), start_alloc)
+        self.assertEqual(len(ba), 0 + 1000)
+        self.assertEqual(ba.__alloc__(), start_alloc)
+
+        # Take over half, alloc shrinks to exact size.
+        self.assertEqual(ba.take_bytes(501), b'0' * 501)
+        self.assertEqual(len(ba), 499)
+        bytes_header_size = sys.getsizeof(b'')
+        self.assertEqual(ba.__alloc__(), 499 + bytes_header_size)
+
+    def test_take_bytes_reentrant_resize(self):
+        # gh-153570: n.__index__() can resize the bytearray, so take_bytes()
+        # must re-read the size afterwards.  It cached the size before the
+        # call and used it for the bounds check and the buffer reads, so a
+        # reentrant clear() returned freed memory (a use-after-free read).
+        def take(target, resize, n):
+            class Evil:
+                def __index__(self):
+                    resize(target)
+                    return n
+            return target.take_bytes(Evil())
+
+        # clear() during __index__: nothing is left to take.
+        ba = bytearray(b'abcdefgh')
+        with self.assertRaises(IndexError):
+            take(ba, lambda b: b.clear(), 8)
+        self.assertEqual(ba, b'')
+
+        # shrink during __index__: n past the new size is out of range.
+        ba = bytearray(b'abcdefgh')
+        with self.assertRaises(IndexError):
+            take(ba, lambda b: b.__delitem__(slice(4, None)), 8)
+        self.assertEqual(ba, b'abcd')
+
+        # grow during __index__: the take runs against the new, larger size.
+        ba = bytearray(b'abcd')
+        self.assertEqual(take(ba, lambda b: b.extend(b'efgh'), 8), b'abcdefgh')
+        self.assertEqual(ba, b'')
 
     @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_setitem(self):
@@ -2073,9 +2261,45 @@ class ByteArrayTest(BaseBytesTest, unittest.TestCase):
 
         self.assertRaises(BufferError, ba.hex, S(b':'))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
-    def test_mod(self):
-        return super().test_mod()
+    def test_no_init_called(self):
+        # A bytearray created without calling bytearray.__init__
+        # should not crash the interpreter (see gh-153419).
+        def bytearray_new():
+            return bytearray.__new__(bytearray)
+
+        bytearray_new().insert(0, 1)
+        bytearray_new().extend(b"x")
+        bytearray_new().extend([1, 2, 3])
+        bytearray_new().resize(4)
+        bytearray_new().__init__(5)
+        bytearray_new().__init__(b"xyz")
+        bytearray_new().take_bytes()
+        bytearray_new().take_bytes(0)
+
+        a = bytearray_new()
+        a.append(1)
+
+        a = bytearray_new()
+        a += b"x"
+
+        a = bytearray_new()
+        a[:] = b"xyz"
+
+    def test_reinit_length(self):
+        # There is a shortcut taken when resizing, where alloc/2 < newsize.
+        # In this case, the existing buffer is reused, rather than reset.
+        # If this happens when newsize == 0 and alloc == 1, then various
+        # code assumptions can be violated.  This test should catch those
+        # in debug builds. (see gh-153419)
+        a = bytearray(1)
+        a.__init__()
+        self.assertEqual(a, b"")
+
+    def test_reinit_with_view(self):
+        a = bytearray()
+        with memoryview(a):
+            self.assertRaises(BufferError, a.__init__, "x", "ascii")
+        self.assertEqual(a, b"")
 
 
 class AssortedBytesTest(unittest.TestCase):
@@ -2105,7 +2329,7 @@ class AssortedBytesTest(unittest.TestCase):
         self.assertEqual(f(bytearray([7, 8, 9, 10, 11, 12, 13])),
                             r"bytearray(b'\x07\x08\t\n\x0b\x0c\r')")
         self.assertEqual(f(bytearray(b'"')), """bytearray(b'"')""") # '"'
-        self.assertEqual(f(bytearray(b"'")), r'''bytearray(b"\'")''') # "\'"
+        self.assertEqual(f(bytearray(b"'")), '''bytearray(b"'")''') # "'"
         self.assertEqual(f(bytearray(b"'\"")), r"""bytearray(b'\'"')""") # '\'"'
         self.assertEqual(f(bytearray(b"\"'\"")), r"""bytearray(b'"\'"')""") # '"\'"'
         self.assertEqual(f(bytearray(b'\'"\'')), r"""bytearray(b'\'"\'')""") # '\'"\''
@@ -2698,6 +2922,18 @@ class FreeThreadingTest(unittest.TestCase):
             b.wait()
             a.resize(10)
 
+        def take_bytes(b, a):  # MODIFIES!
+            b.wait()
+            c = a.take_bytes()
+            assert not c or c[0] == 48  # '0'
+
+        def take_bytes_n(b, a):  # MODIFIES!
+            b.wait()
+            try:
+                c = a.take_bytes(10)
+                assert c == b'0123456789'
+            except IndexError: pass
+
         def check(funcs, a=None, *args):
             if a is None:
                 a = bytearray(b'0' * 0x400000)
@@ -2762,6 +2998,10 @@ class FreeThreadingTest(unittest.TestCase):
 
         check([clear] + [resize] * 10)
 
+        check([clear] + [take_bytes] * 10)
+        check([take_bytes_n] * 10, bytearray(b'0123456789' * 0x400))
+        check([take_bytes_n] * 10, bytearray(b'0123456789' * 5))
+
         check([clear] + [contains] * 10)
         check([clear] + [subscript] * 10)
         check([clear2] + [ass_subscript2] * 10, None, bytearray(b'0' * 0x400000))
@@ -2834,5 +3074,18 @@ class FreeThreadingTest(unittest.TestCase):
         with threading_helper.start_threads(threads):
             pass
 
+    @threading_helper.reap_threads
+    @threading_helper.requires_working_threading()
+    def test_free_threading_bytearray_resize_other_thread(self):
+        # Shrinking a bytearray whose buffer another thread owns must not
+        # adopt the immortal single-byte bytes object a the buffer.
+        ba = bytearray(b'abc')
+        thread = threading.Thread(target=ba.resize, args=(1,))
+        with threading_helper.start_threads([thread]):
+            pass
+        ba[0] = ord('X')
+        self.assertEqual(ba, bytearray(b'X'))
+        self.assertEqual(ord(b'a'), ord('a'))
+
 if __name__ == "__main__":
     unittest.main()

crates/vm/src/builtins/bytearray.rs

@@ -585,7 +585,21 @@ impl Py<PyByteArray> {
 
     #[pymethod]
     fn clear(&self, vm: &VirtualMachine) -> PyResult<()> {
-        self.try_resizable(vm)?.elements.clear();
+        let old = match self.resize_access() {
+            ResizeAccess::Exported { len: 0 } => return Ok(()),
+            ResizeAccess::Exported { .. } => {
+                return Err(
+                    vm.new_buffer_error("Existing exports of data: object cannot be re-sized")
+                );
+            }
+            ResizeAccess::Exclusive(mut inner) => {
+                if inner.elements.is_empty() {
+                    return Ok(());
+                }
+                core::mem::take(&mut inner.elements)
+            }
+        };
+        drop(old);
         Ok(())
     }
 

crates/vm/src/bytes_inner.rs

@@ -420,17 +420,24 @@ impl ByteInnerTranslateOptions {
 
 pub(crate) type ByteInnerSplitOptions = anystr::SplitArgs<PyBytesInner>;
 
-fn bytearray_repr_char_len(ch: u8) -> usize {
+fn bytearray_repr_char_len(ch: u8, quote: u8) -> usize {
+    if ch == quote {
+        return 2;
+    }
     match ch {
-        b'\'' | b'\\' | b'\t' | b'\r' | b'\n' => 2,
+        b'\\' | b'\t' | b'\r' | b'\n' => 2,
         0x20..=0x7e => 1,
         _ => 4, // \xHH
     }
 }
 
-fn write_bytearray_repr_char(ch: u8, buf: &mut String) {
+fn write_bytearray_repr_char(ch: u8, quote: u8, buf: &mut String) {
+    if ch == quote {
+        buf.push('\\');
+        buf.push(ch as char);
+        return;
+    }
     match ch {
-        b'\'' => buf.push_str(r#"\'"#),
         b'\\' => buf.push_str(r#"\\"#),
         b'\t' => buf.push_str(r#"\t"#),
         b'\n' => buf.push_str(r#"\n"#),
@@ -478,7 +485,7 @@ impl PyBytesInner {
             .elements
             .iter()
             .try_fold(0usize, |len, &ch| {
-                len.checked_add(bytearray_repr_char_len(ch))
+                len.checked_add(bytearray_repr_char_len(ch, quote as u8))
             })
             .ok_or_else(|| Self::new_repr_overflow_error(vm))?;
         let len = class_name
@@ -492,7 +499,7 @@ impl PyBytesInner {
         buf.push('b');
         buf.push(quote);
         for &ch in &self.elements {
-            write_bytearray_repr_char(ch, &mut buf);
+            write_bytearray_repr_char(ch, quote as u8, &mut buf);
         }
         buf.push(quote);
         buf.push(')');
@@ -1354,7 +1361,7 @@ pub struct ByteInnerReplaceOptions {
     old: PyBytesInner,
     #[pyarg(positional)]
     new: PyBytesInner,
-    #[pyarg(positional, default = -1)]
+    #[pyarg(any, default = -1)]
     count: isize,
 }
 
@@ -1437,7 +1444,7 @@ fn hex_impl(bytes: &[u8], sep: u8, bytes_per_sep: isize) -> String {
     let len = bytes.len();
 
     let buf = if bytes_per_sep < 0 {
-        let bytes_per_sep = core::cmp::min(len, (-bytes_per_sep) as usize);
+        let bytes_per_sep = core::cmp::min(len, bytes_per_sep.unsigned_abs());
         let chunks = (len - 1) / bytes_per_sep;
         let chunked = chunks * bytes_per_sep;
         let unchunked = len - chunked;

extra_tests/snippets/builtin_bytearray.py

@@ -733,6 +733,17 @@ a = bytearray(b"abcd")
 a.clear()
 assert len(a) == 0
 
+# Clearing an empty exported buffer is a no-op and keeps its export alive.
+empty = bytearray()
+with memoryview(empty) as view:
+    assert empty.clear() is None
+    assert bytes(view) == b""
+    assert empty.clear() is None
+    with assert_raises(BufferError):
+        empty.append(1)
+empty.append(1)
+assert empty == b"\x01"
+
 b = bytearray(b"test")
 assert len(b) == 4
 b.pop()

scripts/update_lib/cmd_migrate.py

@@ -22,11 +22,13 @@ import argparse
 import ast
 import builtins
 import collections
+import copy
 import hashlib
 import json
 import pathlib
 import shutil
 import sys
+import textwrap
 
 sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
 
@@ -49,6 +51,144 @@ else:
 EXCEPTION_LOOP = """for excname in PYTHON2_EXCEPTIONS:
     NAME_MAPPING[("exceptions", excname)] = ("builtins", excname)
 """
+BYTES_CAPI_UPSTREAM = (
+    "    _testlimitedcapi = import_helper.import_module('_testlimitedcapi')\n"
+)
+BYTES_CAPI_LOCAL = """    # TODO: RUSTPYTHON; import_helper.import_module here cause the entire test stopping
+    # _testlimitedcapi = import_helper.import_module('_testlimitedcapi')
+    try:
+        import _testlimitedcapi
+    except ImportError:
+        _testlimitedcapi = None
+"""
+BYTES_ALLOCATION_COMMENTS = {
+    "test_alloc": """    # NOTE: RUSTPYTHON:
+    #
+    # The second instance of self.assertGreater was replaced with
+    # self.assertGreaterEqual since, in RustPython, the underlying storage
+    # is a Vec which doesn't require trailing null byte.
+""",
+    "test_init_alloc": """    # NOTE: RUSTPYTHON:
+    #
+    # The usages of self.assertGreater were replaced with
+    # self.assertGreaterEqual since, in RustPython, the underlying storage
+    # is a Vec which doesn't require trailing null byte.
+""",
+}
+
+
+def _bytes_bindings(root, name):
+    """Find bindings conservatively, including nested or indirect shadows."""
+    for node in ast.walk(root):
+        if isinstance(node, ast.Name) and isinstance(node.ctx, (ast.Store, ast.Del)):
+            bound = node.id
+        elif isinstance(node, ast.alias):
+            bound = node.asname or node.name.split(".")[0]
+        elif isinstance(
+            node,
+            (
+                ast.FunctionDef,
+                ast.AsyncFunctionDef,
+                ast.ClassDef,
+                ast.ExceptHandler,
+                ast.MatchAs,
+                ast.MatchStar,
+            ),
+        ):
+            bound = node.name
+        elif isinstance(node, ast.MatchMapping):
+            bound = node.rest
+        elif isinstance(node, ast.arg):
+            bound = node.arg
+        elif isinstance(node, (ast.Global, ast.Nonlocal)):
+            bound = name if name in node.names else None
+        else:
+            continue
+        if bound == name:
+            yield node
+
+
+def _bytes_owner(contents):
+    tree = ast.parse(contents)
+    owner = _unique_anchor(
+        contents,
+        tree,
+        lambda node: isinstance(node, ast.ClassDef) and node.name == "ByteArrayTest",
+        direct=True,
+    )
+    if list(_bytes_bindings(tree, "ByteArrayTest")) != [owner]:
+        raise ValueError("Shadowed bytes test class")
+    return owner
+
+
+def _verify_bytes_capi(contents: str, block: str) -> ast.stmt:
+    owner = _bytes_owner(contents)
+    expected = ast.parse(textwrap.dedent(block)).body[0]
+    anchor = _unique_anchor(
+        contents, owner, lambda node: ast.dump(node) == ast.dump(expected), direct=True
+    )
+    allowed = set(ast.walk(anchor))
+    if any(node not in allowed for node in _bytes_bindings(owner, "_testlimitedcapi")):
+        raise ValueError("Shadowed optional bytes C API import")
+    start = anchor.lineno - 1 - (2 if block == BYTES_CAPI_LOCAL else 0)
+    anchored_text = "".join(
+        contents.splitlines(keepends=True)[start : anchor.end_lineno]
+    )
+    if start < 0 or anchored_text != block:
+        raise ValueError("Missing, changed, or ambiguous optional bytes C API import")
+    return anchor
+
+
+def _preserve_bytes_allocations(old: str, new: str) -> str:
+    """Keep only the existing three Vec allocation assertion adaptations."""
+    owners = [_bytes_owner(contents) for contents in (old, new)]
+    replacements = []
+    for name, count in (("test_alloc", 1), ("test_init_alloc", 2)):
+        methods = [
+            _unique_anchor(
+                contents,
+                owner,
+                lambda node: (
+                    isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+                    and node.name == name
+                ),
+                direct=True,
+            )
+            for contents, owner in zip((old, new), owners)
+        ]
+        if any(
+            list(_bytes_bindings(owner, name)) != [method]
+            for owner, method in zip(owners, methods)
+        ):
+            raise ValueError("Shadowed bytes allocation test")
+        if any(not isinstance(method, ast.FunctionDef) for method in methods):
+            raise ValueError("Changed bytes allocation test kind")
+        normalized = copy.deepcopy(methods[0])
+        expected = ast.parse("self.assertGreaterEqual(alloc, len(b))", mode="eval").body
+        matches = [
+            node
+            for node in ast.walk(normalized)
+            if isinstance(node, ast.Call) and ast.dump(node) == ast.dump(expected)
+        ]
+        if len(matches) != count:
+            raise ValueError("Changed local bytes allocation contract")
+        for node in matches:
+            node.func.attr = "assertGreater"
+        if ast.dump(normalized) != ast.dump(methods[1]):
+            raise ValueError("Changed canonical bytes allocation contract")
+        old_lines = old.splitlines(keepends=True)
+        comment = BYTES_ALLOCATION_COMMENTS[name]
+        start = methods[0].lineno - 1 - len(comment.splitlines())
+        if start < 0 or "".join(old_lines[start : methods[0].lineno - 1]) != comment:
+            raise ValueError("Changed local bytes allocation explanation")
+        local = "".join(old_lines[start : methods[0].end_lineno])
+        replacements.append((methods[1].lineno - 1, methods[1].end_lineno, local))
+    lines = new.splitlines(keepends=True)
+    for start, end, local in sorted(replacements, reverse=True):
+        lines[start:end] = [local]
+    result = "".join(lines)
+    ast.parse(result)
+    return result
 
 
 def _unique_anchor(contents, root, predicate, *, direct=False, exact=None):
@@ -153,6 +293,12 @@ def preserve_migration_customizations(path: pathlib.Path, old: str, new: str) ->
         _verify_call(old, LOCAL_CALL)
         _verify_call(new, CANONICAL_CALL)
         return new.replace(CANONICAL_CALL, LOCAL_CALL, 1)
+    if parts == ("Lib", "test", "test_bytes.py"):
+        _verify_bytes_capi(old, BYTES_CAPI_LOCAL)
+        anchor = _verify_bytes_capi(new, BYTES_CAPI_UPSTREAM)
+        lines = new.splitlines(keepends=True)
+        lines[anchor.lineno - 1 : anchor.end_lineno] = [BYTES_CAPI_LOCAL]
+        return _preserve_bytes_allocations(old, "".join(lines))
     return new
 
 

scripts/update_lib/tests/test_migrate.py

@@ -11,6 +11,9 @@ from pathlib import Path
 from unittest.mock import patch
 
 from update_lib.cmd_migrate import (
+    BYTES_ALLOCATION_COMMENTS,
+    BYTES_CAPI_LOCAL,
+    BYTES_CAPI_UPSTREAM,
     CANONICAL_CALL,
     EXCEPTION_LOOP,
     JIT_ERROR,
@@ -37,6 +40,240 @@ from update_lib.patch_spec import (
 )
 
 
+class TestBytesPreservation(unittest.TestCase):
+    path = Path("Lib/test/test_bytes.py")
+    prefix = "class ByteArrayTest(BaseBytesTest, unittest.TestCase):\n"
+    allocations = """
+    def test_alloc(self):
+        b = bytearray()
+        alloc = b.__alloc__()
+        self.assertGreaterEqual(alloc, 0)
+        # existing allocation adaptation
+        self.assertGreaterEqual(alloc, len(b))
+
+    def test_init_alloc(self):
+        b = bytearray()
+        def g():
+            alloc = b.__alloc__()
+            self.assertGreaterEqual(alloc, len(b))
+            yield 1
+        b.__init__(g())
+        alloc = b.__alloc__()
+        self.assertGreaterEqual(alloc, len(b))
+"""
+    canonical_allocations = allocations.replace(
+        "assertGreaterEqual(alloc, len(b))", "assertGreater(alloc, len(b))"
+    ).replace("        # existing allocation adaptation\n", "")
+    for name, comment in BYTES_ALLOCATION_COMMENTS.items():
+        allocations = allocations.replace(
+            f"    def {name}(", comment + f"    def {name}("
+        )
+    old = prefix + BYTES_CAPI_LOCAL + allocations
+    canonical = prefix + BYTES_CAPI_UPSTREAM + canonical_allocations
+
+    def assert_rejected(self, changed, *, local):
+        with self.assertRaises(ValueError):
+            preserve_migration_customizations(
+                self.path,
+                changed if local else self.old,
+                self.canonical if local else changed,
+            )
+
+    def test_preserves_existing_blocks_and_is_idempotent(self):
+        source = self.canonical + "\nnew_canonical_content = True\n"
+        expected = self.old + "\nnew_canonical_content = True\n"
+        result = preserve_migration_customizations(self.path, self.old, source)
+        self.assertEqual(result, expected)
+        self.assertEqual(
+            preserve_migration_customizations(self.path, result, source), result
+        )
+
+    def test_earlier_nested_import_and_decoy_text_are_untouched(self):
+        earlier = (
+            "class BaseBytesTest:\n    def test_sq_item(self):\n    "
+            + BYTES_CAPI_UPSTREAM
+        )
+        decoy = 'description = """\n' + BYTES_CAPI_UPSTREAM + '"""\n'
+        prefix = earlier + decoy
+        self.assertEqual(
+            preserve_migration_customizations(
+                self.path, prefix + self.old, prefix + self.canonical
+            ),
+            prefix + self.old,
+        )
+
+    def test_import_missing_duplicate_changed_and_moved_fail_closed(self):
+        for local, block, allocations in (
+            (True, BYTES_CAPI_LOCAL, self.allocations),
+            (False, BYTES_CAPI_UPSTREAM, self.canonical_allocations),
+        ):
+            source = self.prefix + block + allocations
+            for changed in (
+                self.prefix + "    pass\n" + allocations,
+                source + block,
+                source + source,
+                source.replace("ByteArrayTest", "OtherTest"),
+                source.replace("_testlimitedcapi", "_other"),
+                source.replace("ImportError", "Exception")
+                if local
+                else source.replace("import_module", "other"),
+                "if True:\n"
+                + "".join("    " + line for line in source.splitlines(True)),
+            ):
+                with self.subTest(local=local, changed=changed):
+                    self.assert_rejected(changed, local=local)
+
+    def test_import_and_owner_shadows_fail_closed(self):
+        shadows = (
+            "    _testlimitedcapi = None\n",
+            "    del _testlimitedcapi\n",
+            "    import other as _testlimitedcapi\n",
+            "    def _testlimitedcapi(self): pass\n",
+            "    async def _testlimitedcapi(self): pass\n",
+            "    class _testlimitedcapi: pass\n",
+            "    try:\n        pass\n"
+            "    except Exception as _testlimitedcapi:\n        pass\n",
+            "    match None:\n        case _testlimitedcapi: pass\n",
+            "    global _testlimitedcapi\n",
+            "ByteArrayTest = Other\n",
+            "del ByteArrayTest\n",
+            "from other import ByteArrayTest\n",
+            "def ByteArrayTest(): pass\n",
+            "async def ByteArrayTest(): pass\n",
+            "try:\n    pass\nexcept Exception as ByteArrayTest:\n    pass\n",
+            "match None:\n    case ByteArrayTest: pass\n",
+        )
+        for local, source in ((True, self.old), (False, self.canonical)):
+            for shadow in shadows:
+                with self.subTest(local=local, shadow=shadow):
+                    self.assert_rejected(source + shadow, local=local)
+
+    def test_decoy_cannot_validate_changed_real_import(self):
+        for local, block, allocations in (
+            (True, BYTES_CAPI_LOCAL, self.allocations),
+            (False, BYTES_CAPI_UPSTREAM, self.canonical_allocations),
+        ):
+            decoy = '    """\n' + block + '    """\n'
+            changed = (
+                self.prefix
+                + decoy
+                + block.replace("_testlimitedcapi =", "_testlimitedcapi  =")
+                + allocations
+            )
+            with self.subTest(local=local):
+                self.assert_rejected(changed, local=local)
+        self.assert_rejected(
+            self.canonical.replace(
+                BYTES_CAPI_UPSTREAM,
+                BYTES_CAPI_UPSTREAM.rstrip() + "; unrelated = 1\n",
+            ),
+            local=False,
+        )
+        self.assert_rejected(
+            self.old.replace("# TODO: RUSTPYTHON;", "# changed existing comment;"),
+            local=True,
+        )
+
+    def test_changed_allocation_methods_fail_closed(self):
+        for local, source in ((True, self.old), (False, self.canonical)):
+            for changed in (
+                source.replace("len(b)", "len(other)", 1),
+                source.replace(
+                    "def test_alloc(self):",
+                    "def test_alloc(self):\n        changed = True",
+                ),
+                source.replace("def test_alloc(self):", "def renamed(self):"),
+                source.replace("def test_alloc(self):", "async def test_alloc(self):"),
+                source + "\n    def test_alloc(self): pass\n",
+            ):
+                with self.subTest(local=local, changed=changed):
+                    self.assert_rejected(changed, local=local)
+        for changed in (
+            self.old.replace(
+                "assertGreaterEqual(alloc, len(b))", "assertEqual(alloc, len(b))", 1
+            ),
+            self.old.replace("self.assertGreaterEqual(alloc, len(b))", "pass", 1),
+            self.old.replace(
+                "self.assertGreaterEqual(alloc, len(b))",
+                "self.assertGreaterEqual(alloc, len(b)); "
+                "self.assertGreaterEqual(alloc, len(b))",
+                1,
+            ),
+        ):
+            self.assert_rejected(changed, local=True)
+
+    def test_allocation_method_shadows_fail_closed(self):
+        for name in ("test_alloc", "test_init_alloc"):
+            for local, source in ((True, self.old), (False, self.canonical)):
+                for shadow in (
+                    f"    {name} = None\n",
+                    f"    del {name}\n",
+                    f"    from other import {name}\n",
+                    f"    async def {name}(self): pass\n",
+                    f"    class {name}: pass\n",
+                    "    try:\n        pass\n"
+                    f"    except Exception as {name}:\n        pass\n",
+                    f"    match None:\n        case {name}: pass\n",
+                ):
+                    with self.subTest(name=name, local=local, shadow=shadow):
+                        self.assert_rejected(source + shadow, local=local)
+
+    def test_changed_allocation_explanation_fails_closed(self):
+        for comment in BYTES_ALLOCATION_COMMENTS.values():
+            for replacement in ("", comment.replace("RUSTPYTHON", "changed")):
+                with self.subTest(comment=comment, replacement=replacement):
+                    self.assert_rejected(
+                        self.old.replace(comment, replacement), local=True
+                    )
+
+    def test_other_paths_are_untouched(self):
+        for path in ("Lib/test/other.py", "Lib/other/test_bytes.py"):
+            self.assertEqual(
+                preserve_migration_customizations(Path(path), "old", "new"), "new"
+            )
+
+    def test_file_migration_preserves_guards_and_rejection_writes_nothing(self):
+        guard = (
+            "\n    @unittest.expectedFailure  # TODO: RUSTPYTHON\n"
+            "    def test_example(self):\n        pass\n"
+        )
+        canonical_test = "\n    def test_example(self):\n        pass\n"
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            output = root / self.path
+            output.parent.mkdir(parents=True)
+            output.write_text("import unittest\n" + self.old + guard, encoding="utf-8")
+            source = root / "canonical.py"
+            source.write_text(
+                "import unittest\n" + self.canonical + canonical_test, encoding="utf-8"
+            )
+            before_source = source.read_bytes()
+            before_guards = patches_to_json(extract_patches(output.read_text()))
+            module_file = root / "scripts/update_lib/cmd_migrate.py"
+            with patch("update_lib.cmd_migrate.__file__", str(module_file)):
+                patch_file(source, output, verbose=False)
+                self.assertEqual(
+                    output.read_text(), "import unittest\n" + self.old + guard
+                )
+                self.assertEqual(source.read_bytes(), before_source)
+                self.assertEqual(
+                    patches_to_json(extract_patches(output.read_text())), before_guards
+                )
+                self.assertEqual(
+                    patch_single_content(source, output), output.read_text()
+                )
+                before_output = output.read_bytes()
+                source.write_text(
+                    source.read_text().replace("ByteArrayTest", "OtherTest"),
+                    encoding="utf-8",
+                )
+                rejected_source = source.read_bytes()
+                with self.assertRaises(ValueError):
+                    patch_file(source, output, verbose=False)
+                self.assertEqual(output.read_bytes(), before_output)
+                self.assertEqual(source.read_bytes(), rejected_source)
+
+
 class TestSupportAssignments(unittest.TestCase):
     names = ["_available", "requires_available", "requires_unavailable"]
     exports = names[1:]