# ============================================================ # SC-04 扩充：计划员约束规则库黄金测试（可重生 ✅） # 1) 默认通用规则 = 内建 C1..C13 目录，扩充规则不动它 # 2) 计划员一句话 → 规则（按类别 / 按范围归档） # 3) 没接入排产引擎的规则如实标注，不假装生效 # 4) 绑定可配内置项时真正落到排产引擎，并说清执行来源 # 5) 新增 / 启用 / 停用 / 删除 全生命周期 + 非法载荷拒绝 # ============================================================ from __future__ import annotations import pytest from server.aps_domain.constraint_rules import ( all_rules, apply_rule_save, confirmation_for_rule_save, constraint_library, normalize_rule_payload, rule_effect, tenant_rules_path, ) from server.aps_domain.constraints import ( DEFAULT_CONSTRAINT_DEFS, apply_profile_save, get_constraint, get_constraint_profile, ) from server.state.seed import seed_world @pytest.fixture def rules_home(tmp_path, monkeypatch): """每条用例独占一个数据根，避免全厂规则文件互相串。""" monkeypatch.setenv("APS_HOME", str(tmp_path)) return tmp_path def _add(world, tenant, **kw): payload = {"op": "add", "name": kw.pop("name"), "category": kw.pop("category", "other")} payload.update(kw) return apply_rule_save(world, payload, tenant) # ---------------- 1) 默认通用规则是基线，扩充不动它 ---------------- def test_default_library_is_builtin_catalog_only(rules_home): world = seed_world() lib = constraint_library(world, "t1") assert lib["summary"]["builtinTotal"] == len(DEFAULT_CONSTRAINT_DEFS) assert lib["summary"]["customTotal"] == 0 assert lib["summary"]["engineBound"] == 0 assert lib["groups"] == [] def test_adding_unbound_rule_does_not_touch_builtin_profile(rules_home): world = seed_world() before = [(c["kind"], c["enabled"]) for c in get_constraint_profile(world)["constraints"]] _add(world, "t1", name="夜班不安排换型", category="process", scope="project", kind="soft") after = [(c["kind"], c["enabled"]) for c in get_constraint_profile(world)["constraints"]] assert before == after assert len(world["constraintRules"]) == 1 # ---------------- 2) 计划员一句话 → 规则，按类别/范围归档 ---------------- def test_project_rule_is_archived_with_category_and_scope(rules_home): world = seed_world() out = _add(world, "t1", name="夜班不安排换型", category="process", scope="project", kind="soft", sourceText="以后夜班别安排换型") assert out["ruleId"] == "R-001" lib = constraint_library(world, "t1") assert [g["id"] for g in lib["groups"]] == ["process"] rule = lib["custom"][0] assert rule["name"] == "夜班不安排换型" assert rule["categoryLabel"] == "工艺" assert rule["scopeLabel"] == "仅本项目" assert rule["kindLabel"] == "尽量满足" assert rule["source"]["text"] == "以后夜班别安排换型" def test_tenant_rule_lands_in_tenant_file_and_stays_tenant_scoped(rules_home): world = seed_world() _add(world, "tenant-a", name="全厂周末不排产", category="calendar", scope="tenant", kind="hard") assert tenant_rules_path("tenant-a").exists() assert not world.get("constraintRules") assert constraint_library(world, "tenant-a")["summary"]["customTotal"] == 1 assert constraint_library(world, "tenant-b")["summary"]["customTotal"] == 0 assert constraint_library(world, "tenant-a")["custom"][0]["scopeLabel"] == "全厂" def test_rule_ids_increment_across_scopes(rules_home): world = seed_world() _add(world, "t1", name="本项目第一条", category="other", scope="project") _add(world, "t1", name="全厂第二条", category="other", scope="tenant") rules = all_rules(world, "t1") assert sorted(r["id"] for r in rules) == ["R-001", "R-002"] assert {r["scope"] for r in rules} == {"project", "tenant"} # ---------------- 3) 没接入引擎就直说 ---------------- def test_unbound_rule_is_reported_as_record_only(rules_home): world = seed_world() out = _add(world, "t1", name="喷涂线每天最多八小时", category="capacity", scope="project") rule = world["constraintRules"][0] effect = rule_effect(world, rule) assert effect["bound"] is False and effect["enforced"] is False assert effect["text"] == "已记录，暂未接入排产引擎" assert "暂未接入排产引擎" in out["message"] assert constraint_library(world, "t1")["summary"]["recordOnly"] == 1 # ---------------- 4) 绑定内置项时真正生效，并说清来源 ---------------- def test_binding_to_kind_configurable_constraint_applies_to_engine(rules_home): world = seed_world() assert get_constraint(world, "C6_material_kit")["kind"] == "soft" _add(world, "t1", name="缺料必须挡住发布", category="material", scope="project", kind="hard", boundConstraintId="C6_material_kit") assert get_constraint(world, "C6_material_kit")["kind"] == "hard" rule = world["constraintRules"][0] effect = rule_effect(world, rule) assert effect["enforced"] is True assert "物料齐套" in effect["text"] and "硬约束" in effect["text"] assert constraint_library(world, "t1")["summary"]["engineBound"] == 1 def test_binding_to_non_configurable_hard_rule_never_downgrades_it(rules_home): world = seed_world() _add(world, "t1", name="前序没做完不能开下道", category="process", scope="project", kind="soft", boundConstraintId="C1_precedence") c1 = get_constraint(world, "C1_precedence") assert c1["kind"] == "hard" and c1["enabled"] is True assert rule_effect(world, world["constraintRules"][0])["enforced"] is True def test_disabled_builtin_marks_bound_rule_as_not_effective(rules_home): world = seed_world() _add(world, "t1", name="缺料先提示我", category="material", scope="project", kind="soft", boundConstraintId="C6_material_kit") apply_profile_save(world, {"constraints": {"C6_material_kit": {"enabled": False}}}) effect = rule_effect(world, world["constraintRules"][0]) assert effect["enforced"] is False assert "已关闭" in effect["text"] # ---------------- 5) 生命周期与校验 ---------------- def test_rule_lifecycle_enable_disable_delete(rules_home): world = seed_world() _add(world, "t1", name="插单必须先经我确认", category="order", scope="project") rid = world["constraintRules"][0]["id"] apply_rule_save(world, {"op": "disable", "ruleId": rid}, "t1") assert world["constraintRules"][0]["enabled"] is False assert constraint_library(world, "t1")["custom"][0]["enabled"] is False apply_rule_save(world, {"op": "enable", "ruleId": rid}, "t1") assert world["constraintRules"][0]["enabled"] is True out = apply_rule_save(world, {"op": "delete", "ruleId": rid}, "t1") assert world["constraintRules"] == [] assert "已删除" in out["message"] def test_tenant_rule_toggle_writes_back_to_tenant_file(rules_home): world = seed_world() _add(world, "tenant-a", name="全厂周末不排产", category="calendar", scope="tenant") rid = all_rules(world, "tenant-a")[0]["id"] apply_rule_save(world, {"op": "disable", "ruleId": rid}, "tenant-a") assert all_rules(world, "tenant-a")[0]["enabled"] is False assert not world.get("constraintRules") @pytest.mark.parametrize("payload", [ {"op": "add", "name": ""}, {"op": "add", "name": "超" * 41}, {"op": "add", "name": "合法内容", "category": "unknown"}, {"op": "add", "name": "合法内容", "scope": "site"}, {"op": "add", "name": "合法内容", "kind": "must"}, {"op": "add", "name": "合法内容", "boundConstraintId": "C99_nope"}, {"op": "enable"}, {"op": "archive", "ruleId": "R-001"}, ]) def test_invalid_payloads_are_rejected(payload): with pytest.raises(ValueError): normalize_rule_payload(payload) def test_missing_rule_id_is_rejected_on_apply(rules_home): world = seed_world() with pytest.raises(ValueError): apply_rule_save(world, {"op": "delete", "ruleId": "R-404"}, "t1") with pytest.raises(ValueError): confirmation_for_rule_save(world, {"op": "disable", "ruleId": "R-404"}, "t1") # ---------------- 6) 确认卡说人话 ---------------- def test_confirmation_card_states_scope_strength_and_engine_linkage(rules_home): world = seed_world() title, lines = confirmation_for_rule_save( world, {"op": "add", "name": "夜班不换型", "category": "process", "scope": "tenant", "kind": "soft"}, "t1") body = "\n".join(lines) assert "夜班不换型" in title assert "工艺" in body and "全厂" in body and "尽量满足" in body assert "暂未接入排产引擎" in body _, lines_bound = confirmation_for_rule_save( world, {"op": "add", "name": "缺料挡发布", "category": "material", "kind": "hard", "boundConstraintId": "C6_material_kit"}, "t1") assert "物料齐套" in "\n".join(lines_bound) # ---------------- 7) Pi 工具目录接线 + 确认后落库 ---------------- def test_pi_tool_catalog_exposes_constraint_rule_tools(): from server.agent_core import fallback_lane catalog = {t["name"]: t for t in fallback_lane._primary_tool_catalog()} assert catalog["constraint.rule.query"]["power"] == "P0" assert catalog["constraint.rule.save"]["power"] == "P2" schema = catalog["constraint.rule.save"]["paramsSchema"] assert schema["additionalProperties"] is False assert "add" in schema["properties"]["op"]["enum"] assert "tenant" in schema["properties"]["scope"]["enum"] assert "工艺" in catalog["constraint.rule.query"]["description"] or \ "规则库" in catalog["constraint.rule.query"]["description"] async def test_pi_intent_stages_rule_and_confirmation_writes_it(rules_home, tmp_path): from server.aps_domain.workflow import execute_confirmed, handle_intent from server.contracts import IntentResult from server.state.store import WorldStore store = WorldStore(path=tmp_path / "world.json") reply = await handle_intent( store, "s-rules", IntentResult( intent="constraint.rule.save", params={"op": "add", "name": "夜班不安排换型", "category": "process", "scope": "project", "kind": "soft", "sourceText": "以后夜班别安排换型"}, confidence=1.0, source="LLM", ), ) assert reply.blocks, "P2 动作必须出确认卡" block = reply.blocks[0] assert block.props["power"] == "P2" assert not store.data.get("constraintRules"), "确认前不得写规则库" message = execute_confirmed(store, str(block.props["confirmId"]), True, actor="planner-1") assert "R-001" in message assert store.data["constraintRules"][0]["name"] == "夜班不安排换型" audits = [e for e in store.data["auditEvents"] if e.get("action") == "constraint.rule.save" and e.get("result") == "SUCCESS"] assert audits, "批准执行必须落审计" async def test_pi_rule_query_lists_defaults_and_extensions(rules_home, tmp_path): from server.aps_domain.workflow import handle_intent from server.contracts import IntentResult from server.state.store import WorldStore world = seed_world() _add(world, "platform", name="喷涂线每天最多八小时", category="capacity", scope="project") store = WorldStore(path=tmp_path / "world.json") store.data = world reply = await handle_intent( store, "s-rules", IntentResult(intent="constraint.rule.query", params={}, confidence=1.0, source="LLM"), ) assert "默认通用约束" in reply.text assert "喷涂线每天最多八小时" in reply.text assert "暂未接入排产引擎" in reply.text def test_gateway_endpoints_expose_and_apply_planner_rules(rules_home, monkeypatch): from fastapi.testclient import TestClient import server.gateway.app as gateway_module from tests.auth_provider import install_test_auth install_test_auth(monkeypatch, "tenant-rule-library") client = TestClient(gateway_module.create_app()) assert client.post("/api/auth/login", json={"username": "planner"}).status_code == 200 lib = client.get("/api/constraints/library") assert lib.status_code == 200 summary = lib.json()["summary"] assert summary["builtinTotal"] == len(DEFAULT_CONSTRAINT_DEFS) assert summary["customTotal"] == 0 staged = client.post("/api/constraints/rules/stage", json={ "sessionId": "s-http-rules", "payload": {"op": "add", "name": "夜班不安排换型", "category": "process", "scope": "project", "kind": "soft"}, }) assert staged.status_code == 200 body = staged.json() assert not body.get("error"), body confirm_id = body["block"]["props"]["confirmId"] assert client.get("/api/constraints/library").json()["summary"]["customTotal"] == 0, \ "暂存阶段不得写规则库" approved = client.post("/api/actions/confirm", json={ "sessionId": "s-http-rules", "confirmId": confirm_id, "approve": True, }) assert approved.status_code == 200 assert "R-001" in approved.json()["message"] after = client.get("/api/constraints/library").json() assert after["summary"]["customTotal"] == 1 assert after["custom"][0]["name"] == "夜班不安排换型" assert after["custom"][0]["categoryLabel"] == "工艺"