99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
# ============================================================
|
||
# M-G 端到端验收：数据入库 → 工时维护 → 柔性排产 → 外部 skill 试排 → 知识问答
|
||
# （康尼现场 Excel 在仓库外，链路用演示厂数据走通同一代码路径）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
import pytest
|
||
|
||
from server.contracts import AgentReply, IntentResult
|
||
from server.state.seed import ensure_flex_seed, seed_world
|
||
|
||
|
||
class FakeStore:
|
||
def __init__(self):
|
||
self.data = seed_world()
|
||
ensure_flex_seed(self.data)
|
||
self._counters: dict[str, int] = {}
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
self._counters[kind] = self._counters.get(kind, 0) + 1
|
||
return self._counters[kind]
|
||
|
||
def save(self) -> None:
|
||
pass
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _isolate(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "master.db"))
|
||
monkeypatch.setenv("APS_SKILLS_PATH", str(tmp_path / "skills.json"))
|
||
monkeypatch.delenv("APS_DB_DISABLED", raising=False)
|
||
import server.agent_core.skills as sk
|
||
from server.db.database import reset_engine
|
||
sk._registry = None
|
||
reset_engine()
|
||
yield
|
||
sk._registry = None
|
||
reset_engine()
|
||
|
||
|
||
def test_end_to_end_master_data_to_scheduling_to_knowledge():
|
||
store = FakeStore()
|
||
sid = "e2e"
|
||
|
||
# ---- 1. 数据地基：主数据入库 → 投影回 world 等价 ----
|
||
from server.db.database import init_db
|
||
from server.db.sync import db_has_master, db_to_world, world_to_db
|
||
init_db()
|
||
world_to_db(store.data, project_code="e2e-demo")
|
||
assert db_has_master("e2e-demo")
|
||
projected: dict = {}
|
||
db_to_world(projected, project_code="e2e-demo")
|
||
assert len(projected["flexOrders"]) == len(store.data["flexOrders"])
|
||
assert len(projected["flexRoutings"]) == len(store.data["flexRoutings"])
|
||
|
||
# ---- 2. 工时维护：制造一个「待维护」缺口，走 P2 确认卡补齐 ----
|
||
for e in store.data["flexEquipment"]:
|
||
e.get("opStdTime", {}).pop("OP-WELD", None)
|
||
from server.aps_domain.readiness import check_readiness
|
||
before = check_readiness(store.data)
|
||
assert before["summary"]["blocked"] > 0, "摘掉工时后应判定不可排"
|
||
|
||
from server.aps_domain.workflow import execute_confirmed, handle_intent
|
||
staged = asyncio.run(handle_intent(
|
||
store, sid,
|
||
IntentResult(intent="flex.time.update",
|
||
params={"operationCode": "OP-WELD", "stdMin": 25},
|
||
confidence=1.0, source="LLM"),
|
||
actor="e2e"))
|
||
assert isinstance(staged, AgentReply)
|
||
confirm = [b for b in staged.blocks if b.type == "confirm-card"]
|
||
assert confirm, "补工时应出 P2 确认卡"
|
||
msg = execute_confirmed(store, confirm[0].props["confirmId"], approve=True, actor="e2e")
|
||
assert "工时已更新" in msg
|
||
assert check_readiness(store.data)["summary"]["blocked"] == 0, "补齐工时后应全绿"
|
||
|
||
# ---- 3. 排产：全绿 → flex.schedule 工具请求 → 产出工单 ----
|
||
schedule_reply = asyncio.run(handle_intent(
|
||
store, sid,
|
||
IntentResult(intent="flex.schedule", params={}, confidence=1.0, source="LLM"),
|
||
actor="e2e"))
|
||
assert isinstance(schedule_reply, AgentReply)
|
||
assert store.data.get("flexWorkOrders"), "试排应产出柔性工单"
|
||
|
||
# ---- 4. 外部 skill 试排：algo.stub 经中性 DTO 写回草稿 ----
|
||
from server.engines.external_engine import run_external_flex
|
||
summary = run_external_flex(store, skill_id="algo.stub")
|
||
assert summary["woCount"] >= 1 and summary["runId"]
|
||
|
||
# ---- 5. 知识问答：检索命中必带出处+版本 ----
|
||
from server.knowledge.assets import get_knowledge
|
||
from server.knowledge.retrieval import hybrid_search
|
||
units = get_knowledge().iter_search_units()
|
||
hits = hybrid_search(units, "车削类零件的工艺路线怎么生成", top_k=3)
|
||
assert hits, "机加工工艺知识应命中"
|
||
assert all(h.get("assetId") and h.get("version") for h in hits), "命中必带出处与版本"
|