Update some tests to 3.14.5 (#7991)

Lib/test/test_android.py

@@ -1,5 +1,4 @@
 import io
-import platform
 import queue
 import re
 import subprocess
@@ -17,17 +16,11 @@ from unittest.mock import patch
 if sys.platform != "android":
     raise unittest.SkipTest("Android-specific")
 
-api_level = platform.android_ver().api_level
-
 # (name, level, fileno)
 STREAM_INFO = [("stdout", "I", 1), ("stderr", "W", 2)]
 
 
 # Test redirection of stdout and stderr to the Android log.
-@unittest.skipIf(
-    api_level < 23 and platform.machine() == "aarch64",
-    "SELinux blocks reading logs on older ARM64 emulators"
-)
 class TestAndroidOutput(unittest.TestCase):
     maxDiff = None
 
@@ -42,31 +35,41 @@ class TestAndroidOutput(unittest.TestCase):
             for line in self.logcat_process.stdout:
                 self.logcat_queue.put(line.rstrip("\n"))
             self.logcat_process.stdout.close()
+
         self.logcat_thread = Thread(target=logcat_thread)
         self.logcat_thread.start()
 
-        from ctypes import CDLL, c_char_p, c_int
-        android_log_write = getattr(CDLL("liblog.so"), "__android_log_write")
-        android_log_write.argtypes = (c_int, c_char_p, c_char_p)
-        ANDROID_LOG_INFO = 4
+        try:
+            from ctypes import CDLL, c_char_p, c_int
+            android_log_write = getattr(CDLL("liblog.so"), "__android_log_write")
+            android_log_write.argtypes = (c_int, c_char_p, c_char_p)
+            ANDROID_LOG_INFO = 4
 
-        # Separate tests using a marker line with a different tag.
-        tag, message = "python.test", f"{self.id()} {time()}"
-        android_log_write(
-            ANDROID_LOG_INFO, tag.encode("UTF-8"), message.encode("UTF-8"))
-        self.assert_log("I", tag, message, skip=True, timeout=5)
+            # Separate tests using a marker line with a different tag.
+            tag, message = "python.test", f"{self.id()} {time()}"
+            android_log_write(
+                ANDROID_LOG_INFO, tag.encode("UTF-8"), message.encode("UTF-8"))
+            self.assert_log("I", tag, message, skip=True)
+        except:
+            # If setUp throws an exception, tearDown is not automatically
+            # called. Avoid leaving a dangling thread which would keep the
+            # Python process alive indefinitely.
+            self.tearDown()
+            raise
 
     def assert_logs(self, level, tag, expected, **kwargs):
         for line in expected:
             self.assert_log(level, tag, line, **kwargs)
 
-    def assert_log(self, level, tag, expected, *, skip=False, timeout=0.5):
-        deadline = time() + timeout
+    def assert_log(self, level, tag, expected, *, skip=False):
+        deadline = time() + LOOPBACK_TIMEOUT
         while True:
             try:
                 line = self.logcat_queue.get(timeout=(deadline - time()))
             except queue.Empty:
-                self.fail(f"line not found: {expected!r}")
+                raise self.failureException(
+                    f"line not found: {expected!r}"
+                ) from None
             if match := re.fullmatch(fr"(.)/{tag}: (.*)", line):
                 try:
                     self.assertEqual(level, match[1])
@@ -81,35 +84,42 @@ class TestAndroidOutput(unittest.TestCase):
         self.logcat_process.wait(LOOPBACK_TIMEOUT)
         self.logcat_thread.join(LOOPBACK_TIMEOUT)
 
+        # Avoid an irrelevant warning about threading._dangling.
+        self.logcat_thread = None
+
     @contextmanager
-    def unbuffered(self, stream):
-        stream.reconfigure(write_through=True)
+    def reconfigure(self, stream, **settings):
+        original_settings = {key: getattr(stream, key, None) for key in settings.keys()}
+        stream.reconfigure(**settings)
         try:
             yield
         finally:
-            stream.reconfigure(write_through=False)
+            stream.reconfigure(**original_settings)
 
-    # In --verbose3 mode, sys.stdout and sys.stderr are captured, so we can't
-    # test them directly. Detect this mode and use some temporary streams with
-    # the same properties.
     def stream_context(self, stream_name, level):
-        # https://developer.android.com/ndk/reference/group/logging
-        prio = {"I": 4, "W": 5}[level]
-
         stack = ExitStack()
         stack.enter_context(self.subTest(stream_name))
+
+        # In --verbose3 mode, sys.stdout and sys.stderr are captured, so we can't
+        # test them directly. Detect this mode and use some temporary streams with
+        # the same properties.
         stream = getattr(sys, stream_name)
         native_stream = getattr(sys, f"__{stream_name}__")
         if isinstance(stream, io.StringIO):
+            # https://developer.android.com/ndk/reference/group/logging
+            prio = {"I": 4, "W": 5}[level]
             stack.enter_context(
                 patch(
                     f"sys.{stream_name}",
-                    TextLogStream(
-                        prio, f"python.{stream_name}", native_stream.fileno(),
-                        errors="backslashreplace"
+                    stream := TextLogStream(
+                        prio, f"python.{stream_name}", native_stream,
                     ),
                 )
             )
+
+        # The tests assume the stream is initially buffered.
+        stack.enter_context(self.reconfigure(stream, write_through=False))
+
         return stack
 
     def test_str(self):
@@ -136,7 +146,7 @@ class TestAndroidOutput(unittest.TestCase):
                     self.assert_logs(level, tag, lines)
 
                 # Single-line messages,
-                with self.unbuffered(stream):
+                with self.reconfigure(stream, write_through=True):
                     write("", [])
 
                     write("a")
@@ -166,14 +176,18 @@ class TestAndroidOutput(unittest.TestCase):
 
                 # Multi-line messages. Avoid identical consecutive lines, as
                 # they may activate "chatty" filtering and break the tests.
-                write("\nx", [""])
+                #
+                # Additional spaces will appear in the output where necessary to
+                # protect leading newlines.
+                write("\nx", [" "])
                 write("\na\n", ["x", "a"])
-                write("\n", [""])
+                write("\n", [" "])
+                write("\n\n", [" ", " "])
                 write("b\n", ["b"])
-                write("c\n\n", ["c", ""])
+                write("c\n\n", ["c", " "])
                 write("d\ne", ["d"])
                 write("xx", [])
-                write("f\n\ng", ["exxf", ""])
+                write("f\n\ng", ["exxf", " "])
                 write("\n", ["g"])
 
                 # Since this is a line-based logging system, line buffering
@@ -183,16 +197,17 @@ class TestAndroidOutput(unittest.TestCase):
 
                 # However, buffering can be turned off completely if you want a
                 # flush after every write.
-                with self.unbuffered(stream):
-                    write("\nx", ["", "x"])
-                    write("\na\n", ["", "a"])
-                    write("\n", [""])
+                with self.reconfigure(stream, write_through=True):
+                    write("\nx", [" ", "x"])
+                    write("\na\n", [" ", "a"])
+                    write("\n", [" "])
+                    write("\n\n", [" ", " "])
                     write("b\n", ["b"])
-                    write("c\n\n", ["c", ""])
+                    write("c\n\n", ["c", " "])
                     write("d\ne", ["d", "e"])
                     write("xx", ["xx"])
-                    write("f\n\ng", ["f", "", "g"])
-                    write("\n", [""])
+                    write("f\n\ng", ["f", " ", "g"])
+                    write("\n", [" "])
 
                 # "\r\n" should be translated into "\n".
                 write("hello\r\n", ["hello"])
@@ -312,19 +327,16 @@ class TestAndroidOutput(unittest.TestCase):
                 # currently use `logcat -v tag`, which shows each line as if it
                 # was a separate log entry, but strips a single trailing
                 # newline.
-                #
-                # On newer versions of Android, all three of the above tools (or
-                # maybe Logcat itself) will also strip any number of leading
-                # newlines.
-                write(b"\nx", ["", "x"] if api_level < 30 else ["x"])
-                write(b"\na\n", ["", "a"] if api_level < 30 else ["a"])
-                write(b"\n", [""])
+                write(b"\nx", [" ", "x"])
+                write(b"\na\n", [" ", "a"])
+                write(b"\n", [" "])
+                write(b"\n\n", [" ", ""])
                 write(b"b\n", ["b"])
                 write(b"c\n\n", ["c", ""])
                 write(b"d\ne", ["d", "e"])
                 write(b"xx", ["xx"])
                 write(b"f\n\ng", ["f", "", "g"])
-                write(b"\n", [""])
+                write(b"\n", [" "])
 
                 # "\r\n" should be translated into "\n".
                 write(b"hello\r\n", ["hello"])

Lib/test/test_binascii.py

@@ -222,6 +222,7 @@ class BinASCIITest(unittest.TestCase):
         assertInvalidLength(b'a' * (4 * 87 + 1))
         assertInvalidLength(b'A\tB\nC ??DE')  # only 5 valid characters
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: Error not raised by a2b_uu
     def test_uu(self):
         MAX_UU = 45
         for backtick in (True, False):
@@ -242,6 +243,10 @@ class BinASCIITest(unittest.TestCase):
         self.assertEqual(binascii.a2b_uu(b"\xff"), b"\x00"*31)
         self.assertRaises(binascii.Error, binascii.a2b_uu, b"\xff\x00")
         self.assertRaises(binascii.Error, binascii.a2b_uu, b"!!!!")
+        self.assertRaises(binascii.Error, binascii.a2b_uu,
+                          self.type2test(b""))
+        self.assertRaises(binascii.Error, binascii.a2b_uu,
+                          self.type2test(b"#86)C")[:0])
         self.assertRaises(binascii.Error, binascii.b2a_uu, 46*b"!")
 
         # Issue #7701 (crash on a pydebug build)
@@ -440,6 +445,7 @@ class BinASCIITest(unittest.TestCase):
         self.assertConversion(binary, converted, restored,
                               quotetabs=quotetabs, istext=istext, header=header)
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: Error not raised by a2b_uu
     def test_empty_string(self):
         # A test for SF bug #1022953.  Make sure SystemError is not raised.
         empty = self.type2test(b'')
@@ -449,6 +455,9 @@ class BinASCIITest(unittest.TestCase):
                 binascii.crc_hqx(empty, 0)
                 continue
             f = getattr(binascii, func)
+            if func == 'a2b_uu':
+                self.assertRaises(binascii.Error, f, empty)
+                continue
             try:
                 f(empty)
             except Exception as err:

Lib/test/test_builtin.py

@@ -293,6 +293,27 @@ class BuiltinTest(ComplexesAreIdenticalMixin, unittest.TestCase):
         self.assertEqual(overridden_outputs, ['all', 'any', 'tuple'])
 
 
+    def test_builtin_call_async_genexpr_no_crash(self):
+        async def f_all():
+            return all(await 2 for _ in [])
+
+        async def f_any():
+            return any(await 2 for _ in [])
+
+        async def f_tuple():
+            return tuple(await 2 for _ in [])
+
+        async def f_list():
+            return list(await 2 for _ in [])
+
+        async def f_set():
+            return set(await 2 for _ in [])
+
+        for f in (f_all, f_any, f_tuple, f_list, f_set):
+            with self.subTest(func=f.__name__):
+                with self.assertRaises(TypeError):
+                    run_yielding_async_fn(f)
+
     def test_ascii(self):
         self.assertEqual(ascii(''), '\'\'')
         self.assertEqual(ascii(0), '0')
@@ -606,7 +627,7 @@ class BuiltinTest(ComplexesAreIdenticalMixin, unittest.TestCase):
         exec(co, glob)
         self.assertEqual(type(glob['ticker']()), AsyncGeneratorType)
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: <_ast.Name object at 0xb40000731e3d1360> is not an instance of <class '_ast.Constant'>
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: <_ast.Name object at 0xb40000731e3d1360> is not an instance of <class '_ast.Constant'>
     def test_compile_ast(self):
         args = ("a*__debug__", "f.py", "exec")
         raw = compile(*args, flags = ast.PyCF_ONLY_AST).body[0]
@@ -970,7 +991,7 @@ class BuiltinTest(ComplexesAreIdenticalMixin, unittest.TestCase):
         self.assertRaisesRegex(NameError, "name 'superglobal' is not defined",
                                eval, code, ns)
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON; wrong error message
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; wrong error message
     def test_exec_builtins_mapping_import(self):
         code = compile("import foo.bar", "test", "exec")
         ns = {'__builtins__': types.MappingProxyType({})}
@@ -979,7 +1000,7 @@ class BuiltinTest(ComplexesAreIdenticalMixin, unittest.TestCase):
         exec(code, ns)
         self.assertEqual(ns['foo'], ('foo.bar', ns, ns, None, 0))
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: AttributeError not raised by eval
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: AttributeError not raised by eval
     def test_eval_builtins_mapping_reduce(self):
         # list_iterator.__reduce__() calls _PyEval_GetBuiltin("iter")
         code = compile("x.__reduce__()", "test", "eval")

Lib/test/test_defaultdict.py

@@ -204,5 +204,20 @@ class TestDefaultDict(unittest.TestCase):
         self.assertEqual(test_dict[key], 2)
         self.assertEqual(count, 2)
 
+    def test_repr_recursive_factory(self):
+        # gh-145492: defaultdict.__repr__ should not cause infinite recursion
+        # when the factory's __repr__ calls repr() on the defaultdict.
+        class ProblematicFactory:
+            def __call__(self):
+                return {}
+            def __repr__(self):
+                repr(dd)
+                return f"ProblematicFactory for {dd}"
+
+        dd = defaultdict(ProblematicFactory())
+        # Should not raise RecursionError
+        r = repr(dd)
+        self.assertIn("ProblematicFactory for", r)
+
 if __name__ == "__main__":
     unittest.main()

Lib/test/test_mmap.py

@@ -1,9 +1,12 @@
 from test.support import (
-    requires, _2G, _4G, gc_collect, cpython_only, is_emscripten
+    requires, _2G, _4G, gc_collect, cpython_only, is_emscripten, is_apple,
+    in_systemd_nspawn_sync_suppressed,
 )
 from test.support.import_helper import import_module
 from test.support.os_helper import TESTFN, unlink
+from test.support.script_helper import assert_python_ok
 import unittest
+import errno
 import os
 import re
 import itertools
@@ -11,6 +14,7 @@ import random
 import socket
 import string
 import sys
+import textwrap
 import weakref
 
 # Skip test if we can't import mmap.
@@ -32,6 +36,10 @@ if is_emscripten:
 class MmapTests(unittest.TestCase):
 
     def setUp(self):
+        # TODO: RUSTPYTHON; Remove this once windows doesn't get errored on setup:/
+        if os.name == "nt":
+            raise unittest.SkipTest("TODO: RUSTPYTHON; Error during class setUp")
+
         if os.path.exists(TESTFN):
             os.unlink(TESTFN)
 
@@ -41,6 +49,7 @@ class MmapTests(unittest.TestCase):
         except OSError:
             pass
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; AttributeError: 'mmap' object has no attribute 'seekable'
     def test_basic(self):
         # Test mmap module on Unix systems and Windows
 
@@ -93,11 +102,12 @@ class MmapTests(unittest.TestCase):
             self.assertEqual(end, PAGESIZE + 6)
 
         # test seeking around (try to overflow the seek implementation)
-        m.seek(0,0)
+        self.assertTrue(m.seekable())
+        self.assertEqual(m.seek(0, 0), 0)
         self.assertEqual(m.tell(), 0)
-        m.seek(42,1)
+        self.assertEqual(m.seek(42, 1), 42)
         self.assertEqual(m.tell(), 42)
-        m.seek(0,2)
+        self.assertEqual(m.seek(0, 2), len(m))
         self.assertEqual(m.tell(), len(m))
 
         # Try to seek to negative position...
@@ -162,7 +172,7 @@ class MmapTests(unittest.TestCase):
 
             # Ensuring that readonly mmap can't be write() to
             try:
-                m.seek(0,0)
+                m.seek(0, 0)
                 m.write(b'abc')
             except TypeError:
                 pass
@@ -171,7 +181,7 @@ class MmapTests(unittest.TestCase):
 
             # Ensuring that readonly mmap can't be write_byte() to
             try:
-                m.seek(0,0)
+                m.seek(0, 0)
                 m.write_byte(b'd')
             except TypeError:
                 pass
@@ -255,15 +265,79 @@ class MmapTests(unittest.TestCase):
             # Try writing with PROT_EXEC and without PROT_WRITE
             prot = mmap.PROT_READ | getattr(mmap, 'PROT_EXEC', 0)
             with open(TESTFN, "r+b") as f:
-                m = mmap.mmap(f.fileno(), mapsize, prot=prot)
-                self.assertRaises(TypeError, m.write, b"abcdef")
-                self.assertRaises(TypeError, m.write_byte, 0)
-                m.close()
+                try:
+                    m = mmap.mmap(f.fileno(), mapsize, prot=prot)
+                except PermissionError:
+                    # on macOS 14, PROT_READ | PROT_EXEC is not allowed
+                    pass
+                else:
+                    self.assertRaises(TypeError, m.write, b"abcdef")
+                    self.assertRaises(TypeError, m.write_byte, 0)
+                    m.close()
+
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
+    @unittest.skipIf(os.name == 'nt', 'trackfd not present on Windows')
+    def test_trackfd_parameter(self):
+        size = 64
+        with open(TESTFN, "wb") as f:
+            f.write(b"a"*size)
+        for close_original_fd in True, False:
+            with self.subTest(close_original_fd=close_original_fd):
+                with open(TESTFN, "r+b") as f:
+                    with mmap.mmap(f.fileno(), size, trackfd=False) as m:
+                        if close_original_fd:
+                            f.close()
+                        self.assertEqual(len(m), size)
+                        with self.assertRaises(OSError) as err_cm:
+                            m.size()
+                        self.assertEqual(err_cm.exception.errno, errno.EBADF)
+                        with self.assertRaises(ValueError):
+                            m.resize(size * 2)
+                        with self.assertRaises(ValueError):
+                            m.resize(size // 2)
+                        self.assertEqual(m.closed, False)
+
+                        # Smoke-test other API
+                        m.write_byte(ord('X'))
+                        m[2] = ord('Y')
+                        m.flush()
+                        with open(TESTFN, "rb") as f:
+                            self.assertEqual(f.read(4), b'XaYa')
+                        self.assertEqual(m.tell(), 1)
+                        m.seek(0)
+                        self.assertEqual(m.tell(), 0)
+                        self.assertEqual(m.read_byte(), ord('X'))
+
+                self.assertEqual(m.closed, True)
+                self.assertEqual(os.stat(TESTFN).st_size, size)
+
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
+    @unittest.skipIf(os.name == 'nt', 'trackfd not present on Windows')
+    def test_trackfd_neg1(self):
+        size = 64
+        with mmap.mmap(-1, size, trackfd=False) as m:
+            with self.assertRaises(OSError):
+                m.size()
+            with self.assertRaises(ValueError):
+                m.resize(size // 2)
+            self.assertEqual(len(m), size)
+            m[0] = ord('a')
+            assert m[0] == ord('a')
+
+    @unittest.skipIf(os.name != 'nt', 'trackfd only fails on Windows')
+    def test_no_trackfd_parameter_on_windows(self):
+        # 'trackffd' is an invalid keyword argument for this function
+        size = 64
+        with self.assertRaises(TypeError):
+            mmap.mmap(-1, size, trackfd=True)
+        with self.assertRaises(TypeError):
+            mmap.mmap(-1, size, trackfd=False)
 
     def test_bad_file_desc(self):
         # Try opening a bad file descriptor...
         self.assertRaises(OSError, mmap.mmap, -2, 4096)
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_tougher_find(self):
         # Do a tougher .find() test.  SF bug 515943 pointed out that, in 2.2,
         # searching for data with embedded \0 bytes didn't work.
@@ -282,6 +356,7 @@ class MmapTests(unittest.TestCase):
                 self.assertEqual(m.find(slice + b'x'), -1)
         m.close()
 
+    @unittest.skip("TODO: RUSTPYTHON; panic")
     def test_find_end(self):
         # test the new 'end' parameter works as expected
         with open(TESTFN, 'wb+') as f:
@@ -299,7 +374,29 @@ class MmapTests(unittest.TestCase):
         self.assertEqual(m.find(b'one', 1, -2), -1)
         self.assertEqual(m.find(bytearray(b'one')), 0)
 
+        for i in range(-n-1, n+1):
+            for j in range(-n-1, n+1):
+                for p in [b"o", b"on", b"two", b"ones", b"s"]:
+                    expected = data.find(p, i, j)
+                    self.assertEqual(m.find(p, i, j), expected, (p, i, j))
 
+    def test_find_does_not_access_beyond_buffer(self):
+        try:
+            flags = mmap.MAP_PRIVATE | mmap.MAP_ANONYMOUS
+            PAGESIZE = mmap.PAGESIZE
+            PROT_NONE = 0
+            PROT_READ = mmap.PROT_READ
+        except AttributeError as e:
+            raise unittest.SkipTest("mmap flags unavailable") from e
+        for i in range(0, 2049):
+            with mmap.mmap(-1, PAGESIZE * (i + 1),
+                           flags=flags, prot=PROT_NONE) as guard:
+                with mmap.mmap(-1, PAGESIZE * (i + 2048),
+                               flags=flags, prot=PROT_READ) as fm:
+                    fm.find(b"fo", -2)
+
+
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_rfind(self):
         # test the new 'end' parameter works as expected
         with open(TESTFN, 'wb+') as f:
@@ -360,6 +457,7 @@ class MmapTests(unittest.TestCase):
             self.assertRaises(ValueError, mmap.mmap, f.fileno(), 0,
                               offset=2147418112)
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_move(self):
         # make move works everywhere (64-bit format problem earlier)
         with open(TESTFN, 'wb+') as f:
@@ -407,7 +505,6 @@ class MmapTests(unittest.TestCase):
         m.move(0, 0, 1)
         m.move(0, 0, 0)
 
-
     def test_anonymous(self):
         # anonymous mmap.mmap(-1, PAGE)
         m = mmap.mmap(-1, PAGESIZE)
@@ -420,6 +517,7 @@ class MmapTests(unittest.TestCase):
             m[x] = b
             self.assertEqual(m[x], b)
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_read_all(self):
         m = mmap.mmap(-1, 16)
         self.addCleanup(m.close)
@@ -441,6 +539,7 @@ class MmapTests(unittest.TestCase):
         m.seek(9)
         self.assertEqual(m.read(-42), bytes(range(9, 16)))
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_read_invalid_arg(self):
         m = mmap.mmap(-1, 16)
         self.addCleanup(m.close)
@@ -557,6 +656,7 @@ class MmapTests(unittest.TestCase):
             except OSError:
                 pass
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_subclass(self):
         class anon_mmap(mmap.mmap):
             def __new__(klass, *args, **kwargs):
@@ -609,6 +709,7 @@ class MmapTests(unittest.TestCase):
         self.assertEqual(m[:], b"012barbaz9")
         self.assertRaises(ValueError, m.write, b"ba")
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_non_ascii_byte(self):
         for b in (129, 200, 255): # > 128
             m = mmap.mmap(-1, 1)
@@ -618,6 +719,7 @@ class MmapTests(unittest.TestCase):
             self.assertEqual(m.read_byte(), b)
             m.close()
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @unittest.skipUnless(os.name == 'nt', 'requires Windows')
     def test_tagname(self):
         data1 = b"0123456789"
@@ -646,14 +748,16 @@ class MmapTests(unittest.TestCase):
         m2.close()
         m1.close()
 
+        with self.assertRaisesRegex(TypeError, 'tagname'):
+            mmap.mmap(-1, 8, tagname=1)
+
     @cpython_only
     @unittest.skipUnless(os.name == 'nt', 'requires Windows')
     def test_sizeof(self):
         m1 = mmap.mmap(-1, 100)
         tagname = random_tagname()
         m2 = mmap.mmap(-1, 100, tagname=tagname)
-        self.assertEqual(sys.getsizeof(m2),
-                         sys.getsizeof(m1) + len(tagname) + 1)
+        self.assertGreater(sys.getsizeof(m2), sys.getsizeof(m1))
 
     @unittest.skipUnless(os.name == 'nt', 'requires Windows')
     def test_crasher_on_windows(self):
@@ -708,6 +812,7 @@ class MmapTests(unittest.TestCase):
                               "wrong exception raised in context manager")
         self.assertTrue(m.closed, "context manager failed")
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_weakref(self):
         # Check mmap objects are weakrefable
         mm = mmap.mmap(-1, 16)
@@ -717,6 +822,7 @@ class MmapTests(unittest.TestCase):
         gc_collect()
         self.assertIs(wr(), None)
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_write_returning_the_number_of_bytes_written(self):
         mm = mmap.mmap(-1, 16)
         self.assertEqual(mm.write(b""), 0)
@@ -724,6 +830,7 @@ class MmapTests(unittest.TestCase):
         self.assertEqual(mm.write(b"yz"), 2)
         self.assertEqual(mm.write(b"python"), 6)
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_resize_past_pos(self):
         m = mmap.mmap(-1, 8192)
         self.addCleanup(m.close)
@@ -744,7 +851,7 @@ class MmapTests(unittest.TestCase):
         with self.assertRaises(TypeError):
             m * 2
 
-    @unittest.skipIf(sys.platform.startswith("linux"), "TODO: RUSTPYTHON, memmap2 doesn't throw OSError when offset is not a multiple of mmap.PAGESIZE on Linux")
+    @unittest.skipIf(sys.platform.startswith("linux"), "TODO: RUSTPYTHON; memmap2 doesn't throw OSError when offset is not a multiple of mmap.PAGESIZE on Linux")
     def test_flush_return_value(self):
         # mm.flush() should return None on success, raise an
         # exception on error under all platforms.
@@ -753,11 +860,13 @@ class MmapTests(unittest.TestCase):
         mm.write(b'python')
         result = mm.flush()
         self.assertIsNone(result)
-        if sys.platform.startswith('linux'):
+        if (sys.platform.startswith(('linux', 'android'))
+            and not in_systemd_nspawn_sync_suppressed()):
             # 'offset' must be a multiple of mmap.PAGESIZE on Linux.
             # See bpo-34754 for details.
             self.assertRaises(OSError, mm.flush, 1, len(b'python'))
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     def test_repr(self):
         open_mmap_repr_pat = re.compile(
             r"<mmap.mmap closed=False, "
@@ -794,11 +903,16 @@ class MmapTests(unittest.TestCase):
                     match = closed_mmap_repr_pat.match(repr(mm))
                     self.assertIsNotNone(match)
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
     @unittest.skipUnless(hasattr(mmap.mmap, 'madvise'), 'needs madvise')
     def test_madvise(self):
         size = 2 * PAGESIZE
         m = mmap.mmap(-1, size)
 
+        class Number:
+            def __index__(self):
+                return 2
+
         with self.assertRaisesRegex(ValueError, "madvise start out of bounds"):
             m.madvise(mmap.MADV_NORMAL, size)
         with self.assertRaisesRegex(ValueError, "madvise start out of bounds"):
@@ -807,42 +921,84 @@ class MmapTests(unittest.TestCase):
             m.madvise(mmap.MADV_NORMAL, 0, -1)
         with self.assertRaisesRegex(OverflowError, "madvise length too large"):
             m.madvise(mmap.MADV_NORMAL, PAGESIZE, sys.maxsize)
+        with self.assertRaisesRegex(
+                TypeError, "'str' object cannot be interpreted as an integer"):
+            m.madvise(mmap.MADV_NORMAL, PAGESIZE, "Not a Number")
         self.assertEqual(m.madvise(mmap.MADV_NORMAL), None)
         self.assertEqual(m.madvise(mmap.MADV_NORMAL, PAGESIZE), None)
         self.assertEqual(m.madvise(mmap.MADV_NORMAL, PAGESIZE, size), None)
         self.assertEqual(m.madvise(mmap.MADV_NORMAL, 0, 2), None)
+        self.assertEqual(m.madvise(mmap.MADV_NORMAL, 0, Number()), None)
         self.assertEqual(m.madvise(mmap.MADV_NORMAL, 0, size), None)
 
-    @unittest.skipUnless(os.name == 'nt', 'requires Windows')
-    def test_resize_up_when_mapped_to_pagefile(self):
+    @unittest.expectedFailureIf(sys.platform in ("linux", "win32"), "TODO: RUSTPYTHON")
+    def test_resize_up_anonymous_mapping(self):
         """If the mmap is backed by the pagefile ensure a resize up can happen
         and that the original data is still in place
         """
         start_size = PAGESIZE
         new_size = 2 * start_size
-        data = bytes(random.getrandbits(8) for _ in range(start_size))
+        data = random.randbytes(start_size)
 
-        m = mmap.mmap(-1, start_size)
-        m[:] = data
-        m.resize(new_size)
-        self.assertEqual(len(m), new_size)
-        self.assertEqual(m[:start_size], data[:start_size])
+        with mmap.mmap(-1, start_size) as m:
+            m[:] = data
+            if sys.platform.startswith(('linux', 'android')):
+                # Can't expand a shared anonymous mapping on Linux.
+                # See https://bugzilla.kernel.org/show_bug.cgi?id=8691
+                with self.assertRaises(ValueError):
+                    m.resize(new_size)
+            else:
+                try:
+                    m.resize(new_size)
+                except SystemError:
+                    pass
+                else:
+                    self.assertEqual(len(m), new_size)
+                    self.assertEqual(m[:start_size], data)
+                    self.assertEqual(m[start_size:], b'\0' * (new_size - start_size))
 
-    @unittest.skipUnless(os.name == 'nt', 'requires Windows')
-    def test_resize_down_when_mapped_to_pagefile(self):
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
+    @unittest.skipUnless(os.name == 'posix', 'requires Posix')
+    def test_resize_up_private_anonymous_mapping(self):
+        start_size = PAGESIZE
+        new_size = 2 * start_size
+        data = random.randbytes(start_size)
+
+        with mmap.mmap(-1, start_size, flags=mmap.MAP_PRIVATE) as m:
+            m[:] = data
+            try:
+                m.resize(new_size)
+            except SystemError:
+                pass
+            else:
+                self.assertEqual(len(m), new_size)
+                self.assertEqual(m[:start_size], data)
+                self.assertEqual(m[start_size:], b'\0' * (new_size - start_size))
+
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
+    def test_resize_down_anonymous_mapping(self):
         """If the mmap is backed by the pagefile ensure a resize down up can happen
         and that a truncated form of the original data is still in place
         """
-        start_size = PAGESIZE
+        start_size = 2 * PAGESIZE
         new_size = start_size // 2
-        data = bytes(random.getrandbits(8) for _ in range(start_size))
+        data = random.randbytes(start_size)
 
-        m = mmap.mmap(-1, start_size)
-        m[:] = data
-        m.resize(new_size)
-        self.assertEqual(len(m), new_size)
-        self.assertEqual(m[:new_size], data[:new_size])
+        with mmap.mmap(-1, start_size) as m:
+            m[:] = data
+            try:
+                m.resize(new_size)
+            except SystemError:
+                pass
+            else:
+                self.assertEqual(len(m), new_size)
+                self.assertEqual(m[:], data[:new_size])
+                if sys.platform.startswith(('linux', 'android')):
+                    # Can't expand to its original size.
+                    with self.assertRaises(ValueError):
+                        m.resize(start_size)
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @unittest.skipUnless(os.name == 'nt', 'requires Windows')
     def test_resize_fails_if_mapping_held_elsewhere(self):
         """If more than one mapping is held against a named file on Windows, neither
@@ -867,6 +1023,7 @@ class MmapTests(unittest.TestCase):
         finally:
             f.close()
 
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
     @unittest.skipUnless(os.name == 'nt', 'requires Windows')
     def test_resize_succeeds_with_error_for_second_named_mapping(self):
         """If a more than one mapping exists of the same name, none of them can
@@ -888,6 +1045,168 @@ class MmapTests(unittest.TestCase):
         self.assertEqual(m1[:data_length], data)
         self.assertEqual(m2[:data_length], data)
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
+    def test_mmap_closed_by_int_scenarios(self):
+        """
+        gh-103987: Test that mmap objects raise ValueError
+                for closed mmap files
+        """
+
+        class MmapClosedByIntContext:
+            def __init__(self, access) -> None:
+                self.access = access
+
+            def __enter__(self):
+                self.f = open(TESTFN, "w+b")
+                self.f.write(random.randbytes(100))
+                self.f.flush()
+
+                m = mmap.mmap(self.f.fileno(), 100, access=self.access)
+
+                class X:
+                    def __index__(self):
+                        m.close()
+                        return 10
+
+                return (m, X)
+
+            def __exit__(self, exc_type, exc_value, traceback):
+                self.f.close()
+
+        read_access_modes = [
+            mmap.ACCESS_READ,
+            mmap.ACCESS_WRITE,
+            mmap.ACCESS_COPY,
+            mmap.ACCESS_DEFAULT,
+        ]
+
+        write_access_modes = [
+            mmap.ACCESS_WRITE,
+            mmap.ACCESS_COPY,
+            mmap.ACCESS_DEFAULT,
+        ]
+
+        for access in read_access_modes:
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m[X()]
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m[X() : 20]
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m[X() : 20 : 2]
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m[20 : X() : -2]
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m.read(X())
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m.find(b"1", 1, X())
+
+        for access in write_access_modes:
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m[X() : 20] = b"1" * 10
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m[X() : 20 : 2] = b"1" * 5
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m[20 : X() : -2] = b"1" * 5
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m.move(1, 2, X())
+
+            with MmapClosedByIntContext(access) as (m, X):
+                with self.assertRaisesRegex(ValueError, "mmap closed or invalid"):
+                    m.write_byte(X())
+
+    @unittest.skipUnless(os.name == 'nt', 'requires Windows')
+    @unittest.skipUnless(hasattr(mmap.mmap, '_protect'), 'test needs debug build')
+    def test_access_violations(self):
+        from test.support.os_helper import TESTFN
+
+        code = textwrap.dedent("""
+            import faulthandler
+            import mmap
+            import os
+            import sys
+            from contextlib import suppress
+
+            # Prevent logging access violations to stderr.
+            faulthandler.disable()
+
+            PAGESIZE = mmap.PAGESIZE
+            PAGE_NOACCESS = 0x01
+
+            with open(sys.argv[1], 'bw+') as f:
+                f.write(b'A'* PAGESIZE)
+                f.flush()
+
+                m = mmap.mmap(f.fileno(), PAGESIZE)
+                m._protect(PAGE_NOACCESS, 0, PAGESIZE)
+                with suppress(OSError):
+                    m.read(PAGESIZE)
+                    assert False, 'mmap.read() did not raise'
+                with suppress(OSError):
+                    m.read_byte()
+                    assert False, 'mmap.read_byte() did not raise'
+                with suppress(OSError):
+                    m.readline()
+                    assert False, 'mmap.readline() did not raise'
+                with suppress(OSError):
+                    m.write(b'A'* PAGESIZE)
+                    assert False, 'mmap.write() did not raise'
+                with suppress(OSError):
+                    m.write_byte(0)
+                    assert False, 'mmap.write_byte() did not raise'
+                with suppress(OSError):
+                    m[0]  # test mmap_subscript
+                    assert False, 'mmap.__getitem__() did not raise'
+                with suppress(OSError):
+                    m[0:10]  # test mmap_subscript
+                    assert False, 'mmap.__getitem__() did not raise'
+                with suppress(OSError):
+                    m[0:10:2]  # test mmap_subscript
+                    assert False, 'mmap.__getitem__() did not raise'
+                with suppress(OSError):
+                    m[0] = 1
+                    assert False, 'mmap.__setitem__() did not raise'
+                with suppress(OSError):
+                    m[0:10] = b'A'* 10
+                    assert False, 'mmap.__setitem__() did not raise'
+                with suppress(OSError):
+                    m[0:10:2] = b'A'* 5
+                    assert False, 'mmap.__setitem__() did not raise'
+                with suppress(OSError):
+                    m.move(0, 10, 1)
+                    assert False, 'mmap.move() did not raise'
+                with suppress(OSError):
+                    list(m)  # test mmap_item
+                    assert False, 'mmap.__getitem__() did not raise'
+                with suppress(OSError):
+                    m.find(b'A')
+                    assert False, 'mmap.find() did not raise'
+                with suppress(OSError):
+                    m.rfind(b'A')
+                    assert False, 'mmap.rfind() did not raise'
+        """)
+        rt, stdout, stderr = assert_python_ok("-c", code, TESTFN)
+        self.assertEqual(stdout.strip(), b'')
+        self.assertEqual(stderr.strip(), b'')
+
+
 class LargeMmapTests(unittest.TestCase):
 
     def setUp(self):
@@ -897,7 +1216,7 @@ class LargeMmapTests(unittest.TestCase):
         unlink(TESTFN)
 
     def _make_test_file(self, num_zeroes, tail):
-        if sys.platform[:3] == 'win' or sys.platform == 'darwin':
+        if sys.platform[:3] == 'win' or is_apple:
             requires('largefile',
                 'test requires %s bytes and a long time to run' % str(0x180000000))
         f = open(TESTFN, 'w+b')

Lib/test/test_posix.py

@@ -1934,6 +1934,11 @@ class _PosixSpawnMixin:
         )
         support.wait_process(pid, exitcode=0)
 
+    def test_setpgroup_allow_none(self):
+        path, args = self.NOOP_PROGRAM[0], self.NOOP_PROGRAM
+        pid = self.spawn_func(path, args, os.environ, setpgroup=None)
+        support.wait_process(pid, exitcode=0)
+
     def test_setpgroup_wrong_type(self):
         with self.assertRaises(TypeError):
             self.spawn_func(sys.executable,
@@ -2034,6 +2039,21 @@ class _PosixSpawnMixin:
                             [sys.executable, "-c", "pass"],
                             os.environ, setsigdef=[signal.NSIG, signal.NSIG+1])
 
+    def test_scheduler_allow_none(self):
+        path, args = self.NOOP_PROGRAM[0], self.NOOP_PROGRAM
+        pid = self.spawn_func(path, args, os.environ, scheduler=None)
+        support.wait_process(pid, exitcode=0)
+
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; Wrong error message
+    @support.subTests("scheduler", [object(), 1, [1, 2]])
+    def test_scheduler_wrong_type(self, scheduler):
+        path, args = self.NOOP_PROGRAM[0], self.NOOP_PROGRAM
+        with self.assertRaisesRegex(
+            TypeError,
+            "scheduler must be a tuple or None",
+        ):
+            self.spawn_func(path, args, os.environ, scheduler=scheduler)
+
     @unittest.expectedFailureIf(sys.platform in ("darwin", "linux"), "TODO: RUSTPYTHON; NotImplementedError: scheduler parameter is not yet implemented")
     @requires_sched
     @unittest.skipIf(sys.platform.startswith(('freebsd', 'netbsd')),

Lib/test/test_uuid.py

@@ -1176,6 +1176,47 @@ class CommandLineTestCases:
         self.assertEqual(cm.exception.code, 2)
         self.assertIn("error: Incorrect number of arguments", mock_err.getvalue())
 
+    @mock.patch.object(sys, "argv",
+                       ["", "-u", "uuid3", "-n", "@dns", "-N", "python.org"])
+    def test_cli_uuid3_outputted_with_valid_namespace_and_name(self):
+        stdout = io.StringIO()
+        with contextlib.redirect_stdout(stdout):
+            self.uuid.main()
+
+        output = stdout.getvalue().strip()
+        uuid_output = self.uuid.UUID(output)
+
+        # Output should be in the form of uuid3
+        self.assertEqual(output, str(uuid_output))
+        self.assertEqual(uuid_output.version, 3)
+
+    @mock.patch.object(sys, "argv",
+                       ["", "-u", "uuid3", "-n",
+                        "0d6a16cc-34a7-47d8-b660-214d0ae184d2",
+                        "-N", "some.user"])
+    def test_cli_uuid3_outputted_with_custom_namespace_and_name(self):
+        stdout = io.StringIO()
+        with contextlib.redirect_stdout(stdout):
+            self.uuid.main()
+
+        output = stdout.getvalue().strip()
+        uuid_output = self.uuid.UUID(output)
+
+        # Output should be in the form of uuid3
+        self.assertEqual(output, str(uuid_output))
+        self.assertEqual(uuid_output.version, 3)
+
+    @mock.patch.object(sys, "argv",
+                       ["", "-u", "uuid3", "-n", "any UUID", "-N", "python.org"])
+    @mock.patch('sys.stderr', new_callable=io.StringIO)
+    def test_cli_uuid3_with_invalid_namespace(self, mock_err):
+        with self.assertRaises(SystemExit) as cm:
+            self.uuid.main()
+        # Check that exception code is the same as argparse.ArgumentParser.error
+        self.assertEqual(cm.exception.code, 2)
+        self.assertIn("error: badly formed hexadecimal UUID string",
+                      mock_err.getvalue())
+
     @mock.patch.object(sys, "argv", [""])
     def test_cli_uuid4_outputted_with_no_args(self):
         stdout = io.StringIO()
@@ -1203,23 +1244,9 @@ class CommandLineTestCases:
             uuid_output = self.uuid.UUID(o)
             self.assertEqual(uuid_output.version, 4)
 
-    @mock.patch.object(sys, "argv",
-                       ["", "-u", "uuid3", "-n", "@dns", "-N", "python.org"])
-    def test_cli_uuid3_ouputted_with_valid_namespace_and_name(self):
-        stdout = io.StringIO()
-        with contextlib.redirect_stdout(stdout):
-            self.uuid.main()
-
-        output = stdout.getvalue().strip()
-        uuid_output = self.uuid.UUID(output)
-
-        # Output should be in the form of uuid5
-        self.assertEqual(output, str(uuid_output))
-        self.assertEqual(uuid_output.version, 3)
-
     @mock.patch.object(sys, "argv",
                        ["", "-u", "uuid5", "-n", "@dns", "-N", "python.org"])
-    def test_cli_uuid5_ouputted_with_valid_namespace_and_name(self):
+    def test_cli_uuid5_outputted_with_valid_namespace_and_name(self):
         stdout = io.StringIO()
         with contextlib.redirect_stdout(stdout):
             self.uuid.main()
@@ -1231,6 +1258,33 @@ class CommandLineTestCases:
         self.assertEqual(output, str(uuid_output))
         self.assertEqual(uuid_output.version, 5)
 
+    @mock.patch.object(sys, "argv",
+                       ["", "-u", "uuid5", "-n",
+                        "0d6a16cc-34a7-47d8-b660-214d0ae184d2",
+                        "-N", "some.user"])
+    def test_cli_uuid5_ouputted_with_custom_namespace_and_name(self):
+        stdout = io.StringIO()
+        with contextlib.redirect_stdout(stdout):
+            self.uuid.main()
+
+        output = stdout.getvalue().strip()
+        uuid_output = self.uuid.UUID(output)
+
+        # Output should be in the form of uuid5
+        self.assertEqual(output, str(uuid_output))
+        self.assertEqual(uuid_output.version, 5)
+
+    @mock.patch.object(sys, "argv",
+                       ["", "-u", "uuid5", "-n", "any UUID", "-N", "python.org"])
+    @mock.patch('sys.stderr', new_callable=io.StringIO)
+    def test_cli_uuid5_with_invalid_namespace(self, mock_err):
+        with self.assertRaises(SystemExit) as cm:
+            self.uuid.main()
+        # Check that exception code is the same as argparse.ArgumentParser.error
+        self.assertEqual(cm.exception.code, 2)
+        self.assertIn("error: badly formed hexadecimal UUID string",
+                      mock_err.getvalue())
+
     @mock.patch.object(sys, "argv", ["", "-u", "uuid6"])
     def test_cli_uuid6(self):
         self.do_test_standalone_uuid(6)

Lib/uuid.py

@@ -961,7 +961,7 @@ def main():
                         default="uuid4",
                         help="function to generate the UUID")
     parser.add_argument("-n", "--namespace",
-                        choices=["any UUID", *namespaces.keys()],
+                        metavar=f"{{any UUID,{','.join(namespaces)}}}",
                         help="uuid3/uuid5 only: "
                         "a UUID, or a well-known predefined UUID addressed "
                         "by namespace name")
@@ -983,7 +983,13 @@ def main():
                 f"{args.uuid} requires a namespace and a name. "
                 "Run 'python -m uuid -h' for more information."
             )
-        namespace = namespaces[namespace] if namespace in namespaces else UUID(namespace)
+        if namespace in namespaces:
+            namespace = namespaces[namespace]
+        else:
+            try:
+                namespace = UUID(namespace)
+            except ValueError as exc:
+                parser.error(f"{exc}: {args.namespace!r}")
         for _ in range(args.count):
             print(uuid_func(namespace, name))
     else: