Align sys runtime metadata with Python 3.15 (#9194)

Expose ABI metadata, preserve the canonical sys.flags tuple order and named flags, and report the native lock implementation honestly. Refresh canonical test_sys while preserving RustPython guards.

Assisted-by: OpenAI Codex:model version unavailable

Lib/test/test_sys.py

@@ -6,6 +6,7 @@ import locale
 import operator
 import os
 import random
+import shutil
 import socket
 import struct
 import subprocess
@@ -494,7 +495,7 @@ class SysModuleTest(unittest.TestCase):
             self.assertIs(f, f2)
         self.assertIsNone(sys._getframemodulename(i))
 
-    # sys._current_frames() is a CPython-only gimmick.
+    @support.cpython_only  # sys._current_frames() is a CPython-only gimmick.
     @threading_helper.reap_threads
     @threading_helper.requires_working_threading()
     def test_current_frames(self):
@@ -562,6 +563,38 @@ class SysModuleTest(unittest.TestCase):
             leave_g.set()
             t.join()
 
+    @support.cpython_only
+    @requires_subinterpreters
+    @threading_helper.requires_working_threading()
+    def test_current_frames_other_interpreters(self):
+        # gh-158364: sys._current_frames() would access frames of another
+        # interpreter and crash
+        import threading
+
+        entered = threading.Event()
+        left = threading.Event()
+
+        def park():
+            entered.set()
+            left.wait()
+
+        t = threading.Thread(target=park)
+        with threading_helper.start_threads([t], unlock=left.set):
+            entered.wait()
+            interp = interpreters.create()
+            try:
+                interp.exec(f"""if True:
+                    import sys
+                    import threading
+
+                    frames = sys._current_frames()
+                    assert threading.get_ident() in frames, frames
+                    assert frames[threading.get_ident()].f_globals is globals()
+                    assert {t.ident} not in frames, frames
+                    """)
+            finally:
+                interp.close()
+
     @threading_helper.reap_threads
     @threading_helper.requires_working_threading()
     def test_current_exceptions(self):
@@ -628,6 +661,39 @@ class SysModuleTest(unittest.TestCase):
             leave_g.set()
             t.join()
 
+    @support.cpython_only
+    @requires_subinterpreters
+    @threading_helper.requires_working_threading()
+    def test_current_exceptions_other_interpreters(self):
+        # gh-158364: sys._current_exceptions() would hand out exceptions of
+        # another interpreter and crash
+        import threading
+
+        entered = threading.Event()
+        left = threading.Event()
+
+        def hold():
+            # The thread has to be handling an exception, otherwise
+            # sys._current_exceptions() has nothing to report for it.
+            try:
+                raise ValueError
+            except ValueError:
+                entered.set()
+                left.wait()
+
+        t = threading.Thread(target=hold)
+        with threading_helper.start_threads([t], unlock=left.set):
+            entered.wait()
+            interp = interpreters.create()
+            try:
+                interp.exec(f"""if True:
+                    import sys
+
+                    assert {t.ident} not in sys._current_exceptions()
+                    """)
+            finally:
+                interp.close()
+
     def test_attributes(self):
         self.assertIsInstance(sys.api_version, int)
         self.assertIsInstance(sys.argv, list)
@@ -729,15 +795,25 @@ class SysModuleTest(unittest.TestCase):
         info = sys.thread_info
         self.assertEqual(len(info), 3)
         self.assertIn(info.name, ('nt', 'pthread', 'pthread-stubs', 'solaris', None))
-        self.assertIn(info.lock, ('semaphore', 'mutex+cond', None))
-        if sys.platform.startswith(("linux", "android", "freebsd")):
+        self.assertIn(info.lock, ('pymutex', None))
+        if sys.platform.startswith(("linux", "android", "freebsd", "wasi")):
             self.assertEqual(info.name, "pthread")
         elif sys.platform == "win32":
             self.assertEqual(info.name, "nt")
         elif sys.platform == "emscripten":
             self.assertIn(info.name, {"pthread", "pthread-stubs"})
-        elif sys.platform == "wasi":
-            self.assertEqual(info.name, "pthread-stubs")
+
+    def test_abi_info(self):
+        info = sys.abi_info
+        info_keys = {'pointer_bits', 'free_threaded', 'debug', 'byteorder'}
+        self.assertEqual(set(vars(info)), info_keys)
+        pointer_bits = 64 if sys.maxsize > 2**32 else 32
+        self.assertEqual(info.pointer_bits, pointer_bits)
+        self.assertEqual(info.free_threaded,
+                         bool(sysconfig.get_config_var('Py_GIL_DISABLED')))
+        self.assertEqual(info.debug,
+                         bool(sysconfig.get_config_var('Py_DEBUG')))
+        self.assertEqual(info.byteorder, sys.byteorder)
 
     @unittest.skipUnless(support.is_emscripten, "only available on Emscripten")
     def test_emscripten_info(self):
@@ -849,23 +925,35 @@ class SysModuleTest(unittest.TestCase):
                     '''))
                 self.assertTrue(sys._is_interned(s))
 
-    def test_sys_flags(self):
+    def test_sys_flags_indexable_attributes(self):
         self.assertTrue(sys.flags)
-        attrs = ("debug",
+        # We've stopped assigning sequence indices to new sys.flags attributes:
+        # https://github.com/python/cpython/issues/122575#issuecomment-2416497086
+        indexable_attrs = ("debug",
                  "inspect", "interactive", "optimize",
                  "dont_write_bytecode", "no_user_site", "no_site",
                  "ignore_environment", "verbose", "bytes_warning", "quiet",
                  "hash_randomization", "isolated", "dev_mode", "utf8_mode",
                  "warn_default_encoding", "safe_path", "int_max_str_digits")
-        for attr in attrs:
+        for attr_idx, attr in enumerate(indexable_attrs):
             self.assertHasAttr(sys.flags, attr)
             attr_type = bool if attr in ("dev_mode", "safe_path") else int
             self.assertEqual(type(getattr(sys.flags, attr)), attr_type, attr)
+            attr_value = getattr(sys.flags, attr)
+            self.assertEqual(sys.flags[attr_idx], attr_value,
+                             msg=f"sys.flags .{attr} vs [{attr_idx}]")
         self.assertTrue(repr(sys.flags))
-        self.assertEqual(len(sys.flags), len(attrs))
+        self.assertEqual(len(sys.flags), 18, msg="Do not increase, see GH-122575")
 
         self.assertIn(sys.flags.utf8_mode, {0, 1, 2})
 
+    def test_sys_flags_name_only_attributes(self):
+        # non-tuple sequence fields (name only sys.flags attributes)
+        self.assertIsInstance(sys.flags.gil, int|type(None))
+        self.assertIsInstance(sys.flags.thread_inherit_context, int|type(None))
+        self.assertIsInstance(sys.flags.context_aware_warnings, int|type(None))
+        self.assertIsInstance(sys.flags.lazy_imports, int|type(None))
+
     def assert_raise_on_new_sys_type(self, sys_attr):
         # Users are intentionally prevented from creating new instances of
         # sys.flags, sys.version_info, and sys.getwindowsversion.
@@ -1295,6 +1383,7 @@ class SysModuleTest(unittest.TestCase):
         for name in sys.stdlib_module_names:
             self.assertIsInstance(name, str)
 
+    @unittest.skipUnless(hasattr(sys, '_stdlib_dir'), 'need sys._stdlib_dir')
     def test_stdlib_dir(self):
         os = import_helper.import_fresh_module('os')
         marker = getattr(os, '__file__', None)
@@ -1339,8 +1428,29 @@ class SysModuleTest(unittest.TestCase):
     def test_disable_gil_abi(self):
         self.assertEqual('t' in sys.abiflags, support.Py_GIL_DISABLED)
 
+    def test_int_max_str_digits(self):
+        old_limit = sys.get_int_max_str_digits()
+        self.assertIsInstance(old_limit, int)
+        self.assertGreaterEqual(old_limit, 0)
+        self.addCleanup(sys.set_int_max_str_digits, old_limit)
+
+        sys.set_int_max_str_digits(0)
+        self.assertEqual(sys.get_int_max_str_digits(), 0)
+
+        sys.set_int_max_str_digits(2_048)
+        self.assertEqual(sys.get_int_max_str_digits(), 2_048)
+
+        with self.assertRaises(ValueError):
+            # the minimum is 640 digits
+            sys.set_int_max_str_digits(5)
+        with self.assertRaises(ValueError):
+            sys.set_int_max_str_digits(-2)
+        with self.assertRaises(TypeError):
+            sys.set_int_max_str_digits(2_048.0)
+
 
 @test.support.cpython_only
+@test.support.force_not_colorized_test_class
 class UnraisableHookTest(unittest.TestCase):
     def test_original_unraisablehook(self):
         _testcapi = import_helper.import_module('_testcapi')
@@ -1572,7 +1682,7 @@ class SizeofTest(unittest.TestCase):
         samples = [b'', b'u'*100000]
         for sample in samples:
             x = bytearray(sample)
-            check(x, vsize('n2Pi') + x.__alloc__())
+            check(x, vsize('n2PiP') + x.__alloc__())
         # bytearray_iterator
         check(iter(bytearray()), size('nP'))
         # bytes
@@ -1714,9 +1824,10 @@ class SizeofTest(unittest.TestCase):
         check(int(PyLong_BASE**2), vsize('') + 3*self.longdigit)
         # module
         if support.Py_GIL_DISABLED:
-            check(unittest, size('PPPPPP'))
+            md_gil = '?'
         else:
-            check(unittest, size('PPPPP'))
+            md_gil = ''
+        check(unittest, size('PPPP?' + md_gil + 'NPPPPP'))
         # None
         check(None, size(''))
         # NotImplementedType
@@ -1776,7 +1887,7 @@ class SizeofTest(unittest.TestCase):
         check((1,2,3), vsize('') + self.P + 3*self.P)
         # type
         # static type: PyTypeObject
-        fmt = 'P2nPI13Pl4Pn9Pn12PIPc'
+        fmt = 'P2nPI13Pl4Pn9Pn12PI2Pc'
         s = vsize(fmt)
         check(int, s)
         typeid = 'n' if support.Py_GIL_DISABLED else ''
@@ -1870,7 +1981,10 @@ class SizeofTest(unittest.TestCase):
         check(S(), set(), '3P')
         class FS(frozenset):
             __slots__ = 'a', 'b', 'c'
-        check(FS(), frozenset(), '3P')
+
+        class mytuple(tuple):
+            pass
+        check(FS([mytuple()]), frozenset([mytuple()]), '3P')
         from collections import OrderedDict
         class OD(OrderedDict):
             __slots__ = 'a', 'b', 'c'
@@ -1883,7 +1997,7 @@ class SizeofTest(unittest.TestCase):
         check = self.check_sizeof
         # _ast.AST
         import _ast
-        check(_ast.AST(), size('P'))
+        check(_ast.Module(), size('3P'))
         try:
             raise TypeError
         except TypeError as e:
@@ -1894,10 +2008,16 @@ class SizeofTest(unittest.TestCase):
         # symtable entry
         # XXX
         # sys.flags
-        # FIXME: The +3 is for the 'gil', 'thread_inherit_context' and
-        # 'context_aware_warnings' flags and will not be necessary once
-        # gh-122575 is fixed
-        check(sys.flags, vsize('') + self.P + self.P * (3 + len(sys.flags)))
+        # FIXME: The non_sequence_fields adjustment is for these flags:
+        # - 'gil'
+        # - 'thread_inherit_context'
+        # - 'context_aware_warnings'
+        # - 'lazy_imports'
+        # Not needing to increment this every time we add a new field
+        # per GH-122575 would be nice...
+        # Q: What is the actual point of this sys.flags C size derived from PyStructSequence_Field array assertion?
+        non_sequence_fields = 4
+        check(sys.flags, vsize('') + self.P + self.P * (non_sequence_fields + len(sys.flags)))
 
     def test_asyncgen_hooks(self):
         old = sys.get_asyncgen_hooks()
@@ -1962,7 +2082,8 @@ class TestRemoteExec(unittest.TestCase):
         test.support.reap_children()
 
     def _run_remote_exec_test(self, script_code, python_args=None, env=None,
-                              prologue='',
+                              python_executable=None, prologue='',
+                              after_ready=None,
                               script_path=os_helper.TESTFN + '_remote.py'):
         # Create the script that will be remotely executed
         self.addCleanup(os_helper.unlink, script_path)
@@ -2010,7 +2131,10 @@ sock.close()
 ''')
 
         # Start the target process and capture its output
