codekingpro/portable-devtools

1# -*- coding: utf-8 -*-
2"""
3Tests for greenlet.
4
5"""
6import os
7import sys
8import sysconfig
9import unittest
10
11from gc import collect
12from gc import get_objects
13from threading import active_count as active_thread_count
14from time import sleep
15from time import time
16
17import psutil
18
19from greenlet import greenlet as RawGreenlet
20from greenlet import getcurrent
21
22from greenlet._greenlet import get_pending_cleanup_count
23from greenlet._greenlet import get_total_main_greenlets
24
25from . import leakcheck
26
27PY312 = sys.version_info[:2] >= (3, 12)
28PY313 = sys.version_info[:2] >= (3, 13)
29# XXX: First tested on 3.14a7. Revisit all uses of this on later versions to ensure they
30# are still valid.
31PY314 = sys.version_info[:2] >= (3, 14)
32
33WIN = sys.platform.startswith("win")
34RUNNING_ON_GITHUB_ACTIONS = os.environ.get('GITHUB_ACTIONS')
35RUNNING_ON_TRAVIS = os.environ.get('TRAVIS') or RUNNING_ON_GITHUB_ACTIONS
36RUNNING_ON_APPVEYOR = os.environ.get('APPVEYOR')
37RUNNING_ON_CI = RUNNING_ON_TRAVIS or RUNNING_ON_APPVEYOR
38RUNNING_ON_MANYLINUX = os.environ.get('GREENLET_MANYLINUX')
39
40# Is the current interpreter free-threaded?) Note that this
41# isn't the same as whether the GIL is enabled, this is the build-time
42# value. Certain CPython details, like the garbage collector,
43# work very differently on potentially-free-threaded builds than
44# standard builds.
45RUNNING_ON_FREETHREAD_BUILD = bool(sysconfig.get_config_var("Py_GIL_DISABLED"))
46
47class TestCaseMetaClass(type):
48    # wrap each test method with
49    # a) leak checks
50    def __new__(cls, classname, bases, classDict):
51        # pylint and pep8 fight over what this should be called (mcs or cls).
52        # pylint gets it right, but we can't scope disable pep8, so we go with
53        # its convention.
54        # pylint: disable=bad-mcs-classmethod-argument
55        check_totalrefcount = True
56
57        # Python 3: must copy, we mutate the classDict. Interestingly enough,
58        # it doesn't actually error out, but under 3.6 we wind up wrapping
59        # and re-wrapping the same items over and over and over.
60        for key, value in list(classDict.items()):
61            if key.startswith('test') and callable(value):
62                classDict.pop(key)
63                if check_totalrefcount:
64                    value = leakcheck.wrap_refcount(value)
65                classDict[key] = value
66        return type.__new__(cls, classname, bases, classDict)
67
68
69class TestCase(unittest.TestCase, metaclass=TestCaseMetaClass):
70
71    cleanup_attempt_sleep_duration = 0.001
72    cleanup_max_sleep_seconds = 1
73
74    def wait_for_pending_cleanups(self,
75                                  initial_active_threads=None,
76                                  initial_main_greenlets=None):
77        initial_active_threads = initial_active_threads or self.threads_before_test
78        initial_main_greenlets = initial_main_greenlets or self.main_greenlets_before_test
79        sleep_time = self.cleanup_attempt_sleep_duration
80        # NOTE: This is racy! A Python-level thread object may be dead
81        # and gone, but the C thread may not yet have fired its
82        # destructors and added to the queue. There's no particular
83        # way to know that's about to happen. We try to watch the
84        # Python threads to make sure they, at least, have gone away.
85        # Counting the main greenlets, which we can easily do deterministically,
86        # also helps.
87
88        # Always sleep at least once to let other threads run
89        sleep(sleep_time)
90        quit_after = time() + self.cleanup_max_sleep_seconds
91        # TODO: We could add an API that calls us back when a particular main greenlet is deleted?
92        # It would have to drop the GIL
93        while (
94                get_pending_cleanup_count()
95                or active_thread_count() > initial_active_threads
96                or (not self.expect_greenlet_leak
97                    and get_total_main_greenlets() > initial_main_greenlets)):
98            sleep(sleep_time)
99            if time() > quit_after:
100                print("Time limit exceeded.")
101                print("Threads: Waiting for only", initial_active_threads,
102                      "-->", active_thread_count())
103                print("MGlets : Waiting for only", initial_main_greenlets,
104                      "-->", get_total_main_greenlets())
105                break
106        collect()
107
108    def count_objects(self, kind=list, exact_kind=True):
109        # pylint:disable=unidiomatic-typecheck
110        # Collect the garbage.
111        for _ in range(3):
112            collect()
113        if exact_kind:
114            return sum(
115                1
116                for x in get_objects()
117                if type(x) is kind
118            )
119        # instances
120        return sum(
121            1
122            for x in get_objects()
123            if isinstance(x, kind)
124        )
125
126    greenlets_before_test = 0
127    threads_before_test = 0
128    main_greenlets_before_test = 0
129    expect_greenlet_leak = False
130
131    def count_greenlets(self):
132        """
133        Find all the greenlets and subclasses tracked by the GC.
134        """
135        return self.count_objects(RawGreenlet, False)
136
137    def setUp(self):
138        # Ensure the main greenlet exists, otherwise the first test
139        # gets a false positive leak
140        super().setUp()
141        getcurrent()
142        self.threads_before_test = active_thread_count()
143        self.main_greenlets_before_test = get_total_main_greenlets()
144        self.wait_for_pending_cleanups(self.threads_before_test, self.main_greenlets_before_test)
145        self.greenlets_before_test = self.count_greenlets()
146
147    def tearDown(self):
148        if getattr(self, 'skipTearDown', False):
149            return
150
151        self.wait_for_pending_cleanups(self.threads_before_test, self.main_greenlets_before_test)
152        super().tearDown()
153
154    def get_expected_returncodes_for_aborted_process(self):
155        import signal
156        # The child should be aborted in an unusual way. On POSIX
157        # platforms, this is done with abort() and signal.SIGABRT,
158        # which is reflected in a negative return value; however, on
159        # Windows, even though we observe the child print "Fatal
160        # Python error: Aborted" and in older versions of the C
161        # runtime "This application has requested the Runtime to
162        # terminate it in an unusual way," it always has an exit code
163        # of 3. This is interesting because 3 is the error code for
164        # ERROR_PATH_NOT_FOUND; BUT: the C runtime abort() function
165        # also uses this code.
166        #
167        # If we link to the static C library on Windows, the error
168        # code changes to '0xc0000409' (hex(3221226505)), which
169        # apparently is STATUS_STACK_BUFFER_OVERRUN; but "What this
170        # means is that nowadays when you get a
171        # STATUS_STACK_BUFFER_OVERRUN, it doesn’t actually mean that
172        # there is a stack buffer overrun. It just means that the
173        # application decided to terminate itself with great haste."
174        #
175        #
176        # On windows, we've also seen '0xc0000005' (hex(3221225477)).
177        # That's "Access Violation"
178        #
179        # See
180        # https://devblogs.microsoft.com/oldnewthing/20110519-00/?p=10623
181        # and
182        # https://docs.microsoft.com/en-us/previous-versions/k089yyh0(v=vs.140)?redirectedfrom=MSDN
183        # and
184        # https://devblogs.microsoft.com/oldnewthing/20190108-00/?p=100655
185        expected_exit = (
186            -signal.SIGABRT,
187            # But beginning on Python 3.11, the faulthandler
188            # that prints the C backtraces sometimes segfaults after
189            # reporting the exception but before printing the stack.
190            # This has only been seen on linux/gcc.
191            -signal.SIGSEGV,
192        ) if not WIN else (
193            3,
194            0xc0000409,
195            0xc0000005,
196        )
197        return expected_exit
198
199    def get_process_uss(self):
200        """
201        Return the current process's USS in bytes.
202
203        uss is available on Linux, macOS, Windows. Also known as
204        "Unique Set Size", this is the memory which is unique to a
205        process and which would be freed if the process was terminated
206        right now.
207
208        If this is not supported by ``psutil``, this raises the
209        :exc:`unittest.SkipTest` exception.
210        """
211        try:
212            return psutil.Process().memory_full_info().uss
213        except AttributeError as e:
214            raise unittest.SkipTest("uss not supported") from e
215
216    def run_script(self, script_name, show_output=True):
217        import subprocess
218        script = os.path.join(
219            os.path.dirname(__file__),
220            script_name,
221        )
222
223        try:
224            return subprocess.check_output([sys.executable, script],
225                                           encoding='utf-8',
226                                           stderr=subprocess.STDOUT)
227        except subprocess.CalledProcessError as ex:
228            if show_output:
229                print('-----')
230                print('Failed to run script', script)
231                print('~~~~~')
232                print(ex.output)
233                print('------')
234            raise
235
236
237    def assertScriptRaises(self, script_name, exitcodes=None):
238        import subprocess
239        with self.assertRaises(subprocess.CalledProcessError) as exc:
240            output = self.run_script(script_name, show_output=False)
241            __traceback_info__ = output
242            # We're going to fail the assertion if we get here, at least
243            # preserve the output in the traceback.
244
245        if exitcodes is None:
246            exitcodes = self.get_expected_returncodes_for_aborted_process()
247        self.assertIn(exc.exception.returncode, exitcodes)
248        return exc.exception
249