281 lines
16 KiB
Python
281 lines
16 KiB
Python
"""Real source file through production upload, review, confirmation and editing APIs."""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from tests.auth_provider import install_test_auth
|
||
from tests.pi_primary_adapter import install_fake_pi_tool_adapter
|
||
from tests.workbook_acceptance import load_expectations, source_for_test
|
||
|
||
EXPECTED = load_expectations()
|
||
DIGEST = EXPECTED["sourceSha256"]
|
||
|
||
|
||
@pytest.fixture
|
||
def intake_client(tmp_path, monkeypatch):
|
||
source = source_for_test()
|
||
monkeypatch.setenv("APS_SEED_DEMO", "0")
|
||
monkeypatch.setenv("APS_HOME", str(tmp_path))
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "master.db"))
|
||
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
|
||
monkeypatch.setenv("APS_DATA_DIR", str(tmp_path / "data"))
|
||
install_test_auth(monkeypatch, "round87-workflow")
|
||
install_fake_pi_tool_adapter(monkeypatch)
|
||
from server.gateway.app import create_app
|
||
with TestClient(create_app()) as client:
|
||
response = client.post("/api/auth/login", json={"method": "password", "username": "planner", "password": "test"})
|
||
assert response.status_code == 200
|
||
created = client.post("/api/projects", json={"name": "锐扬资料验收"}).json()
|
||
project_id, session_id = created["project"]["id"], created["session"]["id"]
|
||
raw = source.read_bytes()
|
||
assert hashlib.sha256(raw).hexdigest() == DIGEST
|
||
upload = client.post(f"/api/projects/{project_id}/files/upload", files=[("files", (source.name, raw,
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"))])
|
||
assert upload.status_code == 200, upload.text
|
||
assert upload.json()["saved"] == [source.name]
|
||
yield client, project_id, session_id
|
||
assert hashlib.sha256(source.read_bytes()).hexdigest() == DIGEST
|
||
|
||
|
||
def _chat(client, project_id, session_id, text):
|
||
response = client.post("/api/chat", json={"text": text, "sessionId": session_id, "projectId": project_id})
|
||
assert response.status_code == 200, response.text
|
||
events = [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith("data: ")]
|
||
return [event["block"] for event in events if event.get("type") == "block"], response.text
|
||
|
||
|
||
def _confirm(client, session_id, block, approve=True):
|
||
response = client.post("/api/actions/confirm", json={"sessionId": session_id,
|
||
"confirmId": block["props"]["confirmId"], "approve": approve})
|
||
assert response.status_code == 200, response.text
|
||
return response.json()
|
||
|
||
|
||
def test_complete_file_review_does_not_write_until_adoption(intake_client):
|
||
client, project_id, session_id = intake_client
|
||
before = client.get("/api/master").json()
|
||
blocks, _text = _chat(client, project_id, session_id, "分析一下数据文件")
|
||
review = next(block for block in blocks if block["type"] == "folder-pack")
|
||
assert review["props"]["summary"] == EXPECTED["reviewCounts"]
|
||
assert review["props"]["entityCounts"]["personnel"] == EXPECTED["entityCounts"]["personnel"]
|
||
assert review["props"]["entityCounts"]["sandboxOrders"] == EXPECTED["entityCounts"]["sandboxOrders"]
|
||
assert len(review["props"]["sheetSummary"]) == len(EXPECTED["sheets"])
|
||
assert client.get("/api/master").json()["materials"] == before["materials"]
|
||
card = next(block for block in blocks if block["type"] == "confirm-card")
|
||
assert card["props"]["action"] == "import.commit"
|
||
confirmed = _confirm(client, session_id, card)
|
||
assert confirmed["refresh"] is True, confirmed
|
||
assert "资料已采用" in confirmed["message"], confirmed
|
||
master = client.get("/api/master").json()
|
||
assert len(master["materials"]) == EXPECTED["entityCounts"]["materials"]
|
||
assert len(master["equipment"]) == EXPECTED["entityCounts"]["equipment"]
|
||
assert sum(len(route["steps"]) for route in master["routings"]) == EXPECTED["entityCounts"]["routing"]
|
||
flex = client.get("/api/flex/world").json()
|
||
assert len(flex["orders"]) == EXPECTED["entityCounts"]["orders"]
|
||
assert sum(material.get("inTransit") or 0 for material in flex["materials"]) == EXPECTED["inventory"]["inTransit"]
|
||
assert not flex.get("versions")
|
||
|
||
|
||
def test_imported_master_edit_survives_repeat_review_and_reimport(intake_client):
|
||
client, project_id, session_id = intake_client
|
||
blocks, _ = _chat(client, project_id, session_id, "分析一下数据文件")
|
||
adopted = _confirm(client, session_id, next(block for block in blocks if block["type"] == "confirm-card"))
|
||
assert adopted["refresh"] is True
|
||
assert "资料已采用" in adopted["message"], adopted
|
||
master = client.get("/api/master").json()
|
||
material = next(row for row in master["materials"] if row["code"] == EXPECTED["materialEditCode"])
|
||
result = client.post("/api/master/stage", json={"sessionId": session_id,
|
||
"action": "master.material.upsert", "payload": {"id": material["id"], "stock": 1234}}).json()
|
||
assert "block" in result, result
|
||
assert _confirm(client, session_id, result["block"])["refresh"] is True
|
||
blocks, text = _chat(client, project_id, session_id, "分析一下数据文件")
|
||
assert not any(block["type"] == "confirm-card" for block in blocks), text
|
||
assert next(row for row in client.get("/api/flex/world").json()["materials"]
|
||
if row["code"] == material["code"])["stock"] == 1234
|
||
step = master["routings"][0]["steps"][0]
|
||
staged = client.post("/api/master/stage", json={"sessionId": session_id,
|
||
"action": "master.routing.upsert", "payload": {"stepId": step["id"], "runTimePerUnit": 12.5}}).json()
|
||
assert "block" in staged, staged
|
||
assert _confirm(client, session_id, staged["block"])["refresh"] is True
|
||
updated = client.get("/api/master").json()
|
||
assert next(s for r in updated["routings"] for s in r["steps"] if s["id"] == step["id"])["runTimePerUnit"] == 12.5
|
||
|
||
|
||
def test_rejected_adoption_leaves_master_empty(intake_client):
|
||
client, project_id, session_id = intake_client
|
||
blocks, _ = _chat(client, project_id, session_id, "根据这些数据排产")
|
||
card = next(block for block in blocks if block["type"] == "confirm-card")
|
||
_confirm(client, session_id, card, approve=False)
|
||
assert client.get("/api/master").json()["materials"] == []
|
||
|
||
|
||
def test_trial_response_reports_actual_blocked_orders_and_keeps_edits(intake_client):
|
||
client, project_id, session_id = intake_client
|
||
blocks, _ = _chat(client, project_id, session_id, "分析一下数据文件")
|
||
_confirm(client, session_id, next(block for block in blocks if block["type"] == "confirm-card"))
|
||
blocks, text = _chat(client, project_id, session_id, "根据这些数据排产")
|
||
result = next(block for block in blocks if block["type"] == "flex-schedule")
|
||
plan = EXPECTED["trialPlan"]
|
||
assert result["props"]["trialOnly"] is True
|
||
assert result["props"]["stats"]["orderCount"] == EXPECTED["entityCounts"]["orders"]
|
||
# 试排把「未登记/未确认」资料降级为显式假设后仍要排出草稿；只有真实缺料订单未排出。
|
||
assert result["props"]["stats"]["blockedOrderCount"] == len(plan["blockedOrderNos"])
|
||
assert result["props"]["stats"]["woCount"] > 0
|
||
lines = {row["orderNo"]: row.get("status") for row in result["props"]["lines"]}
|
||
assert {no for no, status in lines.items() if status == "scheduled"} == set(plan["scheduledOrderNos"])
|
||
assert {no for no, status in lines.items() if status == "blocked"} == set(plan["blockedOrderNos"])
|
||
assumptions = [c for c in result["props"]["conflicts"] if c.get("severity") == plan["assumptionSeverity"]]
|
||
assert {c["type"] for c in assumptions} == set(plan["assumptionTypes"])
|
||
assert all(str(c.get("description") or "").startswith(plan["assumptionDescriptionPrefix"]) for c in assumptions)
|
||
blocking = [c for c in result["props"]["conflicts"] if c.get("severity") == "CRITICAL"]
|
||
assert {c["type"] for c in blocking} <= set(plan["blockedReasonTypes"])
|
||
assert {c["orderNo"] for c in blocking} == set(plan["blockedOrderNos"])
|
||
assert "完整工序计划" not in text
|
||
assert not result["props"]["downloadUrl"]
|
||
assert not any(block["type"] == "confirm-card" for block in blocks)
|
||
|
||
|
||
def test_adopted_review_checks_current_master_not_original_warnings(monkeypatch):
|
||
source = source_for_test()
|
||
from server.aps_domain.importers import apply_import_commit, preview_file
|
||
from server.aps_domain.planning_intake import profile_intake_reply
|
||
from server.state.seed import empty_world
|
||
from tests.golden.test_guidance import _MemStore
|
||
|
||
raw = source.read_bytes()
|
||
preview = preview_file(source.name, raw, empty_world())
|
||
store = _MemStore(empty_world())
|
||
apply_import_commit(store.data, store.next_id, preview["batches"])
|
||
store.data["flexPersonnel"].append(dict(EXPECTED["trial"]["supplementPerson"]))
|
||
report = {"files": [{"name": source.name, **{k: preview[k] for k in ("profile", "source", "entityCounts", "sheetSummary", "canCommit")}}],
|
||
"batches": preview["batches"]}
|
||
monkeypatch.setattr("server.aps_domain.folder_pack.prepare_folder_schedule", lambda *_: report)
|
||
reply = profile_intake_reply(store, "s", schedule_requested=False, schedule_current=lambda: None)
|
||
assert reply.blocks[0].props["adopted"] is True
|
||
assert not any(issue.get("code") == "NO_PERSONNEL_SKILL" for issue in reply.blocks[0].props["diagnostics"])
|
||
assert any(issue.get("code") == "NO_PERSONNEL_SKILL" for issue in reply.blocks[0].props["sourceDiagnostics"])
|
||
|
||
|
||
def _intake_file(name: str, *, can_commit: bool, digest: str) -> dict:
|
||
return {"name": name, "profile": "ruiyang-aps-v1", "canCommit": can_commit,
|
||
"source": {"sha256": digest}, "profileDigest": "profile-digest-1",
|
||
"entityCounts": {"orders": 6}, "sheetSummary": [],
|
||
"okCount": 258 if can_commit else 32, "errorCount": 0 if can_commit else 5}
|
||
|
||
|
||
def _patch_folder_report(monkeypatch, files: list[dict], batches: list[dict] | None = None) -> None:
|
||
report = {"files": files, "batches": batches or [], "projectName": "新工厂排产项目B",
|
||
"totalErrors": 5, "sourceManifestDigest": "manifest-digest"}
|
||
monkeypatch.setattr("server.aps_domain.folder_pack.prepare_folder_schedule",
|
||
lambda *_: report)
|
||
|
||
|
||
def test_broken_template_beside_valid_workbook_is_ignored(monkeypatch):
|
||
"""回归：工程目录里混着空模板时，仍然采用唯一一份可排产工作簿（2026-09-12 现场）。"""
|
||
from server.aps_domain.planning_intake import profile_intake_reply
|
||
from server.state.seed import empty_world
|
||
from tests.golden.test_guidance import _MemStore
|
||
|
||
valid = _intake_file("湖南锐扬APS精简演示数据.xlsx", can_commit=True, digest="a" * 64)
|
||
template = _intake_file("APS精简演示模板.xlsx", can_commit=False, digest="b" * 64)
|
||
_patch_folder_report(monkeypatch, [template, valid])
|
||
|
||
reply = profile_intake_reply(_MemStore(empty_world()), "s", schedule_requested=True,
|
||
schedule_current=lambda: None)
|
||
assert "多份数据文件" not in reply.text
|
||
assert "资料检查完成" in reply.text
|
||
assert reply.blocks[0].props["source"]["sha256"] == valid["source"]["sha256"]
|
||
assert [block.type for block in reply.blocks] == ["folder-pack", "confirm-card"]
|
||
|
||
|
||
def test_planner_named_broken_workbook_still_reported(monkeypatch):
|
||
"""用户点名哪一份就按哪一份核对：点名坏文件时给出错误卡，不静默换文件。"""
|
||
from server.aps_domain.planning_intake import profile_intake_reply
|
||
from server.state.seed import empty_world
|
||
from tests.golden.test_guidance import _MemStore
|
||
|
||
valid = _intake_file("湖南锐扬APS精简演示数据.xlsx", can_commit=True, digest="a" * 64)
|
||
template = _intake_file("APS精简演示模板.xlsx", can_commit=False, digest="b" * 64)
|
||
_patch_folder_report(monkeypatch, [template, valid])
|
||
|
||
reply = profile_intake_reply(
|
||
_MemStore(empty_world()), "s", schedule_requested=False,
|
||
schedule_current=lambda: None, query="用 APS精简演示模板.xlsx 核对一下",
|
||
)
|
||
assert "字段或关联错误" in reply.text
|
||
assert reply.blocks[0].props["source"]["sha256"] == template["source"]["sha256"]
|
||
assert not any(block.type == "confirm-card" for block in reply.blocks)
|
||
|
||
|
||
def test_two_usable_workbooks_ask_planner_to_pick_one(monkeypatch):
|
||
"""两份都能用时必须让用户点名，点序号后按点名那一份继续。"""
|
||
from server.aps_domain.planning_intake import profile_intake_reply
|
||
from server.state.seed import empty_world
|
||
from tests.golden.test_guidance import _MemStore
|
||
|
||
first = _intake_file("甲方案.xlsx", can_commit=True, digest="c" * 64)
|
||
second = _intake_file("乙方案.xlsx", can_commit=True, digest="d" * 64)
|
||
_patch_folder_report(monkeypatch, [first, second])
|
||
store = _MemStore(empty_world())
|
||
|
||
reply = profile_intake_reply(store, "s", schedule_requested=False,
|
||
schedule_current=lambda: None)
|
||
choice = next(block for block in reply.blocks if block.type == "clarify")
|
||
assert [option["label"] for option in choice.props["options"]] == ["甲方案.xlsx", "乙方案.xlsx"]
|
||
assert "请回复序号或文件名" in reply.text
|
||
|
||
picked = profile_intake_reply(store, "s", schedule_requested=False,
|
||
schedule_current=lambda: None, query="2. 乙方案.xlsx")
|
||
assert picked.blocks[0].props["source"]["sha256"] == second["source"]["sha256"]
|
||
|
||
|
||
def test_confirmation_only_carries_batches_of_the_selected_workbook(monkeypatch):
|
||
"""同格式的两份工作簿：选中的那份才进确认载荷，避免重复表导致采用失败。"""
|
||
from server.agent_core import harness
|
||
from server.aps_domain.planning_intake import profile_intake_reply
|
||
from server.state.seed import empty_world
|
||
from tests.golden.test_guidance import _MemStore
|
||
|
||
def _batch(source_file: str, role: str) -> dict:
|
||
return {"sourceFile": source_file, "sourceProfile": "ruiyang-aps-v1", "role": role,
|
||
"kind": role, "sheet": role, "fieldMap": []}
|
||
|
||
valid = _intake_file("湖南锐扬APS精简演示数据.xlsx", can_commit=True, digest="a" * 64)
|
||
template = _intake_file("APS精简演示模板.xlsx", can_commit=False, digest="b" * 64)
|
||
batches = [_batch(valid["name"], "orders"), _batch(template["name"], "orders")]
|
||
_patch_folder_report(monkeypatch, [template, valid], batches)
|
||
|
||
reply = profile_intake_reply(_MemStore(empty_world()), "s", schedule_requested=False,
|
||
schedule_current=lambda: None)
|
||
assert reply.blocks[0].props["totalErrors"] == 0
|
||
assert [item["params"]["batches"] for item in harness.list_pending()] == [[batches[0]]]
|
||
|
||
|
||
def test_pending_adoption_card_leads_with_conclusion_and_a_locked_scheduling_entry(monkeypatch):
|
||
"""回归：现场那张"排产资料核对"卡首屏只给结论、关键数和下一步，不再堆原始字段。"""
|
||
from server.aps_domain.planning_intake import profile_intake_reply
|
||
from server.state.seed import empty_world
|
||
from tests.golden.test_guidance import _MemStore
|
||
|
||
valid = _intake_file("湖南锐扬APS精简演示数据.xlsx", can_commit=True, digest="e" * 64)
|
||
_patch_folder_report(monkeypatch, [valid])
|
||
|
||
reply = profile_intake_reply(_MemStore(empty_world()), "s", schedule_requested=False,
|
||
schedule_current=lambda: None)
|
||
props = reply.blocks[0].props
|
||
summary = props["analysisSummary"]
|
||
assert summary["status"] == "review"
|
||
assert summary["statusLabel"] == "待核对采用"
|
||
assert summary["headline"] == "资料已读取，确认采用后才会写入项目主数据。"
|
||
assert summary["nextStep"]["label"] == "开始排产"
|
||
assert summary["nextStep"]["enabled"] is False
|
||
assert "确认采用" in summary["nextStep"]["reason"]
|
||
assert [metric["label"] for metric in summary["metrics"]] == ["订单记录", "产品和物料", "加工步骤", "设备记录"]
|
||
assert [metric["value"] for metric in summary["metrics"]] == [6, 0, 0, 0]
|
||
assert "尚未写入主数据" in summary["recap"]
|
||
assert [block.type for block in reply.blocks] == ["folder-pack", "confirm-card"]
|