255 lines
11 KiB
Python
255 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""play-loop 契约校验器（PlaySpec C5 / VerdictFeedback C6）。
|
||
|
||
零依赖：只用 Python 标准库（json/re/sys/pathlib）。之所以不引 jsonschema/ajv——本仓
|
||
没装 jsonschema、也没 vendor ajv，而 W-ARCH① 红线③要求零新依赖；且两份契约的消费方
|
||
（cheap-worker gate_judge.py、tier2 run.py）本就是 Python，落一份 stdlib 校验器与消费侧同栈。
|
||
本文件自带一个 JSON-Schema Draft-2020-12 子集校验器，只实现两份 schema 真正用到的关键字：
|
||
type / required / properties / patternProperties / additionalProperties /
|
||
enum / const / pattern / minimum / maximum / minItems / maxItems / minLength / maxLength /
|
||
items / anyOf / oneOf / allOf / if-then-else / $ref(仅 #/$defs/<name> 本地引用)。
|
||
关键字集是刻意收窄的：schema 由本任务作者，只用上述关键字，正负样本套件即本校验器的回归证明——
|
||
若某关键字实现错了，要么正样本会被误拦、要么负样本会漏过，跑套件即暴露。
|
||
|
||
用法：
|
||
1) 单点校验：python3 validate.py <schema.json> <sample.json> [<sample2.json> ...]
|
||
逐样本打印 PASS/FAIL 与错误路径；全部合规 exit 0，否则 exit 1。
|
||
2) 套件校验：python3 validate.py --suite [samples 根目录，默认脚本旁 samples/]
|
||
走 samples/<contract>/valid（应全 PASS）与 samples/<contract>/invalid（应全 FAIL），
|
||
<contract> 目录名对应同名 schema（play-spec → play-spec.schema.json）。
|
||
打印每契约『正样本 N 过 / 负样本 M 拦』与总计；期望全部满足 exit 0，否则 exit 1。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
_HERE = Path(__file__).resolve().parent
|
||
|
||
|
||
# ── JSON 类型判定（区分 integer/number/boolean：Python 里 bool 是 int 子类，须显式排除）──
|
||
def _json_type_ok(value, t: str) -> bool:
|
||
if t == "object":
|
||
return isinstance(value, dict)
|
||
if t == "array":
|
||
return isinstance(value, list)
|
||
if t == "string":
|
||
return isinstance(value, str)
|
||
if t == "boolean":
|
||
return isinstance(value, bool)
|
||
if t == "integer":
|
||
return isinstance(value, int) and not isinstance(value, bool)
|
||
if t == "number":
|
||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||
if t == "null":
|
||
return value is None
|
||
return False
|
||
|
||
|
||
def _resolve_ref(ref: str, root: dict):
|
||
"""解析本地 $ref（仅支持 #/$defs/<name> 形态；其余抛错，防静默漏校验）。"""
|
||
if not ref.startswith("#/"):
|
||
raise ValueError(f"仅支持本地 #/$defs 引用，收到：{ref}")
|
||
node = root
|
||
for part in ref[2:].split("/"):
|
||
# JSON Pointer 转义还原（~1→/，~0→~），本仓 key 无特殊字符，兜个底而已。
|
||
part = part.replace("~1", "/").replace("~0", "~")
|
||
if not isinstance(node, dict) or part not in node:
|
||
raise ValueError(f"$ref 指向不存在的节点：{ref}")
|
||
node = node[part]
|
||
return node
|
||
|
||
|
||
def validate(schema, instance, root: dict, path: str = "#") -> list:
|
||
"""按 schema 校验 instance，返回错误串列表（空=合规）。schema 为 dict 或 bool。"""
|
||
# 布尔 schema：true=恒过，false=恒否（本仓未用，兜底）。
|
||
if isinstance(schema, bool):
|
||
return [] if schema else [f"{path}: 布尔 schema false，恒不合规"]
|
||
if not isinstance(schema, dict):
|
||
return [f"{path}: schema 非法（既非对象也非布尔）"]
|
||
|
||
errors: list = []
|
||
|
||
# $ref：解析后就地校验（本仓 $ref 节点自带完整约束，直接接管本层）。
|
||
if "$ref" in schema:
|
||
target = _resolve_ref(schema["$ref"], root)
|
||
return validate(target, instance, root, path)
|
||
|
||
# type
|
||
if "type" in schema:
|
||
types = schema["type"]
|
||
types = types if isinstance(types, list) else [types]
|
||
if not any(_json_type_ok(instance, t) for t in types):
|
||
errors.append(f"{path}: 类型应为 {types}，实际 {type(instance).__name__}")
|
||
|
||
# const / enum
|
||
if "const" in schema and instance != schema["const"]:
|
||
errors.append(f"{path}: 应恒等于 {schema['const']!r}，实际 {instance!r}")
|
||
if "enum" in schema and instance not in schema["enum"]:
|
||
errors.append(f"{path}: 应属枚举 {schema['enum']}，实际 {instance!r}")
|
||
|
||
# 字符串约束
|
||
if isinstance(instance, str):
|
||
if "minLength" in schema and len(instance) < schema["minLength"]:
|
||
errors.append(f"{path}: 字符串长度应 ≥ {schema['minLength']}，实际 {len(instance)}")
|
||
if "maxLength" in schema and len(instance) > schema["maxLength"]:
|
||
errors.append(f"{path}: 字符串长度应 ≤ {schema['maxLength']}，实际 {len(instance)}")
|
||
if "pattern" in schema and re.search(schema["pattern"], instance) is None:
|
||
errors.append(f"{path}: 应匹配正则 /{schema['pattern']}/，实际 {instance!r}")
|
||
|
||
# 数值约束（排除 bool）
|
||
if isinstance(instance, (int, float)) and not isinstance(instance, bool):
|
||
if "minimum" in schema and instance < schema["minimum"]:
|
||
errors.append(f"{path}: 应 ≥ {schema['minimum']}，实际 {instance}")
|
||
if "maximum" in schema and instance > schema["maximum"]:
|
||
errors.append(f"{path}: 应 ≤ {schema['maximum']}，实际 {instance}")
|
||
|
||
# 数组约束
|
||
if isinstance(instance, list):
|
||
if "minItems" in schema and len(instance) < schema["minItems"]:
|
||
errors.append(f"{path}: 数组元素数应 ≥ {schema['minItems']}，实际 {len(instance)}")
|
||
if "maxItems" in schema and len(instance) > schema["maxItems"]:
|
||
errors.append(f"{path}: 数组元素数应 ≤ {schema['maxItems']}，实际 {len(instance)}")
|
||
if "items" in schema:
|
||
for i, item in enumerate(instance):
|
||
errors += validate(schema["items"], item, root, f"{path}/{i}")
|
||
|
||
# 对象约束
|
||
if isinstance(instance, dict):
|
||
for key in schema.get("required", []):
|
||
if key not in instance:
|
||
errors.append(f"{path}: 缺必填字段 '{key}'")
|
||
props = schema.get("properties", {})
|
||
pattern_props = schema.get("patternProperties", {})
|
||
add_props = schema.get("additionalProperties", True)
|
||
for key, val in instance.items():
|
||
handled = False
|
||
if key in props:
|
||
errors += validate(props[key], val, root, f"{path}/{key}")
|
||
handled = True
|
||
for pat, sub in pattern_props.items():
|
||
if re.search(pat, key) is not None:
|
||
errors += validate(sub, val, root, f"{path}/{key}")
|
||
handled = True
|
||
if not handled:
|
||
if add_props is False:
|
||
errors.append(f"{path}: 不允许的额外字段 '{key}'（additionalProperties=false）")
|
||
elif isinstance(add_props, dict):
|
||
errors += validate(add_props, val, root, f"{path}/{key}")
|
||
|
||
# 组合子：allOf / anyOf / oneOf
|
||
for i, sub in enumerate(schema.get("allOf", [])):
|
||
errors += validate(sub, instance, root, f"{path}/allOf[{i}]")
|
||
if "anyOf" in schema:
|
||
branch_errs = [validate(sub, instance, root, f"{path}/anyOf[{i}]")
|
||
for i, sub in enumerate(schema["anyOf"])]
|
||
if not any(len(e) == 0 for e in branch_errs):
|
||
errors.append(f"{path}: anyOf 无一分支匹配（各分支首错：" +
|
||
"；".join(e[0] for e in branch_errs if e) + "）")
|
||
if "oneOf" in schema:
|
||
matched = sum(1 for sub in schema["oneOf"]
|
||
if len(validate(sub, instance, root, path)) == 0)
|
||
if matched != 1:
|
||
errors.append(f"{path}: oneOf 应恰好 1 分支匹配，实际 {matched}")
|
||
|
||
# 条件子：if-then-else（if 本身不产错，命中则套 then、否则套 else）
|
||
if "if" in schema:
|
||
if len(validate(schema["if"], instance, root, path)) == 0:
|
||
if "then" in schema:
|
||
errors += validate(schema["then"], instance, root, f"{path}/then")
|
||
else:
|
||
if "else" in schema:
|
||
errors += validate(schema["else"], instance, root, f"{path}/else")
|
||
|
||
return errors
|
||
|
||
|
||
def _load(p: Path):
|
||
return json.loads(p.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _validate_file(schema: dict, sample_path: Path) -> list:
|
||
"""校验单个样本文件，返回错误列表（JSON 解析失败也算一条错误）。"""
|
||
try:
|
||
instance = _load(sample_path)
|
||
except Exception as e: # noqa: BLE001 —— 解析失败即样本不合格，如实报，不抛断链
|
||
return [f"#: JSON 解析失败：{e}"]
|
||
return validate(schema, instance, schema)
|
||
|
||
|
||
def _run_suite(samples_root: Path) -> int:
|
||
"""套件模式：正样本应全过、负样本应全拦。返回进程退出码（0=期望全满足）。"""
|
||
contracts = [
|
||
("play-spec", _HERE / "play-spec.schema.json"),
|
||
("verdict-feedback", _HERE / "verdict-feedback.schema.json"),
|
||
]
|
||
total_pass = total_block = 0
|
||
all_ok = True
|
||
for name, schema_path in contracts:
|
||
schema = _load(schema_path)
|
||
base = samples_root / name
|
||
valid_dir, invalid_dir = base / "valid", base / "invalid"
|
||
valids = sorted(valid_dir.glob("*.json")) if valid_dir.is_dir() else []
|
||
invalids = sorted(invalid_dir.glob("*.json")) if invalid_dir.is_dir() else []
|
||
print(f"\n══ {name}（{schema_path.name}）══")
|
||
|
||
n_pass = 0
|
||
for f in valids:
|
||
errs = _validate_file(schema, f)
|
||
if errs:
|
||
all_ok = False
|
||
print(f" ✗ 正样本本应过却被拦：{f.name}")
|
||
for e in errs[:4]:
|
||
print(f" {e}")
|
||
else:
|
||
n_pass += 1
|
||
print(f" ✓ 正样本过：{f.name}")
|
||
|
||
n_block = 0
|
||
for f in invalids:
|
||
errs = _validate_file(schema, f)
|
||
if errs:
|
||
n_block += 1
|
||
# 打印首条错误，证明是按预期规则拦下的（非偶然）。
|
||
print(f" ✓ 负样本拦：{f.name} —— {errs[0]}")
|
||
else:
|
||
all_ok = False
|
||
print(f" ✗ 负样本本应拦却放过：{f.name}")
|
||
|
||
print(f" 小计：正样本 {n_pass}/{len(valids)} 过，负样本 {n_block}/{len(invalids)} 拦")
|
||
total_pass += n_pass
|
||
total_block += n_block
|
||
|
||
print(f"\n══ 总计：正样本 {total_pass} 过 / 负样本 {total_block} 拦 —— "
|
||
f"{'全部符合预期' if all_ok else '存在未达预期项（见上 ✗）'}")
|
||
return 0 if all_ok else 1
|
||
|
||
|
||
def main(argv: list) -> int:
|
||
if len(argv) >= 2 and argv[1] == "--suite":
|
||
root = Path(argv[2]).resolve() if len(argv) >= 3 else (_HERE / "samples")
|
||
return _run_suite(root)
|
||
if len(argv) < 3:
|
||
print(__doc__)
|
||
return 2
|
||
schema = _load(Path(argv[1]).resolve())
|
||
exit_code = 0
|
||
for sample_arg in argv[2:]:
|
||
sample_path = Path(sample_arg).resolve()
|
||
errs = _validate_file(schema, sample_path)
|
||
if errs:
|
||
exit_code = 1
|
||
print(f"FAIL {sample_path.name}")
|
||
for e in errs:
|
||
print(f" {e}")
|
||
else:
|
||
print(f"PASS {sample_path.name}")
|
||
return exit_code
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv))
|