190 lines
9.2 KiB
Python
190 lines
9.2 KiB
Python
"""Scheduling entry must reach a real nonempty plan, not only a blocked response.
|
|
|
|
The real source is selected externally. A positive case adds only the explicitly
|
|
configured test worker; source inventory, orders and execution facts stay intact.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
import pytest
|
|
|
|
from server.auth.context import bind_identity, reset_identity
|
|
from tests.auth_provider import TestAuthProvider
|
|
from tests.golden.test_ruiyang_intake_workflow import (
|
|
_chat,
|
|
_confirm,
|
|
)
|
|
from tests.golden.test_ruiyang_intake_workflow import (
|
|
intake_client as intake_client, # noqa: PLC0414 -- explicit pytest fixture re-export
|
|
)
|
|
from tests.workbook_acceptance import load_expectations
|
|
|
|
|
|
def _review(client, project_id, session_id):
|
|
blocks, text = _chat(client, project_id, session_id, "分析一下数据文件")
|
|
review = next(block for block in blocks if block["type"] == "folder-pack")
|
|
return review, blocks, text
|
|
|
|
|
|
def _entry(review, action_id="start_schedule"):
|
|
actions = review["props"].get("nextActions") or []
|
|
assert actions, "资料检查后必须保留可操作的排产入口"
|
|
action = next(action for action in actions if action["id"] == action_id)
|
|
if action_id == "start_schedule":
|
|
assert action["label"] == "开始排产"
|
|
assert action["command"] == "立即排产"
|
|
elif action_id == "trial_schedule":
|
|
assert action["label"] == "试排"
|
|
assert action["command"] == "立即排产"
|
|
else:
|
|
assert action["label"] == "查看缺少资料"
|
|
assert action["command"] == "排产还缺什么"
|
|
assert action["enabled"] is True
|
|
return action
|
|
|
|
|
|
def _adopt(client, project_id, session_id):
|
|
_review_block, blocks, _text = _review(client, project_id, session_id)
|
|
card = next(block for block in blocks if block["type"] == "confirm-card")
|
|
result = _confirm(client, session_id, card)
|
|
assert result["refresh"] is True, result
|
|
|
|
|
|
def _supply_snapshot(world):
|
|
return {row["code"]: {key: row.get(key) for key in ("stock", "inTransit", "expectedArrivalDate")}
|
|
for row in world.get("flexMaterials", [])}
|
|
|
|
|
|
def _add_only_configured_test_worker(project_id):
|
|
"""Seed one test fact in the authenticated isolated project, never production."""
|
|
from server.state.store import get_store
|
|
|
|
expected = load_expectations()
|
|
identity = TestAuthProvider("round87-workflow")._identity("planner")
|
|
token = bind_identity(identity)
|
|
try:
|
|
store = get_store()
|
|
assert store.world_key == project_id
|
|
before = {"inventory": _supply_snapshot(store.data),
|
|
"orders": copy.deepcopy(store.data["flexOrders"]),
|
|
"wip": copy.deepcopy(store.data["flexWip"]),
|
|
"peopleCount": len(store.data["flexPersonnel"])}
|
|
worker = copy.deepcopy(expected["trial"]["supplementPerson"])
|
|
worker["sourceRef"] = {"kind": "isolated_test_supplement"}
|
|
assert not any(person["code"] == worker["code"] for person in store.data["flexPersonnel"])
|
|
store.data["flexPersonnel"].append(worker)
|
|
store.save()
|
|
return before
|
|
finally:
|
|
reset_identity(token)
|
|
|
|
|
|
def test_entry_before_adoption_reaches_review_without_scheduling(intake_client):
|
|
client, project_id, session_id = intake_client
|
|
review, _blocks, _text = _review(client, project_id, session_id)
|
|
assert review["props"]["adopted"] is False
|
|
assert review["props"]["canSchedule"] is False
|
|
action = _entry(review, "review_missing")
|
|
blocks, text = _chat(client, project_id, session_id, action["command"])
|
|
assert any(block["type"] == "confirm-card" and block["props"]["action"] == "import.commit" for block in blocks), text
|
|
assert not any(block["type"] == "flex-schedule" for block in blocks)
|
|
assert client.get("/api/master").json()["materials"] == []
|
|
assert client.get("/api/flex/world").json()["workOrders"] == []
|
|
|
|
|
|
def test_original_source_keeps_enabled_entry_and_reports_real_blockers(intake_client):
|
|
client, project_id, session_id = intake_client
|
|
_adopt(client, project_id, session_id)
|
|
review, blocks, _text = _review(client, project_id, session_id)
|
|
assert review["props"]["adopted"] is True
|
|
assert review["props"]["canSchedule"] is False
|
|
assert review["props"]["trialAvailable"] is True
|
|
_entry(review, "trial_schedule")
|
|
assert not any(block["type"] == "confirm-card" for block in blocks)
|
|
action = _entry(review, "review_missing")
|
|
planned, text = _chat(client, project_id, session_id, action["command"])
|
|
refreshed = next(block for block in planned if block["type"] == "folder-pack")
|
|
assert refreshed["props"]["canSchedule"] is False
|
|
assert refreshed["props"]["nextActions"][0]["id"] == "review_missing"
|
|
assert not any(block["type"] == "flex-schedule" for block in planned), text
|
|
assert not any(block["type"] == "confirm-card" for block in planned)
|
|
|
|
|
|
@pytest.mark.parametrize("command", ["立即排产", "在 APS 系统中手动触发排产"])
|
|
def test_real_chat_entry_generates_nonempty_plan_with_only_explicit_worker_supplement(intake_client, command):
|
|
client, project_id, session_id = intake_client
|
|
expected = load_expectations()
|
|
_adopt(client, project_id, session_id)
|
|
before = _add_only_configured_test_worker(project_id)
|
|
review, _blocks, _text = _review(client, project_id, session_id)
|
|
assert review["props"]["canSchedule"] is True
|
|
_entry(review)
|
|
planned, text = _chat(client, project_id, session_id, command)
|
|
result = next(block["props"] for block in planned if block["type"] == "flex-schedule")
|
|
assert result["trialOnly"] is True and result["productionReady"] is False
|
|
assert result["stats"]["woCount"] > 0, text
|
|
assert 0 < result["stats"]["vlCount"] < expected["entityCounts"]["orders"]
|
|
assert result["stats"]["blockedOrderCount"] > 0, "测试补充不能掩盖其他订单的真实缺项"
|
|
assert not any(block["type"] == "confirm-card" for block in planned), text
|
|
flex = client.get("/api/flex/world").json()
|
|
assert len(flex["workOrders"]) == result["stats"]["woCount"]
|
|
assert all(row["equipmentCode"] and row["plannedStartTime"] and row["plannedEndTime"]
|
|
for row in flex["workOrders"])
|
|
assert not set(expected["sandboxOrderNos"]) & {row["flexOrderNo"] for row in flex["workOrders"]}
|
|
assert result["conflicts"], "未解决的在制、库存缺项必须继续展示"
|
|
|
|
from server.state.store import get_store
|
|
token = bind_identity(TestAuthProvider("round87-workflow")._identity("planner"))
|
|
try:
|
|
world = get_store().data
|
|
assert _supply_snapshot(world) == before["inventory"]
|
|
assert world["flexOrders"] == before["orders"]
|
|
assert world["flexWip"] == before["wip"]
|
|
assert len(world["flexPersonnel"]) == before["peopleCount"] + 1
|
|
version = world["flexScheduleVersions"][-1]
|
|
assert version["woCount"] == result["stats"]["woCount"]
|
|
assert version["inputSnapshot"]["planningContext"]["sourceSha256"] == expected["sourceSha256"]
|
|
finally:
|
|
reset_identity(token)
|
|
|
|
|
|
def test_review_action_survives_actual_message_persistence_and_reload(intake_client):
|
|
client, project_id, session_id = intake_client
|
|
_adopt(client, project_id, session_id)
|
|
review, _blocks, _text = _review(client, project_id, session_id)
|
|
_entry(review, "trial_schedule")
|
|
messages = [{"role": "agent", "text": "资料核对结果", "blocks": [review]}]
|
|
saved = client.put(f"/api/sessions/{session_id}/messages", json={"messages": messages})
|
|
assert saved.status_code == 200, saved.text
|
|
loaded = client.get(f"/api/sessions/{session_id}/messages")
|
|
assert loaded.status_code == 200, loaded.text
|
|
recovered = loaded.json()["messages"][0]["blocks"][0]
|
|
assert recovered == review
|
|
action = _entry(recovered, "trial_schedule")
|
|
blocks, text = _chat(client, project_id, session_id, action["command"])
|
|
assert any(block["type"] == "flex-schedule" for block in blocks), text
|
|
|
|
|
|
def test_adopted_source_with_real_blockers_exposes_and_runs_trial(intake_client):
|
|
"""现场回归：有真实阻断时仍能从卡片发起草稿试排；假设项逐条留痕，真实缺料订单保持未排出。"""
|
|
client, project_id, session_id = intake_client
|
|
expected = load_expectations()
|
|
plan = expected["trialPlan"]
|
|
_adopt(client, project_id, session_id)
|
|
review, _blocks, _text = _review(client, project_id, session_id)
|
|
action = _entry(review, "trial_schedule")
|
|
|
|
planned, text = _chat(client, project_id, session_id, action["command"])
|
|
result = next(block["props"] for block in planned if block["type"] == "flex-schedule")
|
|
assert result["trialOnly"] is True
|
|
assert result["productionReady"] is False
|
|
assert result["stats"]["orderCount"] == expected["entityCounts"]["orders"]
|
|
assert result["stats"]["blockedOrderCount"] == len(plan["blockedOrderNos"])
|
|
assert result["stats"]["woCount"] > 0
|
|
assert {row["orderNo"] for row in result["lines"] if row.get("status") == "scheduled"} == set(plan["scheduledOrderNos"])
|
|
assert {row["orderNo"] for row in result["lines"] if row.get("status") == "blocked"} == set(plan["blockedOrderNos"])
|
|
assert {c["type"] for c in result["conflicts"] if c.get("severity") == plan["assumptionSeverity"]} == set(plan["assumptionTypes"])
|
|
assert result["conflicts"], text
|
|
assert not result["downloadUrl"]
|