aps-agent/tests/golden/test_dxf_drawing.py

# ============================================================
# 工程图纸 DXF 自动识别黄金测试（domain-dxf-drawing）
# 覆盖：样例图纸解析（标题栏/位置号/技术说明）、文件名交叉校验、
#       物料/BOM/工艺路线候选安全边界（missing 不编造）、SVG 预览、
#       工程目录发现、多模态提取器接线与入库 fail closed、
#       /api/dxf/* 端点、P2 确认卡 staging 与确认后写主数据。
# 样例：demand/dxf-samples/5060102101-001-e.dxf
#       （MTU 12V 183 TB32 发动机安装图，AC1015，德英双语标题栏）
# ============================================================
from __future__ import annotations

import os

import pytest
from fastapi.testclient import TestClient

from server.agent_core import harness
from server.aps_domain.dxf_drawing import (
    discover_dxf_files,
    drawing_to_master_candidates,
    dxf_svg_cached,
    dxf_to_svg,
    get_cached_drawing,
    parse_dxf,
    parse_dxf_bytes,
)
from server.aps_domain.masterdata import confirmation_for_master_action
from server.aps_domain.multimodal import candidates_to_batches, default_registry
from server.aps_domain.workflow import execute_confirmed
from server.auth.context import IdentityContext, bind_identity, reset_identity
from server.state.seed import seed_world

_SAMPLE = os.path.join(os.path.dirname(__file__), "..", "..",
                       "demand", "dxf-samples", "5060102101-001-e.dxf")
_SAMPLE = os.path.abspath(_SAMPLE)
_PLANNER = IdentityContext(1001, "planner", "计划员", "tenant-dxf", roles=("planner",))


def _next_id(world):
    counters: dict[str, int] = {}

    def next_id(kind: str) -> int:
        table = "auditEvents" if kind == "audit" else kind + "s"
        if kind not in counters:
            rows = world.get(table, [])
            counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0)
        counters[kind] += 1
        return counters[kind]

    return next_id


class _WorldStore:
    def __init__(self, world, next_id_fn):
        self.data = world
        self._next_id_fn = next_id_fn
        self.tenant_uuid = "tenant-dxf"
        self.world_key = "personal-1001"

    def next_id(self, kind: str) -> int:
        return self._next_id_fn(kind)

    def save(self) -> None:
        pass


@pytest.fixture(scope="module")
def parsed():
    assert os.path.isfile(_SAMPLE), f"样例图纸缺失：{_SAMPLE}"
    return parse_dxf(_SAMPLE)


# ---------------- 解析：标题栏 / 位置号 / 技术说明 ----------------

def test_parse_sample_title_block(parsed):
    assert parsed["dxfVersion"] == "AC1015"
    tb = parsed["titleBlock"]
    assert tb["found"] is True
    assert tb["source"] == "block-attribs"
    assert tb["blockName"] == "SF_STD_A"
    fields = tb["fields"]
    dn = fields["drawingNumber"]
    assert dn["value"] == "506 010 21 01"
    assert dn["normalized"] == "5060102101"
    assert dn["confidence"] == pytest.approx(0.95)     # 属性 + 文件名一致互证
    assert "12V 183 TB32" in fields["title"]["value"]
    assert fields["scale"]["value"] == "1:5"
    assert fields["sheet"]["value"] == "1"
    assert fields["drawingType"]["value"] == "EB/Installation"
    assert fields["applicableTo"]["value"] == "OM 444 LA"
    assert fields["revision"]["value"] == "e"          # 文件名后缀 + B_FREI 互证
    assert fields["revision"]["confidence"] == pytest.approx(0.9)
    # 未填写字段保持空 + missing（不编造材质）
    assert fields["materialSpec"]["value"] == ""
    assert fields["materialSpec"]["missing"] is True


def test_parse_positions_notes_and_stats(parsed):
    positions = {p["position"] for p in parsed["positionNumbers"]}
    assert {"14.001", "06.020", "15.108", "09.152"} <= positions
    assert len(parsed["positionNumbers"]) >= 30
    assert len(parsed["technicalNotes"]) >= 5
    assert parsed["entityStats"]["LINE"] > 0
    assert parsed["entityStats"]["ARC"] > 0
    assert any(l["name"] == "RAHM2" for l in parsed["layers"])
    assert parsed["extents"]["maxX"] > parsed["extents"]["minX"]
    assert parsed["warnings"] == []


def test_parse_fail_closed():
    with pytest.raises(ValueError, match="不存在"):
        parse_dxf(r"D:\no-such-dir\no-such-file.dxf")
    bad = os.path.join(os.path.dirname(_SAMPLE), "not-a-dxf.dxf")
    with open(bad, "w", encoding="utf-8") as fh:
        fh.write("这不是 DXF 内容")
    try:
        with pytest.raises(ValueError, match="解析失败"):
            parse_dxf(bad)
    finally:
        os.unlink(bad)


