"""`dwg_review.phases.evidence` 单元测试。""" import base64 import io import pytest from dwg_review.phases.evidence import ( _apply_keep, _content_predicate, _is_truncated, _plan_tiles, _text_boxes, bbox_to_window, collect_frame, maximized_drawing_area, ) from tests.fakes import ( STANDARD_TOOLS, FakeMcpClient, err, frame, make_ctx, ok_image, ok_text, tool_schema, ) def test_bbox_to_window_mapping(): assert bbox_to_window([1, 2, 3, 4]) == { "min_x": 1.0, "min_y": 2.0, "max_x": 3.0, "max_y": 4.0, } def test_bbox_to_window_non_numeric_known_gap(): """已知瑕疵：4 元素但含非数值的 bbox 会抛 ValueError，上游 collect_frame 只判长度。""" with pytest.raises(ValueError): bbox_to_window([0, "a", 10, 10]) def test_collect_frame_returns_none_for_missing_bbox(): ctx = make_ctx(client=FakeMcpClient()) assert collect_frame(ctx, 1, {}) is None assert collect_frame(ctx, 1, {"bbox": None}) is None def test_collect_frame_returns_none_for_bad_bbox_length(): ctx = make_ctx(client=FakeMcpClient()) assert collect_frame(ctx, 1, {"bbox": [0, 0, 10]}) is None def build_evidence_client(image=ok_image("QUJD")): return FakeMcpClient( responses={ "query_entities": ok_text({"total": 5, "items": [{"handle": "A"}]}), "get_text_content": ok_text({"total": 1, "items": [{"text": "你好"}]}), "get_dimensions": ok_text({"total": 2, "items": []}), "get_block_references": ok_text({"total": 0, "items": []}), "get_polylines_info": ok_text({"total": 3, "items": []}), "get_viewport_screenshot": image, } ) def test_collect_frame_assembles_evidence(): ctx = make_ctx(client=build_evidence_client()) ev = collect_frame(ctx, 2, frame([0, 0, 100, 50])) assert ev["index"] == 2 assert ev["window"] == {"min_x": 0.0, "min_y": 0.0, "max_x": 100.0, "max_y": 50.0} assert ev["entities"]["total"] == 5 assert ev["texts"]["total"] == 1 assert ev["dimensions"]["total"] == 2 assert ev["blocks"]["total"] == 0 assert ev["polylines"]["total"] == 3 assert ev["source_type"] == "cad" assert ev["images"] == [{"b64": "QUJD", "mime": "image/png", "label": "overall"}] def test_collect_frame_calls_expected_tools(): ctx = make_ctx(client=build_evidence_client(), frame_limit=99) collect_frame(ctx, 1, frame([1, 2, 3, 4])) client = ctx.client win = {"min_x": 1.0, "min_y": 2.0, "max_x": 3.0, "max_y": 4.0} assert client.arguments_of("zoom_to_window") == dict(win, margin=0.02) for name in ( "query_entities", "get_text_content", "get_dimensions", "get_block_references", "get_polylines_info", ): assert client.arguments_of(name) == {"window": win, "limit": 99} assert client.arguments_of("get_viewport_screenshot") == {} def test_collect_frame_without_image(): ctx = make_ctx(client=build_evidence_client(image=ok_text({}))) ev = collect_frame(ctx, 1, frame([0, 0, 10, 10])) assert ev["images"] == [] def test_collect_frame_tolerates_empty_structured_data(): ctx = make_ctx(client=FakeMcpClient()) # 全部默认 ok_text({}) ev = collect_frame(ctx, 1, frame([0, 0, 10, 10])) assert ev["entities"] == {} assert ev["images"] == [] def test_sub_windows_2x2(): from dwg_review.phases.evidence import sub_windows win = {"min_x": 0.0, "min_y": 0.0, "max_x": 10.0, "max_y": 10.0} subs = sub_windows(win, 4) assert len(subs) == 4 # 行优先：左上 → 右上 → 左下 → 右下 assert subs[0] == {"min_x": 0.0, "min_y": 5.0, "max_x": 5.0, "max_y": 10.0} assert subs[1] == {"min_x": 5.0, "min_y": 5.0, "max_x": 10.0, "max_y": 10.0} assert subs[2] == {"min_x": 0.0, "min_y": 0.0, "max_x": 5.0, "max_y": 5.0} assert subs[3] == {"min_x": 5.0, "min_y": 0.0, "max_x": 10.0, "max_y": 5.0} def test_sub_windows_disabled(): from dwg_review.phases.evidence import sub_windows win = {"min_x": 0.0, "min_y": 0.0, "max_x": 10.0, "max_y": 10.0} assert sub_windows(win, 0) == [] assert sub_windows(win, 1) == [] def test_collect_frame_captures_multiple_images(monkeypatch): from dwg_review import config from dwg_review.phases.evidence import collect_frame monkeypatch.setattr(config, "EVIDENCE_DETAIL_TILES", 4) ctx = make_ctx(client=build_evidence_client()) ev = collect_frame(ctx, 1, frame([0, 0, 100, 50])) assert [img["label"] for img in ev["images"]] == [ "overall", "tile-1", "tile-2", "tile-3", "tile-4", ] def test_collect_dispatches_cad_source(): from dwg_review.phases.evidence import collect ctx = make_ctx(client=build_evidence_client()) ev = collect(ctx, 1, frame([0, 0, 10, 10])) assert ev is not None assert ev["source_type"] == "cad" def test_collect_returns_none_for_unsupported_source(): from dwg_review.phases.evidence import collect ctx = make_ctx(client=FakeMcpClient()) assert collect(ctx, 1, {"source_type": "pdf", "bbox": [0, 0, 1, 1]}) is None # --------------------------------------------------------------------------- # 自适应瓦片规划 # --------------------------------------------------------------------------- @pytest.fixture def tiling_env(monkeypatch): """把瓦片相关配置钉死，使用例不受外部环境变量影响。""" from dwg_review import config for name, value in { "TILING_ENABLED": True, "TEXT_HEIGHT_MM": 2.0, "TEXT_READABLE_PX": 12, "LINE_READABLE_PX": 6, "TILE_MAX_COUNT": 36, "TILE_OVERLAP": 0.0, "HEATMAP_CELL_DIV": 32, "HEATMAP_CLUSTER_DILATE": 0, "HEATMAP_MAX_AREA_FRAC": 0.5, "IMAGE_MAX_EDGE": 1280, "EVIDENCE_MAX_IMAGES": 5, "EVIDENCE_DETAIL_TILES": 4, }.items(): monkeypatch.setattr(config, name, value) return config def _png_b64(w=2000, h=1414): from PIL import Image buf = io.BytesIO() Image.new("RGB", (w, h), "black").save(buf, format="PNG") return base64.b64encode(buf.getvalue()).decode("ascii") def _tiling_client(b64, entity_type="LINE"): return FakeMcpClient( responses={ "query_entities": ok_text( {"total": 1, "items": [{"handle": "A", "type": entity_type, "bbox": [0, 0, 60000, 20000]}]} ), "get_text_content": ok_text({"total": 0, "items": []}), "get_dimensions": ok_text({"total": 0, "items": []}), "get_block_references": ok_text({"total": 0, "items": []}), "get_polylines_info": ok_text({"total": 0, "items": []}), "get_viewport_screenshot": ok_image(b64), } ) def test_is_truncated(): assert _is_truncated({"total": 5, "items": [1]}) is True assert _is_truncated({"total": 1, "items": [1]}) is False assert _is_truncated({"items": [1, 2]}) is False assert _is_truncated(None, {}, {"total": 3, "items": []}) is True assert _is_truncated() is False def test_text_boxes_includes_std_and_custom(): entities = {"items": [ {"type": "LINE", "bbox": [0, 0, 1, 1]}, {"type": "AECDLABEL", "bbox": [2, 2, 3, 3]}, {"type": "TEXT", "bbox": [4, 4, 5, 5]}, ]} texts = {"items": [{"bbox": [6, 6, 7, 7]}]} dimensions = {"items": [{"bbox": [8, 8, 9, 9]}]} boxes = _text_boxes(entities, texts, dimensions) for expected in ((2, 2, 3, 3), (4, 4, 5, 5), (6, 6, 7, 7), (8, 8, 9, 9)): assert tuple(float(v) for v in expected) in boxes assert (0.0, 0.0, 1.0, 1.0) not in boxes # 普通线不算文字 def test_plan_tiles_degrades(tiling_env): win = {"min_x": 0.0, "min_y": 0.0, "max_x": 100.0, "max_y": 50.0} f = frame([0, 0, 100, 50], sheet_size="A0", sheet_scale=100.0) boxes = [(0.0, 0.0, 10.0, 10.0)] assert _plan_tiles(win, f, boxes, [], False, (100, 50)) != [] # 正常路径非空 assert _plan_tiles(win, None, boxes, [], False, (100, 50)) == [] # 缺图框 assert _plan_tiles(win, f, [], [], False, (100, 50)) == [] # 无实体 assert _plan_tiles(win, f, boxes, [], False, None) == [] # 无像素 assert _plan_tiles(win, f, boxes, [], True, (100, 50)) == [] # 数据截断 # 比例未知（sheet_scale=0 且 sheet_size 不在纸张表）→ 无法规划 unknown = frame([0, 0, 100, 50], sheet_size="ZZ", sheet_scale=0.0) assert _plan_tiles(win, unknown, boxes, [], False, (100, 50)) == [] def test_plan_tiles_disabled_by_config(tiling_env, monkeypatch): tiling_env.TILING_ENABLED = False win = {"min_x": 0.0, "min_y": 0.0, "max_x": 100.0, "max_y": 50.0} f = frame([0, 0, 100, 50], sheet_size="A0", sheet_scale=100.0) assert _plan_tiles(win, f, [(0.0, 0.0, 10.0, 10.0)], [], False, (100, 50)) == [] def test_capture_images_tiles_line_cluster(tiling_env): """线条簇：单张可覆盖 → 2 张瓦片（行优先）。""" pytest.importorskip("PIL") ctx = make_ctx(client=_tiling_client(_png_b64())) ev = collect_frame( ctx, 1, frame([0, 0, 120000, 40000], sheet_size="A0", sheet_scale=100.0) ) assert [i["label"] for i in ev["images"]] == ["overall", "tile-1", "tile-2"] def test_capture_images_tiles_text_cluster_is_finer(tiling_env): """文字簇用更细尺度 → 瓦片更多，并受 EVIDENCE_MAX_IMAGES 截断。""" pytest.importorskip("PIL") ctx = make_ctx(client=_tiling_client(_png_b64(), entity_type="AECDLABEL")) ev = collect_frame( ctx, 1, frame([0, 0, 120000, 40000], sheet_size="A0", sheet_scale=100.0) ) labels = [i["label"] for i in ev["images"]] assert labels == ["overall", "tile-1", "tile-2", "tile-3", "tile-4"] # 上限 5 张 def test_capture_images_falls_back_without_scale(tiling_env): """缺 sheet_size/sheet_scale → 退回固定 2x2 象限，并按内容过滤空白象限。""" pytest.importorskip("PIL") client = _tiling_client(_png_b64()) # 实体完全落在**左下**象限内（不贴边界） client.responses["query_entities"] = ok_text( {"total": 1, "items": [{"handle": "A", "type": "LINE", "bbox": [0, 0, 50000, 15000]}]} ) ctx = make_ctx(client=client) ev = collect_frame(ctx, 1, frame([0, 0, 120000, 40000], sheet_size="ZZ")) assert [i["label"] for i in ev["images"]] == ["overall", "tile-1"] def test_capture_images_fallback_keeps_all_when_no_boxes(tiling_env): """拿不到实体 bbox（无法判断内容）→ 回退路径不过滤，四象限全拍。""" pytest.importorskip("PIL") client = _tiling_client(_png_b64()) client.responses["query_entities"] = ok_text({"total": 0, "items": []}) ctx = make_ctx(client=client) ev = collect_frame(ctx, 1, frame([0, 0, 120000, 40000], sheet_size="ZZ")) assert [i["label"] for i in ev["images"]] == [ "overall", "tile-1", "tile-2", "tile-3", "tile-4", ] def test_capture_images_falls_back_when_truncated(tiling_env): """实体被截断 → 禁用内容过滤，退回固定策略。""" pytest.importorskip("PIL") client = _tiling_client(_png_b64()) client.responses["query_entities"] = ok_text( {"total": 500, "items": [{"handle": "A", "type": "LINE", "bbox": [0, 0, 60000, 20000]}]} ) ctx = make_ctx(client=client) ev = collect_frame( ctx, 1, frame([0, 0, 120000, 40000], sheet_size="A0", sheet_scale=100.0) ) assert [i["label"] for i in ev["images"]] == [ "overall", "tile-1", "tile-2", "tile-3", "tile-4", ] # --------------------------------------------------------------------------- # 内容过滤 / 绘图区最大化 # --------------------------------------------------------------------------- def test_apply_keep_filters_and_falls_back(): regions = [ {"min_x": 0.0, "min_y": 0.0, "max_x": 1.0, "max_y": 1.0}, {"min_x": 1.0, "min_y": 1.0, "max_x": 2.0, "max_y": 2.0}, ] keep_first = lambda w: w["min_x"] == 0.0 # noqa: E731 assert _apply_keep(regions, keep_first) == regions[:1] assert _apply_keep(regions, None) == regions # 无谓词 → 原样 assert _apply_keep(regions, lambda w: False) == regions # 全被滤掉 → 退回原集合 assert _apply_keep([], keep_first) == [] def test_content_predicate_none_without_boxes(): assert _content_predicate([]) is None assert _content_predicate(None) is None def test_content_predicate_detects_intersection(): keep = _content_predicate([(0.0, 0.0, 5.0, 5.0)]) assert keep({"min_x": 4.0, "min_y": 4.0, "max_x": 9.0, "max_y": 9.0}) is True assert keep({"min_x": 6.0, "min_y": 6.0, "max_x": 9.0, "max_y": 9.0}) is False @pytest.fixture def drawing_area_env(monkeypatch): """打开绘图区最大化，并把等待时间归零（用例不真的 sleep）。""" from dwg_review import config monkeypatch.setattr(config, "DRAWING_AREA_MAXIMIZE", True) monkeypatch.setattr(config, "DRAWING_AREA_ENTER_WAIT", 0.0) monkeypatch.setattr(config, "DRAWING_AREA_RESTORE_WAIT", 0.0) return config def _area_client(responses=None): """带 set_drawing_area 工具的假客户端。""" return FakeMcpClient( tools=STANDARD_TOOLS + [tool_schema("set_drawing_area", "绘图区最大化")], responses=responses, ) def test_maximized_drawing_area_calls_max_then_restore(drawing_area_env): client = _area_client() ctx = make_ctx(client=client) with maximized_drawing_area(ctx): assert client.called_names() == ["set_drawing_area"] # 进入时只 max assert [c["arguments"] for c in client.called("set_drawing_area")] == [ {"mode": "max"}, {"mode": "restore"}, ] def test_maximized_drawing_area_degrades_without_tool(drawing_area_env): """旧版插件没有该工具 → 静默降级，一次调用都不发。""" client = FakeMcpClient() ctx = make_ctx(client=client) with maximized_drawing_area(ctx): pass assert client.calls == [] def test_maximized_drawing_area_disabled_by_config(drawing_area_env): drawing_area_env.DRAWING_AREA_MAXIMIZE = False client = _area_client() ctx = make_ctx(client=client) with maximized_drawing_area(ctx): pass assert client.calls == [] def test_maximized_drawing_area_restores_even_if_body_raises(drawing_area_env): client = _area_client() ctx = make_ctx(client=client) with pytest.raises(RuntimeError): with maximized_drawing_area(ctx): raise RuntimeError("boom") assert client.arguments_of("set_drawing_area") == {"mode": "restore"} def test_maximized_drawing_area_degrades_when_max_fails(drawing_area_env): """max 调用失败 → 视为没进入最大化，不再发 restore，也不抛异常。""" client = _area_client(responses={"set_drawing_area": err("no window")}) ctx = make_ctx(client=client) with maximized_drawing_area(ctx): pass assert [c["arguments"] for c in client.called("set_drawing_area")] == [{"mode": "max"}] def test_maximized_drawing_area_restore_failure_does_not_mask_body(drawing_area_env): """restore 失败不应掩盖业务异常。""" client = _area_client() ctx = make_ctx(client=client) def boom(args): if args.get("mode") == "restore": raise RuntimeError("restore failed") return ok_text({}) client.responses["set_drawing_area"] = boom with pytest.raises(ValueError): with maximized_drawing_area(ctx): raise ValueError("business error")