DKChan
5f51a916d4
- data_config: ConfigDB 表加载+跨表护栏零诊断/进度坐标 ceil-clamp 解析与警告去重/变异池门控/shop·meta 消费面/manifest↔目录同步/越域 fatal 退出码(独立进程 expect=1,deferred 触发保 quit 通道)/统一随机流(派生确定性/计数器账本/标签不参与派生/负种子归一化) - battle_settlement: DamageMath 四纯函数(池乘区/克制分量/掷点查档/吃掷点分量)+DamageController 伤害单据五配方+B/A 双结账模式+来源敌死作废+requires 前置门+主角倒下 halt+AC 合流三通道+池 clamp 与 refresh 政策 - contracts: 五配方/三目标/双政策词表封闭+掷点标签∈RNG 封闭词表(EC12)+单据九栏缺省+增益布尔受理口径(EC1/EC2) - 红线: 期望值全由表读值推导(零玩法数值字面量)/零自建 RNG(夹具种子仅求复现) - 轻闸: 冒烟 557/0 + ui_input_probe 十一断言全绿(基线零漂移)
199 lines
8.4 KiB
GDScript
199 lines
8.4 KiB
GDScript
# tests/unit/data_config/config_db_test.gd — 数据配置单元测试：表加载/进度坐标 ceil 解析/变异池门控/消费面访问器
|
||
# 运行：godot --headless --path game --script res://tests/unit/data_config/config_db_test.gd
|
||
# 协议（ADR-0004 + tests/README）：extends SceneTree + quit(code)；--script 模式无 autoload——显式实例化。
|
||
# 红线：期望值由表读值推导，不写死玩法数值（tests/README 协议要点）。
|
||
extends SceneTree
|
||
|
||
const ConfigDBScript := preload("res://core/config_db.gd")
|
||
|
||
var _failures := 0
|
||
var _checks := 0
|
||
|
||
|
||
func _init() -> void:
|
||
var db = ConfigDBScript.new()
|
||
var load_ok: bool = db.load_tables("res://data/tables", true)
|
||
_check("全部表加载 + 校验通过（含跨表护栏零 error 诊断）", load_ok and db.diagnostics_count("error") == 0)
|
||
_check("is_ready 通道随加载置位", db.is_ready)
|
||
if not load_ok:
|
||
for d in db.diagnostics():
|
||
if d.severity == "error":
|
||
push_error("[config_db_test] 加载 error: [%s] %s" % [d.category, d.message])
|
||
else:
|
||
_test_progress_resolution(db)
|
||
_test_variant_pool_gate(db)
|
||
_test_shop_accessors(db)
|
||
_test_meta_accessors(db)
|
||
_test_manifest_dir_sync(db)
|
||
if _failures > 0:
|
||
push_error("[config_db_test] %d/%d 项失败" % [_failures, _checks])
|
||
quit(1)
|
||
return
|
||
print("[config_db_test] all green (%d checks)" % _checks)
|
||
quit(0)
|
||
|
||
|
||
func _check(desc: String, ok: bool) -> void:
|
||
_checks += 1
|
||
if ok:
|
||
print(" [PASS] " + desc)
|
||
else:
|
||
_failures += 1
|
||
push_error(" [FAIL] " + desc)
|
||
|
||
|
||
# ---- 进度坐标 ceil 解析（R1：向上下取整保证偏难；越最大行 clamp 取最大值 + 警告去重）----
|
||
|
||
func _test_progress_resolution(db) -> void:
|
||
var bg_rows: Array = db.board_gen.rows
|
||
# 行键全集自表推导（禁写死关卡/深度数值）
|
||
var level_keys: Array = []
|
||
var depth_keys_by_level: Dictionary = {}
|
||
for row in bg_rows:
|
||
var lv := int(row.level_index)
|
||
if not level_keys.has(lv):
|
||
level_keys.append(lv)
|
||
if not depth_keys_by_level.has(lv):
|
||
depth_keys_by_level[lv] = []
|
||
var dv := int(row.node_depth)
|
||
if not depth_keys_by_level[lv].has(dv):
|
||
depth_keys_by_level[lv].append(dv)
|
||
level_keys.sort()
|
||
var min_l: int = level_keys[0]
|
||
var max_l: int = level_keys[-1]
|
||
var min_depths: Array = depth_keys_by_level[min_l]
|
||
min_depths.sort()
|
||
|
||
# 精确命中：请求 = 表内现存 (L,D) → 原行返回（字段逐位相等）
|
||
var probe_row: Dictionary = bg_rows[0]
|
||
var got: Dictionary = db.resolve_board_gen(int(probe_row.level_index), int(probe_row.node_depth))
|
||
_check("进度解析精确命中返回原行（board_gen 首行）",
|
||
int(got.get("level_index", -99)) == int(probe_row.level_index)
|
||
and int(got.get("node_depth", -99)) == int(probe_row.node_depth)
|
||
and String(got.get("board_type", "")) == String(probe_row.get("board_type", "")))
|
||
|
||
# enemy 行集同通道：同 (L,D) 精确行存在（关卡节点地图 前置——battle 节点双表精确行契约）
|
||
var en_got: Dictionary = db.resolve_enemy_rowset(int(probe_row.level_index), int(probe_row.node_depth))
|
||
_check("enemy 行集同坐标解析非空（双表精确行契约）", not en_got.is_empty())
|
||
|
||
# 深度 clamp：请求深度超最大行 → 取该关最大深度行 + 警告诊断一次（去重）
|
||
db.reset_diagnostics()
|
||
var deep: Dictionary = db.resolve_board_gen(min_l, int(min_depths[-1]) + 1)
|
||
_check("深度越最大行 clamp 取该关最大深度行",
|
||
int(deep.get("level_index", -99)) == min_l and int(deep.get("node_depth", -99)) == int(min_depths[-1]))
|
||
_check("clamp 解析发警告诊断（resolve_depth_clamp）", db.diagnostics_count("warning") >= 1)
|
||
var found_clamp := false
|
||
for d in db.diagnostics():
|
||
if d.category == "resolve_depth_clamp":
|
||
found_clamp = true
|
||
_check("警告类别 = resolve_depth_clamp", found_clamp)
|
||
var warn_before: int = db.diagnostics_count("warning")
|
||
db.resolve_board_gen(min_l, int(min_depths[-1]) + 1)
|
||
_check("同键重复请求警告去重（四元组注册表）", db.diagnostics_count("warning") == warn_before)
|
||
|
||
# 关卡 clamp：请求关卡超最大行 → 取最大关卡行
|
||
db.reset_diagnostics()
|
||
var high: Dictionary = db.resolve_board_gen(max_l + 1, 0)
|
||
_check("关卡越最大行 clamp 取最大关卡行", int(high.get("level_index", -99)) == max_l)
|
||
|
||
|
||
# ---- 变异池门控（unlock_level 闸 + ∅ board_types 通配）----
|
||
|
||
func _test_variant_pool_gate(db) -> void:
|
||
var rows: Array = db.variation.rows
|
||
var gated_row := {}
|
||
var wildcard_row := {}
|
||
for row in rows:
|
||
var tags: Array = row.get("board_types", [])
|
||
if gated_row.is_empty() and not tags.is_empty():
|
||
gated_row = row
|
||
if wildcard_row.is_empty() and tags.is_empty():
|
||
wildcard_row = row
|
||
if not gated_row.is_empty():
|
||
var u: int = int(gated_row.unlock_level)
|
||
var tag: String = String(gated_row.board_types[0])
|
||
var before: Array = db.variant_pool(u - 1, tag)
|
||
var after: Array = db.variant_pool(u, tag)
|
||
_check("变异池 unlock 前不含未解锁行（level=unlock−1）", not _rows_contain(before, gated_row))
|
||
_check("变异池 unlock 当级含该行", _rows_contain(after, gated_row))
|
||
if not wildcard_row.is_empty():
|
||
var u2: int = int(wildcard_row.unlock_level)
|
||
var any_pool: Array = db.variant_pool(u2, "nonexistent_board_type_tag")
|
||
_check("∅ board_types 行通配（任一 board_type 命中）", _rows_contain(any_pool, wildcard_row))
|
||
|
||
|
||
func _rows_contain(pool: Array, row: Dictionary) -> bool:
|
||
for r in pool:
|
||
if r == row:
|
||
return true
|
||
return false
|
||
|
||
|
||
# ---- ShopTable 消费面（行集解析单点——runner/编排禁自扫表的契约消费侧验证）----
|
||
|
||
func _test_shop_accessors(db) -> void:
|
||
var products: Array = db.shop_products()
|
||
if products.is_empty():
|
||
_check("ShopTable products 行族非空（表已落籍前提）", false)
|
||
return
|
||
_check("ShopTable products 行族非空", true)
|
||
var first: Dictionary = products[0]
|
||
var got: Dictionary = db.shop_product_row(String(first.product_id))
|
||
_check("shop_product_row 按 product_id 精确命中首行", got == first)
|
||
_check("shop_product_row 未知 id 返回空字典（容错通道）", db.shop_product_row("no_such_product_id__").is_empty())
|
||
var tier_rows: Array = db._shop_rows("gp_drops")
|
||
if not tier_rows.is_empty():
|
||
var tier0: String = String(tier_rows[0].tier)
|
||
_check("shop_gp_drop_row 按 tier 命中", not db.shop_gp_drop_row(tier0).is_empty())
|
||
var bonus_rows: Array = db._shop_rows("level_gp_bonus")
|
||
if not bonus_rows.is_empty():
|
||
var blv: int = int(bonus_rows[0].level_index)
|
||
_check("shop_level_gp_bonus 按关卡序号命中行值", db.shop_level_gp_bonus(blv) == int(bonus_rows[0].bonus))
|
||
_check("shop_level_gp_bonus 未登记关卡返回 0（兜底通道）", db.shop_level_gp_bonus(-1) == 0)
|
||
var price_rows: Array = db._shop_rows("price_level_bonus")
|
||
if not price_rows.is_empty():
|
||
var plv: int = int(price_rows[0].level_index)
|
||
_check("shop_price_level_bonus 按关卡序号命中行值", db.shop_price_level_bonus(plv) == int(price_rows[0].bonus))
|
||
|
||
|
||
# ---- MetaTable 消费面 ----
|
||
|
||
func _test_meta_accessors(db) -> void:
|
||
var forge_rows: Array = db._meta_rows("forge_products")
|
||
if forge_rows.is_empty():
|
||
_check("MetaTable forge_products 行族非空", false)
|
||
else:
|
||
var attr0: String = String(forge_rows[0].attr)
|
||
_check("meta_forge_row 按 attr 命中首行", db.meta_forge_row(attr0) == forge_rows[0])
|
||
_check("meta_forge_row 未知 attr 返回空字典", db.meta_forge_row("no_such_attr__").is_empty())
|
||
var key_rows: Array = db._meta_rows("boss_keystone")
|
||
if not key_rows.is_empty():
|
||
var klv: int = int(key_rows[0].level_index)
|
||
_check("meta_boss_keystone_row 按关卡序号命中", db.meta_boss_keystone_row(klv) == key_rows[0])
|
||
var conv_rows: Array = db._meta_rows("conversion")
|
||
if not conv_rows.is_empty():
|
||
var cattr: String = String(conv_rows[0].attr)
|
||
_check("meta_conversion_row 按 attr 命中", db.meta_conversion_row(cattr) == conv_rows[0])
|
||
|
||
|
||
# ---- TABLE_MANIFEST ↔ data/tables 目录同步（各系统→D2 单点契约的登记面）----
|
||
|
||
func _test_manifest_dir_sync(db) -> void:
|
||
var manifest: Dictionary = db.TABLE_MANIFEST
|
||
var dir := DirAccess.open("res://data/tables")
|
||
var disk_files: Array = []
|
||
if dir != null:
|
||
dir.list_dir_begin()
|
||
var f := dir.get_next()
|
||
while f != "":
|
||
if not dir.current_is_dir() and f.ends_with(".tres"):
|
||
disk_files.append(f)
|
||
f = dir.get_next()
|
||
dir.list_dir_end()
|
||
var synced := disk_files.size() == manifest.size()
|
||
if synced:
|
||
for key in manifest.keys():
|
||
if not disk_files.has(String(manifest[key])):
|
||
synced = false
|
||
_check("TABLE_MANIFEST 与 data/tables 目录逐文件同步（无缺登/无悬登）", synced)
|