def test_parse_bytes_uses_display_name():
    with open(_SAMPLE, "rb") as fh:
        data = fh.read()
    parsed = parse_dxf_bytes("5060102101-001-e.dxf", data)
    assert parsed["filename"] == "5060102101-001-e.dxf"
    assert parsed["fileMeta"]["revision"] == "e"
    assert parsed["titleBlock"]["fields"]["revision"]["value"] == "e"
    entry = get_cached_drawing(parsed["drawingId"])
    assert entry is not None and entry.get("tempPath")


# ---------------- 候选：安全边界（missing 不编造） ----------------

def test_master_candidates_boundary(parsed):
    cands = drawing_to_master_candidates(parsed)
    material = cands["material"]
    assert material["code"] == "5060102101-001"        # 张次沿用文件名 001 写法
    assert "12V 183 TB32" in material["name"]
    assert material["type"] == "FINISHED_PRODUCT"      # 安装图推断（置信封顶 0.6）
    assert material["requiresConfirm"] is True
    assert material["confidence"] <= 0.95
    assert material["materialSpec"] == ""              # 图纸未承载：不编造材质
    assert any("材质" in x for x in material["notFromDrawing"])

    bom = cands["bom"]
    assert bom["complete"] is False
    assert bom["requiresConfirm"] is True
    assert len(bom["items"]) >= 30
    item = bom["items"][0]
    assert item["missing"] == ["materialCode", "name", "quantity"]
    assert item["quantity"] is None                    # 数量绝不臆造

    routing = cands["routing"]
    assert routing["complete"] is False
    assert routing["steps"] == []                      # 工序/工时留空待人工编制
    assert "人工确认" in routing["note"]


# ---------------- SVG 预览与目录发现 ----------------

def test_svg_preview(parsed):
    svg = dxf_svg_cached(parsed["drawingId"])
    assert svg.startswith("<?xml") and "<svg" in svg
    assert len(svg) > 100_000
    svg2 = dxf_to_svg(_SAMPLE)
    assert "<svg" in svg2
    with pytest.raises(ValueError, match="未解析"):
        dxf_svg_cached("0" * 16)


def test_discover_dxf_files():
    result = discover_dxf_files(os.path.dirname(_SAMPLE))
    assert result["count"] >= 1
    hit = next(f for f in result["files"]
               if f["filename"] == "5060102101-001-e.dxf")
    assert hit["fileMeta"]["drawingNumberNormalized"] == "5060102101"
    assert hit["fileMeta"]["sheet"] == "001"
    assert hit["fileMeta"]["revision"] == "e"
    with pytest.raises(ValueError, match="不存在"):
        discover_dxf_files(r"D:\no-such-dir\x")


# ---------------- 多模态接线：提取可用 / 入库 fail closed ----------------

def test_multimodal_dxf_extractor():
    registry = default_registry(world=seed_world())
    assert "dxf_drawing" in registry.list_kinds()
    cands = registry.extract("dxf_drawing", {"path": _SAMPLE})
    assert len(cands) == 1
    c = cands[0]
    assert c.kind == "dxf_drawing" and c.source == "DXF_PARSE"
    assert c.complete is True
    assert c.confidence >= 0.9
    assert c.value["material"]["code"] == "5060102101-001"
    assert c.value["positionCount"] >= 30
    # DXF 候选不经 importers 批次：显式报错并指引 /api/dxf/stage
    with pytest.raises(ValueError, match="/api/dxf/stage"):
        candidates_to_batches([c])


# ---------------- P2 确认卡：确认前不写世界，确认后写主数据 ----------------

def test_dxf_stage_then_confirm_writes_material(parsed):
    """候选 → master.material.upsert P2 卡 → 批准写物料 / 驳回不写。"""
    world = seed_world()
    next_id = _next_id(world)
    material = drawing_to_master_candidates(parsed)["material"]
    payload = {
        "code": material["code"], "name": material["name"],
        "type": material["type"], "unit": material["unit"],
        "spec": material["spec"],
    }
    title, lines = confirmation_for_master_action(
        world, "master.material.upsert", payload)
    assert material["code"] in title or material["name"] in title
    assert lines
    token = bind_identity(_PLANNER)
    try:
        block = harness.stage_confirmation(
            "dxf-test", "master.material.upsert", payload,
            title=title, summary_lines=lines)
        cid = str(block.props["confirmId"])
        assert not any(m["code"] == material["code"] for m in world["materials"])
        store = _WorldStore(world, next_id)
        message = execute_confirmed(store, cid, True, actor="planner")
        assert "物料" in message or "完成" in message
        written = next(m for m in world["materials"] if m["code"] == material["code"])
        assert written["name"] == material["name"]
        assert any(e["action"] == "master.material.upsert"
                   for e in world["auditEvents"])
    finally:
        reset_identity(token)


def test_dxf_stage_validation_fail_closed(parsed):
    """缺名称/编码的候选不出卡：confirmation_for_master_action 直接报错。"""
    world = seed_world()
    with pytest.raises(ValueError):
        confirmation_for_master_action(
            world, "master.material.upsert", {"code": "", "name": ""})


