cad-agent/tests/test_annotator.py

"""`dwg_review.annotator` 单元测试。"""

import pytest

from dwg_review.annotator import _pick_point, annotate, clear
from dwg_review.mcp_client import McpError, McpToolError
from tests.fakes import FakeMcpClient, make_ctx, ok_text


# ---------------------------------------------------------------------------
#  _pick_point
# ---------------------------------------------------------------------------
def test_pick_point_from_bbox_center():
    assert _pick_point({"bbox": [0, 0, 10, 20]}) == [5.0, 10.0, 0.0]


def test_pick_point_from_position():
    assert _pick_point({"position": [1, 2, 3]}) == [1.0, 2.0, 3.0]


def test_pick_point_from_position_without_z():
    assert _pick_point({"position": [1, 2]}) == [1.0, 2.0, 0.0]


def test_pick_point_prefers_bbox_over_position():
    assert _pick_point({"bbox": [0, 0, 10, 10], "position": [99, 99]}) == [5.0, 5.0, 0.0]


def test_pick_point_returns_none_when_unlocatable():
    assert _pick_point({}) is None
    assert _pick_point({"bbox": [0, 0, 1]}) is None
    assert _pick_point("not-a-dict") is None


# ---------------------------------------------------------------------------
#  annotate
# ---------------------------------------------------------------------------
def build_client(detail=None, created=None, fail_leader=False):
    responses = {"get_entity_detail": detail if detail is not None else ok_text({"bbox": [0, 0, 10, 10]})}
    if fail_leader:
        responses["add_leader_annotation"] = McpToolError("引线失败")
    else:
        responses["add_leader_annotation"] = ok_text(
            created if created is not None else {"created": [{"handle": "NEW1"}]}
        )
    responses["highlight_entities"] = ok_text({})
    return FakeMcpClient(responses=responses)


def test_annotate_dry_run_does_not_call_cad():
    logs = []
    ctx = make_ctx(client=FakeMcpClient())
    created, highlighted = annotate(
        ctx, [{"handle": "A", "severity": "high", "message": "x"}], dry_run=True, log=logs.append
    )
    assert created == [] and highlighted == []
    assert ctx.client.calls == []
    assert any("[dry-run]" in line for line in logs)


def test_annotate_skips_issues_without_handle():
    ctx = make_ctx(client=FakeMcpClient())
    created, highlighted = annotate(
        ctx, [{"severity": "high", "message": "无 handle"}], log=lambda *a: None
    )
    assert created == [] and highlighted == []
    assert ctx.client.calls == []


def test_annotate_creates_leader_and_highlights():
    ctx = make_ctx(client=build_client())
    created, highlighted = annotate(
        ctx, [{"handle": "A", "severity": "high", "message": "重叠"}], log=lambda *a: None
    )
    assert created == ["NEW1"]
    assert highlighted == ["A"]
    names = ctx.client.called_names()
    assert "get_entity_detail" in names
    assert "add_leader_annotation" in names
    assert "highlight_entities" in names


def test_annotate_leader_payload_shape():
    ctx = make_ctx(client=build_client())
    annotate(ctx, [{"handle": "A", "severity": "high", "message": "m"}], log=lambda *a: None)
    args = ctx.client.arguments_of("add_leader_annotation")
    assert args["layer"] == "AI_REVIEW"
    assert len(args["points"]) == 2
    assert args["points"][0] == [5.0, 5.0, 0.0]  # 实体定位点
    assert args["text_position"] == args["points"][1]
    assert args["text"].startswith("[high]")


def test_annotate_skips_when_unlocatable():
    logs = []
    ctx = make_ctx(client=build_client(detail=ok_text({})))
    created, highlighted = annotate(
        ctx, [{"handle": "A", "message": "x"}], log=logs.append
    )
    assert created == [] and highlighted == []
    assert "add_leader_annotation" not in ctx.client.called_names()
    assert any("无法定位坐标" in line for line in logs)


def test_annotate_tool_error_does_not_highlight():
    logs = []
    ctx = make_ctx(client=build_client(fail_leader=True))
    created, highlighted = annotate(
        ctx, [{"handle": "A", "message": "x"}], log=logs.append
    )
    assert created == [] and highlighted == []
    assert "highlight_entities" not in ctx.client.called_names()
    assert any("批注失败" in line for line in logs)


def test_annotate_handles_multiple_created_entities():
    client = build_client(created={"created": [{"handle": "N1"}, {"handle": "N2"}]})
    ctx = make_ctx(client=client)
    created, _ = annotate(ctx, [{"handle": "A", "message": "x"}], log=lambda *a: None)
    assert created == ["N1", "N2"]


def test_annotate_created_not_deduped_known_gap():
    """已知瑕疵：同一 handle 出现两次会创建两条批注（created 无去重，仅 highlighted 去重）。"""
    ctx = make_ctx(client=build_client())
    created, highlighted = annotate(
        ctx,
        [
            {"handle": "A", "severity": "high", "message": "第一次"},
            {"handle": "A", "severity": "high", "message": "第二次"},
        ],
        log=lambda *a: None,
    )
    assert created == ["NEW1", "NEW1"]  # 重复批注（期望去重）
    assert highlighted == ["A"]  # 高亮已去重


def test_annotate_propagates_protocol_error_on_detail():
    ctx = make_ctx(client=FakeMcpClient(responses={"get_entity_detail": McpError(-32602, "bad")}))
    # annotate 只兜 add_leader/highlight 的错；detail 的协议错误会向上冒泡（现状行为）
    with pytest.raises(McpError):
        annotate(ctx, [{"handle": "A", "message": "x"}], log=lambda *a: None)


# ---------------------------------------------------------------------------
#  clear
# ---------------------------------------------------------------------------
def test_clear_deletes_and_restores():
    ctx = make_ctx(client=FakeMcpClient())
    clear(ctx, handles=["N1"], highlighted=["A"], log=lambda *a: None)
    assert ctx.client.arguments_of("delete_entities") == {"handles": ["N1"]}
    assert ctx.client.arguments_of("highlight_entities") == {"handles": ["A"], "restore": True}


def test_clear_noop_without_arguments():
    ctx = make_ctx(client=FakeMcpClient())
    clear(ctx, log=lambda *a: None)
    assert ctx.client.calls == []


def test_clear_swallows_tool_errors():
    logs = []
    client = FakeMcpClient(responses={"delete_entities": McpToolError("删不掉")})
    ctx = make_ctx(client=client)
    clear(ctx, handles=["N1"], log=logs.append)
    assert any("撤销失败" in line for line in logs)