144 lines
5.8 KiB
Python
144 lines
5.8 KiB
Python
# ============================================================
|
||
# M-F 黄金测试：引导式排产向导（缺路线→模板推荐→P2生成→试排放行）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from server.agent_core import dialog, harness
|
||
from server.contracts import AgentReply, IntentResult
|
||
from server.state.seed import seed_world
|
||
|
||
|
||
class FakeStore:
|
||
def __init__(self):
|
||
self.data = seed_world()
|
||
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.delenv("APS_DB_DISABLED", raising=False)
|
||
from server.db.database import reset_engine
|
||
reset_engine()
|
||
dialog._SESSIONS.clear()
|
||
yield
|
||
dialog._SESSIONS.clear()
|
||
reset_engine()
|
||
|
||
|
||
def test_wizard_all_green_offers_run_then_releases_schedule_intent():
|
||
store = FakeStore()
|
||
reply = dialog.start_wizard(store, "w1")
|
||
assert isinstance(reply, AgentReply)
|
||
assert "可排" in reply.text or "齐备" in reply.text
|
||
out = dialog.pre_route(store, "w1", "排")
|
||
assert isinstance(out, IntentResult)
|
||
assert out.intent == "flex.schedule"
|
||
assert out.params.get("sortMode") == "BOTTLENECK"
|
||
|
||
|
||
def test_wizard_recommends_template_for_missing_routing():
|
||
store = FakeStore()
|
||
# 新产品 + 订单，无工艺路线 → 触发模板推荐
|
||
store.data["flexMaterials"].append({
|
||
"code": "SHAFT-01", "name": "传动轴（车削件）", "type": "FINISHED_PRODUCT",
|
||
"unit": "件", "stock": 0, "inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0})
|
||
store.data["flexOrders"].append({
|
||
"id": 99, "orderNo": "FO-9901", "productCode": "SHAFT-01", "quantity": 20,
|
||
"dueDate": "2099-12-31", "priority": 1, "wbs": "", "productionController": "测试",
|
||
"status": "RELEASED"})
|
||
|
||
reply = dialog.start_wizard(store, "w2")
|
||
assert "没有工艺路线" in reply.text
|
||
assert reply.blocks and reply.blocks[0].type == "wizard"
|
||
tpls = reply.blocks[0].props.get("templates") or []
|
||
assert tpls, "应推荐至少一个行业模板"
|
||
# 产品名含「车削」→ 车削模板应在推荐里
|
||
assert any(t["code"] == "TPL-TURNING" for t in tpls)
|
||
|
||
# 用户选 1 → 出 P2 确认卡
|
||
reply2 = dialog.pre_route(store, "w2", "1")
|
||
assert isinstance(reply2, AgentReply)
|
||
confirm_blocks = [b for b in reply2.blocks if b.type == "confirm-card"]
|
||
assert confirm_blocks, "选模板后应出确认卡"
|
||
confirm_id = confirm_blocks[0].props["confirmId"]
|
||
|
||
# 批准 → 路线生成（工时标「模板」）
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
msg = execute_confirmed(store, confirm_id, approve=True, actor="tester")
|
||
assert "工艺路线已生成" in msg
|
||
steps = [r for r in store.data["flexRoutings"] if r["productCode"] == "SHAFT-01"]
|
||
assert steps
|
||
assert all(s["stdTimeSource"] == "模板" for s in steps)
|
||
|
||
# 继续排产 → 全绿（或仅剩警告）→ 给试排确认
|
||
reply3 = dialog.pre_route(store, "w2", "继续排产")
|
||
assert isinstance(reply3, AgentReply)
|
||
assert "可排" in reply3.text or "齐备" in reply3.text
|
||
|
||
|
||
def test_wizard_asks_time_for_unmaintained_steps():
|
||
store = FakeStore()
|
||
# 摘掉 OP-WELD 设备默认工时 → PDU 产品该步「待维护」
|
||
for e in store.data["flexEquipment"]:
|
||
e.get("opStdTime", {}).pop("OP-WELD", None)
|
||
reply = dialog.start_wizard(store, "w3")
|
||
assert "没有工时" in reply.text or "工序" in reply.text
|
||
assert dialog._SESSIONS["w3"]["wizard"]["step"] == "fill_time"
|
||
|
||
# 回复分钟数 → 出 P2 工时确认卡
|
||
reply2 = dialog.pre_route(store, "w3", "12")
|
||
assert isinstance(reply2, AgentReply)
|
||
confirm_blocks = [b for b in reply2.blocks if b.type == "confirm-card"]
|
||
assert confirm_blocks
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
msg = execute_confirmed(store, confirm_blocks[0].props["confirmId"], approve=True, actor="tester")
|
||
assert "工时已更新" in msg
|
||
weld_steps = [r for r in store.data["flexRoutings"] if r["operationCode"] == "OP-WELD"]
|
||
assert all(s.get("stdTimePerUnit") == 12.0 for s in weld_steps)
|
||
assert all(s.get("stdTimeSource") == "实测" for s in weld_steps)
|
||
|
||
|
||
def test_wizard_default_calendar():
|
||
store = FakeStore()
|
||
store.data["flexCalendar"] = []
|
||
reply = dialog.start_wizard(store, "w4")
|
||
assert "班次日历" in reply.text
|
||
reply2 = dialog.pre_route(store, "w4", "默认日历")
|
||
assert isinstance(reply2, AgentReply)
|
||
assert "默认日历" in reply2.text
|
||
assert store.data["flexCalendar"], "应写入默认日历"
|
||
|
||
|
||
def test_wizard_cancel():
|
||
store = FakeStore()
|
||
dialog.start_wizard(store, "w5")
|
||
reply = dialog.pre_route(store, "w5", "取消")
|
||
assert isinstance(reply, AgentReply)
|
||
assert "退出" in reply.text
|
||
assert dialog._SESSIONS["w5"]["wizard"] is None
|
||
|
||
|
||
def test_wizard_intent_rules():
|
||
from server.agent_core.intent import parse_fast
|
||
world = seed_world()
|
||
for phrase in ("我要排产", "怎么排产", "排产向导"):
|
||
r = parse_fast(phrase, world)
|
||
assert r is not None and r.intent == "schedule.wizard", phrase
|
||
# 「帮我排产 / 给我排产」是祈使开排，不再进向导菜单
|
||
for phrase in ("帮我排产", "给我排产", "我要你排产", "排产吧"):
|
||
r = parse_fast(phrase, world)
|
||
assert r is not None and r.intent in ("flex.schedule", "folder.schedule"), phrase
|
||
# 既有口令不受影响
|
||
r = parse_fast("柔性排产", world)
|
||
assert r is not None and r.intent == "flex.schedule"
|