iomgaa
61cace71c4
Unrecognizable bytes are rejected before any request. The result package is downloaded as a stream with a cumulative size cap, and error statuses are classified before the size check. Parsing runs in a worker thread so cancellation and call deadlines stay effective. parse_layout shares the first protocol stage without behavior change.
828 lines
34 KiB
Python
828 lines
34 KiB
Python
"""MonkeyOcrTransport 测试(M3 设计 §4): 双端点协议、两段解析、数值防御、错误翻译。
|
|
|
|
fixtures 按 2026-07-21 真实服务取证响应**脱敏二次构造**(设计 §11):
|
|
保留结构骨架/数值/元素类型,OCR 识别文本一律替换为合成占位(零业务假设 +
|
|
P5 医疗数据不入库)。
|
|
"""
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
import threading
|
|
import zipfile
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from polygateway.errors import (
|
|
RequestRejectedError,
|
|
ResultInvalidError,
|
|
SourceDeadError,
|
|
TransientError,
|
|
)
|
|
from polygateway.transports import monkey_ocr
|
|
from polygateway.transports.monkey_ocr import (
|
|
_DEFAULT_REJECTED_MARKERS,
|
|
MonkeyOcrTransport,
|
|
_classify_status,
|
|
_parse_middle_json,
|
|
_sniff_media,
|
|
_translate_http_errors,
|
|
_upload_part,
|
|
)
|
|
from polygateway.types import OcrDocumentElement, OcrDocumentTransportResult, SourceConfig
|
|
from tests.unit._document_fixtures import _package as _doc_package
|
|
from tests.unit._document_fixtures import _page as _doc_page
|
|
from tests.unit._document_fixtures import _table_block, _text_block
|
|
|
|
|
|
def _source(**overrides):
|
|
base = {
|
|
"name": "monkey_1",
|
|
"provider": "monkey",
|
|
"base_url": "http://10.77.0.20:7866",
|
|
"api_key": "none", # MonkeyOCR 无鉴权,占位惯例(CHS .env 同款)
|
|
"model": "monkey-ocr",
|
|
"timeout_s": 120.0,
|
|
}
|
|
base.update(overrides)
|
|
return SourceConfig(**base)
|
|
|
|
|
|
def _text_body(content="LINE-1\nLINE-2"):
|
|
# 真实形态: {"success":true,"task_type":"text","content":...,"message":...}
|
|
return {
|
|
"success": True,
|
|
"task_type": "text",
|
|
"content": content,
|
|
"message": "Text extraction completed successfully",
|
|
}
|
|
|
|
|
|
def _parse_body(download_url="/static/sample_parsed_1.zip"):
|
|
# 真实形态含 output_dir/files;解析只消费 success 与 download_url
|
|
return {
|
|
"success": True,
|
|
"message": "image parsing (standard) completed successfully",
|
|
"output_dir": "/app/tmp/x",
|
|
"files": ["u_middle.json"],
|
|
"download_url": download_url,
|
|
}
|
|
|
|
|
|
def _page(para_blocks, page_size=(759.0, 540.0), page_idx=0):
|
|
return {
|
|
"page_idx": page_idx,
|
|
"page_size": list(page_size),
|
|
"para_blocks": para_blocks,
|
|
"tables": [b for b in para_blocks if b.get("type") == "table"],
|
|
}
|
|
|
|
|
|
def _block(type_, bbox):
|
|
body = {"type": type_, "bbox": list(bbox), "index": 0}
|
|
if type_ == "text":
|
|
body["lines"] = [{"spans": [{"content": "LINE-SYNTH"}]}]
|
|
return body
|
|
|
|
|
|
_DEFAULT_PAGES = [
|
|
_page(
|
|
[
|
|
_block("table", (41, 48, 218, 282)),
|
|
_block("image", (240, 30, 700, 500)),
|
|
_block("text", (50, 300, 200, 340)),
|
|
]
|
|
)
|
|
]
|
|
|
|
|
|
def _middle_bytes(pages=_DEFAULT_PAGES, top=None):
|
|
payload = {"pdf_info": pages, "_parse_type": "ocr", "_version_name": "0.0.1"}
|
|
if top is not None:
|
|
payload = top
|
|
return json.dumps(payload).encode()
|
|
|
|
|
|
def _zip_bytes(middle=None, member="sample_middle.json"):
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
zf.writestr(member, middle if middle is not None else _middle_bytes())
|
|
zf.writestr("sample.md", "# synth")
|
|
return buf.getvalue()
|
|
|
|
|
|
def _transport_for(handler):
|
|
mock = httpx.MockTransport(handler)
|
|
return MonkeyOcrTransport(
|
|
client_factory=lambda src: httpx.AsyncClient(base_url=src.base_url, transport=mock)
|
|
)
|
|
|
|
|
|
def _routes(text_resp=None, parse_resp=None, zip_resp=None, health_resp=None):
|
|
"""按路径分发的 mock handler;None 项返回 500 便于暴露误路由。"""
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
path = request.url.path
|
|
if path == "/ocr/text":
|
|
return text_resp or httpx.Response(500)
|
|
if path == "/parse":
|
|
return parse_resp or httpx.Response(500)
|
|
if path.startswith("/static/"):
|
|
return zip_resp or httpx.Response(500)
|
|
if path == "/health":
|
|
return health_resp or httpx.Response(500)
|
|
return httpx.Response(500)
|
|
|
|
return handler
|
|
|
|
|
|
class TestRecognizeText:
|
|
async def test_normal(self):
|
|
t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body())))
|
|
r = await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
assert r.text == "LINE-1\nLINE-2"
|
|
assert r.raw["task_type"] == "text"
|
|
|
|
async def test_empty_content_legal(self):
|
|
t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body(""))))
|
|
r = await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
assert r.text == ""
|
|
|
|
async def test_content_missing_result_invalid(self):
|
|
body = _text_body()
|
|
del body["content"]
|
|
t = _transport_for(_routes(text_resp=httpx.Response(200, json=body)))
|
|
with pytest.raises(ResultInvalidError):
|
|
await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
|
|
async def test_content_non_str_result_invalid(self):
|
|
t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body(123))))
|
|
with pytest.raises(ResultInvalidError):
|
|
await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
|
|
async def test_multipart_shape(self):
|
|
seen = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen["content_type"] = request.headers.get("content-type", "")
|
|
seen["body"] = request.read()
|
|
return httpx.Response(200, json=_text_body())
|
|
|
|
t = _transport_for(handler)
|
|
await t.recognize_text(image=b"IMAGE-BYTES", source=_source(), call_id="c1")
|
|
assert seen["content_type"].startswith("multipart/form-data")
|
|
assert b"IMAGE-BYTES" in seen["body"]
|
|
assert b'filename="image.jpg"' in seen["body"] # CHS invokers.py:496 固定名
|
|
|
|
|
|
_PDF_BODY = b"%PDF-1.4\n%synthetic body"
|
|
_JUNK = b"x" * 200
|
|
|
|
|
|
class TestSniffMedia:
|
|
"""上传类型按内容判定(#34 设计 §3 Q2): 图片魔数优先 → 前 1KB 内 `%PDF-` → 未知。"""
|
|
|
|
@pytest.mark.parametrize(
|
|
("data", "expected"),
|
|
[
|
|
pytest.param(_PDF_BODY, "pdf", id="pdf"),
|
|
pytest.param(_JUNK + _PDF_BODY, "pdf", id="pdf-with-junk-prefix"),
|
|
pytest.param(b"\xff\xd8\xff\xe0" + b"0" * 20, "image", id="jpeg"),
|
|
pytest.param(b"\x89PNG\r\n\x1a\n" + b"0" * 20, "image", id="png"),
|
|
pytest.param(b"II*\x00" + b"0" * 20, "image", id="tiff-le"),
|
|
pytest.param(b"MM\x00*" + b"0" * 20, "image", id="tiff-be"),
|
|
pytest.param(b"BM" + b"0" * 20, "image", id="bmp"),
|
|
pytest.param(b"GIF89a" + b"0" * 20, "image", id="gif89a"),
|
|
pytest.param(b"GIF87a" + b"0" * 20, "image", id="gif87a"),
|
|
pytest.param(b"RIFF\x00\x00\x00\x00WEBPVP8 ", "image", id="webp"),
|
|
# 图片魔数优先: 含 XMP 元数据的图片可能在 1KB 内带 `%PDF-` 字样
|
|
pytest.param(
|
|
b"\xff\xd8\xff\xe0" + b"<xmp>%PDF-</xmp>", "image", id="jpeg-mentions-pdf"
|
|
),
|
|
pytest.param(b"<html><body>paywall</body></html>", "unknown", id="html"),
|
|
pytest.param(b"GARBAGE-NOT-AN-IMAGE", "unknown", id="garbage"),
|
|
pytest.param(b"", "unknown", id="empty"),
|
|
# 1KB 窗口之外的 `%PDF-` 不算
|
|
pytest.param(b"x" * 1100 + b"%PDF-", "unknown", id="header-beyond-window"),
|
|
],
|
|
)
|
|
def test_sniff(self, data, expected):
|
|
assert _sniff_media(data) == expected
|
|
|
|
def test_upload_part_labels_pdf_only(self):
|
|
assert _upload_part(_PDF_BODY) == ("doc.pdf", _PDF_BODY, "application/pdf")
|
|
assert _upload_part(_JUNK + _PDF_BODY)[0] == "doc.pdf"
|
|
|
|
@pytest.mark.parametrize(
|
|
"data", [b"\xff\xd8\xff\xe0abc", b"\x89PNG\r\n\x1a\nabc", b"junk", b""]
|
|
)
|
|
def test_everything_else_keeps_the_legacy_label(self, data):
|
|
"""非 PDF 路径字节级不变: 服务对图片按内容解码(2026-10-21 实测 PNG/TIFF/BMP 标 image.jpg 成功)。"""
|
|
assert _upload_part(data) == ("image.jpg", data, "image/jpeg")
|
|
|
|
|
|
class TestUploadLabelPerMethod:
|
|
"""三个调用点共用同一个标签点: 漏掉任何一个,E4 的静默错误结果就在那里复活。"""
|
|
|
|
@staticmethod
|
|
def _capturing_transport(seen):
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/ocr/text":
|
|
seen["body"] = request.read()
|
|
return httpx.Response(200, json=_text_body())
|
|
if request.url.path == "/parse":
|
|
seen["body"] = request.read()
|
|
return httpx.Response(200, json=_parse_body())
|
|
return httpx.Response(200, content=_zip_bytes())
|
|
|
|
return _transport_for(handler)
|
|
|
|
async def test_recognize_text_labels_pdf_as_pdf(self):
|
|
seen = {}
|
|
t = self._capturing_transport(seen)
|
|
await t.recognize_text(image=_PDF_BODY, source=_source(), call_id="c1")
|
|
assert b'filename="doc.pdf"' in seen["body"]
|
|
assert b"Content-Type: application/pdf" in seen["body"]
|
|
|
|
async def test_parse_layout_labels_pdf_as_pdf(self):
|
|
seen = {}
|
|
t = self._capturing_transport(seen)
|
|
await t.parse_layout(image=_PDF_BODY, source=_source(), call_id="c1")
|
|
assert b'filename="doc.pdf"' in seen["body"]
|
|
assert b"Content-Type: application/pdf" in seen["body"]
|
|
|
|
async def test_parse_layout_keeps_image_label_for_images(self):
|
|
seen = {}
|
|
t = self._capturing_transport(seen)
|
|
await t.parse_layout(image=b"\xff\xd8\xff\xe0abc", source=_source(), call_id="c1")
|
|
assert b'filename="image.jpg"' in seen["body"]
|
|
assert b"Content-Type: image/jpeg" in seen["body"]
|
|
|
|
|
|
class TestParseLayout:
|
|
async def test_full_chain_relative_url(self):
|
|
t = _transport_for(
|
|
_routes(
|
|
parse_resp=httpx.Response(200, json=_parse_body()),
|
|
zip_resp=httpx.Response(200, content=_zip_bytes()),
|
|
)
|
|
)
|
|
r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
|
|
types = [e.type for e in r.elements]
|
|
assert types == ["table", "image", "text"]
|
|
assert r.elements[0].bbox == (41.0, 48.0, 218.0, 282.0)
|
|
assert r.elements[0].page_index == 0
|
|
assert r.page_sizes == [(759.0, 540.0)]
|
|
|
|
async def test_absolute_download_url(self):
|
|
url = "http://10.77.0.20:7866/static/abs.zip"
|
|
t = _transport_for(
|
|
_routes(
|
|
parse_resp=httpx.Response(200, json=_parse_body(url)),
|
|
zip_resp=httpx.Response(200, content=_zip_bytes()),
|
|
)
|
|
)
|
|
r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
|
|
assert len(r.elements) == 3
|
|
|
|
async def test_success_false_rejected_with_status_200(self):
|
|
body = {"success": False, "message": "unsupported"}
|
|
t = _transport_for(_routes(parse_resp=httpx.Response(200, json=body)))
|
|
with pytest.raises(RequestRejectedError) as ei:
|
|
await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
|
|
# 设计 §10.2 有意修复: 200 响应确证服务活着,熔断按"响应即健康"记成功
|
|
assert ei.value.status_code == 200
|
|
|
|
async def test_download_url_missing_transient(self):
|
|
body = _parse_body()
|
|
del body["download_url"]
|
|
t = _transport_for(_routes(parse_resp=httpx.Response(200, json=body)))
|
|
with pytest.raises(TransientError):
|
|
await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
|
|
|
|
async def test_non_json_response_transient(self):
|
|
t = _transport_for(_routes(parse_resp=httpx.Response(200, content=b"<html>")))
|
|
with pytest.raises(TransientError):
|
|
await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
|
|
|
|
async def test_no_table_page_legal(self):
|
|
pages = [_page([_block("text", (1, 2, 30, 40))])]
|
|
t = _transport_for(
|
|
_routes(
|
|
parse_resp=httpx.Response(200, json=_parse_body()),
|
|
zip_resp=httpx.Response(200, content=_zip_bytes(_middle_bytes(pages))),
|
|
)
|
|
)
|
|
r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
|
|
assert [e.type for e in r.elements] == ["text"] # 无 table = 合法,不抛
|
|
|
|
|
|
class TestMiddleJsonDefense:
|
|
"""数值防御纯函数(CHS invokers.py:427-479 全量下沉)。"""
|
|
|
|
def _expect_invalid(self, zip_bytes):
|
|
with pytest.raises(ResultInvalidError):
|
|
_parse_middle_json(zip_bytes)
|
|
|
|
def test_bad_zip(self):
|
|
self._expect_invalid(b"not-a-zip")
|
|
|
|
def test_missing_middle_member(self):
|
|
self._expect_invalid(_zip_bytes(member="other.json"))
|
|
|
|
def test_pdf_info_not_list(self):
|
|
self._expect_invalid(_zip_bytes(_middle_bytes(top={"pdf_info": {}})))
|
|
|
|
def test_middle_not_json(self):
|
|
self._expect_invalid(_zip_bytes(b"{broken"))
|
|
|
|
@pytest.mark.parametrize(
|
|
"page_size", [[-1, 540], [float("nan"), 540], [True, 540], [759], "bad"]
|
|
)
|
|
def test_page_size_invalid(self, page_size):
|
|
page = _page([_block("table", (1, 2, 3, 4))])
|
|
page["page_size"] = page_size
|
|
self._expect_invalid(_zip_bytes(_middle_bytes([page])))
|
|
|
|
def test_bbox_order_invalid(self):
|
|
self._expect_invalid(
|
|
_zip_bytes(_middle_bytes([_page([_block("table", (218, 48, 41, 282))])]))
|
|
)
|
|
|
|
def test_bbox_int_degenerate(self):
|
|
# float 合法但 int() 后宽度为零: 专为 CHS shim 的裁剪路径兜底
|
|
self._expect_invalid(
|
|
_zip_bytes(_middle_bytes([_page([_block("table", (1.2, 1.2, 1.8, 5))])]))
|
|
)
|
|
|
|
def test_bbox_non_finite(self):
|
|
self._expect_invalid(
|
|
_zip_bytes(_middle_bytes([_page([_block("table", (1, 2, float("inf"), 4))])]))
|
|
)
|
|
|
|
def test_type_missing(self):
|
|
block = {"bbox": [1, 2, 30, 40], "index": 0}
|
|
self._expect_invalid(_zip_bytes(_middle_bytes([_page([block])])))
|
|
|
|
def test_para_blocks_absent_defaults_empty(self):
|
|
page = _page([])
|
|
del page["para_blocks"]
|
|
elements, page_sizes = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
|
|
assert elements == [] and page_sizes == [(759.0, 540.0)]
|
|
|
|
def test_page_idx_missing_falls_back_to_enumeration(self):
|
|
page = _page([_block("text", (1, 2, 30, 40))])
|
|
del page["page_idx"]
|
|
elements, _ = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
|
|
assert elements[0].page_index == 0
|
|
|
|
@pytest.mark.parametrize("bad_idx", [True, -3, "2", 1.5])
|
|
def test_page_idx_non_int_falls_back(self, bad_idx):
|
|
# bool/负数/非 int 一律回退枚举序(与 _finite_number 拒 bool 的防御口径一致)
|
|
page = _page([_block("text", (1, 2, 30, 40))])
|
|
page["page_idx"] = bad_idx
|
|
elements, _ = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
|
|
assert elements[0].page_index == 0
|
|
|
|
|
|
class TestErrorTranslation:
|
|
@pytest.mark.parametrize(
|
|
("status", "exc_type"),
|
|
[
|
|
(502, TransientError),
|
|
(429, TransientError),
|
|
(401, SourceDeadError),
|
|
(403, SourceDeadError),
|
|
(404, RequestRejectedError),
|
|
],
|
|
)
|
|
async def test_http_status(self, status, exc_type):
|
|
t = _transport_for(_routes(text_resp=httpx.Response(status)))
|
|
with pytest.raises(exc_type) as ei:
|
|
await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
assert ei.value.status_code == status
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "exc_type"),
|
|
[
|
|
(502, TransientError),
|
|
(429, TransientError), # 与 5xx 共用分支,但仍单列: 漏分支正是 issue #10 的成因
|
|
(401, SourceDeadError),
|
|
(404, RequestRejectedError),
|
|
],
|
|
)
|
|
async def test_body_survives_every_branch(self, status, exc_type):
|
|
"""issue #10: OCR 侧 message 原本只有 HTTP 状态码,拒绝理由同样丢失。"""
|
|
body = '{"detail":"unsupported image mode CMYK"}'
|
|
t = _transport_for(_routes(text_resp=httpx.Response(status, content=body.encode())))
|
|
with pytest.raises(exc_type) as ei:
|
|
await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
assert ei.value.body_text == body
|
|
assert str(ei.value).endswith(f" | {body}")
|
|
|
|
def test_unread_body_degrades_without_changing_class(self):
|
|
"""取不到 body 时降级空串: 绝不能让 ResponseNotRead 逃出错误四分类。
|
|
|
|
直接测纯函数而非走 MockTransport——真实客户端对非 stream 请求总会读完
|
|
响应,未读态只可能在将来给 OCR 加 stream 时出现,而那正是要防的场景。
|
|
"""
|
|
|
|
class _Unread(httpx.SyncByteStream):
|
|
def __iter__(self):
|
|
yield b"body"
|
|
|
|
exc = httpx.HTTPStatusError(
|
|
"404",
|
|
request=httpx.Request("POST", "http://ocr.example/ocr/text"),
|
|
response=httpx.Response(404, stream=_Unread()),
|
|
)
|
|
err = _classify_status(exc, "monkey_1", "text")
|
|
assert isinstance(err, RequestRejectedError)
|
|
assert err.body_text == ""
|
|
assert str(err) == "monkey_1 OCR text HTTP 404"
|
|
|
|
async def test_connect_error_transient(self):
|
|
def handler(request):
|
|
raise httpx.ConnectError("refused", request=request)
|
|
|
|
t = _transport_for(handler)
|
|
with pytest.raises(TransientError) as ei:
|
|
await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
assert ei.value.status_code is None
|
|
|
|
async def test_read_timeout_transient_with_cause(self):
|
|
def handler(request):
|
|
raise httpx.ReadTimeout("slow", request=request)
|
|
|
|
t = _transport_for(handler)
|
|
with pytest.raises(TransientError) as ei:
|
|
await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
# __cause__ 链保留,RetryMW._failure_reason 依赖它归类 timeout
|
|
assert isinstance(ei.value.__cause__, httpx.TimeoutException)
|
|
|
|
|
|
def _http_error(status, body):
|
|
request = httpx.Request("POST", "http://ocr.example/parse")
|
|
return httpx.HTTPStatusError(
|
|
str(status),
|
|
request=request,
|
|
response=httpx.Response(status, content=body.encode(), request=request),
|
|
)
|
|
|
|
|
|
def _parsing_failed(reason):
|
|
# 真实形态(2026-10-21 实测): 服务把一切异常包成 500 + {"detail": "Parsing failed: …"}
|
|
return json.dumps({"detail": f"Parsing failed: {reason}"})
|
|
|
|
|
|
class TestRejectedMarkers:
|
|
"""#34 设计 §6: 确定性坏输入的 500 在文档路径归「请求被拒」,其余分类一概不动。
|
|
|
|
共享的 `_classify_status` 默认行为字节级不变——医院线对坏图的重试/熔断行为属公共行为,
|
|
推广须另行设计(issue #35)。
|
|
"""
|
|
|
|
@pytest.mark.parametrize("marker", _DEFAULT_REJECTED_MARKERS)
|
|
def test_known_deterministic_failures_become_rejected_with_status_500(self, marker):
|
|
exc = _http_error(500, _parsing_failed(marker.capitalize()))
|
|
err = _classify_status(exc, "monkey_1", "parse", _DEFAULT_REJECTED_MARKERS)
|
|
assert isinstance(err, RequestRejectedError)
|
|
assert err.status_code == 500 # 服务有响应: 熔断按「响应即健康」记成功
|
|
assert marker.capitalize() in err.body_text # 响应体摘要仍随该分支
|
|
|
|
def test_marker_matching_is_case_insensitive(self):
|
|
exc = _http_error(500, '{"detail": "PARSING FAILED: FAILED TO OPEN STREAM"}')
|
|
err = _classify_status(exc, "monkey_1", "parse", _DEFAULT_REJECTED_MARKERS)
|
|
assert isinstance(err, RequestRejectedError)
|
|
|
|
@pytest.mark.parametrize(
|
|
"body",
|
|
[
|
|
pytest.param('{"detail": "Failed to open stream"}', id="no-parsing-failed-prefix"),
|
|
pytest.param(
|
|
_parsing_failed("500: Model not initialized"), id="real-fault-same-prefix"
|
|
),
|
|
pytest.param(
|
|
_parsing_failed("400: Unsupported file type: .webp"), id="unsupported-type"
|
|
),
|
|
pytest.param(_parsing_failed("CUDA out of memory"), id="oom"),
|
|
pytest.param("", id="empty-body"),
|
|
],
|
|
)
|
|
def test_everything_else_stays_transient(self, body):
|
|
"""`Parsing failed:` 前缀不能单独当判据: 服务把真故障也包成同一前缀。
|
|
`Unsupported file type` 有意不在白名单——嗅探后它只可能是契约变化或我方标签 bug。"""
|
|
err = _classify_status(
|
|
_http_error(500, body), "monkey_1", "parse", _DEFAULT_REJECTED_MARKERS
|
|
)
|
|
assert isinstance(err, TransientError)
|
|
|
|
@pytest.mark.parametrize("body", [_parsing_failed(m) for m in _DEFAULT_REJECTED_MARKERS])
|
|
def test_default_call_without_markers_is_unchanged(self, body):
|
|
err = _classify_status(_http_error(500, body), "monkey_1", "parse")
|
|
assert isinstance(err, TransientError)
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "exc_type"),
|
|
[
|
|
(429, TransientError),
|
|
(502, TransientError),
|
|
(503, TransientError),
|
|
(401, SourceDeadError),
|
|
(403, SourceDeadError),
|
|
(404, RequestRejectedError),
|
|
],
|
|
)
|
|
def test_other_statuses_are_never_narrowed(self, status, exc_type):
|
|
body = _parsing_failed(_DEFAULT_REJECTED_MARKERS[0])
|
|
err = _classify_status(_http_error(status, body), "m", "parse", _DEFAULT_REJECTED_MARKERS)
|
|
assert type(err) is exc_type
|
|
|
|
def test_translate_http_errors_passes_markers_through(self):
|
|
with (
|
|
pytest.raises(RequestRejectedError),
|
|
_translate_http_errors("monkey_1", "parse", rejected_markers=_DEFAULT_REJECTED_MARKERS),
|
|
):
|
|
raise _http_error(500, _parsing_failed(_DEFAULT_REJECTED_MARKERS[0]))
|
|
# 默认不传: 行为不变
|
|
with pytest.raises(TransientError), _translate_http_errors("monkey_1", "parse"):
|
|
raise _http_error(500, _parsing_failed(_DEFAULT_REJECTED_MARKERS[0]))
|
|
|
|
def test_markers_are_configurable_per_transport(self):
|
|
assert MonkeyOcrTransport()._rejected_markers == _DEFAULT_REJECTED_MARKERS
|
|
custom = ("synthetic marker",)
|
|
assert MonkeyOcrTransport(deterministic_failure_markers=custom)._rejected_markers == custom
|
|
|
|
async def test_layout_and_text_paths_do_not_apply_the_markers(self):
|
|
"""传输实例持有标记,但只有文档路径使用;医院线的两个方法行为不变。"""
|
|
body = _parsing_failed(_DEFAULT_REJECTED_MARKERS[0]).encode()
|
|
t = _transport_for(
|
|
_routes(
|
|
text_resp=httpx.Response(500, content=body),
|
|
parse_resp=httpx.Response(500, content=body),
|
|
)
|
|
)
|
|
with pytest.raises(TransientError):
|
|
await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
|
|
with pytest.raises(TransientError):
|
|
await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
|
|
|
|
|
|
class TestParseDocument:
|
|
"""#34 设计 §5: 本地预检 → POST /parse → 流式 GET(体积上限)→ 线程内解析。"""
|
|
|
|
@staticmethod
|
|
def _doc_zip():
|
|
return _doc_package(
|
|
[_doc_page([_text_block("SYNTH-A"), _table_block("<table>SYNTH</table>")])]
|
|
)
|
|
|
|
def _transport(self, seen=None, *, zip_resp=None, parse_resp=None):
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if seen is not None:
|
|
seen.append((request.method, request.url.path, request.read()))
|
|
if request.url.path == "/parse":
|
|
return parse_resp or httpx.Response(200, json=_parse_body())
|
|
if request.url.path.startswith("/static/"):
|
|
return zip_resp or httpx.Response(200, content=self._doc_zip())
|
|
return httpx.Response(500)
|
|
|
|
return _transport_for(handler)
|
|
|
|
async def test_full_chain_relative_download_url(self):
|
|
t = self._transport()
|
|
result = await t.parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
assert isinstance(result, OcrDocumentTransportResult)
|
|
assert [(e.type, e.text) for e in result.elements] == [
|
|
("text", "SYNTH-A"),
|
|
("table", "<table>SYNTH</table>"),
|
|
]
|
|
assert all(isinstance(e, OcrDocumentElement) for e in result.elements)
|
|
assert result.page_sizes == [(612.0, 792.0)]
|
|
assert result.markdown == "# synth\n"
|
|
assert result.raw == {"success": True, "download_url": "/static/sample_parsed_1.zip"}
|
|
|
|
async def test_absolute_download_url(self):
|
|
url = "http://10.77.0.20:7866/static/abs.zip"
|
|
t = self._transport(parse_resp=httpx.Response(200, json=_parse_body(url)))
|
|
result = await t.parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
assert result.raw["download_url"] == url
|
|
|
|
@pytest.mark.parametrize(
|
|
("document", "filename", "content_type"),
|
|
[
|
|
pytest.param(_PDF_BODY, b"doc.pdf", b"application/pdf", id="pdf"),
|
|
pytest.param(_JUNK + _PDF_BODY, b"doc.pdf", b"application/pdf", id="pdf-junk-prefix"),
|
|
pytest.param(b"\xff\xd8\xff\xe0abc", b"image.jpg", b"image/jpeg", id="jpeg"),
|
|
],
|
|
)
|
|
async def test_upload_label_matrix_for_the_document_entry(
|
|
self, document, filename, content_type
|
|
):
|
|
seen = []
|
|
await self._transport(seen).parse_document(
|
|
document=document, source=_source(), call_id="c1"
|
|
)
|
|
post_body = next(body for method, path, body in seen if path == "/parse")
|
|
assert b'filename="' + filename + b'"' in post_body
|
|
assert b"Content-Type: " + content_type in post_body
|
|
|
|
@pytest.mark.parametrize(
|
|
"document",
|
|
[b"<html>paywall</html>", b"GARBAGE-NOT-AN-IMAGE", b""],
|
|
ids=["html", "garbage", "empty"],
|
|
)
|
|
async def test_unrecognizable_bytes_are_rejected_locally_without_any_request(self, document):
|
|
seen = []
|
|
with pytest.raises(RequestRejectedError) as ei:
|
|
await self._transport(seen).parse_document(
|
|
document=document, source=_source(), call_id="c1"
|
|
)
|
|
assert seen == [] # 服务从未被打扰
|
|
assert ei.value.status_code is None # 服务未响应: 熔断侧只归还探针,不计成败
|
|
|
|
@pytest.mark.parametrize("marker", _DEFAULT_REJECTED_MARKERS)
|
|
async def test_known_deterministic_post_failures_are_rejected_with_status_500(self, marker):
|
|
body = _parsing_failed(marker).encode()
|
|
t = self._transport(parse_resp=httpx.Response(500, content=body))
|
|
with pytest.raises(RequestRejectedError) as ei:
|
|
await t.parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
assert ei.value.status_code == 500 and ei.value.body_text
|
|
|
|
@pytest.mark.parametrize(
|
|
"reason",
|
|
["CUDA out of memory", "400: Unsupported file type: .webp", "500: Model not initialized"],
|
|
)
|
|
async def test_unknown_post_500_stays_transient(self, reason):
|
|
t = self._transport(
|
|
parse_resp=httpx.Response(500, content=_parsing_failed(reason).encode())
|
|
)
|
|
with pytest.raises(TransientError):
|
|
await t.parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
|
|
async def test_success_false_is_rejected_with_status_200(self):
|
|
t = self._transport(
|
|
parse_resp=httpx.Response(200, json={"success": False, "message": "no"})
|
|
)
|
|
with pytest.raises(RequestRejectedError) as ei:
|
|
await t.parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
assert ei.value.status_code == 200
|
|
|
|
async def test_missing_download_url_and_non_json_are_transient(self):
|
|
body = {"success": True, "message": "ok"}
|
|
for resp in (httpx.Response(200, json=body), httpx.Response(200, content=b"not json")):
|
|
t = self._transport(parse_resp=resp)
|
|
with pytest.raises(TransientError):
|
|
await t.parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "exc_type"),
|
|
[
|
|
(401, SourceDeadError),
|
|
(429, TransientError),
|
|
(500, TransientError),
|
|
(404, RequestRejectedError),
|
|
],
|
|
)
|
|
async def test_download_errors_keep_their_class_and_body(self, status, exc_type):
|
|
"""GET 阶段不适用确定性窄化,且响应体摘要随每个分支(issue #10)。"""
|
|
body = _parsing_failed(_DEFAULT_REJECTED_MARKERS[0])
|
|
t = self._transport(zip_resp=httpx.Response(status, content=body.encode()))
|
|
with pytest.raises(exc_type) as ei:
|
|
await t.parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
assert ei.value.status_code == status and ei.value.body_text == body
|
|
|
|
async def test_error_status_with_oversized_content_length_is_still_classified_by_http(
|
|
self, monkeypatch
|
|
):
|
|
"""结果包资源检查不得改变错误分类: 超限的 Content-Length 出现在 401 上,仍是 SourceDead。"""
|
|
monkeypatch.setattr(monkey_ocr, "_MAX_PACKAGE_BYTES", 100)
|
|
resp = httpx.Response(401, content=b"denied", headers={"content-length": "101"})
|
|
with pytest.raises(SourceDeadError):
|
|
await self._transport(zip_resp=resp).parse_document(
|
|
document=_PDF_BODY, source=_source(), call_id="c1"
|
|
)
|
|
|
|
async def test_oversized_declared_content_length_is_rejected(self, monkeypatch):
|
|
monkeypatch.setattr(monkey_ocr, "_MAX_PACKAGE_BYTES", 100)
|
|
resp = httpx.Response(200, content=b"x" * 10, headers={"content-length": "101"})
|
|
with pytest.raises(ResultInvalidError):
|
|
await self._transport(zip_resp=resp).parse_document(
|
|
document=_PDF_BODY, source=_source(), call_id="c1"
|
|
)
|
|
|
|
async def test_oversized_streamed_body_is_rejected_by_cumulative_count(self, monkeypatch):
|
|
"""没有 Content-Length(分块传输)时靠累计字节兜底: 多个小块合起来超限。"""
|
|
monkeypatch.setattr(monkey_ocr, "_MAX_PACKAGE_BYTES", 100)
|
|
|
|
async def chunks():
|
|
for _ in range(3):
|
|
yield b"x" * 60
|
|
|
|
resp = httpx.Response(200, content=chunks())
|
|
with pytest.raises(ResultInvalidError):
|
|
await self._transport(zip_resp=resp).parse_document(
|
|
document=_PDF_BODY, source=_source(), call_id="c1"
|
|
)
|
|
|
|
async def test_bad_package_is_result_invalid(self):
|
|
t = self._transport(zip_resp=httpx.Response(200, content=b"not a zip"))
|
|
with pytest.raises(ResultInvalidError):
|
|
await t.parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
|
|
async def test_cancellation_during_parse_does_not_block_the_event_loop(self, monkeypatch):
|
|
"""解析在线程里跑: 同步解析大对象会占住事件循环,取消与 call_deadline_s 都会失效。
|
|
|
|
线程内用 threading.Event 阻塞,并用 call_soon_threadsafe 通知(线程里不能直接用
|
|
asyncio.Event)。若实现退回同步解析,桩会把循环占满 1 秒后正常返回,
|
|
`task.cancel()` 就取不到 CancelledError——用例转红。
|
|
"""
|
|
loop = asyncio.get_running_loop()
|
|
entered = asyncio.Event()
|
|
release = threading.Event()
|
|
|
|
def blocking_parse(zip_bytes):
|
|
loop.call_soon_threadsafe(entered.set)
|
|
release.wait(timeout=1)
|
|
return [], [(1.0, 1.0)], ""
|
|
|
|
monkeypatch.setattr(monkey_ocr, "parse_document_package", blocking_parse)
|
|
task = asyncio.create_task(
|
|
self._transport().parse_document(document=_PDF_BODY, source=_source(), call_id="c1")
|
|
)
|
|
try:
|
|
await asyncio.wait_for(entered.wait(), timeout=2)
|
|
await asyncio.sleep(0.01) # 线程还卡着,循环必须仍能调度别的协程
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
finally:
|
|
release.set() # 让线程收尾,不泄漏
|
|
|
|
|
|
class TestCheckHealth:
|
|
async def test_healthy(self):
|
|
t = _transport_for(_routes(health_resp=httpx.Response(200, json={"status": "healthy"})))
|
|
assert await t.check_health(source=_source()) is True
|
|
|
|
async def test_500_unhealthy(self):
|
|
t = _transport_for(_routes(health_resp=httpx.Response(500)))
|
|
assert await t.check_health(source=_source()) is False
|
|
|
|
async def test_3xx_unhealthy(self):
|
|
# 设计 §10.1 有意收紧: VT 的 resp.ok 含 3xx,库只认 2xx
|
|
t = _transport_for(_routes(health_resp=httpx.Response(302)))
|
|
assert await t.check_health(source=_source()) is False
|
|
|
|
async def test_connect_failure_false(self):
|
|
def handler(request):
|
|
raise httpx.ConnectError("refused", request=request)
|
|
|
|
t = _transport_for(handler)
|
|
assert await t.check_health(source=_source()) is False
|
|
|
|
async def test_cancellation_passes_through(self):
|
|
def handler(request):
|
|
raise asyncio.CancelledError()
|
|
|
|
t = _transport_for(handler)
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await t.check_health(source=_source())
|
|
|
|
|
|
class TestLifecycle:
|
|
async def test_aclose_idempotent(self):
|
|
t = _transport_for(_routes(health_resp=httpx.Response(200)))
|
|
await t.check_health(source=_source())
|
|
await t.aclose()
|
|
await t.aclose()
|
|
|
|
|
|
@pytest.mark.parametrize("method", ["recognize_text", "parse_layout"])
|
|
async def test_ocr_wire_does_not_send_reasoning_configuration(method):
|
|
"""真实 multipart 与 ZIP 下载两段均不发送源级推理配置。"""
|
|
sent = []
|
|
|
|
def handler(request):
|
|
sent.append(request)
|
|
if request.method == "GET":
|
|
return httpx.Response(200, content=_zip_bytes())
|
|
return httpx.Response(
|
|
200, json=_text_body() if method == "recognize_text" else _parse_body()
|
|
)
|
|
|
|
transport = _transport_for(handler)
|
|
try:
|
|
await getattr(transport, method)(
|
|
image=b"image",
|
|
source=_source(enable_thinking=True, reasoning_effort="high"),
|
|
call_id="wire",
|
|
)
|
|
assert len(sent) == (1 if method == "recognize_text" else 2)
|
|
for request in sent:
|
|
for key in (b"reasoning_effort", b"enable_thinking", b"thinking_budget"):
|
|
assert key not in request.content
|
|
finally:
|
|
await transport.aclose()
|