iomgaa/PolyGateway
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 17 Wiki Activity

feat: add OcrClient.parse_document with full governance

Browse Source 
Reuses the existing retry, breaker, limiter and telemetry loop; the transport capability is checked once before the statistics boundary. Legacy entrances keep their error text. Tests cover the document path through the real transport so parser exceptions, deadlines and probe bookkeeping are proven end to end.
This commit is contained in:
iomgaa
2026-10-22 00:01:45 -04:00
parent 61cace71c4
commit 80bb8aba71
5 changed files with 524 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_ocr_client.py
+360 -2
View File
@@ -6,6 +6,7 @@ retry_exhausted/circuit_open/stalled 三组断言即设计 §6 ③ 的 G1 契约
import asyncio
import httpx
import pytest
from loguru import logger
@@ -21,20 +22,34 @@ from polygateway.errors import (
TransientError,
)
from polygateway.ocr import OcrClient
from polygateway.transports.monkey_ocr import MonkeyOcrTransport
from polygateway.types import (
BackpressurePolicy,
BreakerConfig,
GlobalLimits,
OcrDocumentElement,
OcrDocumentTransportResult,
OcrLayoutElement,
OcrLayoutTransportResult,
OcrTextTransportResult,
RetryPolicy,
SourceConfig,
)
from tests.unit import _document_fixtures as fx
_BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
_TEXT_OK = OcrTextTransportResult(text="LINE-1", raw={"task_type": "text"})
_DOC_OK = OcrDocumentTransportResult(
elements=[
OcrDocumentElement(
type="text", bbox=(1.0, 2.0, 3.0, 4.0), page_index=0, text="SYNTH-DOC-BODY"
)
],
page_sizes=[(612.0, 792.0)],
markdown="# SYNTH-MD",
raw={"success": True},
)
_LAYOUT_OK = OcrLayoutTransportResult(
elements=[OcrLayoutElement(type="table", bbox=(41.0, 48.0, 218.0, 282.0), page_index=0)],
page_sizes=[(759.0, 540.0)],
@@ -72,7 +87,11 @@ class ScriptedOcrTransport:
if action == "hang":
self.entered.set()
await asyncio.Event().wait()
return _TEXT_OK if action == "text" else _LAYOUT_OK
if action == "text":
return _TEXT_OK
if action == "document":
return _DOC_OK
return _LAYOUT_OK
async def recognize_text(self, *, image, source, call_id):
return await self._next("text", source, call_id)
@@ -80,6 +99,9 @@ class ScriptedOcrTransport:
async def parse_layout(self, *, image, source, call_id):
return await self._next("layout", source, call_id)
async def parse_document(self, *, document, source, call_id):
return await self._next("document", source, call_id)
async def check_health(self, *, source):
raise NotImplementedError
@@ -548,7 +570,7 @@ class TestOcrCallerDimensions:
assert [r["tenant_id"] for r in recorder.rows] == ["t1", "t1"]
assert [r["meta"] for r in recorder.rows] == ['{"batch": "b-42"}'] * 2
@pytest.mark.parametrize("method", ["recognize_text", "parse_layout"])
@pytest.mark.parametrize("method", ["recognize_text", "parse_layout", "parse_document"])
async def test_invalid_meta_rejected_before_any_telemetry(self, method):
"""校验必须早于遥测: 链路内的失败都被降级成 warning,放下游等于没有校验。"""
recorder = _MemoryRecorder()
@@ -558,6 +580,14 @@ class TestOcrCallerDimensions:
assert recorder.rows == []
assert client._transport.calls == [] # 连调用都没发出
async def test_parse_document_row_carries_dimensions(self):
"""第三个入口: 只测前两个会漏掉它的透传缺口。"""
recorder = _MemoryRecorder()
client, _, _ = _client([_src()], ["document"], telemetry=recorder)
await client.parse_document(b"%PDF-1.4", tenant_id="t3", meta={"batch": "b-44"})
assert recorder.rows[0]["tenant_id"] == "t3"
assert recorder.rows[0]["meta"] == '{"batch": "b-44"}'
async def test_defaults_land_as_sentinels(self):
recorder = _MemoryRecorder()
client, _, _ = _client([_src()], ["text"], telemetry=recorder)
@@ -566,6 +596,334 @@ class TestOcrCallerDimensions:
assert recorder.rows[0]["meta"] == "{}"
class _LegacyOcrTransport:
"""只实现旧 `OcrTransport` 三方法的下游 fake: 不满足 `OcrDocumentTransport`。"""
def __init__(self):
self.calls = []
async def recognize_text(self, *, image, source, call_id):
self.calls.append("text")
return _TEXT_OK
async def parse_layout(self, *, image, source, call_id):
self.calls.append("layout")
return _LAYOUT_OK
async def check_health(self, *, source):
return True
class TestParseDocument:
"""#34 设计 §4.2/§6: 与 parse_layout 对称的整份文档入口,治理循环原样复用。"""
async def test_result_fields(self):
client, _, gate = _client([_src()], ["document"])
r = await client.parse_document(b"%PDF-1.4")
assert [e.text for e in r.elements] == ["SYNTH-DOC-BODY"]
assert r.markdown == "# SYNTH-MD" and r.page_sizes == [(612.0, 792.0)]
assert r.source_name == "m1"
assert r.usage.prompt_tokens == 0 and r.usage.completion_tokens == 0
assert r.latency_ms >= 0 and r.call_id and r.raw == {"success": True}
assert r.call_stats is not None and r.call_stats.attempts == 1
assert gate.successes == [("m1", True)]
assert client._selector.outcomes == [("m1", True)]
async def test_input_validation_names_the_document(self):
client, _, _ = _client([_src()], [])
with pytest.raises(TypeError, match="document 必须是 bytes"):
await client.parse_document("not-bytes")
with pytest.raises(ValueError, match="document 不能为空"):
await client.parse_document(b"")
async def test_legacy_entrances_keep_their_error_text(self):
"""私有形参改名为 payload 不得改变旧入口的报错文案(含 D9 的业务边界提示)。"""
client, _, _ = _client([_src()], [])
with pytest.raises(TypeError, match=r"image 必须是 bytes\(路径读取/批量拼帧留业务侧,D9\)"):
await client.recognize_text("not-bytes")
with pytest.raises(ValueError, match="image 不能为空"):
await client.parse_layout(b"")
async def test_transport_without_document_support_is_an_assembly_error(self):
"""装配错误当场报,且早于统计边界: 没有遥测行、没有任何 transport 调用。"""
recorder = _MemoryRecorder()
legacy = _LegacyOcrTransport()
client, limiter, _ = _client([_src(rpm=100)], [], transport=legacy, telemetry=recorder)
with pytest.raises(TypeError, match="OcrDocumentTransport"):
await client.parse_document(b"%PDF-1.4")
assert recorder.rows == [] and legacy.calls == []
# 早于准入: 从未向限流后端申请过许可(否则 rpm 已被占用一次)
assert (await limiter.source_stats("m1")).rpm_used == 0
async def test_invalid_call_deadline_is_rejected_like_the_other_entrances(self):
client, _, _ = _client([_src()], ["document"])
with pytest.raises(ValueError, match="call_deadline_s"):
await client.parse_document(b"%PDF-1.4", call_deadline_s=-1)
async def test_transient_retries_then_succeeds(self):
client, _, _ = _client(
[_src()],
[TransientError("boom", status_code=500), "document"],
sleep=lambda _d: asyncio.sleep(0),
)
r = await client.parse_document(b"%PDF-1.4")
assert r.markdown == "# SYNTH-MD"
assert client._selector.outcomes == [("m1", False), ("m1", True)]
async def test_result_invalid_counts_no_attempt_and_no_feed(self):
client, _, gate = _client([_src()], [ResultInvalidError("bad package")])
with pytest.raises(ResultInvalidError):
await client.parse_document(b"%PDF-1.4")
assert gate.successes == [("m1", False)]
assert client._selector.outcomes == []
async def test_deterministic_rejection_with_status_500_counts_as_a_live_service(self):
"""服务有响应(500 + 确定性坏输入标记)= 服务活着: 记成功不入窗口,不喂选源器。"""
client, _, gate = _client(
[_src()], [RequestRejectedError("Parsing failed: is no PDF", status_code=500)]
)
with pytest.raises(RequestRejectedError):
await client.parse_document(b"%PDF-1.4")
assert gate.successes == [("m1", False)]
assert client._selector.outcomes == []
async def test_local_rejection_does_not_touch_the_breaker(self):
client, _, gate = _client([_src()], [RequestRejectedError("unrecognizable")])
with pytest.raises(RequestRejectedError):
await client.parse_document(b"%PDF-1.4")
assert gate.successes == []
async def test_rate_limited_responses_spend_retry_budget_but_not_breaker_failures(self):
"""429 是背压不是故障(熔断两通道均不计): 三次 429 耗尽重试后,熔断仍是 CLOSED。"""
client, _, gate = _client(
[_src()],
[TransientError("busy", status_code=429)] * 3,
sleep=lambda _d: asyncio.sleep(0),
)
with pytest.raises(AllSourcesExhausted):
await client.parse_document(b"%PDF-1.4")
entry = await gate._inner.try_enter("m1", "check")
assert entry.allowed and not entry.is_probe
async def test_read_timeouts_are_counted_as_source_failures(self):
"""对照: 文档路径的 httpx 超时仍是瞬时故障、计熔断(`timeout_s` 不是免责额度)。"""
def timeout_exc():
exc = TransientError("timed out", operation="parse")
exc.__cause__ = httpx.ReadTimeout("read timed out")
return exc
client, _, gate = _client(
[_src()], [timeout_exc() for _ in range(3)], sleep=lambda _d: asyncio.sleep(0)
)
with pytest.raises(AllSourcesExhausted) as ei:
await client.parse_document(b"%PDF-1.4")
assert ei.value.per_source_reasons == {"m1": "timeout"}
entry = await gate._inner.try_enter("m1", "check")
assert not entry.allowed # 阈值 3 的熔断已开路
async def test_cancel_releases_permit_and_probe(self):
clock = FakeClock()
inner = InMemoryGate(config=_BREAKER, now=clock)
gate = RecordingGate(inner)
transport = ScriptedOcrTransport(["hang"])
client, limiter, _ = _client(
[_src(max_concurrency=1)], [], now=clock, breaker=gate, transport=transport
)
entry = await inner.try_enter("m1", "setup")
await inner.record_failure(entry, "source_dead", True)
clock.advance(61.0) # 越过冷却 → 半开探针
task = asyncio.create_task(client.parse_document(b"%PDF-1.4"))
await transport.entered.wait()
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
assert gate.probe_releases >= 1
assert (await limiter.source_stats("m1")).inflight == 0
class TestNoLatencyJudgement:
"""长文档不会被判慢(设计 §7): 库内没有任何基于延迟的熔断/选源判定。
时钟只在 transport 内前进(`ClockAdvancingOcrTransport`),门与 client 共用同一注入钟。
probe_ttl_s 取极大值,避免把探针租约接管混进"慢"的断言。
"""
_CFG = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=1e12)
async def test_slow_success_leaves_breaker_closed_with_no_failures(self):
clock = FakeClock()
gate = RecordingGate(InMemoryGate(config=self._CFG, now=clock))
transport = ClockAdvancingOcrTransport([(100_000.0, "document")], clock)
client, _, _ = _client(
[_src(timeout_s=120.0)], [], now=clock, breaker=gate, transport=transport
)
r = await client.parse_document(b"%PDF-1.4")
assert r.call_stats.attempts == 1 and r.call_stats.generation_ms >= 100_000_000
assert gate._inner._gates["m1"].fails == 0
assert gate.successes == [("m1", True)]
async def test_slow_success_clears_pre_existing_failure_count(self):
"""前置状态不是初始 CLOSED: 已有 2 次失败(阈值 3),一次很慢的成功也只会清零。"""
clock = FakeClock()
inner = InMemoryGate(config=self._CFG, now=clock)
for _ in range(2):
entry = await inner.try_enter("m1", "setup")
await inner.record_failure(entry, "network_error", False)
assert inner._gates["m1"].fails == 2
transport = ClockAdvancingOcrTransport([(100_000.0, "document")], clock)
client, _, _ = _client(
[_src()], [], now=clock, breaker=RecordingGate(inner), transport=transport
)
await client.parse_document(b"%PDF-1.4")
assert inner._gates["m1"].fails == 0
async def test_slow_success_on_the_half_open_probe_closes_the_breaker(self):
clock = FakeClock()
inner = InMemoryGate(config=self._CFG, now=clock)
entry = await inner.try_enter("m1", "setup")
await inner.record_failure(entry, "source_dead", True) # OPEN
clock.advance(61.0) # → HALF_OPEN
transport = ClockAdvancingOcrTransport([(100_000.0, "document")], clock)
client, _, _ = _client(
[_src()], [], now=clock, breaker=RecordingGate(inner), transport=transport
)
await client.parse_document(b"%PDF-1.4")
after = await inner.try_enter("m1", "check")
assert after.allowed and not after.is_probe # 探针成功 → CLOSED,而不是被慢耗尽
class TestDocumentTelemetry:
async def test_success_row(self):
recorder = _MemoryRecorder()
client, _, _ = _client([_src()], ["document"], telemetry=recorder)
await client.parse_document(b"%PDF-RAW-BYTES")
(row,) = recorder.rows
assert row["operation"] == "parse_document"
assert "<ocr:document image_bytes=14>" in row["messages"]
assert "RAW-BYTES" not in row["messages"] # 文档字节绝不入库
assert row["response"] == "<document elements=1 pages=1 md_chars=10>"
assert "SYNTH-MD" not in row["response"] and "SYNTH-DOC-BODY" not in row["response"]
assert row["reasoning_effort"] is None # `reasoning_applies=False` 的落库形态
assert row["error"] is None and row["prompt_tokens"] == 0
async def test_summary_distinguishes_the_three_result_kinds(self):
"""新结果类型必须在 `_summarize` 里显式区分,否则会被静默套成 layout 摘要而漏记 md_chars。"""
assert OcrClient._summarize(_DOC_OK) == "<document elements=1 pages=1 md_chars=10>"
assert OcrClient._summarize(_LAYOUT_OK) == "<elements n=1 pages=1>"
assert OcrClient._summarize(_TEXT_OK) == "LINE-1"
async def test_failed_attempt_row_carries_the_class_prefix(self):
recorder = _MemoryRecorder()
client, _, _ = _client([_src()], [ResultInvalidError("bad package")], telemetry=recorder)
with pytest.raises(ResultInvalidError):
await client.parse_document(b"%PDF-1.4")
assert any(r["error"].startswith("ResultInvalidError:") for r in recorder.rows)
assert {r["operation"] for r in recorder.rows} == {"parse_document"}
def _real_transport(handler):
return MonkeyOcrTransport(
client_factory=lambda src: httpx.AsyncClient(
base_url=src.base_url, transport=httpx.MockTransport(handler)
)
)
def _doc_handler(package: bytes | None = None, *, download=None):
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/parse":
return httpx.Response(
200,
json={"success": True, "message": "ok", "download_url": "/static/doc.zip"},
)
if download is not None:
return download()
return httpx.Response(200, content=package)
return handler
class TestDocumentThroughTheRealTransport:
"""真实 `MonkeyOcrTransport` + `MockTransport` + 真实 `OcrClient`: 证明异常/期限贯穿整条链。
MockTransport 不执行 httpcore 的逐次读超时,所以本组只证明**治理期限与异常分类**,
不证明 read timeout 的真实行为(后者留给活体用例)。
"""
async def test_happy_path_end_to_end(self):
package = fx._package([fx._page([fx._text_block("SYNTH-E2E")])])
client, _, _ = _client([_src()], [], transport=_real_transport(_doc_handler(package)))
r = await client.parse_document(b"%PDF-1.4 synthetic")
assert [e.text for e in r.elements] == ["SYNTH-E2E"]
assert r.markdown == "# synth\n"
@pytest.mark.parametrize("package_kind", ["huge-int", "encrypted"])
async def test_parser_exceptions_become_result_invalid_with_full_bookkeeping(
self, package_kind
):
"""真实解析器抛出的异常(OverflowError / RuntimeError)经真实客户端: 终态遥测行已写、
半开探针已归还、熔断窗口不变——不能只用桩直接抛预制的 `ResultInvalidError`。"""
if package_kind == "huge-int":
block = fx._text_block()
block["bbox"] = [0, 0, 10**400, 1]
package = fx._package([fx._page([block])])
else:
package = fx._encrypted_package()
clock = FakeClock()
inner = InMemoryGate(
config=BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0), now=clock
)
entry = await inner.try_enter("m1", "setup")
await inner.record_failure(entry, "source_dead", True)
clock.advance(61.0) # 半开: 本次调用持有探针
recorder = _MemoryRecorder()
client, _, gate = _client(
[_src()],
[],
now=clock,
breaker=RecordingGate(inner),
telemetry=recorder,
transport=_real_transport(_doc_handler(package)),
)
with pytest.raises(ResultInvalidError):
await client.parse_document(b"%PDF-1.4 synthetic")
assert any(r["error"].startswith("ResultInvalidError:") for r in recorder.rows)
after = await inner.try_enter("m1", "check")
assert after.allowed and not after.is_probe # 探针已按"坏结果≠坏服务"结算,没有悬挂
assert inner._gates["m1"].fails == 0
async def test_sustained_slow_download_hits_the_call_deadline_not_a_transient_error(self):
"""每个分块都在 read 超时内、总时长超过 call_deadline_s: 是期限,不是瞬时故障。"""
closed = []
produced = []
async def slow_body():
try:
for _ in range(200):
await asyncio.sleep(0.01)
produced.append(1)
yield b"x" * 8
finally:
closed.append(True)
handler = _doc_handler(download=lambda: httpx.Response(200, content=slow_body()))
client, limiter, gate = _client(
[_src(max_concurrency=1, timeout_s=120.0)],
[],
transport=_real_transport(handler),
call_deadline_s=0.1,
)
with pytest.raises(CallDeadlineExceeded):
await client.parse_document(b"%PDF-1.4 synthetic")
assert len(produced) >= 2 # 确实进入了多块下载,而不是一开始就超时
assert closed == [True] # 流已关闭
assert (await limiter.source_stats("m1")).inflight == 0 # permit 释放
assert gate.successes == [] # 期限不是服务故障也不是成功
entry = await gate._inner.try_enter("m1", "check")
assert entry.allowed and not entry.is_probe # 熔断窗口不变
class TestAssembly:
_ENV = {
"OCR__MONKEY__1__BASE_URL": "http://10.77.0.20:7866",