Function run_test_with_deadlock_detection

tests/test_wrapper.py:40–136

Run a test with deadlock detection. Args: runner_exe: Path to runner.exe (loads test DLLs) test_dll: Path to test DLL timeout_seconds: Timeout in seconds (default: 20) Returns: Exit code (0 = success, non-zero = failure/timeout)

(
    runner_exe: str,
    test_dll: str,
    timeout_seconds: float = DEFAULT_TIMEOUT,
)

def run_test_with_deadlock_detection(
    runner_exe: str,
    test_dll: str,
    timeout_seconds: float = DEFAULT_TIMEOUT,
) -> int:
    """
    Run a test with deadlock detection.

    Args:
        runner_exe: Path to runner.exe (loads test DLLs)
        test_dll: Path to test DLL
        timeout_seconds: Timeout in seconds (default: 20)

    Returns:
        Exit code (0 = success, non-zero = failure/timeout)
    """
    test_name = Path(test_dll).stem

    print(f"Running test: {test_name} (timeout: {timeout_seconds}s)")

    start_time = time.time()

    # Start the test process (runner.exe loads the test DLL)
    try:
        proc = subprocess.Popen(
            [runner_exe, test_dll],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,  # Line buffered
        )
    except KeyboardInterrupt as ki:
        handle_keyboard_interrupt(ki)
        raise
    except Exception as e:
        print(f"Failed to start test: {e}", file=sys.stderr)
        return 1

    # Start background thread to read output without blocking
    output_lines = []
    reader_thread = threading.Thread(
        target=_output_reader, args=(proc, output_lines), daemon=True
    )
    reader_thread.start()

    # Monitor the process - just poll() and check timeout
    try:
        while True:
            # Check if process has finished
            returncode = proc.poll()
            if returncode is not None:
                # Process finished - wait for reader thread to drain remaining output
                reader_thread.join(timeout=2.0)

                elapsed = time.time() - start_time
                if returncode == 0:
                    print(f"✓ Test passed in {elapsed:.2f}s")
                else: