wehub-resource-sync
140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
"""Tests for hybrid_server non-blocking conversion.
|
|
|
|
Verifies that converter.convert() runs in a thread pool rather than blocking
|
|
the event loop. This is the root cause of issue #301: the Java client's
|
|
3-second health check times out when the server is busy with a synchronous
|
|
conversion call inside an async endpoint.
|
|
"""
|
|
|
|
import asyncio
|
|
import threading
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_docling():
|
|
"""Mock docling modules so tests don't need the actual dependency."""
|
|
mock_converter = MagicMock()
|
|
mock_result = MagicMock()
|
|
mock_result.status.value = "success"
|
|
mock_result.errors = []
|
|
mock_result.input.page_count = 1
|
|
mock_result.document.export_to_dict.return_value = {
|
|
"pages": {"1": {}},
|
|
"body": {},
|
|
}
|
|
|
|
# Track which thread the conversion runs on
|
|
convert_thread_name = {}
|
|
|
|
def tracking_convert(path, page_range=None):
|
|
convert_thread_name["thread"] = threading.current_thread().name
|
|
time.sleep(2)
|
|
return mock_result
|
|
|
|
mock_converter.convert = tracking_convert
|
|
mock_converter._convert_thread = convert_thread_name
|
|
|
|
mock_conversion_status = MagicMock()
|
|
mock_conversion_status.PARTIAL_SUCCESS = "partial_success"
|
|
|
|
with patch.dict("sys.modules", {
|
|
"docling": MagicMock(),
|
|
"docling.datamodel.base_models": MagicMock(
|
|
InputFormat=MagicMock(PDF="pdf"),
|
|
ConversionStatus=mock_conversion_status,
|
|
),
|
|
"docling.datamodel.pipeline_options": MagicMock(),
|
|
"docling.document_converter": MagicMock(),
|
|
"uvicorn": MagicMock(),
|
|
}):
|
|
yield mock_converter
|
|
|
|
|
|
@pytest.fixture
|
|
def app_with_converter(mock_docling):
|
|
"""Create a FastAPI app with the mock converter."""
|
|
import importlib
|
|
from opendataloader_pdf import hybrid_server
|
|
|
|
importlib.reload(hybrid_server)
|
|
|
|
app = hybrid_server.create_app()
|
|
hybrid_server.converter = mock_docling
|
|
return app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_convert_runs_in_thread_pool(app_with_converter, mock_docling):
|
|
"""converter.convert() must run in a worker thread, not on the event loop.
|
|
|
|
When converter.convert() runs directly on the async event loop thread,
|
|
it blocks all concurrent request handling — including /health checks.
|
|
The fix wraps converter.convert() with asyncio.to_thread() so it runs
|
|
in a separate thread.
|
|
|
|
Reproduces issue #301.
|
|
"""
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
transport = ASGITransport(app=app_with_converter)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
event_loop_thread = threading.current_thread().name
|
|
|
|
response = await client.post(
|
|
"/v1/convert/file",
|
|
files={"files": ("test.pdf", b"%PDF-1.4 minimal", "application/pdf")},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
convert_thread = mock_docling._convert_thread.get("thread")
|
|
assert convert_thread is not None, "converter.convert() was not called"
|
|
assert convert_thread != event_loop_thread, (
|
|
f"converter.convert() ran on the event loop thread '{event_loop_thread}'. "
|
|
f"It must run in a worker thread via asyncio.to_thread() to avoid "
|
|
f"blocking /health and other endpoints during long conversions."
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_responds_during_conversion(app_with_converter):
|
|
"""Health endpoint must respond quickly even during active conversion.
|
|
|
|
This is the user-facing symptom of issue #301: the Java CLI gets
|
|
SocketTimeoutException when the hybrid server is busy processing
|
|
another document. Mock sleep is 2s; health must respond under 0.2s.
|
|
"""
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
transport = ASGITransport(app=app_with_converter)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
# Start conversion (takes 2s in mock)
|
|
convert_task = asyncio.create_task(
|
|
client.post(
|
|
"/v1/convert/file",
|
|
files={"files": ("test.pdf", b"%PDF-1.4 minimal", "application/pdf")},
|
|
)
|
|
)
|
|
|
|
# Wait for conversion to start in the worker thread
|
|
await asyncio.sleep(0.3)
|
|
|
|
# Health check should respond quickly — well under the 2s conversion
|
|
start = time.monotonic()
|
|
health_response = await client.get("/health")
|
|
health_time = time.monotonic() - start
|
|
|
|
assert health_response.status_code == 200
|
|
assert health_response.json() == {"status": "ok"}
|
|
assert health_time < 0.2, (
|
|
f"Health endpoint took {health_time:.2f}s during conversion. "
|
|
f"Expected < 0.2s. The event loop is likely blocked."
|
|
)
|
|
|
|
convert_response = await convert_task
|
|
assert convert_response.status_code == 200
|