# ============================================================ # EX-06 资源利用率 + SC-11 班组约束黄金测试 # ============================================================ from __future__ import annotations from server.agent_core.intent import parse_fast from server.aps_domain.analytics import build_utilization_report from server.aps_domain.flex import run_flex_schedule from server.engines.pool_engine import PoolEngine from server.state.seed import seed_world from server.timeutil import add_minutes, fmt_date, today0 class _MemStore: def __init__(self, data): self.data = data def next_id(self, kind: str) -> int: key = f"_c_{kind}" self.data[key] = self.data.get(key, 7000) + 1 return self.data[key] def save(self): pass def test_utilization_equipment_and_teams(): store = _MemStore(seed_world()) run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test") rep = build_utilization_report(store.data, track="flex", days=7) assert rep["track"] == "flex" assert rep["equipment"] assert rep["teams"] assert "avgUtilization" in rep["summary"] assert any(w.get("teamCode") for w in store.data["flexWorkOrders"] if w.get("versionId") == store.data["flexScheduleVersions"][-1]["id"]) def test_no_team_conflict_when_ops_uncovered(): world = seed_world() # 抽空班组覆盖 → 应产生 NO_TEAM world["flexTeams"] = [{ "code": "T-ONLY-PKG", "name": "仅包装", "memberCount": 2, "supportOps": ["OP-PKG"], "skillLevel": "L1", }] start = fmt_date(add_minutes(today0(), 24 * 60)) counters = {"n": 8000} def nid(k): counters["n"] += 1 return counters["n"] r = PoolEngine().solve(world, nid, sort_mode="BOTTLENECK", start_date=start, enforce_teams=True) types = {c.get("conflictType") for c in world.get("flexConflicts", []) if c.get("versionId") == r["versionId"]} assert "NO_TEAM" in types def test_intent_util_view(): world = seed_world() assert parse_fast("打开资源利用率", world).intent == "viewport.mode" assert parse_fast("打开资源利用率", world).params.get("mode") == "util" assert parse_fast("切换到负荷热力图", world).params.get("mode") == "load"