bb1dcc1a58
- 新增 example_llm_call(97)、example_run/example_run_receipt/example_quality_result(98) - 加固：全对象幂等(IF NOT EXISTS/OR REPLACE)、append-only 表加 TRUNCATE 防护、 运行终态闭合约束、token/成本非负与 model_match 一致性 CHECK（已 DROP 空表重 apply 验证） - db.py：新增 execparams 参数化写（%s 服务端绑定+语句白名单+stdin 大对象）； query 改只读连接(拒 INSERT...RETURNING)；tables 用 sql.Identifier 安全拼接；connect() 统一入口 - 表映射登记 97/98、96 标不启用（仅本轮改动，未含他处未提交内容）
46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
#!/usr/bin/env python3
|
||
"""db execparams 参数装载逻辑离线自测（不连库）。
|
||
|
||
跑法（仓库根目录）：.venv/bin/python .claude/skills/db/scripts/test_db_params.py
|
||
"""
|
||
import io
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
import click # noqa: E402
|
||
|
||
import db # noqa: E402
|
||
|
||
|
||
def expect_exc(fn, msg):
|
||
try:
|
||
fn()
|
||
except click.ClickException:
|
||
return
|
||
raise AssertionError(f"应抛 ClickException：{msg}")
|
||
|
||
|
||
def main():
|
||
# --param 按序
|
||
assert db._read_params(False, ("a", "b")) == ["a", "b"]
|
||
assert db._read_params(False, ()) == []
|
||
# --stdin 合法 JSON 数组（含 null/数字）
|
||
assert db._read_params(True, (), stream=io.StringIO('["x", 1, null]')) == ["x", 1, None]
|
||
# 大对象（含中文引号/换行）经 stdin 不破坏
|
||
big = '中文"引号"\n换行\\反斜杠' * 5000
|
||
assert db._read_params(True, (), stream=io.StringIO(json.dumps([big])))[0] == big
|
||
# --stdin 与 --param 互斥
|
||
expect_exc(lambda: db._read_params(True, ("a",)), "--stdin 与 --param 互斥")
|
||
# 非法 JSON
|
||
expect_exc(lambda: db._read_params(True, (), stream=io.StringIO("not json")), "非法 JSON")
|
||
# 非数组
|
||
expect_exc(lambda: db._read_params(True, (), stream=io.StringIO('{"a":1}')), "非数组")
|
||
print("PASS：db execparams 参数装载逻辑全部通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|