codekingpro/portable-devtools

1import gc
2import subprocess
3import unittest
4
5import greenlet
6import objgraph
7
8from . import WIN
9from . import TestCase
10from . import _test_extension_cpp
11
12
13class CPPTests(TestCase):
14    def test_exception_switch(self):
15        greenlets = []
16        for i in range(4):
17            g = greenlet.greenlet(_test_extension_cpp.test_exception_switch)
18            g.switch(i)
19            greenlets.append(g)
20        for i, g in enumerate(greenlets):
21            self.assertEqual(g.switch(), i)
22
23    def _do_test_unhandled_exception(self, target):
24        import os
25        import sys
26        script = os.path.join(
27            os.path.dirname(__file__),
28            'fail_cpp_exception.py',
29        )
30        args = [sys.executable, script, target.__name__ if not isinstance(target, str) else target]
31        __traceback_info__ = args
32        with self.assertRaises(subprocess.CalledProcessError) as exc:
33            subprocess.check_output(
34                args,
35                encoding='utf-8',
36                stderr=subprocess.STDOUT
37            )
38
39        ex = exc.exception
40        expected_exit = self.get_expected_returncodes_for_aborted_process()
41        self.assertIn(ex.returncode, expected_exit)
42        self.assertIn('fail_cpp_exception is running', ex.output)
43        return ex.output
44
45
46    def test_unhandled_nonstd_exception_aborts(self):
47        # verify that plain unhandled throw aborts
48        self._do_test_unhandled_exception(_test_extension_cpp.test_exception_throw_nonstd)
49
50    def test_unhandled_std_exception_aborts(self):
51        # verify that plain unhandled throw aborts
52        self._do_test_unhandled_exception(_test_extension_cpp.test_exception_throw_std)
53
54    @unittest.skipIf(WIN, "XXX: This does not crash on Windows")
55    # Meaning the exception is getting lost somewhere...
56    def test_unhandled_std_exception_as_greenlet_function_aborts(self):
57        # verify that plain unhandled throw aborts
58        output = self._do_test_unhandled_exception('run_as_greenlet_target')
59        self.assertIn(
60            # We really expect this to be prefixed with "greenlet: Unhandled C++ exception:"
61            # as added by our handler for std::exception (see TUserGreenlet.cpp), but
62            # that's not correct everywhere --- our handler never runs before std::terminate
63            # gets called (for example, on arm32).
64            'Thrown from an extension.',
65            output
66        )
67
68    def test_unhandled_exception_in_greenlet_aborts(self):
69        # verify that unhandled throw called in greenlet aborts too
70        self._do_test_unhandled_exception('run_unhandled_exception_in_greenlet_aborts')
71
72
73    def test_leak_test_exception_switch_and_do_in_g2(self):
74        def raiser():
75            raise ValueError("boom")
76
77        gc.collect()
78        before = objgraph.count("greenlet")
79
80        for _ in range(1000):
81            with self.assertRaises(ValueError):
82                _test_extension_cpp.test_exception_switch_and_do_in_g2(raiser)
83
84        gc.collect()
85        after = objgraph.count("greenlet")
86        leaked = after - before
87        self.assertEqual(0, leaked)
88
89
90if __name__ == '__main__':
91    unittest.main()
92