|
|
|
|
|
""" |
|
|
Test script to verify timeout handling in PDF processing |
|
|
""" |
|
|
|
|
|
import asyncio |
|
|
import signal |
|
|
from pathlib import Path |
|
|
|
|
|
def test_timeout_mechanism(): |
|
|
"""Test the timeout mechanism for PDF processing""" |
|
|
print("β±οΈ Testing timeout mechanism for PDF processing...") |
|
|
|
|
|
def timeout_handler(signum, frame): |
|
|
print("β
Timeout handler works correctly") |
|
|
raise TimeoutError("Operation timed out") |
|
|
|
|
|
|
|
|
signal.signal(signal.SIGALRM, timeout_handler) |
|
|
signal.alarm(3) |
|
|
|
|
|
try: |
|
|
|
|
|
print("Starting simulated long operation...") |
|
|
import time |
|
|
time.sleep(5) |
|
|
print("β Operation completed without timeout (unexpected)") |
|
|
except TimeoutError: |
|
|
print("β
Timeout mechanism working correctly") |
|
|
finally: |
|
|
|
|
|
signal.alarm(0) |
|
|
|
|
|
print("\nπ Timeout mechanism test completed") |
|
|
|
|
|
async def test_async_timeout(): |
|
|
"""Test async timeout handling""" |
|
|
print("\nβ±οΈ Testing async timeout handling...") |
|
|
|
|
|
try: |
|
|
|
|
|
async def long_task(): |
|
|
await asyncio.sleep(5) |
|
|
return "Task completed" |
|
|
|
|
|
|
|
|
result = await asyncio.wait_for(long_task(), timeout=3.0) |
|
|
print(f"β Task completed without timeout: {result}") |
|
|
except asyncio.TimeoutError: |
|
|
print("β
Async timeout working correctly") |
|
|
|
|
|
if __name__ == "__main__": |
|
|
test_timeout_mechanism() |
|
|
asyncio.run(test_async_timeout()) |