80 lines
3.7 KiB
Python
80 lines
3.7 KiB
Python
|
"""User journey regressions: accurate counts, one approval/one plan, then review."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from server.aps_domain.folder_pack import analyze_work_dir
|
||
|
|
from server.aps_domain.guidance import build_guidance
|
||
|
|
from server.state.seed import empty_world
|
||
|
|
from tests.golden.test_folder_schedule_security import _approve, _stage
|
||
|
|
from tests.golden.test_folder_schedule_security import (
|
||
|
|
folder_env as folder_env, # noqa: PLC0414 - register pytest fixture
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_mixed_workbook_counts_each_business_category_once(tmp_path, monkeypatch):
|
||
|
|
(tmp_path / "设备与订单.xlsx").write_bytes(b"test workbook")
|
||
|
|
|
||
|
|
class Projects:
|
||
|
|
def snapshot(self, include_messages=False):
|
||
|
|
return {"projects": [{"id": "p", "workDir": str(tmp_path)}],
|
||
|
|
"sessions": [{"id": "s", "projectId": "p"}]}
|
||
|
|
|
||
|
|
batches = [{"sheet": kind, "kind": kind, "okCount": count,
|
||
|
|
"okRows": [{"code": f"{kind}-{i}"} for i in range(count)]}
|
||
|
|
for kind, count in [("orders", 6), ("materials", 87),
|
||
|
|
("routing", 15), ("equipment", 57)]]
|
||
|
|
monkeypatch.setattr("server.state.projects.get_project_store", lambda: Projects())
|
||
|
|
monkeypatch.setattr("server.aps_domain.folder_pack.preview_file", lambda *_args, **_kwargs: {"batches": batches})
|
||
|
|
report = analyze_work_dir(empty_world(), "s")
|
||
|
|
assert report["totalOk"] == 165
|
||
|
|
assert report["kindCounts"] == {"orders": 6, "materials": 87, "routing": 15, "equipment": 57}
|
||
|
|
assert {item["key"]: item["count"] for item in report["coverageDetail"]}["equipment"] == 57
|
||
|
|
|
||
|
|
|
||
|
|
def test_existing_flex_plan_never_prompts_first_schedule():
|
||
|
|
world = empty_world()
|
||
|
|
world["flexScheduleVersions"] = [{"id": 42, "versionNo": "V42", "status": "DRAFT",
|
||
|
|
"trialOnly": True, "productionReady": False, "conflictCount": 0}]
|
||
|
|
result = build_guidance(world)
|
||
|
|
ids = {item["id"] for item in result["suggestions"]}
|
||
|
|
assert {"view_schedule", "export_schedule"} <= ids
|
||
|
|
assert not {"run_delivery", "flex_run", "publish"} & ids
|
||
|
|
assert not any(item["id"] == "no_version" for item in result["signals"])
|
||
|
|
assert "尚未下发" in result["hint"]
|
||
|
|
assert next(item for item in result["suggestions"]
|
||
|
|
if item["id"] == "view_schedule")["command"] == "查看排产结果"
|
||
|
|
|
||
|
|
|
||
|
|
def test_fixed_plan_view_and_blocked_publish():
|
||
|
|
world = empty_world()
|
||
|
|
world["scheduleVersions"] = [{"id": 1, "versionNo": "V1", "status": "DRAFT", "conflictCount": 4}]
|
||
|
|
result = build_guidance(world)
|
||
|
|
assert not any(item["id"] == "publish" for item in result["suggestions"])
|
||
|
|
assert result["suggestions"][0]["id"] == "view_conflicts"
|
||
|
|
|
||
|
|
|
||
|
|
def test_one_confirmation_runs_solver_once_and_reviews_the_committed_version(folder_env, monkeypatch):
|
||
|
|
from server.aps_domain import flex
|
||
|
|
|
||
|
|
store, identity, _approvals, _checkpoints, _path = folder_env
|
||
|
|
original = flex.run_flex_schedule
|
||
|
|
calls = []
|
||
|
|
|
||
|
|
def counted(*args, **kwargs):
|
||
|
|
result = original(*args, **kwargs)
|
||
|
|
calls.append(result)
|
||
|
|
return result
|
||
|
|
|
||
|
|
monkeypatch.setattr(flex, "run_flex_schedule", counted)
|
||
|
|
confirm_id, reply = _stage(store, identity)
|
||
|
|
assert "尚未生成方案" in reply.text
|
||
|
|
assert calls == []
|
||
|
|
message = _approve(store, identity, confirm_id)
|
||
|
|
assert len(calls) == 1
|
||
|
|
assert len(store.data["flexScheduleVersions"]) == 1
|
||
|
|
version = store.data["flexScheduleVersions"][0]
|
||
|
|
assert version["trialOnly"] is True and version["productionReady"] is False
|
||
|
|
assert version["versionNo"] in message and "尚未下发" in message
|
||
|
|
assert "根据这些数据排产" not in message
|
||
|
|
_approve(store, identity, confirm_id)
|
||
|
|
assert len(calls) == 1
|