1  import atexit
       2  import os
       3  import sys
       4  import textwrap
       5  import unittest
       6  from test import support
       7  from test.support import script_helper
       8  
       9  
      10  class ESC[4;38;5;81mGeneralTest(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
      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  
      16  class ESC[4;38;5;81mFunctionalTest(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
      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
      52  class ESC[4;38;5;81mSubinterpreterTest(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
      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      @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
      86      def test_callback_on_subinterpreter_teardown(self):
      87          # This tests if a callback is called on
      88          # subinterpreter teardown.
      89          expected = b"The test has passed!"
      90          r, w = os.pipe()
      91  
      92          code = textwrap.dedent(r"""
      93              import os
      94              import atexit
      95              def callback():
      96                  os.write({:d}, b"The test has passed!")
      97              atexit.register(callback)
      98          """.format(w))
      99          ret = support.run_in_subinterp(code)
     100          os.close(w)
     101          self.assertEqual(os.read(r, len(expected)), expected)
     102          os.close(r)
     103  
     104  
     105  if __name__ == "__main__":
     106      unittest.main()