Chris CoutinhoandClaude Opus 4.8
4dbf362261
OCR is an enhancement, not a gate. Previously, escalating a scanned doc to the OCR tier returned the OCR result unconditionally -- so with DOCUMENT_OCR_ENABLED =true but no backend configured (no gateway URL / no MISTRAL_API_KEY) the OCR processor returned success=False and the whole document was marked failed and skipped: strictly worse than leaving OCR off (where it would at least index the tier-1 text). Now the registry keeps the tier-1 fast result when the OCR escalation doesn't succeed (no backend, API down, empty output), logging a warning. A misconfiguration degrades gracefully instead of dropping scanned docs. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
143 lines
4.7 KiB
Python
143 lines
4.7 KiB
Python
"""Unit tests for the tiered PDF routing in ProcessorRegistry.
|
|
|
|
Covers: default fast-tier routing, the pymupdf rollback toggle, classification
|
|
recording derived from the extraction, and OCR escalation (on/off).
|
|
"""
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from nextcloud_mcp_server.document_processors import registry as reg_mod
|
|
from nextcloud_mcp_server.document_processors.base import (
|
|
DocumentProcessor,
|
|
ProcessingResult,
|
|
)
|
|
from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
class _Fake(DocumentProcessor):
|
|
def __init__(
|
|
self, name: str, tier: str, text: str = "clean text here", success=True
|
|
):
|
|
self._name = name
|
|
self._tier = tier
|
|
self._text = text
|
|
self._success = success
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@property
|
|
def tier(self) -> str:
|
|
return self._tier
|
|
|
|
@property
|
|
def supported_mime_types(self) -> set[str]:
|
|
return {"application/pdf"}
|
|
|
|
async def process(
|
|
self, content, content_type, filename=None, options=None, progress_callback=None
|
|
):
|
|
return ProcessingResult(
|
|
text=self._text,
|
|
metadata={
|
|
"page_count": 1,
|
|
"page_boundaries": [
|
|
{"page": 1, "start_offset": 0, "end_offset": len(self._text)}
|
|
],
|
|
},
|
|
processor=self._name,
|
|
success=self._success,
|
|
)
|
|
|
|
async def health_check(self) -> bool:
|
|
return True
|
|
|
|
|
|
class _Settings:
|
|
def __init__(self, engine="pypdfium2", classify=True, ocr=False):
|
|
self.document_tier1_engine = engine
|
|
self.document_classify_enabled = classify
|
|
self.document_ocr_enabled = ocr
|
|
|
|
|
|
def _registry(*procs: tuple[DocumentProcessor, int]) -> ProcessorRegistry:
|
|
r = ProcessorRegistry()
|
|
for proc, prio in procs:
|
|
r.register(proc, priority=prio)
|
|
return r
|
|
|
|
|
|
async def test_pdf_routes_to_fast_tier(monkeypatch):
|
|
monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings())
|
|
r = _registry((_Fake("fast", "fast"), 20), (_Fake("structured", "structured"), 10))
|
|
res = await r.process(b"%PDF-1.7", "application/pdf")
|
|
assert res.processor == "fast"
|
|
|
|
|
|
async def test_engine_rollback_uses_structured(monkeypatch):
|
|
monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(engine="pymupdf"))
|
|
r = _registry((_Fake("fast", "fast"), 20), (_Fake("structured", "structured"), 10))
|
|
res = await r.process(b"%PDF-1.7", "application/pdf")
|
|
assert res.processor == "structured"
|
|
|
|
|
|
async def test_records_classification(monkeypatch):
|
|
monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings())
|
|
rec = MagicMock()
|
|
monkeypatch.setattr(reg_mod, "record_document_classification", rec)
|
|
r = _registry((_Fake("fast", "fast"), 20))
|
|
await r.process(b"%PDF-1.7", "application/pdf")
|
|
rec.assert_called_once()
|
|
|
|
|
|
async def test_classify_disabled_skips_recording(monkeypatch):
|
|
monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(classify=False))
|
|
rec = MagicMock()
|
|
monkeypatch.setattr(reg_mod, "record_document_classification", rec)
|
|
r = _registry((_Fake("fast", "fast"), 20))
|
|
await r.process(b"%PDF-1.7", "application/pdf")
|
|
rec.assert_not_called()
|
|
|
|
|
|
async def test_ocr_escalation_on_empty_text(monkeypatch):
|
|
monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(ocr=True))
|
|
esc = MagicMock()
|
|
monkeypatch.setattr(reg_mod, "record_document_escalation", esc)
|
|
r = _registry(
|
|
(_Fake("fast", "fast", text=""), 20),
|
|
(_Fake("ocr", "ocr", text="ocr text"), 5),
|
|
)
|
|
res = await r.process(b"%PDF-1.7", "application/pdf")
|
|
assert res.processor == "ocr"
|
|
esc.assert_called_once()
|
|
|
|
|
|
async def test_ocr_failure_falls_back_to_fast(monkeypatch):
|
|
# OCR enabled but the backend can't run (no creds / API down) -> keep the
|
|
# tier-1 result instead of failing the document.
|
|
monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(ocr=True))
|
|
monkeypatch.setattr(reg_mod, "record_document_escalation", MagicMock())
|
|
r = _registry(
|
|
(_Fake("fast", "fast", text=""), 20),
|
|
(_Fake("ocr", "ocr", text="", success=False), 5),
|
|
)
|
|
res = await r.process(b"%PDF-1.7", "application/pdf")
|
|
assert res.processor == "fast"
|
|
assert res.success is True
|
|
|
|
|
|
async def test_no_ocr_escalation_when_disabled(monkeypatch):
|
|
monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(ocr=False))
|
|
r = _registry(
|
|
(_Fake("fast", "fast", text=""), 20),
|
|
(_Fake("ocr", "ocr"), 5),
|
|
)
|
|
res = await r.process(b"%PDF-1.7", "application/pdf")
|
|
# Fast tier is terminal when OCR is disabled.
|
|
assert res.processor == "fast"
|