Update test_atexit.py to 3.14.3 (#7217)

Lib/test/_test_atexit.py

@@ -19,7 +19,9 @@ class GeneralTest(unittest.TestCase):
             atexit.register(func, *args)
             atexit._run_exitfuncs()
 
-            self.assertEqual(cm.unraisable.object, func)
+            self.assertIsNone(cm.unraisable.object)
+            self.assertEqual(cm.unraisable.err_msg,
+                    f'Exception ignored in atexit callback {func!r}')
             self.assertEqual(cm.unraisable.exc_type, exc_type)
             self.assertEqual(type(cm.unraisable.exc_value), exc_type)
 
@@ -45,6 +47,7 @@ class GeneralTest(unittest.TestCase):
                           ('func2', (), {}),
                           ('func1', (1, 2), {})])
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_badargs(self):
         def func():
             pass
@@ -52,12 +55,14 @@ class GeneralTest(unittest.TestCase):
         # func() has no parameter, but it's called with 2 parameters
         self.assert_raises_unraisable(TypeError, func, 1 ,2)
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_raise(self):
         def raise_type_error():
             raise TypeError
 
         self.assert_raises_unraisable(TypeError, raise_type_error)
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_raise_unnormalized(self):
         # bpo-10756: Make sure that an unnormalized exception is handled
         # properly.
@@ -66,6 +71,7 @@ class GeneralTest(unittest.TestCase):
 
         self.assert_raises_unraisable(ZeroDivisionError, div_zero)
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_exit(self):
         self.assert_raises_unraisable(SystemExit, sys.exit)
 
@@ -116,6 +122,7 @@ class GeneralTest(unittest.TestCase):
         atexit._run_exitfuncs()
         self.assertEqual(l, [5])
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_atexit_with_unregistered_function(self):
         # See bpo-46025 for more info
         def func():
@@ -125,12 +132,63 @@ class GeneralTest(unittest.TestCase):
         try:
             with support.catch_unraisable_exception() as cm:
                 atexit._run_exitfuncs()
-                self.assertEqual(cm.unraisable.object, func)
+                self.assertIsNone(cm.unraisable.object)
+                self.assertEqual(cm.unraisable.err_msg,
+                        f'Exception ignored in atexit callback {func!r}')
                 self.assertEqual(cm.unraisable.exc_type, ZeroDivisionError)
                 self.assertEqual(type(cm.unraisable.exc_value), ZeroDivisionError)
         finally:
             atexit.unregister(func)
 
+    @unittest.skip("TODO: RUSTPYTHON; Hangs")
+    def test_eq_unregister_clear(self):
+        # Issue #112127: callback's __eq__ may call unregister or _clear
+        class Evil:
+            def __eq__(self, other):
+                action(other)
+                return NotImplemented
+
+        for action in atexit.unregister, lambda o: atexit._clear():
+            with self.subTest(action=action):
+                atexit.register(lambda: None)
+                atexit.unregister(Evil())
+                atexit._clear()
+
+    @unittest.skip("TODO: RUSTPYTHON; Hangs")
+    def test_eq_unregister(self):
+        # Issue #112127: callback's __eq__ may call unregister
+        def f1():
+            log.append(1)
+        def f2():
+            log.append(2)
+        def f3():
+            log.append(3)
+
+        class Pred:
+            def __eq__(self, other):
+                nonlocal cnt
+                cnt += 1
+                if cnt == when:
+                    atexit.unregister(what)
+                if other is f2:
+                    return True
+                return False
+
+        for what, expected in (
+                (f1, [3]),
+                (f2, [3, 1]),
+                (f3, [1]),
+            ):
+            for when in range(1, 4):
+                with self.subTest(what=what.__name__, when=when):
+                    cnt = 0
+                    log = []
+                    for f in (f1, f2, f3):
+                        atexit.register(f)
+                    atexit.unregister(Pred())
+                    atexit._run_exitfuncs()
+                    self.assertEqual(log, expected)
+
 
 if __name__ == "__main__":
     unittest.main()

Lib/test/test_atexit.py

@@ -1,10 +1,12 @@
 import atexit
 import os
+import subprocess
 import textwrap
 import unittest
+from test.support import os_helper
 from test import support
-from test.support import script_helper
-
+from test.support import SuppressCrashReport, script_helper
+from test.support import threading_helper
 
 class GeneralTest(unittest.TestCase):
     def test_general(self):
@@ -46,6 +48,40 @@ class FunctionalTest(unittest.TestCase):
         self.assertEqual(res.out.decode().splitlines(), ["atexit2", "atexit1"])
         self.assertFalse(res.err)
 
+    @unittest.skip("TODO: RUSTPYTHON; Flakey on CI")
+    @threading_helper.requires_working_threading()
+    @support.requires_resource("cpu")
+    @unittest.skipUnless(support.Py_GIL_DISABLED, "only meaningful without the GIL")
+    def test_atexit_thread_safety(self):
+        # GH-126907: atexit was not thread safe on the free-threaded build
+        source = """
+        from threading import Thread
+
+        def dummy():
+            pass
+
+
+        def thready():
+            for _ in range(100):
+                atexit.register(dummy)
+                atexit._clear()
+                atexit.register(dummy)
+                atexit.unregister(dummy)
+                atexit._run_exitfuncs()
+
+
+        threads = [Thread(target=thready) for _ in range(10)]
+        for thread in threads:
+            thread.start()
+
+        for thread in threads:
+            thread.join()
+        """
+
+        # atexit._clear() has some evil side effects, and we don't
+        # want them to affect the rest of the tests.
+        script_helper.assert_python_ok("-c", textwrap.dedent(source))
+
 
 @support.cpython_only
 class SubinterpreterTest(unittest.TestCase):
@@ -100,6 +136,37 @@ class SubinterpreterTest(unittest.TestCase):
         self.assertEqual(os.read(r, len(expected)), expected)
         os.close(r)
 
+    # Python built with Py_TRACE_REFS fail with a fatal error in
+    # _PyRefchain_Trace() on memory allocation error.
+    @unittest.skipIf(support.Py_TRACE_REFS, 'cannot test Py_TRACE_REFS build')
+    def test_atexit_with_low_memory(self):
+        # gh-140080: Test that setting low memory after registering an atexit
+        # callback doesn't cause an infinite loop during finalization.
+        code = textwrap.dedent("""
+            import atexit
+            import _testcapi
+
+            def callback():
+                print("hello")
+
+            atexit.register(callback)
+            # Simulate low memory condition
+            _testcapi.set_nomemory(0)
+        """)
+
+        with os_helper.temp_dir() as temp_dir:
+            script = script_helper.make_script(temp_dir, 'test_atexit_script', code)
+            with SuppressCrashReport():
+                with script_helper.spawn_python(script,
+                                                stderr=subprocess.PIPE) as proc:
+                    proc.wait()
+                    stdout = proc.stdout.read()
+                    stderr = proc.stderr.read()
+
+        self.assertIn(proc.returncode, (0, 1))
+        self.assertNotIn(b"hello", stdout)
+        self.assertIn(b"MemoryError", stderr)
+
 
 if __name__ == "__main__":
     unittest.main()