85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
# ============================================================
|
||
# M-E 黄金测试：行业工艺路线模板库（结构化 + 知识互链 + 实例化）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from server.state.seed import seed_world
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def db_env(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "master.db"))
|
||
monkeypatch.delenv("APS_DB_DISABLED", raising=False)
|
||
from server.db.database import reset_engine
|
||
reset_engine()
|
||
yield
|
||
reset_engine()
|
||
|
||
|
||
def test_builtin_templates_cover_machining_categories():
|
||
from server.knowledge.routing_templates import list_templates
|
||
tpls = list_templates()
|
||
categories = {t["category"] for t in tpls}
|
||
assert {"车削", "铣削", "钻镗", "磨削", "钣金", "焊接", "装配"} <= categories
|
||
for t in tpls:
|
||
assert t["steps"], t["code"]
|
||
for s in t["steps"]:
|
||
assert s["stdMinLow"] <= s["stdMinDefault"] <= s["stdMinHigh"]
|
||
assert s["equipmentType"]
|
||
|
||
|
||
def test_templates_linked_to_knowledge_assets():
|
||
from server.knowledge.routing_templates import list_templates
|
||
tpls = list_templates()
|
||
linked = [t for t in tpls if t.get("assetId")]
|
||
assert linked, "模板应与知识资产互链（assetId）"
|
||
from server.knowledge.assets import get_knowledge
|
||
asset_ids = {a["assetId"] for a in get_knowledge().assets}
|
||
assert all(t["assetId"] in asset_ids for t in linked)
|
||
|
||
|
||
def test_recommend_by_product_keywords():
|
||
from server.knowledge.routing_templates import recommend_templates
|
||
hits = recommend_templates("传动轴 车削 外圆 螺纹")
|
||
assert hits and hits[0]["code"] == "TPL-TURNING"
|
||
hits = recommend_templates("城轨门机构部装总成")
|
||
assert any(t["code"] == "TPL-RAIL-ASSY" for t in hits)
|
||
|
||
|
||
def test_apply_template_generates_schedulable_routing():
|
||
from server.engines import PoolEngine
|
||
from server.knowledge.routing_templates import apply_template_to_product
|
||
|
||
world = seed_world()
|
||
world["flexOrders"].append({
|
||
"id": 88, "orderNo": "FO-8801", "productCode": "SHAFT-X", "quantity": 10,
|
||
"dueDate": "2099-12-31", "priority": 1, "wbs": "", "productionController": "测试",
|
||
"status": "RELEASED"})
|
||
applied = apply_template_to_product(world, "TPL-TURNING", "SHAFT-X", "传动轴")
|
||
assert applied["steps"] > 0
|
||
steps = [r for r in world["flexRoutings"] if r["productCode"] == "SHAFT-X"]
|
||
assert all(r["stdTimeSource"] == "模板" for r in steps)
|
||
# 能力已补齐 → PoolEngine 可排该单
|
||
counters: dict[str, int] = {}
|
||
|
||
def nid(kind: str) -> int:
|
||
counters[kind] = counters.get(kind, 0) + 1
|
||
return counters[kind]
|
||
|
||
result = PoolEngine().solve(world, nid, sort_mode="BOTTLENECK", order_ids=[88])
|
||
assert result["orderCount"] == 1
|
||
assert result["woCount"] == len(steps)
|
||
|
||
|
||
def test_knowledge_refs_injected_into_problem_dto():
|
||
from server.aps_domain.scheduling_dto import world_to_flex_problem
|
||
world = seed_world()
|
||
problem = world_to_flex_problem(world, include_knowledge=True)
|
||
refs = problem.meta.get("knowledgeRefs")
|
||
assert isinstance(refs, list)
|
||
if refs: # 命中时必须带出处 + 版本（RAG 硬规则）
|
||
assert all(r.get("assetId") and r.get("version") for r in refs)
|
||
assert problem.schemaVersion == "1.1"
|