-        cmd = [sys.executable]
+        if python_executable is None:
+            python_executable = sys.executable
+
+        cmd = [python_executable]
         if python_args:
             cmd.extend(python_args)
         cmd.append(target)
@@ -2035,6 +2159,9 @@ sock.close()
                 response = client_socket.recv(1024)
                 self.assertEqual(response, b"ready")
 
+                if after_ready is not None:
+                    after_ready(proc)
+
                 # Try remote exec on the target process
                 sys.remote_exec(proc.pid, script_path)
 
@@ -2057,6 +2184,19 @@ sock.close()
                 proc.terminate()
                 proc.wait(timeout=SHORT_TIMEOUT)
 
+    def _run_remote_exec_with_deleted_mapping(self, deleted_path, **kwargs):
+        def delete_loaded_mapping(proc):
+            os_helper.unlink(deleted_path)
+            with open(f'/proc/{proc.pid}/maps', encoding='utf-8') as maps:
+                self.assertIn(f'{deleted_path} (deleted)', maps.read())
+
+        script = 'print("Remote script executed successfully!")'
+        returncode, stdout, stderr = self._run_remote_exec_test(
+            script, after_ready=delete_loaded_mapping, **kwargs)
+        self.assertEqual(returncode, 0)
+        self.assertIn(b"Remote script executed successfully!", stdout)
+        self.assertEqual(stderr, b"")
+
     def test_remote_exec(self):
         """Test basic remote exec functionality"""
         script = 'print("Remote script executed successfully!")'
