Method kill

dash/testing/application_runners.py:123–138

(self)

        self._old_threads = list(threading._active.keys())  # type: ignore[reportAttributeAccessIssue]; pylint: disable=W0212

    def kill(self):
        # Kill all the new threads.
        for thread_id in list(threading._active):  # type: ignore[reportAttributeAccessIssue]; pylint: disable=W0212
            if thread_id in self._old_threads:
                continue

            res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
                ctypes.c_long(thread_id), ctypes.py_object(SystemExit)
            )
            if res == 0:
                raise ValueError(f"Invalid thread id: {thread_id}")
            if res > 1:
                ctypes.pythonapi.PyThreadState_SetAsyncExc(
                    ctypes.c_long(thread_id), None
                )
                raise SystemExit("Stopping thread failure")


class ThreadedRunner(BaseDashRunner):