iomgaa
bb621cad4d
Zip construction and member reads now share one narrow boundary so corrupt BZIP2, LZMA and version fields cannot leak native exceptions. Non-success statuses are read before classification, text-bearing blocks require lines, the size cap and reasoning contract are pinned by tests that fail under mutation, and the per-page timing claim is corrected to the observed range.
591 lines
25 KiB
Python
591 lines
25 KiB
Python
"""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 random
|
|
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
|
|
from tests.unit._document_fixtures import (
|
|
_BBOX,
|
|
_body,
|
|
_encrypted_package,
|
|
_image_block,
|
|
_package,
|
|
_page,
|
|
_span,
|
|
_table_block,
|
|
_text_block,
|
|
)
|
|
|
|
|
|
def _parse(pages=None, **kwargs):
|
|
return parse_document_package(_package(pages, **kwargs))
|
|
|
|
|
|
def _assert_rejected_on_purpose(exc: ResultInvalidError) -> None:
|
|
"""拒绝必须出自显式校验,不能是 Python 自己绊倒后被外层边界吸收。
|
|
|
|
外层边界会把 TypeError/KeyError 也归为 ResultInvalidError,所以只断言异常类型,
|
|
「显式校验」与「碰巧在别处失败」无法区分(变异检查证实过: 删掉校验后测试仍绿)。
|
|
原因链里出现 TypeError/KeyError 即说明有校验缺口。
|
|
"""
|
|
assert not isinstance(exc.__cause__, TypeError | KeyError), repr(exc.__cause__)
|
|
|
|
|
|
def _invalid(pages=None, **kwargs):
|
|
with pytest.raises(ResultInvalidError) as ei:
|
|
_parse(pages, **kwargs)
|
|
_assert_rejected_on_purpose(ei.value)
|
|
|
|
|
|
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)
|
|
|
|
@pytest.mark.parametrize(
|
|
("blocks", "reason"),
|
|
[
|
|
pytest.param("not-a-list", "blocks 无效", id="blocks-not-list"),
|
|
pytest.param(["not-a-dict"], "块结构无效", id="sub-block-not-dict"),
|
|
pytest.param([{"bbox": list(_BBOX)}], "块缺少 type", id="sub-block-without-type"),
|
|
pytest.param([{"type": " ", "bbox": list(_BBOX)}], "块缺少 type", id="blank-type"),
|
|
],
|
|
)
|
|
def test_malformed_nested_blocks_are_rejected_with_their_own_reason(self, blocks, reason):
|
|
"""外层边界会把 TypeError 等也归为 ResultInvalid,所以必须断言**原因**,否则"显式校验"
|
|
与"碰巧在别处失败"无法区分(变异检查发现仅断言类型时删掉校验测试仍绿)。"""
|
|
block = {"type": "table", "bbox": list(_BBOX), "blocks": blocks}
|
|
with pytest.raises(ResultInvalidError, match=reason):
|
|
_parse([_page([block])])
|
|
|
|
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-true-off-position"),
|
|
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])
|
|
|
|
@pytest.mark.parametrize(
|
|
"pages",
|
|
[
|
|
pytest.param([_page([], page_idx=False)], id="false-equals-0-on-first-page"),
|
|
pytest.param(
|
|
[_page([], page_idx=0), _page([], page_idx=True)], id="true-equals-1-on-second-page"
|
|
),
|
|
],
|
|
)
|
|
def test_bool_page_idx_is_rejected_even_when_it_equals_the_position(self, pages):
|
|
"""`False == 0`、`True == 1`: 没有显式的 bool 守卫时它们恰好等于页序而被放行。"""
|
|
with pytest.raises(ResultInvalidError, match="page_idx"):
|
|
_parse(pages)
|
|
|
|
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)
|
|
|
|
|
|
class TestExternalDataExceptionBoundary:
|
|
"""外部字节引起的一切已知异常都归 `ResultInvalidError`,绝不漏出原生异常。
|
|
|
|
漏出的后果(`OcrClient._call` 只接 PolyGatewayError): 不写终态遥测行,持有的半开探针
|
|
也不归还。边界必须同时盖住「读取」与「全部结构解析」两段。
|
|
"""
|
|
|
|
def test_encrypted_member(self):
|
|
with pytest.raises(ResultInvalidError):
|
|
parse_document_package(_encrypted_package())
|
|
|
|
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(
|
|
("cap_name", "member"),
|
|
[("MAX_MIDDLE_BYTES", "doc_middle.json"), ("MAX_MARKDOWN_BYTES", "doc.md")],
|
|
)
|
|
def test_declared_size_over_cap_names_the_offending_member(self, cap_name, member, monkeypatch):
|
|
"""断言消息里的成员名: 两个上限互换时,只断言异常类型的旧用例仍会通过。"""
|
|
monkeypatch.setattr(_monkey_document, cap_name, 8)
|
|
with pytest.raises(ResultInvalidError, match=member):
|
|
_parse([_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
|
|
|
|
|
|
class TestZipLayerNeverLeaksNativeExceptions:
|
|
"""独立验证发现: BZIP2/LZMA 成员数据损坏会漏出 `OSError`/`LZMAError`,中央目录版本字段损坏会在
|
|
`ZipFile` 构造阶段漏出 `NotImplementedError`——它们都不在外层边界里。漏出的后果与设计 §5 一致:
|
|
`OcrClient._call` 只接 `PolyGatewayError`,既不写终态遥测行,也不归还半开探针。"""
|
|
|
|
@staticmethod
|
|
def _zip(method: int) -> bytes:
|
|
payload = json.dumps({"pdf_info": [_page([_text_block("SYNTH")])]}).encode() * 20
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", compression=method) as zf:
|
|
zf.writestr("doc_middle.json", payload)
|
|
zf.writestr("doc.md", "# synth " * 50)
|
|
return buf.getvalue()
|
|
|
|
@pytest.mark.parametrize(
|
|
"method",
|
|
[
|
|
pytest.param(zipfile.ZIP_DEFLATED, id="deflated"),
|
|
pytest.param(zipfile.ZIP_BZIP2, id="bzip2"),
|
|
pytest.param(zipfile.ZIP_LZMA, id="lzma"),
|
|
],
|
|
)
|
|
def test_single_bit_flips_never_escape_as_native_exceptions(self, method):
|
|
"""确定性字节翻转(固定种子): 每个结果只能是成功或 `ResultInvalidError`。"""
|
|
base = self._zip(method)
|
|
rng = random.Random(34)
|
|
for _ in range(600):
|
|
data = bytearray(base)
|
|
position = rng.randrange(len(data))
|
|
data[position] ^= 1 << rng.randrange(8)
|
|
try:
|
|
parse_document_package(bytes(data))
|
|
except ResultInvalidError:
|
|
continue
|
|
except Exception as exc: # noqa: BLE001 — 测试要抓的正是任何漏出的原生异常
|
|
pytest.fail(
|
|
f"字节 {position} 翻转后漏出 {type(exc).__module__}.{type(exc).__name__}: {exc}"
|
|
)
|
|
|
|
@pytest.mark.parametrize("method", [zipfile.ZIP_BZIP2, zipfile.ZIP_LZMA])
|
|
def test_corrupt_non_deflate_member_data(self, method):
|
|
"""压缩流内部损坏(而非 CRC 失败): 标准库抛 OSError / LZMAError。"""
|
|
data = bytearray(self._zip(method))
|
|
local = data.index(b"PK\x03\x04")
|
|
name_len, extra_len = struct.unpack_from("<HH", data, local + 26)
|
|
body = local + 30 + name_len + extra_len
|
|
for offset in range(body, body + 12):
|
|
data[offset] ^= 0xFF
|
|
with pytest.raises(ResultInvalidError):
|
|
parse_document_package(bytes(data))
|
|
|
|
def test_zipfile_construction_stage_is_inside_the_boundary(self):
|
|
"""中央目录的 extract_version 损坏: `ZipFile(...)` 构造阶段抛 NotImplementedError。"""
|
|
data = bytearray(self._zip(zipfile.ZIP_STORED))
|
|
central = data.index(b"PK\x01\x02")
|
|
data[central + 6] = 99 # extract_version = 9.9 → "zip file version 9.9"
|
|
with pytest.raises(ResultInvalidError):
|
|
parse_document_package(bytes(data))
|
|
|
|
|
|
class TestTextBearingBlocksRequireLines:
|
|
"""text/title/interline_equation 缺 `lines` 不是「空文本」而是畸形: 与 table 缺内容就报错同口径,
|
|
否则下游分不清「服务识别出空块」和「服务给了坏结构」。"""
|
|
|
|
@pytest.mark.parametrize("block_type", ["text", "title", "interline_equation"])
|
|
def test_missing_lines_is_rejected_with_its_reason(self, block_type):
|
|
block = {"type": block_type, "bbox": list(_BBOX)}
|
|
with pytest.raises(ResultInvalidError, match="缺少 lines"):
|
|
_parse([_page([block])])
|