@@ -2183,6 +2323,74 @@ this is invalid python code
         with self.assertRaises(OSError):
             sys.remote_exec(os.getpid(), "invalid_script_path")
 
+    @unittest.skipUnless(sys.platform == 'linux', 'Linux-only regression test')
+    @unittest.skipUnless(
+        sysconfig.get_config_var('Py_ENABLE_SHARED') == 1,
+        'requires a shared libpython build')
+    def test_remote_exec_deleted_libpython(self):
+        """Test remote exec when the target libpython was deleted."""
+        build_dir = sysconfig.get_config_var('abs_builddir')
+        ldlibrary = sysconfig.get_config_var('LDLIBRARY')
+        instsoname = sysconfig.get_config_var('INSTSONAME')
+        if not build_dir or not ldlibrary or not instsoname:
+            self.skipTest('cannot determine shared libpython location')
+
+        source_libpython = os.path.join(build_dir, instsoname)
+        if not os.path.exists(source_libpython):
+            self.skipTest(f'{source_libpython!r} does not exist')
+
+        with os_helper.temp_dir() as lib_dir:
+            copied_libpython = os.path.join(lib_dir, instsoname)
+            shutil.copy2(source_libpython, copied_libpython)
+            if ldlibrary != instsoname:
+                os.symlink(instsoname, os.path.join(lib_dir, ldlibrary))
+
+            env = os.environ.copy()
+            ld_library_path = env.get('LD_LIBRARY_PATH')
+            env['LD_LIBRARY_PATH'] = lib_dir if not ld_library_path else (
+                lib_dir + os.pathsep + ld_library_path)
+
+            self._run_remote_exec_with_deleted_mapping(copied_libpython,
+                                                       env=env)
+
+    @unittest.skipUnless(sys.platform == 'linux', 'Linux-only regression test')
+    @unittest.skipUnless(
+        sysconfig.get_config_var('Py_ENABLE_SHARED') == 0,
+        'requires a static Python build')
+    def test_remote_exec_deleted_static_executable(self):
+        """Test remote exec when the target static executable was deleted."""
+        build_dir = sysconfig.get_config_var('abs_builddir')
+        stdlib_dir = os.path.dirname(os.path.abspath(os.__file__))
+        if not build_dir or not os.path.isdir(stdlib_dir):
+            self.skipTest('cannot determine build-tree locations')
+
+        pybuilddir_txt = os.path.join(build_dir, 'pybuilddir.txt')
+        if not os.path.exists(pybuilddir_txt):
+            self.skipTest(f'{pybuilddir_txt!r} does not exist')
+
+        with open(pybuilddir_txt, encoding='utf-8') as pybuilddir_file:
+            pybuilddir = pybuilddir_file.read().strip()
+        source_ext_dir = os.path.join(build_dir, pybuilddir)
+        if not os.path.isdir(source_ext_dir):
+            self.skipTest(f'{source_ext_dir!r} does not exist')
+
+        with os_helper.temp_dir() as copied_root:
+            copied_build_dir = os.path.join(copied_root, 'build')
+            copied_pybuilddir = os.path.join(copied_build_dir, pybuilddir)
+            os.makedirs(os.path.dirname(copied_pybuilddir))
+            os.symlink(stdlib_dir, os.path.join(copied_root, 'Lib'))
+            os.symlink(source_ext_dir, copied_pybuilddir)
+            shutil.copy2(pybuilddir_txt,
+                         os.path.join(copied_build_dir, 'pybuilddir.txt'))
+
+            copied_python = os.path.join(copied_build_dir,
+                                         os.path.basename(sys.executable))
+            shutil.copy2(sys.executable, copied_python)
+
+            self._run_remote_exec_with_deleted_mapping(
+                copied_python, python_args=['-S'],
+                python_executable=copied_python)
+
     def test_remote_exec_in_process_without_debug_fails_envvar(self):
         """Test remote exec in a process without remote debugging enabled"""
         script = os_helper.TESTFN + '_remote.py'
