report-detect/archive/temp_scripts/test_paddleocrvl_timeout.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Test script to verify PaddleOCRVL timeout mechanism.

This script creates a simple test to ensure the multiprocessing-based
timeout protection works correctly on Windows.
"""

import multiprocessing
import time


def _run_infinite_process(result_queue):
    """Simulates a process that never finishes (like a hanging PaddleOCRVL)."""
    print("Child process: Starting infinite loop...")
    while True:
        time.sleep(1)  # Simulate a blocking call
        print("Child process: Still running...")


def _quick_process(result_queue):
    """A process that completes quickly (must be at module level for pickle)."""
    result_queue.put({"status": "success", "data": "test_data"})


def test_timeout_mechanism(timeout=5):
    """
    Test that the timeout mechanism correctly terminates a hanging process.

    Args:
        timeout: Timeout in seconds
    """
    print("=" * 80)
    print("PaddleOCRVL Timeout Mechanism Test")
    print("=" * 80)
    print(f"Testing with {timeout}s timeout...")

    result_queue = multiprocessing.Queue()

    # Start a process that will hang
    process = multiprocessing.Process(
        target=_run_infinite_process,
        args=(result_queue,)
    )
    process.start()

    print(f"Main process: Started child process (PID: {process.pid})")

    # Wait for timeout
    start_time = time.time()
    process.join(timeout=timeout)
    elapsed = time.time() - start_time

    print(f"Main process: process.join() returned after {elapsed:.1f}s")

    if process.is_alive():
        print(f"Main process: Child process is still alive (expected)")
        print(f"Main process: Terminating child process...")

        process.terminate()
        process.join(timeout=2)  # Wait up to 2 seconds for cleanup

        if process.is_alive():
            print(f"Main process: Child still alive after terminate(), killing...")
            process.kill()
            process.join(timeout=1)
        else:
            print(f"Main process: Child terminated successfully")

        print(f"Main process: Total elapsed time: {time.time() - start_time:.1f}s")
        print(f"Main process: ** TIMEOUT TEST PASSED **")
        return True
    else:
        print(f"Main process: Child process finished unexpectedly")
        print(f"Main process: ** TIMEOUT TEST FAILED **")
        return False


def test_normal_completion():
    """
    Test that normal process completion works correctly.
    """
    print("\n" + "=" * 80)
    print("Testing Normal Process Completion")
    print("=" * 80)

    result_queue = multiprocessing.Queue()
    process = multiprocessing.Process(
        target=_quick_process,
        args=(result_queue,)
    )
    process.start()
    process.join(timeout=10)

    if not process.is_alive() and not result_queue.empty():
        result = result_queue.get_nowait()
        print(f"Result: {result}")
        print("** NORMAL COMPLETION TEST PASSED **")
        return True
    else:
        print("** NORMAL COMPLETION TEST FAILED **")
        return False


def main():
    """Run all tests."""
    # Test timeout mechanism
    timeout_passed = test_timeout_mechanism(timeout=5)

    # Test normal completion
    normal_passed = test_normal_completion()

    print("\n" + "=" * 80)
    print("TEST SUMMARY")
    print("=" * 80)
    print(f"Timeout mechanism: {'PASSED' if timeout_passed else 'FAILED'}")
    print(f"Normal completion: {'PASSED' if normal_passed else 'FAILED'}")

    if timeout_passed and normal_passed:
        print("\n[OK] All tests passed! The multiprocessing timeout mechanism works correctly.")
        print("  PaddleOCRVL calls will be protected from hanging indefinitely.")
        return 0
    else:
        print("\n[FAIL] Some tests failed! Please review the implementation.")
        return 1


if __name__ == "__main__":
    exit(main())