diff --git a/bbot/test/conftest.py b/bbot/test/conftest.py index c2e8b3448..e82333552 100644 --- a/bbot/test/conftest.py +++ b/bbot/test/conftest.py @@ -240,80 +240,80 @@ def pytest_terminal_summary(terminalreporter, exitstatus, config): # pragma: no # BELOW: debugging for frozen/hung tests -# import psutil -# import traceback -# import inspect - - -# def _print_detailed_info(): # pragma: no cover -# """ -# Debugging pytests hanging -# """ -# print("=== Detailed Thread and Process Information ===\n") -# try: -# print("=== Threads ===") -# for thread in threading.enumerate(): -# print(f"Thread Name: {thread.name}") -# print(f"Thread ID: {thread.ident}") -# print(f"Is Alive: {thread.is_alive()}") -# print(f"Daemon: {thread.daemon}") - -# if hasattr(thread, "_target"): -# target = thread._target -# if target: -# qualname = ( -# f"{target.__module__}.{target.__qualname__}" -# if hasattr(target, "__qualname__") -# else str(target) -# ) -# print(f"Target Function: {qualname}") - -# if hasattr(thread, "_args"): -# args = thread._args -# kwargs = thread._kwargs if hasattr(thread, "_kwargs") else {} -# arg_spec = inspect.getfullargspec(target) - -# all_args = list(args) + [f"{k}={v}" for k, v in kwargs.items()] - -# if inspect.ismethod(target) and arg_spec.args[0] == "self": -# arg_spec.args.pop(0) - -# named_args = list(zip(arg_spec.args, all_args)) -# if arg_spec.varargs: -# named_args.extend((f"*{arg_spec.varargs}", arg) for arg in all_args[len(arg_spec.args) :]) - -# print("Arguments:") -# for name, value in named_args: -# print(f" {name}: {value}") -# else: -# print("Target Function: None") -# else: -# print("Target Function: Unknown") - -# print() - -# print("=== Processes ===") -# current_process = psutil.Process() -# for child in current_process.children(recursive=True): -# print(f"Process ID: {child.pid}") -# print(f"Name: {child.name()}") -# print(f"Status: {child.status()}") -# print(f"CPU Times: {child.cpu_times()}") -# print(f"Memory Info: {child.memory_info()}") -# print() - -# print("=== Current Process ===") -# print(f"Process ID: {current_process.pid}") -# print(f"Name: {current_process.name()}") -# print(f"Status: {current_process.status()}") -# print(f"CPU Times: {current_process.cpu_times()}") -# print(f"Memory Info: {current_process.memory_info()}") -# print() - -# except Exception as e: -# print(f"An error occurred: {str(e)}") -# print("Traceback:") -# traceback.print_exc() +import psutil +import traceback +import inspect + + +def _print_detailed_info(): # pragma: no cover + """ + Debugging pytests hanging + """ + print("=== Detailed Thread and Process Information ===\n") + try: + print("=== Threads ===") + for thread in threading.enumerate(): + print(f"Thread Name: {thread.name}") + print(f"Thread ID: {thread.ident}") + print(f"Is Alive: {thread.is_alive()}") + print(f"Daemon: {thread.daemon}") + + if hasattr(thread, "_target"): + target = thread._target + if target: + qualname = ( + f"{target.__module__}.{target.__qualname__}" + if hasattr(target, "__qualname__") + else str(target) + ) + print(f"Target Function: {qualname}") + + if hasattr(thread, "_args"): + args = thread._args + kwargs = thread._kwargs if hasattr(thread, "_kwargs") else {} + arg_spec = inspect.getfullargspec(target) + + all_args = list(args) + [f"{k}={v}" for k, v in kwargs.items()] + + if inspect.ismethod(target) and arg_spec.args[0] == "self": + arg_spec.args.pop(0) + + named_args = list(zip(arg_spec.args, all_args)) + if arg_spec.varargs: + named_args.extend((f"*{arg_spec.varargs}", arg) for arg in all_args[len(arg_spec.args) :]) + + print("Arguments:") + for name, value in named_args: + print(f" {name}: {value}") + else: + print("Target Function: None") + else: + print("Target Function: Unknown") + + print() + + print("=== Processes ===") + current_process = psutil.Process() + for child in current_process.children(recursive=True): + print(f"Process ID: {child.pid}") + print(f"Name: {child.name()}") + print(f"Status: {child.status()}") + print(f"CPU Times: {child.cpu_times()}") + print(f"Memory Info: {child.memory_info()}") + print() + + print("=== Current Process ===") + print(f"Process ID: {current_process.pid}") + print(f"Name: {current_process.name()}") + print(f"Status: {current_process.status()}") + print(f"CPU Times: {current_process.cpu_times()}") + print(f"Memory Info: {current_process.memory_info()}") + print() + + except Exception as e: + print(f"An error occurred: {str(e)}") + print("Traceback:") + traceback.print_exc() @pytest.hookimpl(tryfirst=True, hookwrapper=True) @@ -331,11 +331,11 @@ def pytest_sessionfinish(session, exitstatus): yield # temporarily suspend stdout capture and print detailed thread info - # capmanager = session.config.pluginmanager.get_plugin("capturemanager") - # if capmanager: - # capmanager.suspend_global_capture(in_=True) + capmanager = session.config.pluginmanager.get_plugin("capturemanager") + if capmanager: + capmanager.suspend_global_capture(in_=True) - # _print_detailed_info() + _print_detailed_info() - # if capmanager: - # capmanager.resume_global_capture() + if capmanager: + capmanager.resume_global_capture()