@@ -2247,9 +2455,10 @@ class TestSysJIT(unittest.TestCase):
 
             def frame_3_jit() -> None:
                 # JITs just before the last loop:
-                for i in range(_testinternalcapi.TIER2_THRESHOLD + 1):
+                # 1 extra iteration for tracing.
+                for i in range(_testinternalcapi.TIER2_THRESHOLD + 2):
                     # Careful, doing this in the reverse order breaks tracing:
-                    expected = {enabled} and i == _testinternalcapi.TIER2_THRESHOLD
+                    expected = {enabled} and i >= _testinternalcapi.TIER2_THRESHOLD
                     assert sys._jit.is_active() is expected
                     frame_2_jit(expected)
                     assert sys._jit.is_active() is expected

crates/vm/src/stdlib/sys.rs

@@ -652,6 +652,16 @@ pub mod sys {
             .to_owned()
     }
 
+    #[pyattr]
+    fn abi_info(vm: &VirtualMachine) -> PyRef<PyNamespace> {
+        py_namespace!(vm, {
+            "pointer_bits" => vm.ctx.new_int(usize::BITS),
+            "free_threaded" => vm.ctx.new_bool(true),
+            "debug" => vm.ctx.new_bool(false),
+            "byteorder" => byteorder(vm),
+        })
+    }
+
     #[pyattr]
     fn _base_executable(vm: &VirtualMachine) -> String {
         vm.state.config.paths.base_executable.clone()
@@ -1637,12 +1647,12 @@ pub mod sys {
         dev_mode: bool,
         /// -X utf8
         utf8_mode: u8,
-        /// -X int_max_str_digits=number
-        int_max_str_digits: i64,
-        /// -P, `PYTHONSAFEPATH`
-        safe_path: bool,
         /// -X warn_default_encoding, PYTHONWARNDEFAULTENCODING
         warn_default_encoding: u8,
+        /// -P, `PYTHONSAFEPATH`
+        safe_path: bool,
+        /// -X int_max_str_digits=number
+        int_max_str_digits: i64,
     }
 
     impl FlagsData {
@@ -1685,13 +1695,18 @@ pub mod sys {
         }
 
         #[pygetset]
-        fn context_aware_warnings(&self, vm: &VirtualMachine) -> bool {
-            vm.state.config.settings.context_aware_warnings
+        const fn gil(&self) -> u8 {
+            0
         }
 
         #[pygetset]
-        fn thread_inherit_context(&self, vm: &VirtualMachine) -> bool {
-            vm.state.config.settings.thread_inherit_context
+        fn context_aware_warnings(&self, vm: &VirtualMachine) -> u8 {
+            vm.state.config.settings.context_aware_warnings as u8
+        }
+
+        #[pygetset]
+        fn thread_inherit_context(&self, vm: &VirtualMachine) -> u8 {
+            vm.state.config.settings.thread_inherit_context as u8
         }
 
         #[pygetset]
@@ -1712,9 +1727,9 @@ pub mod sys {
     impl ThreadInfoData {
         const INFO: Self = Self {
             name: crate::stdlib::_thread::_thread::PYTHREAD_NAME,
-            // As I know, there's only way to use lock as "Mutex" in Rust
-            // with satisfying python document spec.
-            lock: Some("mutex+cond"),
+            // RustPython uses parking_lot rather than CPython's PyMutex.
+            // None denotes a lock implementation without a Python ABI name.
+            lock: None,
             version: None,
         };
     }