Merge pull request #58 from scub-france/fix/pipeline-robustness
Fix/pipeline robustness
This commit is contained in:
Pier-Jean Malandrino
GitHub
commit
24cfd567f2
8 changed files with 482 additions and 13 deletions
|
|
@ -42,11 +42,14 @@ from domain.value_objects import (
|
|||
PageElement,
|
||||
)
|
||||
from infra.bbox import to_topleft_list
|
||||
from infra.settings import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Thread lock — DoclingConverter is not thread-safe
|
||||
# Thread lock — DoclingConverter is not thread-safe.
|
||||
# Uses a timeout to prevent a frozen conversion from blocking all others.
|
||||
_converter_lock = threading.Lock()
|
||||
_LOCK_TIMEOUT = 300 # seconds — fail fast rather than wait forever
|
||||
|
||||
# US Letter page dimensions (points) — fallback when page size is unknown
|
||||
_DEFAULT_PAGE_WIDTH = 612.0
|
||||
|
|
@ -102,6 +105,7 @@ def _build_docling_converter(options: ConversionOptions) -> DoclingConverter:
|
|||
generate_page_images=options.generate_page_images,
|
||||
generate_picture_images=options.generate_picture_images,
|
||||
images_scale=options.images_scale,
|
||||
document_timeout=settings.document_timeout,
|
||||
)
|
||||
|
||||
return DoclingConverter(
|
||||
|
|
@ -114,7 +118,11 @@ def _build_docling_converter(options: ConversionOptions) -> DoclingConverter:
|
|||
def _get_default_converter() -> DoclingConverter:
|
||||
global _default_converter
|
||||
if _default_converter is None:
|
||||
_default_converter = _build_docling_converter(ConversionOptions())
|
||||
try:
|
||||
_default_converter = _build_docling_converter(ConversionOptions())
|
||||
except Exception:
|
||||
_default_converter = None
|
||||
raise
|
||||
return _default_converter
|
||||
|
||||
|
||||
|
|
@ -210,9 +218,22 @@ def _process_content_item(
|
|||
|
||||
|
||||
def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResult:
|
||||
with _converter_lock:
|
||||
acquired = _converter_lock.acquire(timeout=_LOCK_TIMEOUT)
|
||||
if not acquired:
|
||||
raise TimeoutError(
|
||||
f"Could not acquire converter lock within {_LOCK_TIMEOUT}s — "
|
||||
"a previous conversion may be frozen"
|
||||
)
|
||||
try:
|
||||
conv = _select_converter(options)
|
||||
result = conv.convert(file_path)
|
||||
kwargs: dict = {}
|
||||
if settings.max_page_count > 0:
|
||||
kwargs["max_num_pages"] = settings.max_page_count
|
||||
if settings.max_file_size > 0:
|
||||
kwargs["max_file_size"] = settings.max_file_size
|
||||
result = conv.convert(file_path, **kwargs)
|
||||
finally:
|
||||
_converter_lock.release()
|
||||
|
||||
doc = result.document
|
||||
page_count = len(doc.pages)
|
||||
|
|
|
|||
|
|
@ -14,8 +14,11 @@ class Settings:
|
|||
docling_serve_url: str = "http://localhost:5001"
|
||||
docling_serve_api_key: str | None = None
|
||||
conversion_timeout: int = 900
|
||||
document_timeout: float = 120.0 # Docling-level per-document timeout (seconds)
|
||||
max_concurrent_analyses: int = 3
|
||||
max_page_count: int = 0 # 0 = unlimited
|
||||
default_table_mode: str = "accurate" # "accurate" or "fast"
|
||||
max_page_count: int = 0 # 0 = unlimited (upload validation)
|
||||
max_file_size: int = 0 # 0 = unlimited (Docling-level, bytes)
|
||||
upload_dir: str = "./uploads"
|
||||
db_path: str = "./data/docling_studio.db"
|
||||
cors_origins: list[str] = field(
|
||||
|
|
@ -33,8 +36,11 @@ class Settings:
|
|||
docling_serve_url=os.environ.get("DOCLING_SERVE_URL", "http://localhost:5001"),
|
||||
docling_serve_api_key=os.environ.get("DOCLING_SERVE_API_KEY"),
|
||||
conversion_timeout=int(os.environ.get("CONVERSION_TIMEOUT", "900")),
|
||||
document_timeout=float(os.environ.get("DOCUMENT_TIMEOUT", "120.0")),
|
||||
max_concurrent_analyses=int(os.environ.get("MAX_CONCURRENT_ANALYSES", "3")),
|
||||
default_table_mode=os.environ.get("DEFAULT_TABLE_MODE", "accurate"),
|
||||
max_page_count=int(os.environ.get("MAX_PAGE_COUNT", "0")),
|
||||
max_file_size=int(os.environ.get("MAX_FILE_SIZE", "0")),
|
||||
upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"),
|
||||
db_path=os.environ.get("DB_PATH", "./data/docling_studio.db"),
|
||||
cors_origins=[o.strip() for o in cors_raw.split(",")],
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ from typing import TYPE_CHECKING
|
|||
|
||||
from domain.models import AnalysisJob, AnalysisStatus
|
||||
from domain.value_objects import ChunkingOptions, ChunkResult, ConversionOptions, ConversionResult
|
||||
from infra.settings import settings
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from domain.ports import DocumentChunker, DocumentConverter
|
||||
|
|
@ -151,7 +152,10 @@ class AnalysisService:
|
|||
await analysis_repo.update_status(job)
|
||||
logger.info("Analysis started: %s (file: %s)", job_id, filename)
|
||||
|
||||
options = ConversionOptions(**(pipeline_options or {}))
|
||||
opts_dict = pipeline_options or {}
|
||||
if "table_mode" not in opts_dict:
|
||||
opts_dict = {**opts_dict, "table_mode": settings.default_table_mode}
|
||||
options = ConversionOptions(**opts_dict)
|
||||
|
||||
result: ConversionResult = await asyncio.wait_for(
|
||||
self._converter.convert(file_path, options),
|
||||
|
|
@ -187,7 +191,33 @@ class AnalysisService:
|
|||
|
||||
except Exception as e:
|
||||
logger.exception("Analysis failed: %s", job_id)
|
||||
await _mark_failed(job_id, str(e))
|
||||
await _mark_failed(job_id, _classify_error(e))
|
||||
|
||||
|
||||
def _classify_error(exc: Exception) -> str:
|
||||
"""Return a user-friendly error message based on the exception type/content."""
|
||||
msg = str(exc).lower()
|
||||
|
||||
if "invalidcxxcompiler" in msg or "no working c++ compiler" in msg:
|
||||
return "Missing C++ compiler — set TORCHDYNAMO_DISABLE=1 to work around this"
|
||||
|
||||
if "out of memory" in msg or "oom" in msg:
|
||||
return "Out of memory — try a smaller document or disable table structure analysis"
|
||||
|
||||
if "could not acquire converter lock" in msg:
|
||||
return "Server busy — a previous conversion is still running. Please retry later"
|
||||
|
||||
if "pipeline" in msg and "failed" in msg:
|
||||
return "Document processing failed — the document may be corrupted or unsupported"
|
||||
|
||||
if "timeout" in msg:
|
||||
return "Processing took too long — try with fewer pages or simpler options"
|
||||
|
||||
# Fallback: truncate raw error to something reasonable
|
||||
raw = str(exc)
|
||||
if len(raw) > 200:
|
||||
raw = raw[:200] + "…"
|
||||
return raw
|
||||
|
||||
|
||||
_background_tasks: set[asyncio.Task] = set()
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@ class TestBuildConverter:
|
|||
assert opts.generate_page_images is False
|
||||
assert opts.generate_picture_images is False
|
||||
assert opts.images_scale == 1.0
|
||||
assert opts.document_timeout is not None
|
||||
|
||||
def test_ocr_disabled(self):
|
||||
conv = build_converter(ConversionOptions(do_ocr=False))
|
||||
|
|
|
|||
369
document-parser/tests/test_robustness.py
Normal file
369
document-parser/tests/test_robustness.py
Normal file
|
|
@ -0,0 +1,369 @@
|
|||
"""Robustness tests — verify pipeline failure scenarios and protective mechanisms.
|
||||
|
||||
Tests cover: document_timeout (C1), lock timeout (C2), convert limits (C3),
|
||||
default table_mode (H1), converter reset on failure (H5), error classification (M1).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from domain.models import AnalysisJob
|
||||
from domain.value_objects import ConversionOptions, ConversionResult, PageDetail
|
||||
from infra.settings import Settings
|
||||
from services.analysis_service import AnalysisService, _classify_error
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# M1 — _classify_error: user-friendly error messages
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestClassifyError:
|
||||
"""Verify _classify_error maps exceptions to user-friendly messages."""
|
||||
|
||||
def test_cxx_compiler_error(self):
|
||||
exc = RuntimeError("InvalidCxxCompiler: No working C++ compiler found")
|
||||
assert "Missing C++ compiler" in _classify_error(exc)
|
||||
|
||||
def test_no_working_compiler(self):
|
||||
exc = RuntimeError("no working c++ compiler found in torch")
|
||||
assert "Missing C++ compiler" in _classify_error(exc)
|
||||
|
||||
def test_out_of_memory(self):
|
||||
exc = MemoryError("Out of memory allocating tensor")
|
||||
assert "Out of memory" in _classify_error(exc)
|
||||
|
||||
def test_oom_shorthand(self):
|
||||
exc = RuntimeError("OOM during inference on page 5")
|
||||
assert "Out of memory" in _classify_error(exc)
|
||||
|
||||
def test_lock_timeout(self):
|
||||
exc = TimeoutError("Could not acquire converter lock within 300s")
|
||||
assert "Server busy" in _classify_error(exc)
|
||||
|
||||
def test_pipeline_failed(self):
|
||||
exc = RuntimeError("Pipeline StandardPdfPipeline failed on page 3")
|
||||
assert "Document processing failed" in _classify_error(exc)
|
||||
|
||||
def test_timeout_generic(self):
|
||||
exc = TimeoutError("timeout exceeded while processing")
|
||||
assert "Processing took too long" in _classify_error(exc)
|
||||
|
||||
def test_unknown_short_error(self):
|
||||
exc = ValueError("something weird happened")
|
||||
assert _classify_error(exc) == "something weird happened"
|
||||
|
||||
def test_unknown_long_error_truncated(self):
|
||||
long_msg = "x" * 300
|
||||
exc = ValueError(long_msg)
|
||||
result = _classify_error(exc)
|
||||
assert len(result) <= 201
|
||||
assert result.endswith("…")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# H1 — default table_mode injection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDefaultTableMode:
|
||||
"""Verify _run_analysis_inner injects settings.default_table_mode."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_job(self):
|
||||
return AnalysisJob(id="j1", document_id="d1", document_filename="test.pdf")
|
||||
|
||||
def _make_result(self) -> ConversionResult:
|
||||
return ConversionResult(
|
||||
page_count=1,
|
||||
content_markdown="# Test",
|
||||
content_html="<h1>Test</h1>",
|
||||
pages=[PageDetail(page_number=1, width=612.0, height=792.0)],
|
||||
)
|
||||
|
||||
@patch("services.analysis_service.analysis_repo")
|
||||
@patch("services.analysis_service.document_repo")
|
||||
@pytest.mark.asyncio
|
||||
async def test_default_table_mode_injected_when_missing(
|
||||
self, mock_doc_repo, mock_analysis_repo, mock_job
|
||||
):
|
||||
mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job)
|
||||
mock_analysis_repo.update_status = AsyncMock()
|
||||
mock_doc_repo.update_page_count = AsyncMock()
|
||||
|
||||
mock_converter = AsyncMock()
|
||||
mock_converter.convert.return_value = self._make_result()
|
||||
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", {})
|
||||
|
||||
opts = mock_converter.convert.call_args[0][1]
|
||||
assert opts.table_mode == "accurate"
|
||||
|
||||
@patch("services.analysis_service.analysis_repo")
|
||||
@patch("services.analysis_service.document_repo")
|
||||
@pytest.mark.asyncio
|
||||
async def test_default_table_mode_injected_when_none(
|
||||
self, mock_doc_repo, mock_analysis_repo, mock_job
|
||||
):
|
||||
mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job)
|
||||
mock_analysis_repo.update_status = AsyncMock()
|
||||
mock_doc_repo.update_page_count = AsyncMock()
|
||||
|
||||
mock_converter = AsyncMock()
|
||||
mock_converter.convert.return_value = self._make_result()
|
||||
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", None)
|
||||
|
||||
opts = mock_converter.convert.call_args[0][1]
|
||||
assert opts.table_mode == "accurate"
|
||||
|
||||
@patch("services.analysis_service.analysis_repo")
|
||||
@patch("services.analysis_service.document_repo")
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_table_mode_preserved(self, mock_doc_repo, mock_analysis_repo, mock_job):
|
||||
mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job)
|
||||
mock_analysis_repo.update_status = AsyncMock()
|
||||
mock_doc_repo.update_page_count = AsyncMock()
|
||||
|
||||
mock_converter = AsyncMock()
|
||||
mock_converter.convert.return_value = self._make_result()
|
||||
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", {"table_mode": "fast"})
|
||||
|
||||
opts = mock_converter.convert.call_args[0][1]
|
||||
assert opts.table_mode == "fast"
|
||||
|
||||
@patch("services.analysis_service.settings", Settings(default_table_mode="fast"))
|
||||
@patch("services.analysis_service.analysis_repo")
|
||||
@patch("services.analysis_service.document_repo")
|
||||
@pytest.mark.asyncio
|
||||
async def test_custom_default_from_settings(self, mock_doc_repo, mock_analysis_repo, mock_job):
|
||||
mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job)
|
||||
mock_analysis_repo.update_status = AsyncMock()
|
||||
mock_doc_repo.update_page_count = AsyncMock()
|
||||
|
||||
mock_converter = AsyncMock()
|
||||
mock_converter.convert.return_value = self._make_result()
|
||||
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", {})
|
||||
|
||||
opts = mock_converter.convert.call_args[0][1]
|
||||
assert opts.table_mode == "fast"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Docling-dependent tests — skip if docling is not installed
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
docling = pytest.importorskip("docling", reason="docling library not installed")
|
||||
|
||||
from docling.datamodel.base_models import InputFormat # noqa: E402
|
||||
|
||||
import infra.local_converter as lc_mod # noqa: E402
|
||||
|
||||
build_converter = lc_mod._build_docling_converter
|
||||
convert_sync = lc_mod._convert_sync
|
||||
get_default_converter = lc_mod._get_default_converter
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# C1 — document_timeout in PdfPipelineOptions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDocumentTimeout:
|
||||
"""Verify document_timeout from settings is wired into PdfPipelineOptions."""
|
||||
|
||||
def _get_pipeline_options(self, converter):
|
||||
fmt_opt = converter.format_to_options[InputFormat.PDF]
|
||||
return fmt_opt.pipeline_options
|
||||
|
||||
def test_document_timeout_from_settings(self):
|
||||
conv = build_converter(ConversionOptions())
|
||||
opts = self._get_pipeline_options(conv)
|
||||
assert opts.document_timeout == 120.0
|
||||
|
||||
@patch("infra.local_converter.settings", Settings(document_timeout=45.0))
|
||||
def test_custom_document_timeout(self):
|
||||
conv = build_converter(ConversionOptions())
|
||||
opts = self._get_pipeline_options(conv)
|
||||
assert opts.document_timeout == 45.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# C2 — converter lock timeout
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _mock_docling_result():
|
||||
"""Create a minimal mock that satisfies _convert_sync's result processing."""
|
||||
mock_result = MagicMock()
|
||||
mock_result.document.pages = {}
|
||||
mock_result.document.iterate_items.return_value = []
|
||||
mock_result.document.export_to_markdown.return_value = ""
|
||||
mock_result.document.export_to_html.return_value = ""
|
||||
mock_result.document.export_to_dict.return_value = {}
|
||||
return mock_result
|
||||
|
||||
|
||||
class TestConverterLockTimeout:
|
||||
"""Verify _convert_sync raises TimeoutError when lock cannot be acquired."""
|
||||
|
||||
@patch("infra.local_converter._converter_lock")
|
||||
def test_lock_timeout_raises(self, mock_lock):
|
||||
mock_lock.acquire.return_value = False
|
||||
|
||||
with pytest.raises(TimeoutError, match="Could not acquire converter lock"):
|
||||
convert_sync("/tmp/test.pdf", ConversionOptions())
|
||||
|
||||
mock_lock.release.assert_not_called()
|
||||
|
||||
@patch("infra.local_converter._select_converter")
|
||||
@patch("infra.local_converter._converter_lock")
|
||||
def test_lock_released_on_success(self, mock_lock, mock_select):
|
||||
mock_lock.acquire.return_value = True
|
||||
mock_conv = MagicMock()
|
||||
mock_conv.convert.return_value = _mock_docling_result()
|
||||
mock_select.return_value = mock_conv
|
||||
|
||||
convert_sync("/tmp/test.pdf", ConversionOptions())
|
||||
|
||||
mock_lock.release.assert_called_once()
|
||||
|
||||
@patch("infra.local_converter._select_converter")
|
||||
@patch("infra.local_converter._converter_lock")
|
||||
def test_lock_released_on_error(self, mock_lock, mock_select):
|
||||
mock_lock.acquire.return_value = True
|
||||
mock_conv = MagicMock()
|
||||
mock_conv.convert.side_effect = RuntimeError("boom")
|
||||
mock_select.return_value = mock_conv
|
||||
|
||||
with pytest.raises(RuntimeError, match="boom"):
|
||||
convert_sync("/tmp/test.pdf", ConversionOptions())
|
||||
|
||||
mock_lock.release.assert_called_once()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# C3 — max_num_pages and max_file_size forwarding
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestConvertSyncLimits:
|
||||
"""Verify _convert_sync forwards limit settings to conv.convert()."""
|
||||
|
||||
@patch("infra.local_converter._select_converter")
|
||||
@patch("infra.local_converter._converter_lock")
|
||||
@patch("infra.local_converter.settings", Settings(max_page_count=10, max_file_size=0))
|
||||
def test_max_page_count_forwarded(self, mock_lock, mock_select):
|
||||
mock_lock.acquire.return_value = True
|
||||
mock_conv = MagicMock()
|
||||
mock_conv.convert.return_value = _mock_docling_result()
|
||||
mock_select.return_value = mock_conv
|
||||
|
||||
convert_sync("/tmp/test.pdf", ConversionOptions())
|
||||
|
||||
kwargs = mock_conv.convert.call_args.kwargs
|
||||
assert kwargs["max_num_pages"] == 10
|
||||
assert "max_file_size" not in kwargs
|
||||
|
||||
@patch("infra.local_converter._select_converter")
|
||||
@patch("infra.local_converter._converter_lock")
|
||||
@patch("infra.local_converter.settings", Settings(max_page_count=0, max_file_size=5_000_000))
|
||||
def test_max_file_size_forwarded(self, mock_lock, mock_select):
|
||||
mock_lock.acquire.return_value = True
|
||||
mock_conv = MagicMock()
|
||||
mock_conv.convert.return_value = _mock_docling_result()
|
||||
mock_select.return_value = mock_conv
|
||||
|
||||
convert_sync("/tmp/test.pdf", ConversionOptions())
|
||||
|
||||
kwargs = mock_conv.convert.call_args.kwargs
|
||||
assert kwargs["max_file_size"] == 5_000_000
|
||||
assert "max_num_pages" not in kwargs
|
||||
|
||||
@patch("infra.local_converter._select_converter")
|
||||
@patch("infra.local_converter._converter_lock")
|
||||
@patch(
|
||||
"infra.local_converter.settings",
|
||||
Settings(max_page_count=20, max_file_size=10_000_000),
|
||||
)
|
||||
def test_both_limits_forwarded(self, mock_lock, mock_select):
|
||||
mock_lock.acquire.return_value = True
|
||||
mock_conv = MagicMock()
|
||||
mock_conv.convert.return_value = _mock_docling_result()
|
||||
mock_select.return_value = mock_conv
|
||||
|
||||
convert_sync("/tmp/test.pdf", ConversionOptions())
|
||||
|
||||
kwargs = mock_conv.convert.call_args.kwargs
|
||||
assert kwargs["max_num_pages"] == 20
|
||||
assert kwargs["max_file_size"] == 10_000_000
|
||||
|
||||
@patch("infra.local_converter._select_converter")
|
||||
@patch("infra.local_converter._converter_lock")
|
||||
@patch("infra.local_converter.settings", Settings(max_page_count=0, max_file_size=0))
|
||||
def test_zero_limits_not_forwarded(self, mock_lock, mock_select):
|
||||
mock_lock.acquire.return_value = True
|
||||
mock_conv = MagicMock()
|
||||
mock_conv.convert.return_value = _mock_docling_result()
|
||||
mock_select.return_value = mock_conv
|
||||
|
||||
convert_sync("/tmp/test.pdf", ConversionOptions())
|
||||
|
||||
kwargs = mock_conv.convert.call_args.kwargs
|
||||
assert "max_num_pages" not in kwargs
|
||||
assert "max_file_size" not in kwargs
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# H5 — _default_converter reset on init failure
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetDefaultConverterReset:
|
||||
"""Verify _get_default_converter resets on failure and retries on next call."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_default_converter(self):
|
||||
original = lc_mod._default_converter
|
||||
lc_mod._default_converter = None
|
||||
yield
|
||||
lc_mod._default_converter = original
|
||||
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_init_failure_resets_to_none(self, mock_build):
|
||||
mock_build.side_effect = RuntimeError("torch not found")
|
||||
|
||||
with pytest.raises(RuntimeError, match="torch not found"):
|
||||
get_default_converter()
|
||||
|
||||
assert lc_mod._default_converter is None
|
||||
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_retry_after_failure_succeeds(self, mock_build):
|
||||
mock_conv = MagicMock()
|
||||
mock_build.side_effect = [RuntimeError("torch not found"), mock_conv]
|
||||
|
||||
with pytest.raises(RuntimeError):
|
||||
get_default_converter()
|
||||
|
||||
result = get_default_converter()
|
||||
assert result is mock_conv
|
||||
assert mock_build.call_count == 2
|
||||
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_success_caches_converter(self, mock_build):
|
||||
mock_conv = MagicMock()
|
||||
mock_build.return_value = mock_conv
|
||||
|
||||
result1 = get_default_converter()
|
||||
result2 = get_default_converter()
|
||||
|
||||
assert result1 is result2 is mock_conv
|
||||
mock_build.assert_called_once()
|
||||
|
|
@ -14,6 +14,7 @@ class TestSettingsDefaults:
|
|||
assert s.docling_serve_url == "http://localhost:5001"
|
||||
assert s.docling_serve_api_key is None
|
||||
assert s.conversion_timeout == 900
|
||||
assert s.document_timeout == 120.0
|
||||
assert s.max_page_count == 0
|
||||
assert s.upload_dir == "./uploads"
|
||||
assert s.db_path == "./data/docling_studio.db"
|
||||
|
|
@ -36,6 +37,7 @@ class TestSettingsFromEnv:
|
|||
monkeypatch.setenv("DOCLING_SERVE_URL", "http://serve:9000")
|
||||
monkeypatch.setenv("DOCLING_SERVE_API_KEY", "secret-key")
|
||||
monkeypatch.setenv("CONVERSION_TIMEOUT", "120")
|
||||
monkeypatch.setenv("DOCUMENT_TIMEOUT", "60.0")
|
||||
monkeypatch.setenv("MAX_PAGE_COUNT", "20")
|
||||
monkeypatch.setenv("UPLOAD_DIR", "/data/uploads")
|
||||
monkeypatch.setenv("DB_PATH", "/data/test.db")
|
||||
|
|
@ -49,6 +51,7 @@ class TestSettingsFromEnv:
|
|||
assert s.docling_serve_url == "http://serve:9000"
|
||||
assert s.docling_serve_api_key == "secret-key"
|
||||
assert s.conversion_timeout == 120
|
||||
assert s.document_timeout == 60.0
|
||||
assert s.max_page_count == 20
|
||||
assert s.upload_dir == "/data/uploads"
|
||||
assert s.db_path == "/data/test.db"
|
||||
|
|
|
|||
|
|
@ -154,17 +154,49 @@ describe('useAnalysisStore', () => {
|
|||
expect(store.currentAnalysis.status).toBe('FAILED')
|
||||
})
|
||||
|
||||
it('polling stops on fetch error', async () => {
|
||||
it('polling retries on transient errors and stops after MAX_POLL_RETRIES', async () => {
|
||||
const job = { id: 'j1', status: 'PENDING', documentId: 'd1' }
|
||||
api.createAnalysis.mockResolvedValue(job)
|
||||
api.fetchAnalysis.mockRejectedValue(new Error('network'))
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
vi.spyOn(console, 'error').mockImplementation(() => {})
|
||||
|
||||
const store = useAnalysisStore()
|
||||
await store.run('d1')
|
||||
|
||||
// First two errors: still polling
|
||||
await vi.advanceTimersByTimeAsync(2000)
|
||||
expect(store.running).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(2000)
|
||||
expect(store.running).toBe(true)
|
||||
|
||||
// Third error: stops polling
|
||||
await vi.advanceTimersByTimeAsync(2000)
|
||||
expect(store.running).toBe(false)
|
||||
})
|
||||
|
||||
it('polling resets error count on successful fetch', async () => {
|
||||
const job = { id: 'j1', status: 'PENDING', documentId: 'd1' }
|
||||
api.createAnalysis.mockResolvedValue(job)
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
|
||||
// Fail once, succeed, fail once — should NOT stop polling
|
||||
api.fetchAnalysis
|
||||
.mockRejectedValueOnce(new Error('network'))
|
||||
.mockResolvedValueOnce({ ...job, status: 'RUNNING' })
|
||||
.mockRejectedValueOnce(new Error('network'))
|
||||
.mockResolvedValueOnce({ ...job, status: 'COMPLETED' })
|
||||
|
||||
const store = useAnalysisStore()
|
||||
await store.run('d1')
|
||||
|
||||
await vi.advanceTimersByTimeAsync(2000) // error 1
|
||||
expect(store.running).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(2000) // success — resets counter
|
||||
expect(store.running).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(2000) // error 1 again
|
||||
expect(store.running).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(2000) // success — COMPLETED
|
||||
expect(store.running).toBe(false)
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -10,7 +10,8 @@ export const useAnalysisStore = defineStore('analysis', () => {
|
|||
const error = ref<string | null>(null)
|
||||
const pollingInterval = ref<ReturnType<typeof setInterval> | null>(null)
|
||||
const pollingTimeout = ref<ReturnType<typeof setTimeout> | null>(null)
|
||||
const MAX_POLLING_DURATION = 10 * 60 * 1000 // 10 minutes
|
||||
const MAX_POLLING_DURATION = 15 * 60 * 1000 // 15 minutes — aligned with backend timeout
|
||||
const MAX_POLL_RETRIES = 3
|
||||
|
||||
const currentPages = computed<Page[]>(() => {
|
||||
if (!currentAnalysis.value?.pagesJson) return []
|
||||
|
|
@ -87,9 +88,11 @@ export const useAnalysisStore = defineStore('analysis', () => {
|
|||
|
||||
function startPolling(id: string): void {
|
||||
stopPolling()
|
||||
let consecutiveErrors = 0
|
||||
pollingInterval.value = setInterval(async () => {
|
||||
try {
|
||||
const updated = await api.fetchAnalysis(id)
|
||||
consecutiveErrors = 0
|
||||
currentAnalysis.value = updated
|
||||
const idx = analyses.value.findIndex((a) => a.id === id)
|
||||
if (idx !== -1) analyses.value[idx] = updated
|
||||
|
|
@ -98,10 +101,14 @@ export const useAnalysisStore = defineStore('analysis', () => {
|
|||
running.value = false
|
||||
}
|
||||
} catch (e) {
|
||||
error.value = (e as Error).message || 'Polling error'
|
||||
console.error('Polling error', e)
|
||||
stopPolling()
|
||||
running.value = false
|
||||
consecutiveErrors++
|
||||
console.warn(`Polling error (${consecutiveErrors}/${MAX_POLL_RETRIES})`, e)
|
||||
if (consecutiveErrors >= MAX_POLL_RETRIES) {
|
||||
error.value = (e as Error).message || 'Polling error'
|
||||
console.error('Polling abandoned after retries', e)
|
||||
stopPolling()
|
||||
running.value = false
|
||||
}
|
||||
}
|
||||
}, 2000)
|
||||
pollingTimeout.value = setTimeout(() => {
|
||||
|
|
|
|||
Loading…
Reference in a new issue