feat: parse document result packages from middle json only
Content comes from the middle json and its paired markdown; the content list is deliberately ignored because it escapes markdown characters. The external-data exception boundary covers both reading and structure parsing so no native exception escapes the OCR client. The layout parser is unchanged.
This commit is contained in:
@@ -0,0 +1,534 @@
|
||||
"""MonkeyOCR 文档结果包解析测试(#34 设计 §5/§5.1)。
|
||||
|
||||
fixtures 全部**合成**: 保留 2026-10-21 实测的结构骨架(para_blocks 的块类型、
|
||||
table/image 的嵌套 `blocks`、span 字段),一切识别文本替换为 `SYNTH-*` 占位——
|
||||
第三方论文内容许可不明、医疗数据不入库(零业务假设 + P5),真实 ZIP 不提交。
|
||||
|
||||
契约表逐行一条,参数化列明各分支。解析函数是 bytes → 结果的纯函数,失败一律
|
||||
`ResultInvalidError`,**不得漏出原生异常**(漏出会让 `OcrClient._call` 既不写终态遥测
|
||||
行也不归还半开探针)。
|
||||
"""
|
||||
|
||||
import io
|
||||
import json
|
||||
import struct
|
||||
import zipfile
|
||||
|
||||
import pytest
|
||||
|
||||
from polygateway.errors import ResultInvalidError
|
||||
from polygateway.transports import _monkey_document
|
||||
from polygateway.transports._monkey_document import parse_document_package
|
||||
from polygateway.types import OcrDocumentElement
|
||||
|
||||
_BBOX = (10.0, 20.0, 110.0, 40.0)
|
||||
|
||||
|
||||
def _span(content="SYNTH-TEXT", type_="text"):
|
||||
return {"type": type_, "content": content, "bbox": list(_BBOX), "score": 1.0}
|
||||
|
||||
|
||||
def _text_block(content="SYNTH-TEXT", *, type_="text", bbox=_BBOX):
|
||||
return {
|
||||
"type": type_,
|
||||
"bbox": list(bbox),
|
||||
"index": 0,
|
||||
"lines": [{"bbox": list(bbox), "spans": [_span(content)]}],
|
||||
}
|
||||
|
||||
|
||||
def _body(type_, spans, bbox=_BBOX):
|
||||
return {"type": type_, "bbox": list(bbox), "lines": [{"bbox": list(bbox), "spans": spans}]}
|
||||
|
||||
|
||||
def _table_block(html="<table><tr><td>SYNTH-CELL</td></tr></table>", extra_subs=()):
|
||||
body = _body("table_body", [{"type": "table", "html": html, "image_path": "synth.jpg"}])
|
||||
return {"type": "table", "bbox": list(_BBOX), "index": 1, "blocks": [*extra_subs, body]}
|
||||
|
||||
|
||||
def _image_block():
|
||||
body = _body("image_body", [{"type": "image", "image_path": "synth.jpg"}])
|
||||
return {"type": "image", "bbox": list(_BBOX), "index": 2, "blocks": [body]}
|
||||
|
||||
|
||||
def _page(blocks=None, *, page_idx=0, page_size=(612.0, 792.0), **extra):
|
||||
page = {"page_idx": page_idx, "page_size": list(page_size), "discarded_blocks": []}
|
||||
if blocks is not None:
|
||||
page["para_blocks"] = blocks
|
||||
page.update(extra)
|
||||
return page
|
||||
|
||||
|
||||
def _package(
|
||||
pages=None,
|
||||
*,
|
||||
md="# synth\n",
|
||||
middle=None,
|
||||
middle_name="doc_middle.json",
|
||||
md_name=None,
|
||||
extra=(),
|
||||
):
|
||||
"""组装结果包。`middle` 给原始 bytes 时原样写入;`md_name` 默认由 middle 名配对。"""
|
||||
if middle is None:
|
||||
middle = json.dumps({"pdf_info": pages if pages is not None else [_page([])]}).encode()
|
||||
if md_name is None:
|
||||
md_name = middle_name[: -len("_middle.json")] + ".md"
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
zf.writestr(middle_name, middle)
|
||||
if md is not None:
|
||||
zf.writestr(md_name, md if isinstance(md, bytes) else md.encode())
|
||||
for name, data in extra:
|
||||
zf.writestr(name, data)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _parse(pages=None, **kwargs):
|
||||
return parse_document_package(_package(pages, **kwargs))
|
||||
|
||||
|
||||
def _invalid(pages=None, **kwargs):
|
||||
with pytest.raises(ResultInvalidError):
|
||||
_parse(pages, **kwargs)
|
||||
|
||||
|
||||
class TestHappyPath:
|
||||
def test_every_block_type_yields_text_per_contract(self):
|
||||
pages = [
|
||||
_page(
|
||||
[
|
||||
_text_block("SYNTH-TITLE", type_="title"),
|
||||
_text_block("SYNTH-BODY"),
|
||||
_text_block("x^{2}", type_="interline_equation"),
|
||||
_table_block("<table>SYNTH</table>"),
|
||||
_image_block(),
|
||||
]
|
||||
)
|
||||
]
|
||||
elements, page_sizes, markdown = _parse(pages)
|
||||
assert [(e.type, e.text) for e in elements] == [
|
||||
("title", "SYNTH-TITLE"),
|
||||
("text", "SYNTH-BODY"),
|
||||
("interline_equation", "x^{2}"),
|
||||
("table", "<table>SYNTH</table>"),
|
||||
("image", ""),
|
||||
]
|
||||
assert all(isinstance(e, OcrDocumentElement) for e in elements)
|
||||
assert all(e.bbox == _BBOX and e.page_index == 0 for e in elements)
|
||||
assert page_sizes == [(612.0, 792.0)]
|
||||
assert markdown == "# synth\n"
|
||||
|
||||
def test_reading_order_and_page_index_across_pages(self):
|
||||
pages = [
|
||||
_page([_text_block("A")], page_idx=0),
|
||||
_page([_text_block("B"), _text_block("C")], page_idx=1, page_size=(595.0, 842.0)),
|
||||
]
|
||||
elements, page_sizes, _ = _parse(pages)
|
||||
assert [(e.page_index, e.text) for e in elements] == [(0, "A"), (1, "B"), (1, "C")]
|
||||
assert page_sizes == [(612.0, 792.0), (595.0, 842.0)]
|
||||
|
||||
def test_content_list_and_images_are_ignored(self):
|
||||
"""内容只来自 middle: content_list 把 `$` 转义成 `\\$`(展示格式),取它会污染 LaTeX。"""
|
||||
garbage = json.dumps([{"type": "text", "text": "GARBAGE \\$"}]).encode()
|
||||
elements, _, _ = parse_document_package(
|
||||
_package(
|
||||
[_page([_text_block("SYNTH $x$")])],
|
||||
extra=[("doc_content_list.json", garbage), ("images/x.jpg", b"\xff\xd8")],
|
||||
)
|
||||
)
|
||||
assert [e.text for e in elements] == ["SYNTH $x$"]
|
||||
|
||||
def test_discarded_blocks_are_not_in_the_result(self):
|
||||
pages = [_page([_text_block("KEEP")], discarded_blocks=[_text_block("DROPPED")])]
|
||||
elements, _, _ = _parse(pages)
|
||||
assert [e.text for e in elements] == ["KEEP"]
|
||||
|
||||
def test_need_drop_pages_are_neither_read_nor_rejected(self):
|
||||
pages = [_page([_text_block("KEEP")], need_drop=True, drop_reason=["synthetic"])]
|
||||
elements, _, _ = _parse(pages)
|
||||
assert [e.text for e in elements] == ["KEEP"]
|
||||
|
||||
|
||||
class TestTextAssembly:
|
||||
def test_multiple_lines_and_spans_are_joined(self):
|
||||
"""样本全是 1 行 1 span;多行/多 span 路径未经实测,只由构造 fixture 钉住拼接规则。"""
|
||||
block = {
|
||||
"type": "text",
|
||||
"bbox": list(_BBOX),
|
||||
"lines": [
|
||||
{"spans": [_span("A"), _span("B")]},
|
||||
{"spans": [_span("C")]},
|
||||
],
|
||||
}
|
||||
elements, _, _ = _parse([_page([block])])
|
||||
assert elements[0].text == "AB\nC"
|
||||
|
||||
def test_inline_equation_spans_are_wrapped_in_dollars(self):
|
||||
"""上游惯例;样本中从未出现该 span 类型,未经实测。"""
|
||||
block = {
|
||||
"type": "text",
|
||||
"bbox": list(_BBOX),
|
||||
"lines": [{"spans": [_span("see "), _span("a+b", "inline_equation"), _span(".")]}],
|
||||
}
|
||||
elements, _, _ = _parse([_page([block])])
|
||||
assert elements[0].text == "see $a+b$."
|
||||
|
||||
def test_empty_lines_list_is_empty_text(self):
|
||||
block = {"type": "text", "bbox": list(_BBOX), "lines": []}
|
||||
elements, _, _ = _parse([_page([block])])
|
||||
assert elements[0].text == ""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mutate",
|
||||
[
|
||||
pytest.param(lambda b: b.update(lines="not-a-list"), id="lines-not-list"),
|
||||
pytest.param(lambda b: b.update(lines=["not-a-dict"]), id="line-not-dict"),
|
||||
pytest.param(lambda b: b["lines"][0].update(spans="x"), id="spans-not-list"),
|
||||
pytest.param(lambda b: b["lines"][0].update(spans=["x"]), id="span-not-dict"),
|
||||
pytest.param(lambda b: b["lines"][0]["spans"][0].pop("content"), id="content-missing"),
|
||||
pytest.param(lambda b: b["lines"][0]["spans"][0].update(content=7), id="content-int"),
|
||||
pytest.param(
|
||||
lambda b: b["lines"][0]["spans"][0].update(content=None), id="content-null"
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_malformed_text_structure_is_rejected_not_silently_empty(self, mutate):
|
||||
block = _text_block()
|
||||
mutate(block)
|
||||
_invalid([_page([block])])
|
||||
|
||||
|
||||
class TestTableAndImage:
|
||||
def test_table_falls_back_to_latex_when_no_html(self):
|
||||
body = _body("table_body", [{"type": "table", "latex": "\\begin{tabular}SYNTH"}])
|
||||
block = {"type": "table", "bbox": list(_BBOX), "blocks": [body]}
|
||||
elements, _, _ = _parse([_page([block])])
|
||||
assert elements[0].text == "\\begin{tabular}SYNTH"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"block",
|
||||
[
|
||||
pytest.param(
|
||||
{
|
||||
"type": "table",
|
||||
"bbox": list(_BBOX),
|
||||
"blocks": [_body("table_body", [{"type": "table"}])],
|
||||
},
|
||||
id="span-has-neither-html-nor-latex",
|
||||
),
|
||||
pytest.param(
|
||||
{"type": "table", "bbox": list(_BBOX), "blocks": []}, id="no-body-sub-block"
|
||||
),
|
||||
pytest.param({"type": "table", "bbox": list(_BBOX)}, id="no-blocks-key"),
|
||||
],
|
||||
)
|
||||
def test_table_without_content_is_rejected_not_silently_empty(self, block):
|
||||
"""表格无内容与"合法空串"无法在结果里区分,故响亮失败。"""
|
||||
_invalid([_page([block])])
|
||||
|
||||
def test_image_text_is_empty_even_with_path_only_span(self):
|
||||
elements, _, _ = _parse([_page([_image_block()])])
|
||||
assert (elements[0].type, elements[0].text) == ("image", "")
|
||||
|
||||
def test_non_body_sub_blocks_become_independent_elements_after_their_parent(self):
|
||||
"""默认部署不产生 caption/footnote 子块(未经实测);出现时不得静默丢弃。"""
|
||||
caption = _body("table_caption", [_span("SYNTH-CAPTION")], bbox=(1.0, 2.0, 3.0, 4.0))
|
||||
footnote = _body("table_footnote", [_span("SYNTH-NOTE")], bbox=(5.0, 6.0, 7.0, 8.0))
|
||||
block = _table_block(extra_subs=[caption])
|
||||
block["blocks"].append(footnote)
|
||||
elements, _, _ = _parse([_page([block])])
|
||||
assert [(e.type, e.text) for e in elements] == [
|
||||
("table", "<table><tr><td>SYNTH-CELL</td></tr></table>"),
|
||||
("table_caption", "SYNTH-CAPTION"),
|
||||
("table_footnote", "SYNTH-NOTE"),
|
||||
]
|
||||
assert elements[1].bbox == (1.0, 2.0, 3.0, 4.0)
|
||||
assert all(e.page_index == 0 for e in elements)
|
||||
|
||||
def test_empty_non_body_sub_block_is_not_emitted(self):
|
||||
caption = _body("table_caption", [_span("")])
|
||||
elements, _, _ = _parse([_page([_table_block(extra_subs=[caption])])])
|
||||
assert [e.type for e in elements] == ["table"]
|
||||
|
||||
def test_unknown_type_without_lines_or_blocks_keeps_type_and_empty_text(self):
|
||||
block = {"type": "synthetic_widget", "bbox": list(_BBOX)}
|
||||
elements, _, _ = _parse([_page([block])])
|
||||
assert (elements[0].type, elements[0].text) == ("synthetic_widget", "")
|
||||
|
||||
def test_unknown_container_keeps_its_body_text(self):
|
||||
"""未知容器类型的 `*_body` 子块文字并入父元素,不被静默丢弃。"""
|
||||
block = {
|
||||
"type": "synthetic_code",
|
||||
"bbox": list(_BBOX),
|
||||
"blocks": [_body("synthetic_code_body", [_span("SYNTH-CODE")])],
|
||||
}
|
||||
elements, _, _ = _parse([_page([block])])
|
||||
assert (elements[0].type, elements[0].text) == ("synthetic_code", "SYNTH-CODE")
|
||||
|
||||
|
||||
class TestPageStructure:
|
||||
@pytest.mark.parametrize(
|
||||
"payload",
|
||||
[
|
||||
pytest.param({}, id="pdf_info-missing"),
|
||||
pytest.param({"pdf_info": "x"}, id="pdf_info-not-list"),
|
||||
pytest.param({"pdf_info": []}, id="zero-pages"),
|
||||
pytest.param([], id="payload-not-dict"),
|
||||
],
|
||||
)
|
||||
def test_document_level_shape(self, payload):
|
||||
"""零页被拒: 成功解析的文档至少 1 页(与 layout 路径"elements 空合法"不同)。"""
|
||||
_invalid(middle=json.dumps(payload).encode())
|
||||
|
||||
def test_page_not_a_dict(self):
|
||||
_invalid(["not-a-dict"])
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"page_size",
|
||||
[
|
||||
pytest.param(None, id="missing"),
|
||||
pytest.param([612.0], id="one-element"),
|
||||
pytest.param([612.0, 792.0, 1.0], id="three-elements"),
|
||||
pytest.param([0, 792.0], id="zero-width"),
|
||||
pytest.param([612.0, -1], id="negative-height"),
|
||||
pytest.param([True, 792.0], id="bool"),
|
||||
pytest.param(["612", 792.0], id="string"),
|
||||
pytest.param([10**400, 792.0], id="huge-int"),
|
||||
],
|
||||
)
|
||||
def test_page_size_invalid(self, page_size):
|
||||
page = _page([])
|
||||
if page_size is None:
|
||||
del page["page_size"]
|
||||
else:
|
||||
page["page_size"] = page_size
|
||||
_invalid([page])
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"page_idx",
|
||||
[
|
||||
pytest.param("missing", id="missing"),
|
||||
pytest.param("0", id="string"),
|
||||
pytest.param(0.0, id="float"),
|
||||
pytest.param(True, id="bool"),
|
||||
pytest.param(-1, id="negative"),
|
||||
pytest.param(3, id="out-of-position"),
|
||||
],
|
||||
)
|
||||
def test_page_idx_must_equal_position(self, page_idx):
|
||||
"""layout 路径对非法 `page_idx` 回退枚举下标,会让 page_sizes 按 page_index 的索引契约静默错位。"""
|
||||
page = _page([])
|
||||
if page_idx == "missing":
|
||||
del page["page_idx"]
|
||||
else:
|
||||
page["page_idx"] = page_idx
|
||||
_invalid([page])
|
||||
|
||||
def test_second_page_must_not_claim_the_first_index(self):
|
||||
_invalid([_page([], page_idx=0), _page([], page_idx=0)])
|
||||
|
||||
def test_para_blocks_missing_is_a_legal_empty_page(self):
|
||||
elements, page_sizes, _ = _parse([_page()])
|
||||
assert elements == [] and page_sizes == [(612.0, 792.0)]
|
||||
|
||||
def test_para_blocks_not_a_list_is_rejected(self):
|
||||
_invalid([_page("x")])
|
||||
|
||||
|
||||
class TestBlockStructure:
|
||||
@pytest.mark.parametrize(
|
||||
"block",
|
||||
[
|
||||
pytest.param("not-a-dict", id="block-not-dict"),
|
||||
pytest.param({"bbox": list(_BBOX)}, id="type-missing"),
|
||||
pytest.param({"type": 7, "bbox": list(_BBOX)}, id="type-not-str"),
|
||||
pytest.param({"type": "", "bbox": list(_BBOX)}, id="type-empty"),
|
||||
pytest.param({"type": " ", "bbox": list(_BBOX)}, id="type-whitespace"),
|
||||
],
|
||||
)
|
||||
def test_block_shape(self, block):
|
||||
_invalid([_page([block])])
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"bbox",
|
||||
[
|
||||
pytest.param(None, id="missing"),
|
||||
pytest.param("1,2,3,4", id="string"),
|
||||
pytest.param([1, 2, 3], id="length-3"),
|
||||
pytest.param([1, 2, 3, 4, 5], id="length-5"),
|
||||
pytest.param([1, 2, "3", 4], id="string-element"),
|
||||
pytest.param([1, 2, True, 4], id="bool-element"),
|
||||
pytest.param([1, 2, float("nan"), 4], id="nan"),
|
||||
pytest.param([1, 2, float("inf"), 4], id="inf"),
|
||||
pytest.param([10, 2, 5, 4], id="x-reversed"),
|
||||
pytest.param([1, 20, 5, 4], id="y-reversed"),
|
||||
pytest.param([0, 0, 10**400, 1], id="huge-int"),
|
||||
],
|
||||
)
|
||||
def test_bbox_invalid(self, bbox):
|
||||
block = _text_block()
|
||||
if bbox is None:
|
||||
del block["bbox"]
|
||||
else:
|
||||
block["bbox"] = bbox
|
||||
# NaN/Infinity 不是合法 JSON,但 Python 的 json 模块默认接受并往返——真实服务可能发出
|
||||
_invalid([_page([block])])
|
||||
|
||||
@pytest.mark.parametrize("bbox", [(5, 5, 5, 20), (5, 5, 50, 5), (5, 5, 5, 5)])
|
||||
def test_zero_width_or_height_boxes_are_legal(self, bbox):
|
||||
"""整数化退化校验专为 CHS shim 的 int() 裁剪;文档消费者不裁剪,一个退化框不该让整份失败。"""
|
||||
elements, _, _ = _parse([_page([_text_block(bbox=bbox)])])
|
||||
assert elements[0].bbox == tuple(float(v) for v in bbox)
|
||||
|
||||
def test_out_of_page_boxes_are_returned_as_is(self):
|
||||
"""越界与否是图内语义,归调用方(ARCHITECTURE §5.1 边界);库不裁剪。"""
|
||||
elements, _, _ = _parse([_page([_text_block(bbox=(0, 0, 9999, 9999))])])
|
||||
assert elements[0].bbox == (0.0, 0.0, 9999.0, 9999.0)
|
||||
|
||||
|
||||
class TestMembers:
|
||||
def test_missing_middle_json(self):
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
zf.writestr("doc.md", "x")
|
||||
with pytest.raises(ResultInvalidError):
|
||||
parse_document_package(buf.getvalue())
|
||||
|
||||
def test_missing_paired_markdown(self):
|
||||
_invalid(md=None)
|
||||
|
||||
def test_markdown_is_paired_by_name_not_first_match(self):
|
||||
elements, _, markdown = parse_document_package(
|
||||
_package(
|
||||
[_page([_text_block("A")])],
|
||||
md="PAIRED",
|
||||
extra=[("aaa_other.md", "DISTRACTOR")],
|
||||
)
|
||||
)
|
||||
assert markdown == "PAIRED"
|
||||
|
||||
def test_unpaired_markdown_alone_is_not_accepted(self):
|
||||
_invalid(md=None, extra=[("other.md", "DISTRACTOR")])
|
||||
|
||||
def test_first_middle_json_wins_and_pairs_with_its_own_markdown(self):
|
||||
pages_a = json.dumps({"pdf_info": [_page([_text_block("A")])]}).encode()
|
||||
pages_b = json.dumps({"pdf_info": [_page([_text_block("B")])]}).encode()
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
zf.writestr("a_middle.json", pages_a)
|
||||
zf.writestr("b_middle.json", pages_b)
|
||||
zf.writestr("a.md", "MD-A")
|
||||
zf.writestr("b.md", "MD-B")
|
||||
elements, _, markdown = parse_document_package(buf.getvalue())
|
||||
assert [e.text for e in elements] == ["A"] and markdown == "MD-A"
|
||||
|
||||
def test_markdown_bom_is_stripped(self):
|
||||
_, _, markdown = _parse(md=b"\xef\xbb\xbf# synth")
|
||||
assert markdown == "# synth"
|
||||
|
||||
def test_empty_markdown_is_legal(self):
|
||||
_, _, markdown = _parse(md="")
|
||||
assert markdown == ""
|
||||
|
||||
def test_non_utf8_markdown_is_rejected(self):
|
||||
_invalid(md=b"\xff\xfe\x00bad")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"middle",
|
||||
[
|
||||
pytest.param(b"not json at all", id="not-json"),
|
||||
pytest.param(b"", id="empty"),
|
||||
pytest.param(b"\xff\xfe", id="not-utf8"),
|
||||
],
|
||||
)
|
||||
def test_middle_json_unparseable(self, middle):
|
||||
_invalid(middle=middle)
|
||||
|
||||
def test_not_a_zip(self):
|
||||
with pytest.raises(ResultInvalidError):
|
||||
parse_document_package(b"definitely not a zip")
|
||||
|
||||
|
||||
def _patch_first_member_compression(zip_bytes: bytes, method: int) -> bytes:
|
||||
"""把第一个成员的压缩方法号改成 `method`(本地头偏移 8,中央目录偏移 10)。"""
|
||||
data = bytearray(zip_bytes)
|
||||
local = data.index(b"PK\x03\x04")
|
||||
central = data.index(b"PK\x01\x02")
|
||||
struct.pack_into("<H", data, local + 8, method)
|
||||
struct.pack_into("<H", data, central + 10, method)
|
||||
return bytes(data)
|
||||
|
||||
|
||||
def _encrypted_zip() -> bytes:
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
info = zipfile.ZipInfo("doc_middle.json")
|
||||
info.flag_bits |= 0x1 # 加密位:读取时 zipfile 要求密码
|
||||
zf.writestr(info, b'{"pdf_info": []}')
|
||||
zf.writestr("doc.md", "x")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
class TestExternalDataExceptionBoundary:
|
||||
"""外部字节引起的一切已知异常都归 `ResultInvalidError`,绝不漏出原生异常。
|
||||
|
||||
漏出的后果(`OcrClient._call` 只接 PolyGatewayError): 不写终态遥测行,持有的半开探针
|
||||
也不归还。边界必须同时盖住「读取」与「全部结构解析」两段。
|
||||
"""
|
||||
|
||||
def test_encrypted_member(self):
|
||||
with pytest.raises(ResultInvalidError):
|
||||
parse_document_package(_encrypted_zip())
|
||||
|
||||
def test_unknown_compression_method(self):
|
||||
patched = _patch_first_member_compression(_package([_page([])]), 99)
|
||||
with pytest.raises(ResultInvalidError):
|
||||
parse_document_package(patched)
|
||||
|
||||
def test_crc_corruption(self):
|
||||
corrupted = _package([_page([])]).replace(b'"pdf_info"', b'"pdf_infq"', 1)
|
||||
with pytest.raises(ResultInvalidError):
|
||||
parse_document_package(corrupted)
|
||||
|
||||
def test_huge_integer_inside_page_structure_is_not_an_overflow_error(self):
|
||||
"""`float(10**400)` 抛 OverflowError——发生在结构解析段,不在读取段。"""
|
||||
block = _text_block()
|
||||
block["bbox"] = [0, 0, 10**400, 1]
|
||||
_invalid([_page([block])])
|
||||
|
||||
def test_deeply_nested_json_is_not_a_recursion_error(self):
|
||||
deep = b'{"pdf_info": ' + b"[" * 100_000 + b"]" * 100_000 + b"}"
|
||||
_invalid(middle=deep)
|
||||
|
||||
@pytest.mark.parametrize("member", ["middle", "markdown"])
|
||||
def test_declared_size_over_cap_is_rejected(self, member, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
_monkey_document, "MAX_MIDDLE_BYTES" if member == "middle" else "MAX_MARKDOWN_BYTES", 8
|
||||
)
|
||||
_invalid([_page([_text_block("A")])], md="# a markdown longer than eight bytes")
|
||||
|
||||
def test_existing_result_invalid_messages_pass_through_unchanged(self):
|
||||
with pytest.raises(ResultInvalidError, match="至少应有 1 页"):
|
||||
_parse([])
|
||||
|
||||
|
||||
class TestLayoutPathUntouched:
|
||||
"""文档路径更严/更宽的口径不得外溢到医院线在用的 layout 解析。"""
|
||||
|
||||
def test_layout_parser_still_rejects_degenerate_int_boxes(self):
|
||||
from polygateway.transports.monkey_ocr import _parse_middle_json
|
||||
|
||||
block = {"type": "text", "bbox": [5, 5, 5.4, 20], "lines": []}
|
||||
zip_bytes = _package([_page([block])])
|
||||
with pytest.raises(ResultInvalidError):
|
||||
_parse_middle_json(zip_bytes)
|
||||
|
||||
def test_layout_parser_keeps_its_lenient_page_idx_handling(self):
|
||||
"""layout 路径: 合法的非负整数 `page_idx` 原样接受(哪怕与页序不符),非法值才回退
|
||||
枚举下标——这正是文档路径改为严格的原因,但医院线的既有行为不得被连带改动。"""
|
||||
from polygateway.transports.monkey_ocr import _parse_middle_json
|
||||
|
||||
accepted, _ = _parse_middle_json(_package([_page([_text_block()], page_idx=7)]))
|
||||
assert accepted[0].page_index == 7
|
||||
fallback, _ = _parse_middle_json(_package([_page([_text_block()], page_idx="x")]))
|
||||
assert fallback[0].page_index == 0
|
||||
@@ -190,7 +190,9 @@ class TestSniffMedia:
|
||||
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"\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"),
|
||||
@@ -205,7 +207,9 @@ class TestSniffMedia:
|
||||
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""])
|
||||
@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")
|
||||
|
||||
Reference in New Issue
Block a user