533 lines
25 KiB
Python
533 lines
25 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
M3b 离线自测：第12章 event→condition→response → script_type=1 规则 JSON 映射 + 产物导出。
|
||
|
||
运行：cd modules/pbl_compiler && python3 scripts/test_m3b_mapping.py
|
||
不依赖数据库 / ahserver —— 纯函数层 + FakeSor 模拟 sqlor 上下文验证导出幂等性。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import asyncio
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, ROOT)
|
||
|
||
# 绕过包 __init__.py（其 import ahserver，离线环境不可用）：注册同名命名空间包
|
||
import types # noqa: E402
|
||
_pkg = types.ModuleType('pbl_compiler')
|
||
_pkg.__path__ = [os.path.join(ROOT, 'pbl_compiler')]
|
||
sys.modules['pbl_compiler'] = _pkg
|
||
|
||
|
||
def run(coro):
|
||
loop = asyncio.new_event_loop()
|
||
try:
|
||
return loop.run_until_complete(coro)
|
||
finally:
|
||
loop.close()
|
||
|
||
|
||
from pbl_compiler.script_mapping import ( # noqa: E402
|
||
CompileError, build_rules, canonical_dumps, canonical_rules_json, diff_rules,
|
||
has_errors, normalize_condition, normalize_ecr, normalize_event,
|
||
normalize_responses, rules_hash, sha256_text, sort_rules, strip_volatile,
|
||
build_rules_manifest, validate_rules, iter_ecr_sources, OP_ARITY,
|
||
)
|
||
from pbl_compiler import exporter # noqa: E402
|
||
|
||
PASS, FAIL = [], []
|
||
|
||
|
||
def check(name, cond, extra=''):
|
||
(PASS if cond else FAIL).append(name)
|
||
print((' [PASS] ' if cond else ' [FAIL] ') + name + ((' -> ' + str(extra)) if (extra and not cond) else ''))
|
||
|
||
|
||
CTX = {'tenant_id': 'T1', 'blueprint_id': 'bp_demo', 'blueprint_version': 'v3',
|
||
'strict': False}
|
||
|
||
SNAPSHOT = {
|
||
'blueprint_id': 'bp_demo',
|
||
'version': 'v3',
|
||
'rules': [
|
||
{
|
||
'id': 'r_score_pass',
|
||
'name': '得分达标开门',
|
||
'description': '玩家得分>=60 且 已完成任务 时开启第二场景',
|
||
'priority': 200,
|
||
'event': {'type': 'onValueChange', 'source': 'player_1',
|
||
'match': {'field': 'score'}},
|
||
'condition': {'kind': 'all', 'items': [
|
||
{'op': '>=', 'left': 'var:score', 'right': '60'},
|
||
{'field': 'state.task_done', 'value': True},
|
||
]},
|
||
'responses': [
|
||
{'action': 'goto_scene', 'params': {'scene': 'scene_2'}},
|
||
{'action': 'add_score', 'target': 'self', 'params': {'amount': 10}},
|
||
],
|
||
'limits': {'cooldown_ms': 1000, 'max_triggers': 3},
|
||
'tags': ['Main Flow', 'score'],
|
||
'created_at': '2026-09-18 10:00:00',
|
||
},
|
||
{
|
||
'name': '计时器提示',
|
||
'priority': 50,
|
||
'event': {'type': 'timer', 'interval_ms': 5000},
|
||
'condition': 'remaining <= 30',
|
||
'then': [{'action': 'show_message', 'params': {'msg': '时间不多了'}}],
|
||
},
|
||
],
|
||
'scenes': [
|
||
{'id': 'scene_1', 'events': [
|
||
{'id': 'ev_enter', 'type': 'enter', 'source': 'scene_1',
|
||
'responses': [{'action': 'play_bgm', 'params': {'track': 1}}]},
|
||
]},
|
||
],
|
||
}
|
||
|
||
|
||
def t_determinism():
|
||
print('\n[T1] 确定性：同快照两次编译逐字节一致')
|
||
r1, i1 = build_rules(SNAPSHOT, CTX)
|
||
r2, i2 = build_rules(json.loads(json.dumps(SNAPSHOT)), dict(CTX))
|
||
check('两次编译 rules_json 完全相同', canonical_rules_json(r1) == canonical_rules_json(r2))
|
||
check('rules_hash 相同', rules_hash(r1) == rules_hash(r2))
|
||
check('键序无关（打乱 dict 键序后 hash 不变）',
|
||
rules_hash(r1) == rules_hash(json.loads(canonical_rules_json(r1))))
|
||
# 易变字段不影响产物
|
||
s2 = json.loads(json.dumps(SNAPSHOT))
|
||
s2['rules'][0]['created_at'] = '2030-01-01 00:00:00'
|
||
s2['rules'][0]['updated_by'] = 'someone'
|
||
r3, _ = build_rules(s2, CTX)
|
||
check('created_at/updated_by 变更不影响 rules_hash', rules_hash(r1) == rules_hash(r3))
|
||
# 零空白
|
||
txt = canonical_rules_json(r1)
|
||
check('产物零空白（无 ", " / ": "）', ', ' not in txt and ': ' not in txt)
|
||
check('产物无换行', '\n' not in txt)
|
||
|
||
|
||
def t_mapping():
|
||
print('\n[T2] 映射正确性：event / condition / response 归一')
|
||
rules, issues = build_rules(SNAPSHOT, CTX)
|
||
check('共映射 3 条规则（rules 2 + scenes.events 1）', len(rules) == 3, len(rules))
|
||
by_id = {r['rule_id']: r for r in rules}
|
||
r = by_id['r_score_pass']
|
||
check('event.type 归一 onValueChange -> change', r['event']['type'] == 'change', r['event'])
|
||
check('event.source 保留', r['event']['source'] == 'player_1')
|
||
check('event.match 键归一 snake', r['event']['match'] == {'field': 'score'}, r['event']['match'])
|
||
check('condition 为 all 组合子', r['condition']['kind'] == 'all')
|
||
ops = [c['op'] for c in r['condition']['items']]
|
||
check('操作符归一 >= -> gte, 隐式 == -> eq', ops == ['gte', 'eq'], ops)
|
||
check('左操作数 var:score -> {kind:var,name:score}',
|
||
r['condition']['items'][0]['left'] == {'kind': 'var', 'name': 'score'})
|
||
check("右操作数 '60' -> const 60（数字归一）",
|
||
r['condition']['items'][0]['right'] == {'kind': 'const', 'value': 60})
|
||
check('嵌套 var 路径 state.task_done -> var',
|
||
r['condition']['items'][1]['left'] == {'kind': 'var', 'name': 'state.task_done'})
|
||
check('responses 2 个动作', len(r['responses']) == 2)
|
||
check('params 同义键归一 scene -> scene_id',
|
||
r['responses'][0]['params'].get('scene_id') == 'scene_2', r['responses'][0]['params'])
|
||
check('params 同义键归一 amount -> value',
|
||
r['responses'][1]['params'].get('value') == 10, r['responses'][1]['params'])
|
||
check("target 'self' -> kind self", r['responses'][1]['target'] == {'kind': 'self', 'id': ''})
|
||
check('limits 保留', r['limits'] == {'cooldown_ms': 1000, 'max_triggers': 3})
|
||
check('tags 归一 snake + 排序', r['tags'] == ['main_flow', 'score'], r['tags'])
|
||
check('origin 可追溯', r['origin']['blueprint_version'] == 'v3'
|
||
and r['origin']['sub_object_id'] == 'r_score_pass')
|
||
check('schema_version = 1.0', r['schema_version'] == '1.0')
|
||
|
||
# 无 id 的规则 → 内容哈希派生 rule_id
|
||
timer = [x for x in rules if x['event']['type'] == 'timer'][0]
|
||
check('无 id 规则派生 rule_ 前缀哈希 id', timer['rule_id'].startswith('rule_'), timer['rule_id'])
|
||
check('timer interval_ms 提升到 match', timer['event']['match'].get('interval_ms') == 5000)
|
||
check('字符串条件 "remaining <= 30" 解析为 compare lte',
|
||
timer['condition'] == {'kind': 'compare', 'op': 'lte',
|
||
'left': {'kind': 'var', 'name': 'remaining'},
|
||
'right': {'kind': 'const', 'value': 30}}, timer['condition'])
|
||
check("'then' 别名识别为 responses", len(timer['responses']) == 1)
|
||
|
||
# 排序：priority 降序
|
||
prios = [x['priority'] for x in rules]
|
||
check('规则按 priority 降序稳定排序', prios == sorted(prios, reverse=True), prios)
|
||
|
||
|
||
def t_shapes():
|
||
print('\n[T3] 多形态兼容 + 边界')
|
||
# 形态 B：events 顶层
|
||
snap_b = {'events': [{'id': 'e1', 'type': 'click', 'source': 'btn',
|
||
'conditions': [{'field': 'hp', 'op': '<', 'value': 1}],
|
||
'actions': [{'action': 'game_over'}]}]}
|
||
rb, ib = build_rules(snap_b, CTX)
|
||
check('形态B events 顶层可映射', len(rb) == 1 and rb[0]['event']['type'] == 'click', rb)
|
||
check('形态B conditions 别名识别', rb[0]['condition']['op'] == 'lt')
|
||
check('形态B 无 error issue', not has_errors(ib), ib)
|
||
|
||
# 形态 A 单 dict（非数组）
|
||
snap_c = {'rule': {'id': 'r1', 'event': {'type': 'start'},
|
||
'responses': [{'action': 'init_world'}]}}
|
||
rc, ic = build_rules(snap_c, CTX)
|
||
check('单 dict rule 也可映射', len(rc) == 1, rc)
|
||
|
||
# 无条件 → always
|
||
snap_d = {'rules': [{'id': 'r2', 'event': {'type': 'enter'},
|
||
'responses': [{'action': 'log'}]}]}
|
||
rd, _ = build_rules(snap_d, CTX)
|
||
check('缺省条件归一为 always', rd[0]['condition'] == {'kind': 'always'})
|
||
|
||
# 空快照
|
||
try:
|
||
build_rules({}, CTX)
|
||
check('空快照抛 CompileError', False)
|
||
except CompileError as e:
|
||
check('空快照抛 CompileError(E_SNAPSHOT_EMPTY)', e.code == 'E_SNAPSHOT_EMPTY', e.code)
|
||
|
||
# tenant 缺失
|
||
try:
|
||
build_rules(SNAPSHOT, {'blueprint_id': 'b', 'blueprint_version': 'v'})
|
||
check('tenant_id 缺失抛错', False)
|
||
except CompileError as e:
|
||
check('tenant_id 缺失 fail-closed', e.code == 'E_TENANT_REQUIRED', e.code)
|
||
|
||
# 无 ECR
|
||
r_none, i_none = build_rules({'scenes': [{'id': 's'}]}, CTX)
|
||
check('无 ECR 记录 → 0 规则 + W_NO_ECR', r_none == [] and
|
||
any(i['code'] == 'W_NO_ECR' for i in i_none))
|
||
|
||
|
||
def t_failclosed():
|
||
print('\n[T4] fail-closed：strict 模式未登记 action / 缺 event / 缺 response')
|
||
strict_ctx = dict(CTX, strict=True, known_actions={'goto_scene', 'add_score'})
|
||
snap = {'rules': [
|
||
{'id': 'ok', 'event': {'type': 'click'}, 'responses': [{'action': 'goto_scene'}]},
|
||
{'id': 'bad_action', 'event': {'type': 'click'},
|
||
'responses': [{'action': 'launch_missile'}]},
|
||
{'id': 'no_event', 'responses': [{'action': 'goto_scene'}]},
|
||
{'id': 'no_resp', 'event': {'type': 'click'}},
|
||
]}
|
||
rules, issues = build_rules(snap, strict_ctx)
|
||
codes = {i['code'] for i in issues}
|
||
check('未登记 action → E_CAP_UNKNOWN', 'E_CAP_UNKNOWN' in codes, codes)
|
||
check('缺 event → E_EVENT_MISSING', 'E_EVENT_MISSING' in codes)
|
||
check('缺 response → E_RESP_MISSING', 'E_RESP_MISSING' in codes)
|
||
check('strict 下 bad_action 规则被剔除', 'bad_action' not in {r['source_id'] for r in rules})
|
||
check('存在 error issue', has_errors(issues))
|
||
check('无效规则被剔除并记 E_RULE_DROPPED',
|
||
any(i['code'] == 'E_RULE_DROPPED' for i in issues))
|
||
check('no_event / no_resp 规则不进入产物',
|
||
not ({'no_event', 'no_resp'} & {r['source_id'] for r in rules}))
|
||
check('存活规则自身通过 validate_rules（无半成品）',
|
||
not has_errors(validate_rules(rules)), validate_rules(rules))
|
||
# 非 strict：降级 warning，规则保留
|
||
rules2, issues2 = build_rules(snap, dict(strict_ctx, strict=False))
|
||
check('非 strict 降级为 W_CAP_UNKNOWN 且保留规则',
|
||
any(i['code'] == 'W_CAP_UNKNOWN' for i in issues2)
|
||
and 'bad_action' in {r['source_id'] for r in rules2})
|
||
# 未知操作符
|
||
_, i3 = build_rules({'rules': [{'id': 'x', 'event': {'type': 'click'},
|
||
'condition': {'op': '~=', 'left': 'a', 'right': 1},
|
||
'responses': [{'action': 'log'}]}]}, CTX)
|
||
check('未知操作符 → E_COND_OP_UNKNOWN', any(i['code'] == 'E_COND_OP_UNKNOWN' for i in i3))
|
||
# between 缺边界
|
||
_, i4 = build_rules({'rules': [{'id': 'y', 'event': {'type': 'click'},
|
||
'condition': {'op': 'between', 'left': 'a', 'right': [1]},
|
||
'responses': [{'action': 'log'}]}]}, CTX)
|
||
check('between 边界不足 → E_COND_BETWEEN', any(i['code'] == 'E_COND_BETWEEN' for i in i4))
|
||
|
||
|
||
def t_manifest_diff():
|
||
print('\n[T5] manifest / diff / 幂等指纹')
|
||
rules, _ = build_rules(SNAPSHOT, CTX)
|
||
m = build_rules_manifest(rules)
|
||
check('manifest rule_count 正确', m['rule_count'] == len(rules))
|
||
check('manifest script_type=1', m['script_type'] == 1)
|
||
check('manifest rules_hash 与 rules_hash 一致', m['rules_hash'] == rules_hash(rules))
|
||
check('manifest 不含时间戳字段',
|
||
not any(k in canonical_dumps(m) for k in ('created_at', 'timestamp')))
|
||
check('manifest 每条含 content_hash', all(e.get('content_hash') for e in m['rules']))
|
||
|
||
rules2, _ = build_rules(SNAPSHOT, CTX)
|
||
d_same = diff_rules(rules, rules2)
|
||
check('相同规则 diff 全空', d_same['added'] == [] and d_same['removed'] == []
|
||
and d_same['changed'] == [], d_same)
|
||
mod = json.loads(canonical_rules_json(rules))
|
||
mod[0]['priority'] = 999
|
||
d = diff_rules(rules, mod)
|
||
check('修改 priority → changed 命中', d['changed_count'] == 1, d)
|
||
d2 = diff_rules(rules, mod[:1])
|
||
check('删规则 → removed 命中', d2['removed_count'] == len(rules) - 1, d2)
|
||
|
||
|
||
def t_strip():
|
||
print('\n[T6] strip_volatile / 操作数归一')
|
||
o = {'a': 1, 'created_at': 'x', 'nested': {'updated_by': 'y', 'b': [1, {'ts': 2, 'c': 3}]}}
|
||
s = strip_volatile(o)
|
||
check('顶层易变键剥离', 'created_at' not in s)
|
||
check('嵌套易变键剥离', 'updated_by' not in s['nested'] and 'ts' not in s['nested']['b'][1])
|
||
check('正常键保留', s['a'] == 1 and s['nested']['b'][1]['c'] == 3)
|
||
iss = []
|
||
from pbl_compiler.script_mapping import normalize_operand
|
||
check("'$hp' -> var", normalize_operand('$hp', iss, 'p') == {'kind': 'var', 'name': 'hp'})
|
||
check("'{{score}}' -> var", normalize_operand('{{score}}', iss, 'p') == {'kind': 'var', 'name': 'score'})
|
||
check("'3.5' -> const float", normalize_operand('3.5', iss, 'p') == {'kind': 'const', 'value': 3.5})
|
||
check("'true' -> const bool", normalize_operand('true', iss, 'p') == {'kind': 'const', 'value': True})
|
||
check("[1,2] -> list", normalize_operand([1, 2], iss, 'p')['kind'] == 'list')
|
||
check("{'kind':'entity_attr'} 保留", normalize_operand(
|
||
{'kind': 'entity_attr', 'entity': 'e1', 'attr': 'hp'}, iss, 'p') ==
|
||
{'kind': 'entity_attr', 'entity': 'e1', 'attr': 'hp'})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# FakeSor：模拟 sqlor 上下文，验证 exporter 幂等 + tenant 打头
|
||
# ---------------------------------------------------------------------------
|
||
class Row(object):
|
||
def __init__(self, **kw):
|
||
self.__dict__.update(kw)
|
||
|
||
|
||
class FakeSor(object):
|
||
def __init__(self, tables=None):
|
||
self.tables = tables or {}
|
||
self.sqls = []
|
||
self.inserts = []
|
||
self.updates = []
|
||
|
||
async def sqlExe(self, sql, ns=None):
|
||
self.sqls.append((sql, dict(ns or {})))
|
||
low = sql.strip().lower()
|
||
if low.startswith('select 1 from'):
|
||
t = sql.split('`')[1]
|
||
if t not in self.tables:
|
||
raise Exception('no table %s' % t)
|
||
return []
|
||
if low.startswith('select column_name'):
|
||
t = (ns or {}).get('tbl')
|
||
return [Row(COLUMN_NAME=c) for c in sorted(self.tables.get(t, {}).get('cols', []))]
|
||
if low.startswith('select * from'):
|
||
t = sql.split('`')[1]
|
||
return [Row(**r) for r in self.tables.get(t, {}).get('rows', [])]
|
||
if low.startswith('delete from'):
|
||
t = sql.split('`')[1]
|
||
n = len(self.tables.get(t, {}).get('rows', []))
|
||
self.tables.setdefault(t, {'cols': [], 'rows': []})['rows'] = []
|
||
return [Row(deleted=n)]
|
||
return []
|
||
|
||
async def R(self, table, where):
|
||
rows = self.tables.get(table, {}).get('rows', [])
|
||
hit = [r for r in rows if all(str(r.get(k)) == str(v) for k, v in where.items())]
|
||
return [Row(**r) for r in hit]
|
||
|
||
async def C(self, table, ns):
|
||
row = dict(ns)
|
||
row.setdefault('id', 'id_%d' % (len(self.inserts) + 1))
|
||
self.tables.setdefault(table, {'cols': [], 'rows': []})['rows'].append(row)
|
||
self.inserts.append((table, dict(ns)))
|
||
return row
|
||
|
||
async def U(self, table, ns, where):
|
||
self.updates.append((table, dict(ns), dict(where)))
|
||
rows = self.tables.get(table, {}).get('rows', [])
|
||
for r in rows:
|
||
if all(str(r.get(k)) == str(v) for k, v in where.items()):
|
||
r.update(ns)
|
||
return 1
|
||
|
||
async def D(self, table, where):
|
||
return await self.sqlExe('DELETE FROM `%s` WHERE 1' % table, where)
|
||
|
||
|
||
SCRIPT_COLS = ['id', 'tenant_id', 'script_code', 'name', 'script_type', 'content',
|
||
'description', 'group_name', 'enabled', 'content_hash', 'origin',
|
||
'blueprint_id', 'blueprint_version', 'created_at']
|
||
GD_COLS = ['id', 'tenant_id', 'blueprint_id', 'blueprint_version', 'game_definition',
|
||
'gd_hash', 'rules_hash', 'rule_count', 'status', 'compiler_version']
|
||
VER_COLS = ['id', 'tenant_id', 'blueprint_id', 'blueprint_version', 'rules_json',
|
||
'rules_hash', 'rules_manifest', 'rules_count', 'script_ids']
|
||
|
||
|
||
def make_sor():
|
||
return FakeSor({
|
||
'script': {'cols': SCRIPT_COLS, 'rows': []},
|
||
'pbl_game_definition': {'cols': GD_COLS, 'rows': []},
|
||
'pbl_compiler_version': {'cols': VER_COLS, 'rows': []},
|
||
})
|
||
|
||
|
||
def t_export():
|
||
print('\n[T7] 导出：script_type=1 落库 + 幂等 + tenant 打头')
|
||
rules, _ = build_rules(SNAPSHOT, CTX)
|
||
sor = make_sor()
|
||
res = run(
|
||
exporter.export_rules(sor, rules, CTX))
|
||
rows = sor.tables['script']['rows']
|
||
check('导出行数 == 规则数', len(rows) == len(rules), (len(rows), len(rules)))
|
||
check('全部 script_type=1', all(r['script_type'] == 1 for r in rows))
|
||
check('全部带 tenant_id', all(r['tenant_id'] == 'T1' for r in rows))
|
||
check('script_code 唯一', len({r['script_code'] for r in rows}) == len(rows))
|
||
check('content 是确定性 JSON（sort_keys 零空白）',
|
||
all(r['content'] == canonical_dumps(json.loads(r['content'])) for r in rows))
|
||
check('content_hash 与内容一致',
|
||
all(r['content_hash'] == sha256_text(r['content']) for r in rows))
|
||
check('rules_hash 与本地一致', res['rules_hash'] == rules_hash(rules))
|
||
|
||
# 幂等：再导一次 → 0 insert / 0 update
|
||
n_ins = len(sor.inserts)
|
||
res2 = run(
|
||
exporter.export_rules(sor, rules, CTX))
|
||
check('重复导出 skipped == 规则数', res2['skipped'] == len(rules), res2['skipped'])
|
||
check('重复导出 exported == 0', res2['exported'] == 0, res2['exported'])
|
||
check('重复导出无新增 insert', len(sor.inserts) == n_ins)
|
||
check('重复导出无 update', len(sor.updates) == 0)
|
||
check('表行数未增长', len(sor.tables['script']['rows']) == len(rules))
|
||
|
||
# 内容变化 → update 而非 insert
|
||
mod = json.loads(canonical_rules_json(rules))
|
||
mod[0]['priority'] = 777
|
||
res3 = run(
|
||
exporter.export_rules(sor, mod, CTX))
|
||
check('内容变更走 update', res3['exported'] == 1 and len(sor.updates) == 1,
|
||
(res3['exported'], len(sor.updates)))
|
||
check('内容变更不新增行', len(sor.tables['script']['rows']) == len(rules))
|
||
|
||
# 读回一致
|
||
back = run(
|
||
exporter.load_exported_rules(sor, CTX))
|
||
check('load_exported_rules 读回条数一致', len(back) == len(rules), len(back))
|
||
|
||
# tenant 缺失 fail-closed
|
||
try:
|
||
run(
|
||
exporter.export_rules(sor, rules, {'blueprint_version': 'v3'}))
|
||
check('导出缺 tenant 抛错', False)
|
||
except CompileError as e:
|
||
check('导出缺 tenant fail-closed', e.code == 'E_TENANT_REQUIRED', e.code)
|
||
|
||
# dry_run 不落库
|
||
sor_d = make_sor()
|
||
res_d = run(
|
||
exporter.export_rules(sor_d, rules, dict(CTX, dry_run=True)))
|
||
check('dry_run 不落库', len(sor_d.tables['script']['rows']) == 0
|
||
and res_d['skipped'] == len(rules))
|
||
|
||
# 所有 SQL 都带 tenant_id
|
||
check('所有 sqlExe 参数含 tenant_id 或为元数据探测',
|
||
all(('tenant_id' in (ns or {})) or 'information_schema' in sql.lower()
|
||
or sql.strip().lower().startswith('select 1 from')
|
||
or sql.strip().lower().startswith('select * from')
|
||
for sql, ns in sor.sqls), [s for s, n in sor.sqls if 'tenant_id' not in (n or {})][:3])
|
||
|
||
|
||
def t_gd():
|
||
print('\n[T8] game_definition 写入 + 版本回写 + export_all 一站式')
|
||
rules, _ = build_rules(SNAPSHOT, CTX)
|
||
base_gd = {'schema_version': '1.0', 'meta': {'title': 'Demo'}, 'scenes': [],
|
||
'entities': [], 'variables': [], 'ui': [], 'audio': [],
|
||
'scoring': {}, 'flow': {}, 'i18n': {}}
|
||
gd, gd_text, gd_hash = exporter.build_game_definition(rules, CTX, base_gd)
|
||
check('GD 含 rules 顶层键', 'rules' in gd)
|
||
check('GD 保留 M3a 其它顶层键', 'scenes' in gd and 'meta' in gd)
|
||
check('rules.script_type == 1', gd['rules']['script_type'] == 1)
|
||
check('rules.rule_count 正确', gd['rules']['rule_count'] == len(rules))
|
||
check('rules.rules_hash 与 rules_hash 一致', gd['rules']['rules_hash'] == rules_hash(rules))
|
||
check('GD 文本确定性（重算一致）',
|
||
exporter.build_game_definition(json.loads(canonical_rules_json(rules)), CTX, base_gd)[2] == gd_hash)
|
||
check('GD 文本零空白', ', ' not in gd_text and '\n' not in gd_text)
|
||
|
||
sor = make_sor()
|
||
r1 = run(
|
||
exporter.write_game_definition(sor, gd, gd_text, gd_hash, CTX))
|
||
check('首次写入 GD written=True', r1['written'] is True, r1)
|
||
check('GD 表 1 行', len(sor.tables['pbl_game_definition']['rows']) == 1)
|
||
r2 = run(
|
||
exporter.write_game_definition(sor, gd, gd_text, gd_hash, CTX))
|
||
check('同 hash 重复写入 unchanged=True（幂等）',
|
||
r2.get('unchanged') is True and r2['written'] is False, r2)
|
||
check('GD 表仍 1 行', len(sor.tables['pbl_game_definition']['rows']) == 1)
|
||
gd2 = json.loads(gd_text)
|
||
gd2['rules']['rules'][0]['priority'] = 1
|
||
t2 = canonical_dumps(gd2)
|
||
r3 = run(
|
||
exporter.write_game_definition(sor, gd2, t2, sha256_text(t2), CTX))
|
||
check('hash 变化 → 覆盖更新不新增行',
|
||
r3['written'] is True and len(sor.tables['pbl_game_definition']['rows']) == 1)
|
||
|
||
# 版本回写
|
||
exp_res = run(exporter.export_rules(sor, rules, CTX))
|
||
vr = run(
|
||
exporter.write_compiler_version_rules(sor, 'ver_001', rules, CTX, exp_res))
|
||
check('版本回写 written=True', vr['written'] is True, vr)
|
||
vrow = sor.tables['pbl_compiler_version']['rows'][0]
|
||
check('版本行 rules_hash 正确', vrow['rules_hash'] == rules_hash(rules))
|
||
check('版本行 rules_count 正确', vrow['rules_count'] == len(rules))
|
||
check('版本行 rules_json 可解析且确定性',
|
||
json.loads(vrow['rules_json']) is not None
|
||
and vrow['rules_json'] == canonical_rules_json(rules))
|
||
man = json.loads(vrow['rules_manifest'])
|
||
check('版本行 manifest 含 script_ids', len(man['extra']['script_ids']) == len(rules))
|
||
|
||
# export_all 一站式（含 error → 拒绝落库）
|
||
sor_a = make_sor()
|
||
out = run(
|
||
exporter.export_all(sor_a, SNAPSHOT, dict(CTX, strict=False),
|
||
base_gd=base_gd, version_id='ver_002'))
|
||
check('export_all status=ok', out['status'] == 'ok', out.get('issues'))
|
||
check('export_all rule_count 正确', out['rule_count'] == len(rules))
|
||
check('export_all 落库脚本', len(sor_a.tables['script']['rows']) == len(rules))
|
||
check('export_all 写 GD', len(sor_a.tables['pbl_game_definition']['rows']) == 1)
|
||
check('export_all 回写版本', len(sor_a.tables['pbl_compiler_version']['rows']) == 1)
|
||
|
||
bad_snap = {'rules': [{'id': 'b1', 'responses': [{'action': 'x'}]}]} # 缺 event
|
||
sor_b = make_sor()
|
||
try:
|
||
run(
|
||
exporter.export_all(sor_b, bad_snap, CTX, base_gd=base_gd))
|
||
check('export_all 有 error 时拒绝落库', False)
|
||
except CompileError as e:
|
||
check('export_all fail-closed(E_RULE_INVALID)', e.code == 'E_RULE_INVALID', e.code)
|
||
check('失败时脚本表为空（无半成品）', len(sor_b.tables['script']['rows']) == 0)
|
||
check('失败时 GD 表为空', len(sor_b.tables['pbl_game_definition']['rows']) == 0)
|
||
|
||
# rollback
|
||
rb = run(
|
||
exporter.rollback_export(sor_a, CTX))
|
||
check('rollback 清空规则脚本', len(sor_a.tables['script']['rows']) == 0, rb)
|
||
|
||
|
||
def t_ecr_enum():
|
||
print('\n[T9] ECR 源枚举去重 + 路径可追溯')
|
||
snap = {'rules': [{'id': 'r1', 'event': {'type': 'click'}, 'responses': [{'action': 'a'}]}],
|
||
'logic': [{'id': 'r1', 'event': {'type': 'click'}, 'responses': [{'action': 'a'}]}]}
|
||
srcs = iter_ecr_sources(snap)
|
||
check('同内容跨键重复只保留一处', len(srcs) == 1, [p for p, _ in srcs])
|
||
check('source_path 可追溯', srcs[0][0].startswith('snapshot.'), srcs[0][0])
|
||
rules, issues = build_rules(snap, CTX)
|
||
check('去重后无 E_RULE_ID_DUP', not any(i['code'] == 'E_RULE_ID_DUP' for i in issues), issues)
|
||
# 真重复 id 但内容不同 → 报 dup 并派生新 id
|
||
snap2 = {'rules': [
|
||
{'id': 'dup', 'event': {'type': 'click'}, 'responses': [{'action': 'a'}]},
|
||
{'id': 'dup', 'event': {'type': 'enter'}, 'responses': [{'action': 'b'}]}]}
|
||
r2, i2 = build_rules(snap2, CTX)
|
||
check('内容不同的重复 id → E_RULE_ID_DUP', any(i['code'] == 'E_RULE_ID_DUP' for i in i2))
|
||
check('重复 id 派生唯一 rule_id', len({r['rule_id'] for r in r2}) == 2)
|
||
|
||
|
||
def main():
|
||
print('=' * 72)
|
||
print('M3b 自测：script_engine 规则映射与导出（第12章 ECR → script_type=1）')
|
||
print('=' * 72)
|
||
for fn in (t_determinism, t_mapping, t_shapes, t_failclosed, t_manifest_diff,
|
||
t_strip, t_export, t_gd, t_ecr_enum):
|
||
fn()
|
||
print('\n' + '=' * 72)
|
||
print('PASS: %d FAIL: %d' % (len(PASS), len(FAIL)))
|
||
if FAIL:
|
||
print('失败项: ' + ', '.join(FAIL))
|
||
print('=' * 72)
|
||
return 1 if FAIL else 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|