feat: add OCR document types and ports

OcrDocumentElement extends the layout element with text; OcrDocumentResult stays independent because the parent's defaulted call_stats forbids a required markdown field. OcrDocumentTransport is a separate Protocol so existing OcrTransport implementers keep passing isinstance.
This commit is contained in:
2026-10-21 23:20:53 -04:00
parent b31df842a8
commit 398b16e136
6 changed files with 185 additions and 3 deletions
+2
View File
@@ -22,6 +22,8 @@ def test_ocr_public_surface_exported():
"OcrTextResult",
"OcrLayoutResult",
"OcrLayoutElement",
"OcrDocumentResult",
"OcrDocumentElement",
):
assert hasattr(polygateway, name), name
assert name in polygateway.__all__, name
+30
View File
@@ -12,6 +12,7 @@ from polygateway.ports import (
GateState,
GateUpdate,
Middleware,
OcrDocumentTransport,
OcrTransport,
Permit,
ProviderGate,
@@ -193,6 +194,7 @@ class TestReasoningTierIsOnlyOnTheChatPort:
(EmbeddingTransport, "embed"),
(OcrTransport, "recognize_text"),
(OcrTransport, "parse_layout"),
(OcrDocumentTransport, "parse_document"),
],
)
def test_other_transports_have_no_reasoning_tier(self, protocol, method):
@@ -372,6 +374,34 @@ class TestOcrPorts:
assert isinstance(GoodClient(), OcrLayoutPort)
assert isinstance(GoodTransport(), OcrTransport)
def test_document_ports_are_runtime_checkable(self):
from polygateway.ports import OcrDocumentPort, OcrDocumentTransport
class GoodClient:
async def parse_document(self, document): ...
class GoodTransport:
async def parse_document(self, *, document, source, call_id): ...
class Empty: ...
assert isinstance(GoodClient(), OcrDocumentPort)
assert isinstance(GoodTransport(), OcrDocumentTransport)
assert not isinstance(Empty(), OcrDocumentPort)
assert not isinstance(Empty(), OcrDocumentTransport)
def test_ocr_transport_members_are_unchanged(self):
"""新能力走独立 Protocol: 往旧 runtime_checkable Protocol 加成员,会让所有
现有实现(测试替身、下游自带 fake)在 isinstance 上一夜之间不合格。"""
from polygateway.ports import OcrTransport
members = {
name
for name in vars(OcrTransport)
if not name.startswith("_") and callable(getattr(OcrTransport, name))
}
assert members == {"recognize_text", "parse_layout", "check_health"}
def test_missing_method_rejected(self):
from polygateway.ports import OcrTransport
+63
View File
@@ -6,6 +6,7 @@ import inspect
import pytest
from polygateway.types import (
CALL_OPERATIONS,
USAGE_SOURCES,
BackpressurePolicy,
BreakerConfig,
@@ -13,6 +14,10 @@ from polygateway.types import (
Effort,
GlobalLimits,
LLMResponse,
OcrDocumentElement,
OcrDocumentResult,
OcrDocumentTransportResult,
OcrLayoutElement,
RetryPolicy,
SourceConfig,
ThinkingObservation,
@@ -849,3 +854,61 @@ class _FakeMonotonic:
def advance(self, seconds: float) -> None:
self.t += seconds
class TestOcrDocumentTypes:
"""#34 设计 §4.1: 文档元素是版面元素 + 内容;结果独立定义。"""
@staticmethod
def _element(**overrides):
base = {"type": "text", "bbox": (0.0, 0.0, 1.0, 1.0), "page_index": 0, "text": "x"}
base.update(overrides)
return OcrDocumentElement(**base)
def test_document_element_is_a_layout_element(self):
"""继承的价值: 按 bbox 裁剪的下游函数可以同时吃两种元素。"""
assert isinstance(self._element(), OcrLayoutElement)
def test_document_element_never_equals_a_layout_element(self):
"""dataclass 的 `__eq__` 要求同类——已知语义,钉死以免下游拿版面元素去比对文档结果。"""
layout = OcrLayoutElement(type="text", bbox=(0.0, 0.0, 1.0, 1.0), page_index=0)
assert self._element() != layout
def test_new_types_are_frozen(self):
element = self._element()
with pytest.raises(dataclasses.FrozenInstanceError):
element.text = "y" # type: ignore[misc]
transport = OcrDocumentTransportResult(elements=[], page_sizes=[], markdown="", raw={})
with pytest.raises(dataclasses.FrozenInstanceError):
transport.markdown = "y" # type: ignore[misc]
def test_result_is_not_a_subclass_of_layout_result(self):
"""父类末尾 `call_stats` 带默认值,子类无默认字段 `markdown` 会在导入期 TypeError。"""
from polygateway.types import OcrLayoutResult
assert not issubclass(OcrDocumentResult, OcrLayoutResult)
def test_result_carries_provenance_five_plus_markdown(self):
fields = [f.name for f in dataclasses.fields(OcrDocumentResult)]
assert fields == [
"elements",
"page_sizes",
"markdown",
"source_name",
"usage",
"latency_ms",
"call_id",
"raw",
"call_stats",
]
class TestCallOperations:
def test_call_operations_matches_the_literal(self):
"""`CALL_OPERATIONS` 此前没有任何读者,与 `CallOperation` 漂移也无人发现。"""
from typing import get_args
from polygateway.types import CallOperation
assert get_args(CallOperation) == CALL_OPERATIONS
assert "parse_document" in CALL_OPERATIONS