243 lines
12 KiB
Python
243 lines
12 KiB
Python
# ============================================================
|
||
# 检查点世界侧投影黄金测试（moduleId: state-checkpoints, 矩阵 54 行剩余项）
|
||
# 覆盖：建档 → 投影含 summary + 固定/柔性工单 + 产能池字段；
|
||
# 缺失 pair 显式失败（KeyError / HTTP 404 + 中文提示）；
|
||
# 投影不改动仓内数据（深拷贝隔离）；
|
||
# 分支锚定批量投影（Sagan #2：分支 checkpointId → 投影）。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.state.checkpoints import CheckpointStore, world_projection
|
||
from server.state.seed import seed_world
|
||
|
||
|
||
# ---------------- 夹具世界：固定轨 + 柔性轨双轨排产产物 ----------------
|
||
def _dual_track_world():
|
||
"""演示种子世界 + 手工排产产物（固定轨 scheduleVersions + 柔性轨 flexScheduleVersions）。"""
|
||
world = seed_world()
|
||
world["scheduleVersions"] = [{
|
||
"id": 1, "versionNo": "V1", "status": "PUBLISHED", "createdAt": "2026-08-01 09:00",
|
||
"woCount": 2, "poCount": 1, "conflictCount": 1, "totalTardiness": 3.5,
|
||
"avgUtilization": 0.82, "totalCost": 1234, "engineType": "RULE",
|
||
}]
|
||
world["productionOrders"] = [{
|
||
"id": 1, "orderNo": "PO-0001", "salesOrderId": 1, "productId": 1,
|
||
"productName": "智能控制器A型", "productCode": "CTRL-A", "quantity": 100, "unit": "件",
|
||
"lineId": 1, "lineName": "SMT线1", "status": "DRAFT", "priority": 1,
|
||
"schedulingVersionId": 1, "materialKitStatus": "PASSED", "conflictCount": 0,
|
||
"plannedEndDate": "2026-08-10",
|
||
}]
|
||
world["workOrders"] = [
|
||
{"id": 1, "orderNo": "PO-0001-01", "productionOrderId": 1, "productionOrderNo": "PO-0001",
|
||
"operationId": 1, "operationName": "SMT贴片", "sequenceNo": 1, "productId": 1,
|
||
"productName": "智能控制器A型", "quantity": 100, "unit": "件",
|
||
"lineId": 1, "lineName": "SMT线1", "workstationId": 1, "workstationName": "SMT-01",
|
||
"plannedStartTime": "2026-08-03 08:00", "plannedEndTime": "2026-08-03 10:00",
|
||
"status": "PENDING", "kitStatus": "PASSED", "progressPercent": 0, "conflictCount": 0},
|
||
{"id": 2, "orderNo": "PO-0001-02", "productionOrderId": 1, "productionOrderNo": "PO-0001",
|
||
"operationId": 2, "operationName": "DIP插件", "sequenceNo": 2, "productId": 1,
|
||
"productName": "智能控制器A型", "quantity": 100, "unit": "件",
|
||
"lineId": 1, "lineName": "SMT线1", "workstationId": 2, "workstationName": "DIP-01",
|
||
"plannedStartTime": "2026-08-03 10:10", "plannedEndTime": "2026-08-03 12:00",
|
||
"status": "PENDING", "kitStatus": "PASSED", "progressPercent": 0, "conflictCount": 1},
|
||
]
|
||
# 柔性轨：挂一版柔性排产，压接工序落在种子设备 PRESS-01（OP-CRIMP 能力池）
|
||
world["flexScheduleVersions"] = [{
|
||
"id": 7, "versionNo": "F-V1", "sortMode": "DUE_FIRST", "createdAt": "2026-08-01 10:00",
|
||
}]
|
||
world["flexWorkOrders"] = [{
|
||
"id": 91, "versionId": 7, "flexOrderNo": "FO-2601", "orderNo": "FO-2601-01",
|
||
"productCode": "HV-HARNESS", "quantity": 100, "operationName": "压接端子", "seq": 1,
|
||
"equipmentId": 3, "equipmentCode": "PRESS-01", "equipmentName": "气动压接机#1",
|
||
"zone": "ZONE-A", "moldCode": "MOLD-HV-01", "changeoverMin": 20, "moveMin": 10,
|
||
"runMin": 100, "plannedStartTime": "2026-08-03 08:00", "plannedEndTime": "2026-08-03 09:40",
|
||
"isBottleneck": True,
|
||
}]
|
||
return world
|
||
|
||
|
||
def _make_store(tmp_path, world):
|
||
"""临时仓 + 建档，返回 (store, pair_id, 仓指纹)。"""
|
||
store = CheckpointStore(path=str(tmp_path / "ckpt.json"))
|
||
meta = store.create(world, label="投影基线", reason="manual")
|
||
fingerprint = json.dumps(store.pairs, ensure_ascii=False, sort_keys=True)
|
||
return store, meta["pairId"], fingerprint
|
||
|
||
|
||
# ---------------- 1. 投影包含 summary + 固定/柔性工单 + 产能池 ----------------
|
||
def test_checkpoint_world_projection_contains_summary_and_both_tracks(tmp_path):
|
||
"""矩阵 54 行：投影含 summary（KPI+双轨计数）、固定轨工单（订单/工序/开始/结束/设备/模具/交期/状态）、
|
||
柔性轨工单与产能池——足够前端逐字段 diff。"""
|
||
world = _dual_track_world()
|
||
store, pair_id, _ = _make_store(tmp_path, world)
|
||
proj = store.world_projection(pair_id)
|
||
|
||
s = proj["summary"]
|
||
assert s["hasVersion"] is True
|
||
assert s["versionNo"] == "V1" and s["status"] == "PUBLISHED"
|
||
assert s["woCount"] == 2 and s["conflictCount"] == 1
|
||
assert s["totalTardiness"] == 3.5 and s["avgUtilization"] == 0.82
|
||
# 双轨计数（前端 WorldStats 数据源）
|
||
assert s["flexVersionNo"] == "F-V1" and s["flexSortMode"] == "DUE_FIRST"
|
||
assert s["flexWoCount"] == 1 and s["flexOverdue"] == 0 and s["flexConflict"] == 0
|
||
assert s["fixedWoCount"] == 2
|
||
assert s["poolCount"] > 0 and s["bottleneckDaily"] is not None
|
||
|
||
# 固定轨工单：订单/工序/开始/结束/设备/模具/交期/状态
|
||
fixed = proj["gantt"]["workOrders"]
|
||
assert len(fixed) == 2
|
||
wo = next(w for w in fixed if w["orderNo"] == "PO-0001-01")
|
||
assert wo["operationName"] == "SMT贴片"
|
||
assert wo["start"] == "2026-08-03 08:00" and wo["end"] == "2026-08-03 10:00"
|
||
assert wo["workstationName"] == "SMT-01"
|
||
assert wo["mold"] is None # 固定轨无模具概念
|
||
due = next(so for so in world["salesOrders"] if so["id"] == 1)["deliveryDate"]
|
||
assert wo["deliveryDate"] == due # 交期（销售订单 → 生产订单）
|
||
assert wo["status"] == "PENDING" # 状态（工单）
|
||
|
||
# 柔性轨工单：含模具/设备/工序序号
|
||
flex = proj["flexGantt"]["workOrders"]
|
||
assert len(flex) == 1
|
||
fwo = flex[0]
|
||
assert fwo["orderNo"] == "FO-2601" and fwo["operationName"] == "压接端子" and fwo["seq"] == 1
|
||
assert fwo["equipmentCode"] == "PRESS-01"
|
||
assert fwo["moldCode"] == "MOLD-HV-01"
|
||
assert fwo["start"] == "2026-08-03 08:00" and fwo["end"] == "2026-08-03 09:40"
|
||
assert fwo["dueDate"] # 交期（柔性订单）已富化
|
||
|
||
# 产能池：设备数 + 预测利用率
|
||
pools = proj["flexCapacity"]["pools"]
|
||
assert pools, "产能池应非空（种子 OP-CRIMP 能力池）"
|
||
crimp = next(p for p in pools if p["operationCode"] == "OP-CRIMP")
|
||
assert crimp["equipmentCount"] >= 4
|
||
assert crimp["forecast"]["d1"]["utilization"] >= 0
|
||
|
||
|
||
# ---------------- 2. 缺失 pair 显式失败 ----------------
|
||
def test_checkpoint_world_projection_missing_pair_fails(tmp_path):
|
||
"""缺失 pair：投影函数显式抛 KeyError（NotFound 语义，由网关转 404）。"""
|
||
store = CheckpointStore(path=str(tmp_path / "ckpt.json"))
|
||
store.create(_dual_track_world(), label="基线", reason="manual")
|
||
with pytest.raises(KeyError):
|
||
store.world_projection("no-such-pair")
|
||
with pytest.raises(KeyError):
|
||
world_projection("no-such-pair", store=store)
|
||
|
||
|
||
# ---------------- 3. 投影不改动仓内数据（深拷贝） ----------------
|
||
def test_checkpoint_world_projection_does_not_mutate_store(tmp_path):
|
||
"""投影在快照世界深拷贝上计算：仓内 pairs 逐字节不变，含被视图可能补种的柔性层。"""
|
||
store, pair_id, fingerprint = _make_store(tmp_path, _dual_track_world())
|
||
store.world_projection(pair_id) # 投影（内部会跑 flex_gantt_view / capacity_analysis）
|
||
after = json.dumps(store.pairs, ensure_ascii=False, sort_keys=True)
|
||
assert after == fingerprint, "world_projection 改动了仓内数据（违反深拷贝隔离）"
|
||
|
||
|
||
# ---------------- 4. HTTP 端点：单查 + 404 + 分支锚定批量 ----------------
|
||
@pytest.fixture()
|
||
def secure_app(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "tenant.db"))
|
||
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
|
||
monkeypatch.setenv("APS_BRANCH_DIR", str(tmp_path / "branches"))
|
||
monkeypatch.setenv("APS_CHECKPOINT_PATH", str(tmp_path / "ckpt.json"))
|
||
import server.state.checkpoints as ck_mod
|
||
from server.db.database import reset_engine
|
||
from server.state import store as world_store
|
||
from server.state.branches import reset_branch_stores
|
||
from tests.auth_provider import install_test_auth
|
||
from tests.pi_primary_adapter import install_fake_pi_tool_adapter
|
||
|
||
install_test_auth(monkeypatch, "tenant-a-000000000000000000000001")
|
||
install_fake_pi_tool_adapter(monkeypatch)
|
||
world_store._stores.clear()
|
||
ck_mod._checkpoints.clear()
|
||
reset_branch_stores()
|
||
reset_engine()
|
||
from server.gateway.app import create_app
|
||
app = create_app()
|
||
yield app
|
||
reset_engine()
|
||
world_store._stores.clear()
|
||
|
||
|
||
def _login(client: TestClient, username: str = "planner") -> None:
|
||
r = client.post("/api/auth/login", json={
|
||
"method": "password", "username": username, "password": "test",
|
||
})
|
||
assert r.status_code == 200
|
||
|
||
|
||
def _seed_checkpoint(client: TestClient) -> tuple[str, str, str]:
|
||
"""真实链路建档：建项目 → 排产 → 建检查点 → 返回 (sessionId, pairId, branchId)。"""
|
||
created = client.post("/api/projects", json={"id": "proj-proj", "name": "投影验收项目"})
|
||
assert created.status_code == 200
|
||
sid = created.json()["session"]["id"]
|
||
|
||
sched = client.post("/api/chat", json={"sessionId": sid, "text": "试排一版交期优先"})
|
||
assert sched.status_code == 200, sched.text[:200]
|
||
|
||
chat = client.post("/api/chat", json={"sessionId": sid, "text": "建一个检查点"})
|
||
assert chat.status_code == 200, chat.text[:200]
|
||
|
||
tree = client.get(f"/api/sessions/{sid}/branches").json()["tree"]
|
||
active = next(b for b in tree["branches"] if b["id"] == tree["activeNode"])
|
||
pair_id = active["checkpointId"]
|
||
assert pair_id, "建档后活动分支应锚定检查点"
|
||
return sid, pair_id, active["id"]
|
||
|
||
|
||
def test_checkpoint_projection_api_single_and_404(secure_app):
|
||
"""GET /api/checkpoints/{pairId} 返回 {meta, worldProjection}；缺失 pair 404 + 中文提示。"""
|
||
client = TestClient(secure_app)
|
||
_login(client)
|
||
sid, pair_id, _ = _seed_checkpoint(client)
|
||
|
||
r = client.get(f"/api/checkpoints/{pair_id}")
|
||
assert r.status_code == 200, r.text[:300]
|
||
body = r.json()
|
||
assert body["meta"]["pairId"] == pair_id
|
||
proj = body["worldProjection"]
|
||
assert proj["summary"]["hasVersion"] is True
|
||
assert "gantt" in proj and "flexGantt" in proj and "flexCapacity" in proj
|
||
# 固定轨工单已排产（chat 真实排产），投影应带出订单/工序/时间字段
|
||
fixed = proj["gantt"]["workOrders"]
|
||
assert fixed, "chat 排产后固定轨工单投影应非空"
|
||
for wo in fixed:
|
||
assert {"orderNo", "operationName", "start", "end"} <= wo.keys()
|
||
assert "deliveryDate" in wo and "status" in wo and "mold" in wo
|
||
|
||
missing = client.get("/api/checkpoints/does-not-exist")
|
||
assert missing.status_code == 404
|
||
assert "检查点不存在" in missing.json()["detail"]
|
||
|
||
|
||
def test_checkpoint_projection_api_branch_anchor_batch(secure_app):
|
||
"""GET /api/checkpoints?branchIds=&sessionId= 按分支锚点批量投影（Sagan #2：分支 vs 分支 diff）。"""
|
||
client = TestClient(secure_app)
|
||
_login(client)
|
||
sid, pair_id, branch_id = _seed_checkpoint(client)
|
||
|
||
r = client.get("/api/checkpoints", params={"branchIds": branch_id, "sessionId": sid})
|
||
assert r.status_code == 200, r.text[:300]
|
||
body = r.json()
|
||
assert body["missing"] == []
|
||
hit = body["projections"].get(branch_id)
|
||
assert hit and hit["meta"]["pairId"] == pair_id
|
||
assert hit["worldProjection"]["summary"]["hasVersion"] is True
|
||
|
||
# pairIds 批量 + 缺失项上报
|
||
r2 = client.get("/api/checkpoints", params={"pairIds": f"{pair_id},nope-1,nope-2"})
|
||
body2 = r2.json()
|
||
assert body2["projections"][pair_id]["meta"]["pairId"] == pair_id
|
||
assert sorted(body2["missing"]) == ["nope-1", "nope-2"]
|
||
|
||
# 未知分支 id → missing
|
||
r3 = client.get("/api/checkpoints", params={"branchIds": "ghost-branch", "sessionId": sid})
|
||
assert r3.status_code == 200
|
||
assert r3.json()["missing"] == ["ghost-branch"]
|