#!/usr/bin/env python3 # -*- coding: utf-8 -*- """pbl_compiler M3a 自检脚本（QC #1 整改：真实落盘，可执行）。 运行： cd modules/pbl_compiler && python3 scripts/test_m3a_selfcheck.py （或 python3 -m pytest scripts/test_m3a_selfcheck.py -q） 覆盖 QC 退回意见要求的断言组（全部纯函数/纯结构，不依赖真实 DB）： 组1 canonical 确定性：同输入 → canonical_json 字节全等 + sha256 指纹全等（US-11/F-CP-03） 组2 GD 10 顶层键齐全（design 第 9 章） 组3 volatile 字段（createdAt/durationMs/taskNo/id...）不参与指纹（29.6 确定性） 组4 register() 可跑通不崩（QC #3：格式串占位符错位修复验证） 组5 三处同步（QC #2）：__init__ 导出 == init.CONTRACTS == load_path.PATHS 契约名一致 组6 无硬编码 DB（QC #4）：api.py 无 `DB = '...'` 常量，_dbname() 走 get_module_dbname 组7 浮点定点 + sort_keys：键序无关、浮点 6 位定点（canonical 规范化） 组8 registry_hash 确定性：同能力集 → 同 hash；增删能力 → hash 变化 退出码 0 = 全部通过；非 0 = 有断言失败（打印 FAIL 明细）。 """ from __future__ import annotations import ast import io import os import re import sys import types HERE = os.path.dirname(os.path.abspath(__file__)) MOD_ROOT = os.path.dirname(HERE) sys.path.insert(0, MOD_ROOT) # pbl_compiler 包 sys.path.insert(0, os.path.join(os.path.dirname(MOD_ROOT), 'pbl_common')) # pbl_common 依赖 # ---- ahserver 桩（工作空间无平台运行时；仅为 import init.py 不崩）---- if 'ahserver.serverenv' not in sys.modules: _ah = types.ModuleType('ahserver') _se = types.ModuleType('ahserver.serverenv') class _ServerEnv(object): _inst = None def __new__(cls): if cls._inst is None: cls._inst = super(_ServerEnv, cls).__new__(cls) return cls._inst _se.ServerEnv = _ServerEnv _ah.serverenv = _se sys.modules['ahserver'] = _ah sys.modules['ahserver.serverenv'] = _se from pbl_compiler import canonical as cn # noqa: E402 from pbl_compiler import gd_builder as gb # noqa: E402 from pbl_compiler import api as api # noqa: E402 from pbl_compiler import init as init_mod # noqa: E402 import pbl_compiler as pkg # noqa: E402 PASS, FAIL = [], [] def check(name, cond, detail=''): (PASS if cond else FAIL).append(name) print(' [%s] %s%s' % ('PASS' if cond else 'FAIL', name, (' <- ' + detail) if (detail and not cond) else '')) # 一个最小但真实的蓝图版本快照（含 7 类子对象若干 + 易变字段） def _snapshot(): return { 'content': { 'project': [{'id': 'p1', 'title': '火星基地', 'summary': '建造可持续基地'}], 'problem': [{'id': 'q1', 'driving_question': '如何在火星自给自足？'}], 'learning_goals': [ {'id': 'g2', 'text': '掌握生态循环'}, {'id': 'g1', 'text': '理解能源约束'}, ], 'missions': [{'id': 'm1', 'title': '着陆', 'order': 1}], 'roles': [{'id': 'r1', 'name': '工程师'}], 'scenes': [{'id': 's1', 'name': '基地外'}], 'entities': [{'id': 'e1', 'entity_key': 'solar_panel', 'scene_id': 's1'}], 'events': [{'id': 'ev1', 'event_key': 'power_on', 'entity_key': 'solar_panel'}], 'rubrics': [{'id': 'rb1', 'criterion': '可行性', 'weight': 0.5}], } } def _ctx(i=0): return {'blueprint_id': 101, 'blueprint_version_no': 3, 'compiler_version': '1.0.0', 'ruleset_version': 'pbl.rules.v1', 'rules_hash': 'deadbeefcafe1234', 'created_at': '2000-01-01T00:00:00Z', 'duration_ms': i, 'task_no': 'CT:101:3:1.0.0:%05d' % i} def group1_canonical_determinism(): print('组1 canonical 确定性（同输入指纹全等）') obj = {'b': [3, 1, 2], 'a': {'z': 1.5, 'y': 'x'}, 'n': None} c1 = cn.canonical_json(obj, strip=False) c2 = cn.canonical_json(obj, strip=False) check('canonical_json 同输入字节全等', c1 == c2, '%r != %r' % (c1, c2)) f1, f2 = cn.sha256_fingerprint(obj), cn.sha256_fingerprint(obj) check('sha256_fingerprint 同输入全等', f1 == f2 and len(f1) == 64, f1) # 键序无关 obj2 = {'a': {'y': 'x', 'z': 1.5}, 'n': None, 'b': [3, 1, 2]} check('键序不同 canonical 仍全等（sort_keys）', cn.canonical_json(obj, strip=False) == cn.canonical_json(obj2, strip=False)) def group2_gd_top_keys(): print('组2 GD 10 顶层键齐全') gd, fp = gb.build_game_definition(_snapshot(), _ctx(), []) check('GD_TOP_KEYS 恰为 10 个', len(gb.GD_TOP_KEYS) == 10, str(gb.GD_TOP_KEYS)) missing = [k for k in gb.GD_TOP_KEYS if k not in gd] check('GD 含全部 10 顶层键', not missing, 'missing=%s' % missing) check('GD 指纹为 64 位 sha256', isinstance(fp, str) and len(fp) == 64, fp) check('manifest.schema = pbl.game_definition.v1', gd.get('manifest', {}).get('schema') == cn.GD_SCHEMA) def group3_volatile_excluded(): print('组3 volatile 字段不参与指纹') gd_a, fp_a = gb.build_game_definition(_snapshot(), _ctx(0), []) gd_b, fp_b = gb.build_game_definition(_snapshot(), _ctx(999), []) check('created_at/duration_ms/task_no 不同但指纹全等', fp_a == fp_b, '%s != %s' % (fp_a, fp_b)) # 直接验证 strip_volatile 剔除易变键 dirty = {'createdAt': '2020', 'durationMs': 5, 'taskNo': 'X', 'id': 9, 'stable': 'keep'} stripped = cn.strip_volatile(dirty) leaked = [k for k in ('createdAt', 'durationMs', 'taskNo', 'id') if k in stripped] check('strip_volatile 剔除全部易变键', not leaked and stripped.get('stable') == 'keep', 'leaked=%s stripped=%s' % (leaked, stripped)) for k in ('createdAt', 'created_at', 'durationMs', 'duration_ms', 'taskNo', 'task_no', 'id', 'timestamp'): if k not in cn.VOLATILE_KEYS: check('VOLATILE_KEYS 含 %s' % k, False) return check('VOLATILE_KEYS 覆盖时间/耗时/任务号/id', True) def group4_register_runs(): print('组4 register() 可跑通不崩（QC #3）') sys.path.insert(0, HERE) import importlib lp = importlib.import_module('load_path') os.environ['RBAC_SET_PERM'] = '/nonexistent/set_role_perm.py' # 强制走 pending 分支 os.environ.pop('PY', None) try: ret = lp.register() crashed = False err = '' except Exception as exc: # noqa: BLE001 crashed, ret, err = True, None, '%s: %s' % (type(exc).__name__, exc) check('register() 不抛 TypeError（格式串占位符已对齐）', not crashed, err) check('register() 返回 bool（pending 非空 → False）', isinstance(ret, bool), repr(ret)) check('PATHS 含 15 个端点', len(lp.PATHS) == 15, str(len(lp.PATHS))) def group5_three_place_sync(): print('组5 三处同步（QC #2）') api_contracts = set(n for n in api.__all__ if n.startswith('pbl_')) init_contracts = set(init_mod.CONTRACTS.keys()) pkg_exports = set(n for n in pkg.__all__ if n.startswith('pbl_')) sys.path.insert(0, HERE) import importlib lp = importlib.import_module('load_path') rbac_names = set(os.path.basename(p).replace('.dspy', '') for p, _ in lp.PATHS) check('init.CONTRACTS == api 契约（15 个）', init_contracts == api_contracts, 'only_api=%s only_init=%s' % (api_contracts - init_contracts, init_contracts - api_contracts)) check('__init__ 导出 == init.CONTRACTS', pkg_exports == init_contracts, 'missing_export=%s' % (init_contracts - pkg_exports)) check('load_path.PATHS 契约名 == init.CONTRACTS', rbac_names == init_contracts, 'missing_rbac=%s extra_rbac=%s' % (init_contracts - rbac_names, rbac_names - init_contracts)) # QC #2 点名的 8 个此前漏接线契约必须在三处都出现 eight = ['pbl_compiler_task_get', 'pbl_compiler_task_list', 'pbl_game_definition_get', 'pbl_game_definition_get_by_blueprint', 'pbl_compiler_verify_determinism', 'pbl_compiler_version_register', 'pbl_compiler_version_get', 'pbl_compiler_version_diff'] for name in eight: ok = (name in pkg_exports and name in init_contracts and name in rbac_names and callable(getattr(pkg, name, None))) check('QC#2 契约 %s 三处齐全且可调用' % name, ok) # verify_determinism 是 US-11 验收入口，必须可从包直接取到 check('US-11 入口 pbl_compiler_verify_determinism 可达', callable(getattr(pkg, 'pbl_compiler_verify_determinism', None))) def group6_no_hardcoded_db(): print('组6 无硬编码 DB（QC #4）') src = io.open(os.path.join(MOD_ROOT, 'pbl_compiler', 'api.py'), encoding='utf-8').read() check("api.py 无 `DB = '...'` 常量", re.search(r"(?m)^DB\s*=\s*['\"]", src) is None) check('api.py 无模块级 DB 属性', not hasattr(api, 'DB')) check('_dbname() 走 get_module_dbname 解析', callable(api._dbname) and isinstance(api._dbname(), str) and api._dbname()) # AST 扫描：模块级赋值不得出现 DB = 字面量 tree = ast.parse(src) bad = [] for node in tree.body: if isinstance(node, ast.Assign): for t in node.targets: if isinstance(t, ast.Name) and t.id == 'DB' \ and isinstance(node.value, ast.Constant) \ and isinstance(node.value.value, str): bad.append(t.id) check('AST：无模块级 DB = <字符串> 硬编码', not bad, str(bad)) # set_dbname 注入生效（宿主/测试可覆盖，证明非写死） old = api._dbname() api.set_dbname('pbl_injected') injected = api._dbname() api.set_dbname(None) check('set_dbname 注入可覆盖库名', injected == 'pbl_injected', injected) check('set_dbname(None) 回落解析值', api._dbname() == old, api._dbname()) def group7_float_and_sort(): print('组7 浮点定点 + 无空白分隔符') c = cn.canonical_json({'x': 1.00000049, 'y': 2.5}, strip=False) check('canonical 无空白分隔符', ' ' not in c and '\n' not in c, c) check('浮点按 FLOAT_PRECISION 定点', cn.FLOAT_PRECISION == 6) a = cn.canonical_json({'k': 0.1 + 0.2}, strip=False) b = cn.canonical_json({'k': 0.3}, strip=False) check('0.1+0.2 与 0.3 定点后全等', a == b, '%s vs %s' % (a, b)) def group8_registry_hash(): print('组8 registry_hash 确定性') caps = [{'capability_key': 'b', 'version_no': 1}, {'capability_key': 'a', 'version_no': 2}] h1 = cn.registry_hash(caps) h2 = cn.registry_hash(list(reversed(caps))) check('同能力集（顺序无关）registry_hash 全等', h1 == h2 and bool(h1), '%s vs %s' % (h1, h2)) caps2 = caps + [{'capability_key': 'c', 'version_no': 1}] check('新增能力 registry_hash 变化', cn.registry_hash(caps2) != h1) def main(): print('=' * 64) print('pbl_compiler M3a 自检（QC #1/#2/#3/#4 整改验证）') print('=' * 64) for fn in (group1_canonical_determinism, group2_gd_top_keys, group3_volatile_excluded, group4_register_runs, group5_three_place_sync, group6_no_hardcoded_db, group7_float_and_sort, group8_registry_hash): try: fn() except Exception as exc: # noqa: BLE001 import traceback FAIL.append('%s(异常)' % fn.__name__) print(' [ERROR] %s: %s' % (fn.__name__, exc)) traceback.print_exc() print('-' * 64) print('PASS=%d FAIL=%d' % (len(PASS), len(FAIL))) if FAIL: print('FAILED: %s' % FAIL) return 1 print('ALL GREEN') return 0 # ---- pytest 兼容 ---- def test_m3a_selfcheck(): assert main() == 0 if __name__ == '__main__': sys.exit(main())