# ---------------- 网关接线：/api/dxf/* ----------------

class _AuditStore:
    def __init__(self, tenant_uuid: str = "tenant-dxf"):
        self.tenant_uuid = tenant_uuid
        self.world_key = "personal-1001"
        self.data: dict = seed_world()
        self._counter = 0

    def next_id(self, _kind: str) -> int:
        self._counter += 1
        return self._counter

    def save(self) -> None:
        pass


class _CheckpointStore:
    def create(self, *_args, **_kwargs) -> dict:
        return {"pairId": "dxf-pair"}

    def get(self, pair_id: str) -> dict | None:
        return {"pairId": pair_id}


class _ProjectStore:
    def active_world_key(self) -> str:
        return "default"

    def require_active_write(self) -> None:
        pass


@pytest.fixture
def client_factory(monkeypatch):
    import server.gateway.app as gateway_module
    import server.state.projects as projects_module
    from tests.auth_provider import install_test_auth

    provider = install_test_auth(monkeypatch, "tenant-dxf")

    def factory():
        store = _AuditStore()
        monkeypatch.setattr(gateway_module, "get_store", lambda: store)
        monkeypatch.setattr(gateway_module, "get_checkpoints", lambda: _CheckpointStore())
        monkeypatch.setattr(projects_module, "get_project_store", lambda: _ProjectStore())
        client = TestClient(gateway_module.create_app())
        login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
        assert login.status_code == 200, login.text
        return client, store, provider

    return factory


def test_api_parse_path_svg_and_discover(client_factory):
    client, store, _ = client_factory()
    resp = client.post("/api/dxf/parse-path", json={"path": _SAMPLE})
    assert resp.status_code == 200, resp.text
    data = resp.json()
    drawing = data["drawing"]
    assert drawing["titleBlock"]["fields"]["drawingNumber"]["normalized"] == "5060102101"
    assert data["candidates"]["material"]["code"] == "5060102101-001"
    assert any(e["action"] == "dxf.parse" for e in store.data["auditEvents"])

    svg_resp = client.get(f"/api/dxf/{drawing['drawingId']}/svg")
    assert svg_resp.status_code == 200
    assert svg_resp.headers["content-type"].startswith("image/svg+xml")
    assert "<svg" in svg_resp.text

    disc = client.get("/api/dxf/discover", params={"dir": os.path.dirname(_SAMPLE)})
    assert disc.status_code == 200
    assert disc.json()["count"] >= 1

    bad = client.post("/api/dxf/parse-path", json={"path": r"D:\no-such\x.dxf"})
    assert bad.status_code == 200 and "error" in bad.json()
    missing = client.get("/api/dxf/ffffffffffffffff/svg")
    assert missing.status_code == 404


def test_api_dxf_stage_material_card(client_factory):
    client, store, _ = client_factory()
    resp = client.post("/api/dxf/parse-path", json={"path": _SAMPLE})
    drawing = resp.json()["drawing"]
    material = resp.json()["candidates"]["material"]
    before = len(store.data["materials"])
    stage = client.post("/api/dxf/stage", json={
        "drawingId": drawing["drawingId"],
        "target": "material",
        "payload": {"code": material["code"], "name": material["name"],
                    "type": material["type"], "unit": material["unit"],
                    "spec": material["spec"]},
    })
    assert stage.status_code == 200, stage.text
    body = stage.json()
    assert "P2 确认队列" in body["message"]
    assert body["block"]["props"]["confirmId"]
    assert len(store.data["materials"]) == before      # 确认前绝不写世界
    assert any(e["action"] == "dxf.stage.material" for e in store.data["auditEvents"])

    bad = client.post("/api/dxf/stage", json={
        "target": "material", "payload": {"code": "", "name": ""}})
    assert "error" in bad.json()
    unknown = client.post("/api/dxf/stage", json={"target": "magic", "payload": {}})
    assert "error" in unknown.json()


def test_svg_layer_filter_variant(parsed):
    """图层开关：过滤渲染显著小于全量渲染，且变体缓存命中。"""
    all_svg = dxf_svg_cached(parsed["drawingId"])
    only6 = dxf_svg_cached(parsed["drawingId"], include_layers=["6"])
    assert "<svg" in only6
    assert len(only6) < len(all_svg) // 2             # 单图层远小于全图
    again = dxf_svg_cached(parsed["drawingId"], include_layers=["6"])
    assert again is only6                             # 变体缓存命中（同一对象）


def test_api_svg_layer_param(client_factory):
    client, _, _ = client_factory()
    resp = client.post("/api/dxf/parse-path", json={"path": _SAMPLE})
    drawing = resp.json()["drawing"]
    full = client.get(f"/api/dxf/{drawing['drawingId']}/svg")
    filtered = client.get(f"/api/dxf/{drawing['drawingId']}/svg", params={"layers": "6"})
    assert full.status_code == 200 and filtered.status_code == 200
    assert len(filtered.text) < len(full.text) // 2