aps-agent/tests/golden/test_dxf_discovery.py

from __future__ import annotations

import os
import shutil
from pathlib import Path

import pytest

from server.aps_domain.drawing_dxf import inspect_dxf
from server.aps_domain.folder_pack import analyze_work_dir
from server.state.seed import seed_world
from tests.external_data import external_dir

_DEFAULT_SAMPLE = external_dir("RUIYANG_DEMO_DIR", "ruiyang") / "5060102101-001-e(1).dxf"


def _sample_dxf() -> Path:
    try:
        import ezdxf  # noqa: F401
    except ImportError as exc:
        pytest.skip(f"DXF 运行依赖未完整安装：{exc}; 需要 ezdxf 及其 fontTools 依赖")
    path = Path(os.environ.get("APS_DXF_SAMPLE", str(_DEFAULT_SAMPLE)))
    if not path.is_file():
        pytest.skip(f"真实 DXF 样例不存在：{path}；可通过 APS_DXF_SAMPLE 指定")
    return path


def _bind_project(monkeypatch: pytest.MonkeyPatch, work_dir: Path) -> None:
    class FakeProjectStore:
        def snapshot(self, include_messages: bool = False):
            return {
                "projects": [{"id": "p-dxf", "name": "DXF验收项目", "workDir": str(work_dir)}],
                "sessions": [{"id": "s-dxf", "projectId": "p-dxf"}],
                "files": [],
            }

    monkeypatch.setattr("server.state.projects.get_project_store", lambda: FakeProjectStore())


def test_real_dxf_is_discoverable_and_has_stable_identity():
    parsed = inspect_dxf(_sample_dxf())

    assert parsed["contractVersion"] == "drawing-dxf.v1"
    assert parsed["asset"]["kind"] == "DXF"
    assert parsed["asset"]["filename"].lower().endswith(".dxf")
    assert parsed["asset"]["sizeBytes"] > 1_000_000
    assert len(parsed["asset"]["sha256"]) == 64
    assert parsed["asset"]["idempotencyKey"] == f'drawing:{parsed["asset"]["sha256"]}'
    assert parsed["drawing"]["drawingNumber"] == "5060102101-001"
    assert parsed["drawing"]["revision"].lower() == "e"


def test_project_folder_contract_discovers_dxf_without_treating_it_as_tabular(monkeypatch, tmp_path: Path):
    sample = _sample_dxf()
    copied = tmp_path / sample.name
    shutil.copyfile(sample, copied)
    _bind_project(monkeypatch, tmp_path)

    report = analyze_work_dir(seed_world(), "s-dxf")
    drawing_rows = report.get("drawings") or [
        row for row in report.get("files", [])
        if str(row.get("name") or "").lower().endswith(".dxf")
    ]
    if not drawing_rows:
        pytest.skip(
            "依赖未合并：folder_pack.analyze_work_dir 尚未返回 drawings 或 DXF 文件行；"
            "接线后本契约要求工程目录可发现 DXF，且不把它计入 Excel/CSV 有效行。"
        )

    assert any(str(row.get("name") or "").lower() == copied.name.lower() for row in drawing_rows)
    assert report["totalOk"] == 0
    assert report["totalErrors"] == 0


def test_folder_pack_without_dxf_keeps_existing_tabular_behavior(monkeypatch, tmp_path: Path):
    (tmp_path / "订单.csv").write_text(
        "订单号,产品编码,数量,交期,客户\nSO-DXF-REG,P-1,10,2026-08-20,验收客户\n",
        encoding="utf-8-sig",
    )
    (tmp_path / "物料.csv").write_text(
        "编码,名称,类型,单位,库存\nP-1,回归成品,FINISHED_PRODUCT,件,100\n",
        encoding="utf-8-sig",
    )
    _bind_project(monkeypatch, tmp_path)

    report = analyze_work_dir(seed_world(), "s-dxf")

    assert report["ok"] is True
    assert report["totalOk"] >= 1
    assert report["totalErrors"] == 0
    assert any(row["name"] == "订单.csv" for row in report["files"])
    assert not any(str(row.get("name") or "").lower().endswith(".dxf") for row in report["files"])