195 lines
8.4 KiB
Python
195 lines
8.4 KiB
Python
|
# 批量审批端点黄金测试（plan.md §3.3 / §6.10.2）
|
|||
|
|
# ============================================================
|
|||
|
|
# 覆盖：批量批准多条 P2、批量含 P3 首次批准（secondConfirmRequired）、
|
|||
|
|
# 批量驳回、单条失败隔离、note 落库、仅当前用户作用域。
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
from fastapi.testclient import TestClient
|
|||
|
|
|
|||
|
|
from server.agent_core import harness
|
|||
|
|
from server.auth.context import bind_identity, reset_identity
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _AuditStore:
|
|||
|
|
def __init__(self, world_key: str = "personal-1001", tenant_uuid: str = "tenant-batch-test"):
|
|||
|
|
self.world_key = world_key
|
|||
|
|
self.tenant_uuid = tenant_uuid
|
|||
|
|
self.data: dict = {"auditEvents": [], "scheduleVersions": [], "flexScheduleVersions": []}
|
|||
|
|
|
|||
|
|
def next_id(self, _kind: str) -> int:
|
|||
|
|
return len(self.data["auditEvents"]) + 1
|
|||
|
|
|
|||
|
|
def save(self) -> None:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _CheckpointStore:
|
|||
|
|
def __init__(self) -> None:
|
|||
|
|
self.pairs: list[dict] = []
|
|||
|
|
|
|||
|
|
def create(self, *_args, **_kwargs) -> dict:
|
|||
|
|
pair = {"pairId": f"pair-{len(self.pairs) + 1}"}
|
|||
|
|
self.pairs.append(pair)
|
|||
|
|
return pair
|
|||
|
|
|
|||
|
|
def get(self, pair_id: str) -> dict | None:
|
|||
|
|
return next((p for p in self.pairs if p["pairId"] == pair_id), None)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _ProjectStore:
|
|||
|
|
def active_world_key(self) -> str:
|
|||
|
|
return "default"
|
|||
|
|
|
|||
|
|
def require_active_write(self) -> None:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(autouse=True)
|
|||
|
|
def _clear_pending():
|
|||
|
|
harness._approval_store.clear()
|
|||
|
|
yield
|
|||
|
|
harness._approval_store.clear()
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture
|
|||
|
|
def client_factory(monkeypatch):
|
|||
|
|
"""返回 (client, store, provider)；测试认证替换 JMS，登录身份用于 stage/审批。"""
|
|||
|
|
import server.gateway.app as gateway_module
|
|||
|
|
import server.state.projects as projects_module
|
|||
|
|
from tests.auth_provider import install_test_auth
|
|||
|
|
|
|||
|
|
provider = install_test_auth(monkeypatch, "tenant-batch-test")
|
|||
|
|
|
|||
|
|
def factory():
|
|||
|
|
store = _AuditStore()
|
|||
|
|
monkeypatch.setattr(gateway_module, "get_store", lambda: store)
|
|||
|
|
monkeypatch.setattr(gateway_module, "get_checkpoints", lambda: _CheckpointStore())
|
|||
|
|
monkeypatch.setattr(projects_module, "get_project_store", lambda: _ProjectStore())
|
|||
|
|
client = TestClient(gateway_module.create_app())
|
|||
|
|
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
|||
|
|
assert login.status_code == 200, login.text
|
|||
|
|
return client, store, provider
|
|||
|
|
|
|||
|
|
return factory
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _stage(provider, action: str) -> str:
|
|||
|
|
"""用与登录一致的测试身份出卡。"""
|
|||
|
|
identity = provider._identity("planner")
|
|||
|
|
params = {"versionId": 1, "track": "flex"} if action == "mes.dispatch" else {"versionId": 1}
|
|||
|
|
token = bind_identity(identity)
|
|||
|
|
try:
|
|||
|
|
block = harness.stage_confirmation(
|
|||
|
|
"session-batch", action, params, title="批量测试", summary_lines=["批量审批"],
|
|||
|
|
evidence_refs=["schedule-version:1"] if action == "mes.dispatch" else None,
|
|||
|
|
)
|
|||
|
|
finally:
|
|||
|
|
reset_identity(token)
|
|||
|
|
return str(block.props["confirmId"])
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _as(identity, callback):
|
|||
|
|
token = bind_identity(identity)
|
|||
|
|
try:
|
|||
|
|
return callback()
|
|||
|
|
finally:
|
|||
|
|
reset_identity(token)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_confirm_batch_approves_multiple_p2(client_factory):
|
|||
|
|
"""批量批准多条 P2：全部执行成功，审计携带 beforeSnapshot/evidenceRefs。"""
|
|||
|
|
client, store, provider = client_factory()
|
|||
|
|
store.data["scheduleVersions"].append({"id": 1, "versionNo": "V1", "status": "DRAFT",
|
|||
|
|
"poCount": 0, "woCount": 0, "conflictCount": 0,
|
|||
|
|
"totalTardiness": 0})
|
|||
|
|
store.data["productionOrders"] = []
|
|||
|
|
ids = [_stage(provider, "schedule.publish"), _stage(provider, "schedule.publish")]
|
|||
|
|
resp = client.post("/api/actions/confirm-batch",
|
|||
|
|
json={"sessionId": "batch", "confirmIds": ids, "approve": True,
|
|||
|
|
"note": "批量同意"})
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
data = resp.json()
|
|||
|
|
assert len(data["results"]) == 2
|
|||
|
|
assert all(r["ok"] for r in data["results"])
|
|||
|
|
assert all(r["secondConfirmRequired"] is False for r in data["results"])
|
|||
|
|
events = [e for e in store.data["auditEvents"] if e["action"] == "schedule.publish"]
|
|||
|
|
assert len(events) == 2
|
|||
|
|
assert all(e["beforeSnapshot"] is not None for e in events)
|
|||
|
|
assert all("schedule-version:1" in (e.get("evidenceRefs") or []) for e in events)
|
|||
|
|
identity = provider._identity("planner")
|
|||
|
|
history = _as(identity, harness.list_approval_history)
|
|||
|
|
assert len(history) >= 2
|
|||
|
|
assert all(h["note"] == "批量同意" and h["status"] == "APPROVED" for h in history[-2:])
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_confirm_batch_p3_first_approval_marks_second_required(client_factory):
|
|||
|
|
"""批量含 P3：首次批准标记 secondConfirmRequired，不视为失败。"""
|
|||
|
|
client, store, provider = client_factory()
|
|||
|
|
ids = [_stage(provider, "mes.dispatch"), _stage(provider, "mes.dispatch")]
|
|||
|
|
resp = client.post("/api/actions/confirm-batch",
|
|||
|
|
json={"sessionId": "batch", "confirmIds": ids, "approve": True})
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
results = resp.json()["results"]
|
|||
|
|
assert len(results) == 2
|
|||
|
|
assert all(r["secondConfirmRequired"] is True for r in results)
|
|||
|
|
assert all(r["ok"] for r in results)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_confirm_batch_rejects_all(client_factory):
|
|||
|
|
"""批量驳回：全部 DENIED 且无世界写入。"""
|
|||
|
|
client, store, provider = client_factory()
|
|||
|
|
store.data["scheduleVersions"].append({"id": 1, "versionNo": "V1", "status": "DRAFT",
|
|||
|
|
"poCount": 0, "woCount": 0, "conflictCount": 0,
|
|||
|
|
"totalTardiness": 0})
|
|||
|
|
store.data["productionOrders"] = []
|
|||
|
|
ids = [_stage(provider, "schedule.publish"), _stage(provider, "schedule.publish")]
|
|||
|
|
resp = client.post("/api/actions/confirm-batch",
|
|||
|
|
json={"sessionId": "batch", "confirmIds": ids, "approve": False,
|
|||
|
|
"note": "数据不齐"})
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
results = resp.json()["results"]
|
|||
|
|
assert len(results) == 2 and all(r["ok"] for r in results)
|
|||
|
|
assert store.data["scheduleVersions"][0]["status"] == "DRAFT" # 未发布
|
|||
|
|
identity = provider._identity("planner")
|
|||
|
|
history = _as(identity, harness.list_approval_history)
|
|||
|
|
assert len(history) >= 2
|
|||
|
|
assert all(h["note"] == "数据不齐" and h["status"] == "REJECTED" for h in history[-2:])
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_single_confirm_api_carries_note(client_factory):
|
|||
|
|
"""单条 confirm API 透传 note：审批历史记录意见。"""
|
|||
|
|
client, store, provider = client_factory()
|
|||
|
|
store.data["scheduleVersions"].append({"id": 1, "versionNo": "V1", "status": "DRAFT",
|
|||
|
|
"poCount": 0, "woCount": 0, "conflictCount": 0,
|
|||
|
|
"totalTardiness": 0})
|
|||
|
|
store.data["productionOrders"] = []
|
|||
|
|
cid = _stage(provider, "schedule.publish")
|
|||
|
|
resp = client.post("/api/actions/confirm",
|
|||
|
|
json={"sessionId": "single", "confirmId": cid, "approve": True,
|
|||
|
|
"note": "单条同意"})
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
assert resp.json()["secondConfirmRequired"] is False
|
|||
|
|
identity = provider._identity("planner")
|
|||
|
|
history = _as(identity, harness.list_approval_history)
|
|||
|
|
assert history[-1]["note"] == "单条同意"
|
|||
|
|
assert history[-1]["status"] == "APPROVED"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_confirm_batch_isolates_failures(client_factory):
|
|||
|
|
"""单条失败隔离：过期令牌失败不阻断其余成功。"""
|
|||
|
|
client, store, provider = client_factory()
|
|||
|
|
store.data["scheduleVersions"].append({"id": 1, "versionNo": "V1", "status": "DRAFT",
|
|||
|
|
"poCount": 0, "woCount": 0, "conflictCount": 0,
|
|||
|
|
"totalTardiness": 0})
|
|||
|
|
store.data["productionOrders"] = []
|
|||
|
|
good = _stage(provider, "schedule.publish")
|
|||
|
|
bad = "nonexistent-confirm"
|
|||
|
|
resp = client.post("/api/actions/confirm-batch",
|
|||
|
|
json={"sessionId": "batch", "confirmIds": [good, bad], "approve": True})
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
results = {r["confirmId"]: r for r in resp.json()["results"]}
|
|||
|
|
assert results[good]["ok"] is True
|
|||
|
|
assert results[bad]["ok"] is False
|
|||
|
|
assert "不存在" in results[bad]["message"] or "失效" in results[bad]["message"]
|