• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import atexit
2import os
3import sys
4import textwrap
5import unittest
6from test import support
7from test.support import script_helper
8
9
10class GeneralTest(unittest.TestCase):
11    def test_general(self):
12        # Run _test_atexit.py in a subprocess since it calls atexit._clear()
13        script = support.findfile("_test_atexit.py")
14        script_helper.run_test_script(script)
15
16class FunctionalTest(unittest.TestCase):
17    def test_shutdown(self):
18        # Actually test the shutdown mechanism in a subprocess
19        code = textwrap.dedent("""
20            import atexit
21
22            def f(msg):
23                print(msg)
24
25            atexit.register(f, "one")
26            atexit.register(f, "two")
27        """)
28        res = script_helper.assert_python_ok("-c", code)
29        self.assertEqual(res.out.decode().splitlines(), ["two", "one"])
30        self.assertFalse(res.err)
31
32    def test_atexit_instances(self):
33        # bpo-42639: It is safe to have more than one atexit instance.
34        code = textwrap.dedent("""
35            import sys
36            import atexit as atexit1
37            del sys.modules['atexit']
38            import atexit as atexit2
39            del sys.modules['atexit']
40
41            assert atexit2 is not atexit1
42
43            atexit1.register(print, "atexit1")
44            atexit2.register(print, "atexit2")
45        """)
46        res = script_helper.assert_python_ok("-c", code)
47        self.assertEqual(res.out.decode().splitlines(), ["atexit2", "atexit1"])
48        self.assertFalse(res.err)
49
50
51@support.cpython_only
52class SubinterpreterTest(unittest.TestCase):
53
54    def test_callbacks_leak(self):
55        # This test shows a leak in refleak mode if atexit doesn't
56        # take care to free callbacks in its per-subinterpreter module
57        # state.
58        n = atexit._ncallbacks()
59        code = textwrap.dedent(r"""
60            import atexit
61            def f():
62                pass
63            atexit.register(f)
64            del atexit
65        """)
66        ret = support.run_in_subinterp(code)
67        self.assertEqual(ret, 0)
68        self.assertEqual(atexit._ncallbacks(), n)
69
70    def test_callbacks_leak_refcycle(self):
71        # Similar to the above, but with a refcycle through the atexit
72        # module.
73        n = atexit._ncallbacks()
74        code = textwrap.dedent(r"""
75            import atexit
76            def f():
77                pass
78            atexit.register(f)
79            atexit.__atexit = atexit
80        """)
81        ret = support.run_in_subinterp(code)
82        self.assertEqual(ret, 0)
83        self.assertEqual(atexit._ncallbacks(), n)
84
85    def test_callback_on_subinterpreter_teardown(self):
86        # This tests if a callback is called on
87        # subinterpreter teardown.
88        expected = b"The test has passed!"
89        r, w = os.pipe()
90
91        code = textwrap.dedent(r"""
92            import os
93            import atexit
94            def callback():
95                os.write({:d}, b"The test has passed!")
96            atexit.register(callback)
97        """.format(w))
98        ret = support.run_in_subinterp(code)
99        os.close(w)
100        self.assertEqual(os.read(r, len(expected)), expected)
101        os.close(r)
102
103
104if __name__ == "__main__":
105    unittest.main()
106