# -*- coding: utf-8 -*- """pbl_validation 引擎自测（纯函数层，不依赖 DB / 平台）。 运行：python3 modules/pbl_validation/tests/test_validation_engine.py 或： cd modules/pbl_validation && python3 -m pytest tests -q """ from __future__ import annotations import json import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from pbl_validation.constants import ( # noqa: E402 DEFAULT_THRESHOLDS, QUALITY_BASIC, QUALITY_DRAFT, QUALITY_EXCELLENT, QUALITY_GOOD, QUALITY_INCOMPLETE, QUALITY_STATES, resolve_thresholds, ) from pbl_validation.dimensions import ( # noqa: E402 DIM_COUNT, DIMENSION_REGISTRY, SUB_OBJECT_KINDS, dimension_meta, get_checker, ) from pbl_validation.engine import ( # noqa: E402 ValidationError, decide_quality_state, is_publishable, normalize_payload, payload_fingerprint, report_to_row, run_validation, ) PASSED = [] FAILED = [] def check(name, cond, detail=""): if cond: PASSED.append(name) print(" PASS %s" % name) else: FAILED.append((name, detail)) print(" FAIL %s %s" % (name, detail)) def good_blueprint(): """构造一个应判 Q4_EXCELLENT 的完整蓝图。""" objects = [] objects.append({"kind": "objective", "id": "OBJ1", "name": "掌握供需建模", "description": "学生能够建立并求解供需平衡模型"}) objects.append({"kind": "objective", "id": "OBJ2", "name": "团队协作决策", "description": "学生能够在多角色协作中做出权衡决策"}) objects.append({"kind": "role", "id": "R1", "name": "生产经理", "description": "负责产能规划与排产决策"}) objects.append({"kind": "role", "id": "R2", "name": "销售总监", "description": "负责订单承接与价格策略"}) objects.append({"kind": "scene", "id": "S1", "name": "第一季度经营会议", "description": "季度经营决策场景"}) objects.append({"kind": "entity", "id": "E1", "name": "产品", "attributes": [{"name": "单价", "type": "number"}, {"name": "库存", "type": "number"}]}) objects.append({"kind": "entity", "id": "E2", "name": "订单", "attributes": [{"name": "数量", "type": "number"}]}) tasks = [ {"kind": "task", "id": "T1", "name": "需求预测", "objective_id": "OBJ1", "role_id": "R1", "scene_id": "S1", "difficulty": 2, "duration_minutes": 20, "evidence": [{"type": "report", "name": "预测报告"}], "next": "T2"}, {"kind": "task", "id": "T2", "name": "产能决策", "objective_id": "OBJ1", "role_id": "R1", "scene_id": "S1", "difficulty": 3, "duration_minutes": 25, "evidence": [{"type": "decision", "name": "排产方案"}], "branches": [{"next": "T3"}, {"next": "T4"}]}, {"kind": "task", "id": "T3", "name": "扩产投资", "objective_id": "OBJ2", "role_id": "R2", "scene_id": "S1", "difficulty": 4, "duration_minutes": 30, "evidence": [{"type": "plan", "name": "投资计划"}], "next": "T5"}, {"kind": "task", "id": "T4", "name": "外包协作", "objective_id": "OBJ2", "role_id": "R2", "scene_id": "S1", "difficulty": 3, "duration_minutes": 25, "evidence": [{"type": "contract", "name": "外包合同"}], "next": "T5"}, {"kind": "task", "id": "T5", "name": "复盘汇报", "objective_id": "OBJ2", "role_id": "R1", "scene_id": "S1", "difficulty": 2, "duration_minutes": 15, "evidence": [{"type": "presentation", "name": "复盘PPT"}]}, ] objects.extend(tasks) objects.append({"kind": "rule", "id": "RU1", "name": "库存下限约束", "expression": "stock >= 100", "refs": ["E1", "T2"]}) objects.append({"kind": "assessment", "id": "A1", "name": "经营绩效量规", "objective_id": "OBJ1", "criteria": [{"name": "模型正确性", "weight": 0.5}, {"name": "决策合理性", "weight": 0.3}, {"name": "协作表现", "weight": 0.2}]}) return { "root": {"name": "供应链经营沙盘", "driving_question": "如何在需求波动下实现利润最大化？", "duration_minutes": 115}, "objects": objects, } def test_constants(): print("\n[1] 常量与阈值配置化") check("mcRatio 默认 0.7（Q-OPEN-8）", DEFAULT_THRESHOLDS["mcRatio"] == 0.7, str(DEFAULT_THRESHOLDS.get("mcRatio"))) check("5 级质量状态齐备", len(QUALITY_STATES) == 5, str(QUALITY_STATES)) check("质量状态顺序正确", QUALITY_STATES == (QUALITY_DRAFT, QUALITY_INCOMPLETE, QUALITY_BASIC, QUALITY_GOOD, QUALITY_EXCELLENT)) th = resolve_thresholds({"mcRatio": 0.85}) check("阈值覆盖生效", th["mcRatio"] == 0.85, str(th["mcRatio"])) check("阈值覆盖不污染默认表", DEFAULT_THRESHOLDS["mcRatio"] == 0.7) bad = resolve_thresholds({"mcRatio": 1.7, "unknownKey": 1, "minTasks": -3}) check("非法阈值被忽略并记录", len(bad["_ignored"]) == 3 and bad["mcRatio"] == 0.7, json.dumps(bad["_ignored"], ensure_ascii=False)) def test_dimensions_registry(): print("\n[2] 14 维注册表") check("维度数 = 14", DIM_COUNT == 14, str(DIM_COUNT)) check("注册表长度 = 14", len(DIMENSION_REGISTRY) == 14) dims = [d["dim"] for d in DIMENSION_REGISTRY] check("维度编码 D01~D14 连续", dims == ["D%02d" % i for i in range(1, 15)], str(dims)) check("7 类子对象定义齐备", len(SUB_OBJECT_KINDS) == 7, str(SUB_OBJECT_KINDS)) check("每个维度都有可调用 checker", all(callable(d["checker"]) for d in DIMENSION_REGISTRY)) check("get_checker 支持 dim 与 code", get_checker("D08") is not None and get_checker("task_chain") is not None) check("get_checker 非法返回 None", get_checker("D99") is None) meta = dimension_meta() check("dimension_meta 不含函数对象", len(meta) == 14 and all("checker" not in m for m in meta)) check("权重均为正数", all(d["weight"] > 0 for d in DIMENSION_REGISTRY)) def test_normalize(): print("\n[3] 载荷归一化与 fail-closed") bp = good_blueprint() check("dict 原样通过", normalize_payload(bp)["root"]["name"] == "供应链经营沙盘") check("JSON 字符串可解析", normalize_payload(json.dumps(bp, ensure_ascii=False))["root"]["name"] == "供应链经营沙盘") check("list 载荷转 objects", len(normalize_payload(bp["objects"])["objects"]) == 14) for bad, label in ((None, "None"), ("", "空串"), ("{bad json", "非法JSON"), (123, "数字")): try: normalize_payload(bad) check("非法载荷拒绝：%s" % label, False, "未抛异常") except ValidationError as exc: check("非法载荷拒绝：%s" % label, exc.code.startswith("V-INPUT"), exc.code) try: run_validation(bp, tenant_id="") check("tenant_id 缺失 fail-closed", False, "未抛异常") except ValidationError as exc: check("tenant_id 缺失 fail-closed", exc.code == "V-TENANT-001", exc.code) def test_excellent(): print("\n[4] 完整蓝图 → Q4_EXCELLENT") rep = run_validation(good_blueprint(), tenant_id="t1", blueprint_id="BP1", version="v1") check("契约版本正确", rep["contract"] == "pbl.validation.report/1.0", rep["contract"]) check("执行 14 维", rep["dim_count"] == 14, str(rep["dim_count"])) check("质量状态 Q4", rep["quality_state"] == QUALITY_EXCELLENT, "%s / %s" % (rep["quality_state"], rep["quality_reasons"])) check("无 fail 维度", rep["state_summary"]["fail"] == 0, str(rep["state_summary"])) check("无 warn 维度", rep["state_summary"]["warn"] == 0, str(rep["state_summary"])) check("无 blocker", rep["severity_summary"]["blocker"] == 0) check("总分 >= 0.9", rep["total_score"] >= 0.9, str(rep["total_score"])) check("passed=True", rep["passed"] is True) check("发布门禁通过", is_publishable(rep)[0] is True) d08 = [d for d in rep["dimensions"] if d["dim"] == "D08"][0] check("D08 mcRatio = 1.0", d08["metrics"]["mcRatio"] == 1.0, str(d08["metrics"])) check("D08 阈值取 0.7", d08["metrics"]["mcRatio_threshold"] == 0.7) check("D08 识别 2 条完整路径（T2 双分支）", d08["metrics"]["path_count"] == 2, str(d08["metrics"]["path_count"])) check("D08 分支出口总数=2", d08["metrics"]["branch_exits_total"] == 2, str(d08["metrics"]["branch_exits_total"])) check("报告含 14 维元信息", len(rep["dimensions_meta"]) == 14) check("确定性：同载荷两次指纹一致", payload_fingerprint(good_blueprint(), rep["thresholds"]) == payload_fingerprint(good_blueprint(), rep["thresholds"])) def test_mc_ratio_gate(): print("\n[5] mcRatio 阈值门禁（Q-OPEN-8）") bp = good_blueprint() # 把 T2 的一个分支指向不存在的任务 → 该分支出口未被路径覆盖 for o in bp["objects"]: if o.get("id") == "T2": o["branches"] = [{"next": "T3"}, {"next": "T_GHOST"}] rep = run_validation(bp, tenant_id="t1", blueprint_id="BP2") d08 = [d for d in rep["dimensions"] if d["dim"] == "D08"][0] check("mcRatio 降到 0.5", d08["metrics"]["mcRatio"] == 0.5, str(d08["metrics"]["mcRatio"])) check("D08 判 fail", d08["state"] == "fail", d08["state"]) check("命中 V-TASK-004", any(i["code"] == "V-TASK-004" for i in d08["issues"]), json.dumps(d08["issues"], ensure_ascii=False)) check("命中悬空引用 V-TASK-003", any(i["code"] == "V-TASK-003" for i in d08["issues"])) check("质量状态被压到 Q1", rep["quality_state"] == QUALITY_INCOMPLETE, rep["quality_state"]) # 阈值放宽到 0.5 → mcRatio 门禁不再触发（证明阈值配置化生效） rep2 = run_validation(bp, tenant_id="t1", blueprint_id="BP2", threshold_overrides={"mcRatio": 0.5}) d08b = [d for d in rep2["dimensions"] if d["dim"] == "D08"][0] check("阈值放宽后不再命中 V-TASK-004", not any(i["code"] == "V-TASK-004" for i in d08b["issues"])) check("阈值放宽后 D08 阈值=0.5", d08b["metrics"]["mcRatio_threshold"] == 0.5) check("报告回显生效阈值 mcRatio=0.5", rep2["thresholds"]["mcRatio"] == 0.5) def test_rule_json_override(): print("\n[6] rule_json 阈值覆盖优先级") bp = good_blueprint() rep = run_validation(bp, tenant_id="t1", threshold_overrides={"mcRatio": 0.95}, rule_json={"thresholds": {"mcRatio": 0.6}, "weights": {"D08": 2.0}, "disabled_dims": ["D13"]}) check("rule_json 覆盖入参阈值", rep["thresholds"]["mcRatio"] == 0.6, str(rep["thresholds"]["mcRatio"])) d13 = [d for d in rep["dimensions"] if d["dim"] == "D13"][0] check("D13 被禁用 → skip", d13["state"] == "skip", d13["state"]) check("skip 维度不计入权重和", abs(rep["weight_sum"] - sum(d["weight"] for d in rep["dimensions"] if d["state"] != "skip")) < 1e-6) d08 = [d for d in rep["dimensions"] if d["dim"] == "D08"][0] check("D08 权重被规则改为 2.0", d08["weight"] == 2.0, str(d08["weight"])) def test_empty_blueprint(): print("\n[7] 空蓝图 → Q1_INCOMPLETE") rep = run_validation({"root": {}}, tenant_id="t1", blueprint_id="BP_EMPTY") check("空蓝图判 Q1", rep["quality_state"] == QUALITY_INCOMPLETE, "%s %s" % (rep["quality_state"], rep["quality_reasons"])) check("存在 blocker 问题", rep["severity_summary"]["blocker"] > 0, str(rep["severity_summary"])) check("passed=False", rep["passed"] is False) check("发布门禁不通过", is_publishable(rep)[0] is False) check("D01 fail", [d for d in rep["dimensions"] if d["dim"] == "D01"][0]["state"] == "fail") check("D10 缺量规 blocker", any(i["code"] == "V-RUB-001" for i in [d for d in rep["dimensions"] if d["dim"] == "D10"][0]["issues"])) check("总分低于 Q2 线", rep["total_score"] < 0.4, str(rep["total_score"])) def test_partial_blueprint(): print("\n[8] 部分完整蓝图 → Q2/Q3") bp = good_blueprint() # 去掉一半任务的 evidence → D11 覆盖率 0.4 < 0.8 判 fail（非核心维度） n = 0 for o in bp["objects"]: if o.get("kind") == "task": n += 1 if n % 2 == 0: o.pop("evidence", None) rep = run_validation(bp, tenant_id="t1", blueprint_id="BP3") d11 = [d for d in rep["dimensions"] if d["dim"] == "D11"][0] check("D11 覆盖率 0.6", d11["metrics"]["coverage"] == 0.6, str(d11["metrics"])) check("D11 判 fail（低于 0.8 阈值）", d11["state"] == "fail", d11["state"]) check("非核心维度 fail 不触发 R3", not any(r.startswith("R3") for r in rep["quality_reasons"]), str(rep["quality_reasons"])) check("质量状态落在 Q2/Q3", rep["quality_state"] in (QUALITY_BASIC, QUALITY_GOOD), "%s %s" % (rep["quality_state"], rep["quality_reasons"])) def test_safety(): print("\n[9] D14 安全合规扫描") bp = good_blueprint() bp["objects"].append({"kind": "rule", "id": "RU_BAD", "name": "危险规则", "expression": "DROP TABLE pbl_blueprint;"}) rep = run_validation(bp, tenant_id="t1", blueprint_id="BP_SEC") d14 = [d for d in rep["dimensions"] if d["dim"] == "D14"][0] check("命中破坏性 SQL", any(i["code"] == "V-SEC-001" for i in d14["issues"]), json.dumps(d14["issues"], ensure_ascii=False)[:200]) check("D14 判 fail", d14["state"] == "fail") check("blocker 触发 R2 → Q1", rep["quality_state"] == QUALITY_INCOMPLETE and any(r.startswith("R2") for r in rep["quality_reasons"]), str(rep["quality_reasons"])) bp2 = good_blueprint() bp2["root"]["note"] = "apikey = 'sk-abcdef123456'" rep2 = run_validation(bp2, tenant_id="t1") check("命中硬编码凭据", any(i["code"] == "V-SEC-003" for i in [d for d in rep2["dimensions"] if d["dim"] == "D14"][0]["issues"])) def test_rubric_weight(): print("\n[10] D10 量规权重和=1") bp = good_blueprint() for o in bp["objects"]: if o.get("kind") == "assessment": o["criteria"] = [{"name": "A", "weight": 0.5}, {"name": "B", "weight": 0.3}] rep = run_validation(bp, tenant_id="t1") d10 = [d for d in rep["dimensions"] if d["dim"] == "D10"][0] check("权重和 0.8 != 1 命中 V-RUB-004", any(i["code"] == "V-RUB-004" for i in d10["issues"]), json.dumps(d10["issues"], ensure_ascii=False)) check("D10 判 fail", d10["state"] == "fail") check("核心维度 fail → R3 → Q1", rep["quality_state"] == QUALITY_INCOMPLETE and any(r.startswith("R3") for r in rep["quality_reasons"]), str(rep["quality_reasons"])) def test_cycle_and_dangling(): print("\n[11] 任务链环检测") bp = good_blueprint() for o in bp["objects"]: if o.get("id") == "T5": o["next"] = "T1" # 造环 rep = run_validation(bp, tenant_id="t1") d08 = [d for d in rep["dimensions"] if d["dim"] == "D08"][0] check("检出环形路径", d08["metrics"]["cyclic_paths"] > 0, str(d08["metrics"])) check("命中 V-TASK-005", any(i["code"] == "V-TASK-005" for i in d08["issues"])) check("D08 判 fail", d08["state"] == "fail") def test_quality_rules(): print("\n[12] 5 级质量状态规则单测") from pbl_validation.dimensions import DimResult, Issue th = resolve_thresholds() check("R1 无维度 → Q0", decide_quality_state([], th, 0.0)[0] == QUALITY_DRAFT) check("R1 全 skip → Q0", decide_quality_state([DimResult("D01", "x", "skip")], th, 0.0)[0] == QUALITY_DRAFT) blocker_res = [DimResult("D02", "x", "warn", 0.9, issues=[Issue("D02", "V-X", "blocker", "b")])] st, why = decide_quality_state(blocker_res, th, 0.99) check("R2 blocker 优先于高分 → Q1", st == QUALITY_INCOMPLETE and why[0].startswith("R2"), str(why)) core_fail = [DimResult("D08", "x", "fail", 0.0), DimResult("D02", "y", "pass", 1.0)] st, why = decide_quality_state(core_fail, th, 0.95) check("R3 核心维度 fail → Q1", st == QUALITY_INCOMPLETE and why[0].startswith("R3"), str(why)) noncore_fail = [DimResult("D02", "y", "pass", 1.0), DimResult("D12", "x", "fail", 0.0)] st, why = decide_quality_state(noncore_fail, th, 0.30) check("R4 总分 < 0.40 → Q1", st == QUALITY_INCOMPLETE and why[0].startswith("R4"), str(why)) all_pass = [DimResult("D01", "a", "pass"), DimResult("D03", "b", "pass"), DimResult("D08", "c", "pass"), DimResult("D10", "d", "pass"), DimResult("D14", "e", "pass"), DimResult("D02", "f", "pass")] st, why = decide_quality_state(all_pass, th, 0.95) check("R5 高分无 warn → Q4", st == QUALITY_EXCELLENT and why[0].startswith("R5"), str(why)) with_warn = all_pass + [DimResult("D12", "g", "warn", 0.6)] st, why = decide_quality_state(with_warn, th, 0.92) check("R5 有 warn 不给 Q4，落 Q3", st == QUALITY_GOOD and why[0].startswith("R6"), str(why)) st, why = decide_quality_state(with_warn, th, 0.55) check("R7 中等分 → Q2", st == QUALITY_BASIC and why[0].startswith("R7"), str(why)) core_warn = [DimResult("D01", "a", "warn", 0.6), DimResult("D03", "b", "pass"), DimResult("D08", "c", "pass"), DimResult("D10", "d", "pass"), DimResult("D14", "e", "pass")] st, why = decide_quality_state(core_warn, th, 0.80) check("R6 核心维度非全 pass 不给 Q3 → Q2", st == QUALITY_BASIC and why[0].startswith("R7"), str(why)) def test_report_contract(): print("\n[13] 报告契约字段完备性") rep = run_validation(good_blueprint(), tenant_id="t1", blueprint_id="BP1", version="v2") required = ["contract", "engine_version", "report_id", "tenant_id", "blueprint_id", "version", "ruleset", "created_at", "elapsed_ms", "fingerprint", "quality_state", "quality_label", "quality_order", "quality_reasons", "total_score", "weighted_sum", "weight_sum", "dim_count", "dim_total", "state_summary", "severity_summary", "issue_count", "passed", "dimensions", "issues", "thresholds", "engine_errors", "dimensions_meta"] missing = [k for k in required if k not in rep] check("报告含全部契约字段", not missing, str(missing)) check("tenant_id 回显", rep["tenant_id"] == "t1") check("blueprint_id/version 回显", rep["blueprint_id"] == "BP1" and rep["version"] == "v2") check("dim_total = 14", rep["dim_total"] == 14) check("总分 = 加权和/权重和", abs(rep["total_score"] - rep["weighted_sum"] / rep["weight_sum"]) < 1e-3, "%s vs %s" % (rep["total_score"], rep["weighted_sum"] / rep["weight_sum"])) check("报告可 JSON 序列化", isinstance(json.dumps(rep, ensure_ascii=False), str)) d = rep["dimensions"][0] check("维度项字段齐备", all(k in d for k in ("dim", "name", "state", "score", "weight", "weighted_score", "metrics", "issues"))) row = report_to_row(rep, operator="tester") check("结果行 tenant_id 打头", row["tenant_id"] == "t1") check("结果行 quality_state 一致", row["quality_state"] == rep["quality_state"]) check("结果行 dim_scores 为 JSON 串", isinstance(json.loads(row["dim_scores"]), dict)) check("结果行 report_json 可反解", json.loads(row["report_json"])["report_id"] == rep["report_id"]) check("结果行 state=done", row["state"] == "done") def test_dims_subset(): print("\n[14] 指定维度子集执行") rep = run_validation(good_blueprint(), tenant_id="t1", dims=["D08", "task_chain"]) check("去重后只跑 D08", rep["dim_count"] == 1, str(rep["dim_count"])) check("维度为 D08", rep["dimensions"][0]["dim"] == "D08") rep2 = run_validation(good_blueprint(), tenant_id="t1", dims=["D01"]) check("D01 单维可跑", rep2["dim_count"] == 1 and rep2["dimensions"][0]["dim"] == "D01") def test_payload_shapes(): print("\n[15] 多种载荷形态兼容") bp = good_blueprint() kinds = {} for o in bp["objects"]: kinds.setdefault(o["kind"], []).append(o) shape_b = {"root": bp["root"], "kinds": kinds} rep_b = run_validation(shape_b, tenant_id="t1") check("形态B(kinds) 判 Q4", rep_b["quality_state"] == QUALITY_EXCELLENT, "%s %s" % (rep_b["quality_state"], rep_b["quality_reasons"])) shape_c = {"name": "供应链经营沙盘", "driving_question": "如何在需求波动下实现利润最大化？", "duration_minutes": 115, "objectives": [o for o in bp["objects"] if o["kind"] == "objective"], "roles": [o for o in bp["objects"] if o["kind"] == "role"], "scenes": [o for o in bp["objects"] if o["kind"] == "scene"], "entities": [o for o in bp["objects"] if o["kind"] == "entity"], "tasks": [o for o in bp["objects"] if o["kind"] == "task"], "rules": [o for o in bp["objects"] if o["kind"] == "rule"], "assessments": [o for o in bp["objects"] if o["kind"] == "assessment"]} rep_c = run_validation(shape_c, tenant_id="t1") check("形态C(顶层复数键) 判 Q4", rep_c["quality_state"] == QUALITY_EXCELLENT, "%s %s" % (rep_c["quality_state"], rep_c["quality_reasons"])) wrapped = {"blueprint": bp} rep_w = run_validation(wrapped, tenant_id="t1") check("形态D({blueprint:...}) 判 Q4", rep_w["quality_state"] == QUALITY_EXCELLENT, "%s %s" % (rep_w["quality_state"], rep_w["quality_reasons"])) def test_api_layer(): print("\n[16] 契约接口层（无 DB 环境 fail-closed / 离线兜底）") from pbl_validation import api as vapi check("API_REGISTRY 含 5 个契约", len(vapi.API_REGISTRY) == 5, str(sorted(vapi.API_REGISTRY))) expect = {"pbl_validation_run", "pbl_validation_get", "pbl_validation_list", "pbl_validation_rule_list", "pbl_validation_rule_save"} check("契约名与规格一致", set(vapi.API_REGISTRY) == expect, str(set(vapi.API_REGISTRY) ^ expect)) check("内置规则 14 条", len(vapi.builtin_rules()) == 14) check("内置规则 script_type=1", all(r["script_type"] == 1 for r in vapi.builtin_rules())) check("内置规则 rule_json 可解析", all(isinstance(json.loads(r["rule_json"]), dict) for r in vapi.builtin_rules())) # tenant_id 缺失 → fail-closed（不抛异常，返回错误结构） res = vapi.pbl_validation_run({"blueprint": good_blueprint()}) check("run 缺 tenant_id 返回错误", res["success"] is False and res["code"] in ("V-TENANT-001", "V-DB-001", "V-DB-002"), str(res.get("code"))) res2 = vapi.pbl_validation_run({"tenant_id": "t1"}) check("run 缺 blueprint 返回 V-INPUT-001", res2["success"] is False and res2["code"] == "V-INPUT-001", str(res2)) res3 = vapi.pbl_validation_rule_save({"tenant_id": "t1", "dim": "D99"}) check("rule_save 非法 dim 被拒", res3["success"] is False and res3["code"] == "V-INPUT-021", str(res3.get("code"))) res4 = vapi.pbl_validation_rule_save({"tenant_id": "t1", "dim": "D08", "thresholds": {"mcRatio": 5}}) check("rule_save 非法阈值被拒（fail-closed）", res4["success"] is False and res4["code"] == "V-INPUT-022", str(res4.get("code"))) res5 = vapi.pbl_validation_list({"tenant_id": "t1", "quality_state": "Q9"}) check("list 非法 quality_state 被拒", res5["success"] is False and res5["code"] == "V-INPUT-011", str(res5.get("code"))) res6 = vapi.pbl_validation_get({"tenant_id": "t1"}) check("get 缺 id 被拒", res6["success"] is False and res6["code"] == "V-INPUT-010", str(res6.get("code"))) # 无 DB 环境下 run 应离线兜底（内置规则）或明确报 DB 错误，不允许静默成功 res7 = vapi.pbl_validation_run({"tenant_id": "t1", "blueprint": good_blueprint(), "persist": 0}) if res7["success"]: check("离线兜底：run 成功且用内置规则", res7["data"]["rules_from_builtin"] is True and res7["data"]["rule_count"] == 14 and res7["data"]["quality_state"] == QUALITY_EXCELLENT, json.dumps({k: res7["data"][k] for k in ("rules_from_builtin", "rule_count", "quality_state")}, ensure_ascii=False)) check("离线兜底：publishable=True", res7["data"]["publishable"] is True) else: check("无 DB 时明确报错（不静默）", res7["code"] in ("V-DB-001", "V-DB-002"), str(res7)) def main(): print("=" * 72) print("pbl_validation 引擎自测（14 维 + 5 级质量状态 + 阈值配置化）") print("=" * 72) for fn in (test_constants, test_dimensions_registry, test_normalize, test_excellent, test_mc_ratio_gate, test_rule_json_override, test_empty_blueprint, test_partial_blueprint, test_safety, test_rubric_weight, test_cycle_and_dangling, test_quality_rules, test_report_contract, test_dims_subset, test_payload_shapes, test_api_layer): try: fn() except Exception as exc: import traceback traceback.print_exc() FAILED.append((fn.__name__, "%s: %s" % (type(exc).__name__, exc))) print("\n" + "=" * 72) print("PASSED: %d FAILED: %d" % (len(PASSED), len(FAILED))) if FAILED: for name, detail in FAILED: print(" - %s : %s" % (name, detail)) print("=" * 72) return 1 if FAILED else 0 if __name__ == "__main__": sys.exit(main())