Function _check_tracing_other_threads

tests_python/test_utilities.py:229–282

()

def _check_tracing_other_threads():
    import pydevd_tracing
    import time
    from tests_python.debugger_unittest import wait_for_condition
    import _thread

    # This method is called in a subprocess, so, make sure we exit properly even if we somehow
    # deadlock somewhere else.
    def dump_threads_and_kill_on_timeout():
        time.sleep(10)
        from _pydevd_bundle import pydevd_utils

        pydevd_utils.dump_threads()
        time.sleep(1)
        import os

        os._exit(77)

    _thread.start_new_thread(dump_threads_and_kill_on_timeout, ())

    def method():
        while True:
            trace_func = sys.gettrace()
            if trace_func:
                threading.current_thread().trace_func = trace_func
                break
            time.sleep(0.01)

    def dummy_thread_method():
        threads.append(threading.current_thread())
        method()

    threads = []
    threads.append(threading.Thread(target=method))
    threads[-1].daemon = True
    threads[-1].start()
    _thread.start_new_thread(dummy_thread_method, ())

    wait_for_condition(lambda: len(threads) == 2, msg=lambda: "Found threads: %s" % (threads,))

    def tracing_func(frame, event, args):
        return tracing_func

    assert pydevd_tracing.set_trace_to_threads(tracing_func) == 0

    def check_threads_tracing_func():
        for t in threads:
            if getattr(t, "trace_func", None) != tracing_func:
                return False
        return True

    wait_for_condition(check_threads_tracing_func)

    assert tracing_func == sys.gettrace()


def _build_launch_env():