aps-agent/tests/golden/test_mom_pack.py

# ============================================================
# MOM 主数据收集表导入黄金测试
# ============================================================
from __future__ import annotations

from pathlib import Path

from server.importers.mom_pack import import_mom_excel, is_mom_workbook, parse_mom_workbook
from server.state.seed import empty_world


def _mom_path() -> Path:
    root = Path(r"D:/ItemSpace/14.工业智核")
    files = [
        p for p in root.rglob("*.xlsx")
        if ("MOM" in p.name or "主数据收集" in p.name) and not p.name.startswith("~$")
    ]
    assert files, "测试机需有湖南锐扬 MOM 主数据收集表.xlsx"
    return files[0]


def test_detect_and_parse_ruiyang_mom():
    path = _mom_path()
    assert is_mom_workbook(str(path))
    pack = parse_mom_workbook(str(path))
    assert pack["stats"]["materials"] >= 10
    assert pack["stats"]["bom"] >= 5
    assert pack["stats"]["equipment"] >= 5
    assert pack["stats"]["orders"] >= 1
    assert pack["stats"]["rootProduct"]
    equipment_codes = [row["code"] for row in pack["flexEquipment"]]
    assert len(equipment_codes) == len(set(equipment_codes))
    assert not any(code.endswith(("编码", "编号")) for code in equipment_codes)


def test_import_mom_fills_world_counts():
    path = _mom_path()
    world = empty_world()
    result = import_mom_excel(world, str(path), replace=True)
    assert result["applied"]["materials"] >= 10
    assert len(world.get("flexMaterials") or []) >= 10
    assert len(world.get("flexEquipment") or []) >= 5
    assert len(world.get("flexOrders") or []) >= 1
    assert len(world.get("flexBom") or []) >= 5
    # 自制件有默认工艺占位
    assert len(world.get("flexRoutings") or []) >= 5


def test_deep_analysis_imports_real_mom_and_knowledge(tmp_path, monkeypatch):
    import uuid

    from server.aps_domain.project_analyze import analyze_project_deep
    from server.auth.context import IdentityContext, bind_identity, reset_identity
    from server.knowledge.assets import KnowledgeStore

    path = _mom_path()

    class FakePS:
        def snapshot(self, include_messages=False):
            return {
                "projects": [{"id": "p-mom", "name": "锐扬 MOM 项目", "workDir": str(path.parent)}],
                "sessions": [{"id": "s-mom", "projectId": "p-mom"}],
                "files": [],
            }

    counters: dict[str, int] = {}

    def next_id(kind: str) -> int:
        counters[kind] = counters.get(kind, 0) + 1
        return counters[kind]

    identity = IdentityContext(
        user_id=88,
        username="mom_owner",
        fullname="锐扬主数据用户",
        tenant_uuid=f"tenant-mom-{uuid.uuid4().hex[:10]}",
        roles=("planner",),
    )
    token = bind_identity(identity)
    try:
        kb = KnowledgeStore(path=str(tmp_path / "mom-knowledge.json"))
        kb.assets = []
        kb._write()
        monkeypatch.setattr("server.state.projects.get_project_store", lambda: FakePS())
        monkeypatch.setattr("server.knowledge.assets.get_knowledge", lambda: kb)

        world = empty_world()
        result = analyze_project_deep(world, "s-mom", next_id=next_id)

        assert any(str(source).startswith("MOM:") for source in result["sources"])
        assert result["summary"]["materials"] >= 10
        assert result["summary"]["equipment"] >= 5
        assert result["summary"]["bom"] >= 5
        assert result["owner"]["userId"] == 88
        assert result["knowledgeIngest"].get("assetId")
        assert result["knowledgeIngest"].get("chunkCount", 0) >= 1
        assert any(f.get("name") == path.name for f in (result.get("folder") or {}).get("files") or [])
    finally:
        reset_identity(token)


def test_deep_analysis_accepts_explicit_mom_path_without_project_context(tmp_path, monkeypatch):
    import uuid

    from server.aps_domain.project_analyze import analyze_project_deep
    from server.auth.context import IdentityContext, bind_identity, reset_identity
    from server.knowledge.assets import KnowledgeStore

    path = _mom_path()
    kb = KnowledgeStore(path=str(tmp_path / "explicit-path-knowledge.json"))
    kb.assets = []
    kb._write()
    monkeypatch.setattr("server.aps_domain.project_analyze._project_ctx", lambda _sid: (None, ""))
    monkeypatch.setattr("server.knowledge.assets.get_knowledge", lambda: kb)
    monkeypatch.setattr("server.knowledge.embedding.index_units", lambda _units: None)

    counters: dict[str, int] = {}

    def next_id(kind: str) -> int:
        counters[kind] = counters.get(kind, 0) + 1
        return counters[kind]

    identity = IdentityContext(
        user_id=89,
        username="explicit_mom_owner",
        fullname="路径导入用户",
        tenant_uuid=f"tenant-explicit-mom-{uuid.uuid4().hex[:10]}",
        roles=("planner",),
    )
    token = bind_identity(identity)
    try:
        world = empty_world()
        result = analyze_project_deep(
            world,
            "missing-project-session",
            query=f'分析 "{path}" 并写入主数据和知识库',
            next_id=next_id,
        )
    finally:
        reset_identity(token)

    assert result["workDir"] == str(path.parent)
    assert result["sourcePaths"] == [str(path)]
    assert result["summary"]["materials"] >= 10
    assert result["summary"]["equipment"] >= 5
    assert result["summary"]["bom"] >= 5
    assert result["owner"]["userId"] == 89
    assert result["knowledgeIngest"].get("assetId")
    raw_chunks = [
        chunk.get("text", "")
        for asset in kb.assets
        for chunk in (asset.get("chunks") or [])
    ]
    assert any(f"源文件 {path.name}" in text for text in raw_chunks)