diff --git a/core/model-framework/__init__.py b/core/model-framework/__init__.py new file mode 100644 index 0000000..1310a26 --- /dev/null +++ b/core/model-framework/__init__.py @@ -0,0 +1,37 @@ +# -*- coding: utf-8 -*- +"""iAOP-Core · AI 模型框架（AI Model Framework）—— 配置驱动的可模板化模型内核。 + +对应 PRD 5.3「③ AI 模型框架」与 EPIC #5（平台化改造 RISK 项）： +固定主干网络 + 可配置超参；Model Recipe 插件注册；FeatureSpec 声明式特征； +所有可变量外置为 **JSON 超参包**，同一框架切换模板仅改此包（零改码）。 + +模块组成（按 EPIC #5 拆分的 ≤0.5d 子任务逐步落地）： +- hyperparam 超参包 JSON Schema 校验器（Issue #39）：加载 + 校验超参包， + 返回逐条校验报告，供配置台与训练/推理流水线复用。 +- （后续）recipe Model Recipe 插件接口与样例协议（Issue #34） +- （后续）feature FeatureSpec 声明式特征定义引擎（Issue #35） +- （后续）registry 模型模板注册 / 加载 / 版本机制（Issue #41） + +超参包 JSON 结构见 PRD 5.3「超参包 JSON 完整示例」与 ``config/`` 下样例。 + +测试：`python -m unittest discover -s tests -v`（在 core/model-framework 目录下执行）。 +""" +from __future__ import annotations + +from .hyperparam import ( + HyperparamPack, + ValidationIssue, + ValidationReport, + load_pack, + validate_pack, + validate_pack_file, +) + +__all__ = [ + "HyperparamPack", + "ValidationIssue", + "ValidationReport", + "load_pack", + "validate_pack", + "validate_pack_file", +] diff --git a/core/model-framework/config/quality_predict_ti.example.json b/core/model-framework/config/quality_predict_ti.example.json new file mode 100644 index 0000000..db16457 --- /dev/null +++ b/core/model-framework/config/quality_predict_ti.example.json @@ -0,0 +1,16 @@ +{ + "model_id": "quality_predict_ti", + "template": "iAOP-Template-Ti", + "algorithm": "xgboost", + "features": [ + {"name": "EMA_CLF_TEMP_5m", "spec": "EMA(CLF-01.TEMP, 5m)"}, + {"name": "RollingStd_CL2_10", "spec": "RollingStd(CLF-01.CL2, 10)"}, + {"name": "ROC_FURNACE_PRESS", "spec": "RateOfChange(CLF-01.PRES)"} + ], + "target": "Ti_purity", + "objective": "reg:squarederror", + "hyperparams": {"max_depth": 6, "eta": 0.1, "n_estimators": 300}, + "train_window": "180d", + "alarm_threshold": {"type": "zscore", "k": 3.0}, + "drift_check": {"method": "psi", "limit": 0.2} +} diff --git a/core/model-framework/hyperparam.py b/core/model-framework/hyperparam.py new file mode 100644 index 0000000..86a60a1 --- /dev/null +++ b/core/model-framework/hyperparam.py @@ -0,0 +1,339 @@ +# -*- coding: utf-8 -*- +"""超参包（Hyperparam Pack）JSON Schema 校验器。 + +对应 issue #39（父 EPIC #5「③ AI 模型框架 配置化重构」）与 PRD 5.3 +「超参包驱动」：所有可变量（输入特征清单、算法选型、超参、训练窗口、告警阈值、 +目标函数）外置为 JSON 超参包，同一框架切换模板仅改此包。 + +本模块对超参包做 **结构化校验**（不依赖 jsonschema 第三方库，零外部依赖， +便于边缘 / 离线环境运行），返回逐条问题的校验报告，便于配置台一次性展示全部问题、 +也便于训练/推理流水线在加载包时 fail-fast。 + +校验维度（对齐 PRD 5.3 超参包示例与 5.3「配置点」）： +1. 顶层结构：必填字段 model_id / template / algorithm / features / target； +2. model_id / template：非空字符串； +3. algorithm：必须在合法算法集合内（xgboost / lightgbm / dnn / lstm / gnn / + isolation_forest / zscore / linear / ridge）；未知算法报错并提示已知项 + （对齐 PRD「新增结构走插件注册」——校验期即暴露非法选型）； +4. features：非空列表；每项含 name（非空、包内唯一）与 spec（非空字符串， + FeatureSpec 文本，由 issue #35 引擎解释，此处只做存在性校验）； +5. target：非空字符串； +6. objective（可选）：若填写必须为已知目标函数（reg:squarederror / + binary:logistic / multi:softmax / regression / classification 等）； +7. hyperparams（可选）：若提供必须为对象（dict），且不含空键； +8. train_window（可选）：若填写必须形如 ``<正整数><单位>``（单位 d/w/h/m）， + 如 ``180d`` / ``4w``； +9. alarm_threshold（可选）：若提供必须为对象，且含 ``type`` + （zscore / quantile / absolute）与对应阈值键； +10. drift_check（可选）：若提供必须为对象，含 ``method``（psi / ks / chi2） + 与 ``limit``（0~1 之间）。 + +注：返回报告而非抛异常，便于配置台聚合展示；``load_pack`` 在校验失败时抛 +``ValueError`` 供流水线 fail-fast。 +""" +from __future__ import annotations + +import json +import os +import re +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + +# --------------------------------------------------------------------------- +# 合法取值集合（对齐 PRD 5.3「四类模型模板」+ 默认主干网络 + Recipe 插件扩展点） +# --------------------------------------------------------------------------- +#: 已注册算法（默认主干 + Recipe 可选结构）。新增结构应走插件注册（issue #34/#41）， +#: 此处枚举的是内核自带项；校验期遇到未知 algorithm 会报错并列出已知项，避免 +#: 静默落到错误分支。 +KNOWN_ALGORITHMS: Tuple[str, ...] = ( + # 监督回归 / 分类（质量预测） + "xgboost", + "lightgbm", + "linear", + "ridge", + # 神经网络主干（默认 DNN；Recipe 可选 LSTM/GNN） + "dnn", + "lstm", + "gnn", + # 无监督异常 / 杂质预警 + "isolation_forest", + "zscore", +) + +#: 已知目标函数（xgboost 风格 + 通用风格）。 +KNOWN_OBJECTIVES: Tuple[str, ...] = ( + "reg:squarederror", + "reg:squaredlogerror", + "binary:logistic", + "multi:softmax", + "multi:softprob", + "regression", + "classification", +) + +#: 训练窗口合法时间单位。 +_TRAIN_WINDOW_RE = re.compile(r"^\d+(\.\d+)?[dwhm]$") +#: FeatureSpec 仅校验非空文本（具体语法由 issue #35 引擎解释）。 +_FEATURE_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +@dataclass +class ValidationIssue: + """单条校验问题。""" + + code: str # 错误码：missing_field / bad_type / unknown_algorithm / ... + path: str # JSON 路径，如 ``features[1].name`` / ``algorithm`` + message: str # 人类可读描述 + + +@dataclass +class ValidationReport: + """校验报告。""" + + issues: List[ValidationIssue] = field(default_factory=list) + + @property + def ok(self) -> bool: + return not self.issues + + def summary(self) -> str: + if self.ok: + return "超参包校验通过" + by_code: Dict[str, int] = {} + for it in self.issues: + by_code[it.code] = by_code.get(it.code, 0) + 1 + parts = [f"{code}×{n}" for code, n in sorted(by_code.items())] + return f"超参包校验未通过（{len(self.issues)} 个问题：" + "，".join(parts) + "）" + + +@dataclass +class HyperparamPack: + """解析后的超参包（校验通过的视图）。 + + 字段对齐 PRD 5.3 超参包示例；可选字段缺失时为 ``None``。 + """ + + model_id: str + template: str + algorithm: str + features: List[Dict[str, str]] + target: str + objective: Optional[str] = None + hyperparams: Optional[Dict[str, Any]] = None + train_window: Optional[str] = None + alarm_threshold: Optional[Dict[str, Any]] = None + drift_check: Optional[Dict[str, Any]] = None + raw: Dict[str, Any] = field(default_factory=dict) + + @property + def feature_names(self) -> List[str]: + return [f["name"] for f in self.features] + + +# --------------------------------------------------------------------------- +# 校验主逻辑 +# --------------------------------------------------------------------------- +def validate_pack(pack: Dict[str, Any]) -> ValidationReport: + """校验一个已解析为 dict 的超参包，返回报告（不抛异常）。""" + report = ValidationReport() + + def add(code: str, path: str, message: str) -> None: + report.issues.append(ValidationIssue(code, path, message)) + + # 0. 顶层必须是对象 + if not isinstance(pack, dict): + add("bad_type", "$", f"超参包根节点必须为 JSON 对象，实际为 {type(pack).__name__}") + return report + + # 1. 必填字段 + required: Tuple[str, ...] = ("model_id", "template", "algorithm", "features", "target") + for key in required: + if key not in pack: + add("missing_field", key, f"缺少必填字段：{key}") + + # 2. 标量字段类型与取值 + model_id = pack.get("model_id") + if "model_id" in pack: + if not isinstance(model_id, str) or not model_id.strip(): + add("bad_type", "model_id", "model_id 必须为非空字符串") + elif not _FEATURE_NAME_RE.match(model_id): + add("bad_format", "model_id", "model_id 含非法字符（仅字母/数字/下划线，首字符非数字）") + + template = pack.get("template") + if "template" in pack and (not isinstance(template, str) or not template.strip()): + add("bad_type", "template", "template 必须为非空字符串") + + algorithm = pack.get("algorithm") + if "algorithm" in pack: + if not isinstance(algorithm, str) or not algorithm.strip(): + add("bad_type", "algorithm", "algorithm 必须为非空字符串") + elif algorithm not in KNOWN_ALGORITHMS: + known = "、".join(KNOWN_ALGORITHMS) + add( + "unknown_algorithm", + "algorithm", + f"未知算法 '{algorithm}'；已知项：{known}（新增结构请走 Model Recipe 插件注册）", + ) + + target = pack.get("target") + if "target" in pack and (not isinstance(target, str) or not target.strip()): + add("bad_type", "target", "target 必须为非空字符串") + + # 3. objective（可选） + objective = pack.get("objective") + if objective is not None: + if not isinstance(objective, str) or not objective.strip(): + add("bad_type", "objective", "objective 若填写必须为非空字符串") + elif objective not in KNOWN_OBJECTIVES: + known = "、".join(KNOWN_OBJECTIVES) + add("unknown_objective", "objective", f"未知目标函数 '{objective}'；已知项：{known}") + + # 4. features（必填，非空列表） + features = pack.get("features") + if features is None: + # missing_field 已在步骤 1 记录 + pass + elif not isinstance(features, list): + add("bad_type", "features", f"features 必须为数组，实际为 {type(features).__name__}") + elif len(features) == 0: + add("empty_features", "features", "features 不能为空（模型至少需要一个输入特征）") + else: + seen_names: Dict[str, int] = {} + for i, feat in enumerate(features): + fpath = f"features[{i}]" + if not isinstance(feat, dict): + add("bad_type", fpath, f"特征项必须为对象，实际为 {type(feat).__name__}") + continue + name = feat.get("name") + spec = feat.get("spec") + if not isinstance(name, str) or not name.strip(): + add("missing_field", f"{fpath}.name", "特征缺少 name 或为空") + else: + if not _FEATURE_NAME_RE.match(name): + add("bad_format", f"{fpath}.name", f"特征名 '{name}' 含非法字符") + if name in seen_names: + add( + "dup_feature", + f"{fpath}.name", + f"特征名 '{name}' 重复（首次出现在 features[{seen_names[name]}]）", + ) + else: + seen_names[name] = i + if not isinstance(spec, str) or not spec.strip(): + add("missing_field", f"{fpath}.spec", f"特征 '{name}' 缺少 spec（FeatureSpec 声明）或为空") + + # 5. hyperparams（可选，对象） + hyperparams = pack.get("hyperparams") + if hyperparams is not None: + if not isinstance(hyperparams, dict): + add("bad_type", "hyperparams", f"hyperparams 必须为对象，实际为 {type(hyperparams).__name__}") + else: + for k, v in hyperparams.items(): + if not isinstance(k, str) or not k.strip(): + add("bad_format", "hyperparams", "hyperparams 含空键") + + # 6. train_window（可选，<数><单位>） + train_window = pack.get("train_window") + if train_window is not None: + if not isinstance(train_window, str) or not _TRAIN_WINDOW_RE.match(train_window): + add( + "bad_format", + "train_window", + "train_window 必须形如 '<正数><单位>'（单位 d/w/h/m），如 '180d'、'4w'", + ) + + # 7. alarm_threshold（可选，对象，含 type） + alarm = pack.get("alarm_threshold") + if alarm is not None: + if not isinstance(alarm, dict): + add("bad_type", "alarm_threshold", f"alarm_threshold 必须为对象，实际为 {type(alarm).__name__}") + else: + atype = alarm.get("type") + known_alarm_types = ("zscore", "quantile", "absolute") + if not isinstance(atype, str) or atype not in known_alarm_types: + add( + "unknown_alarm_type", + "alarm_threshold.type", + f"alarm_threshold.type 必须为 {known_alarm_types} 之一", + ) + if atype == "zscore" and "k" not in alarm: + add("missing_field", "alarm_threshold.k", "zscore 阈值缺少 k") + if atype == "quantile" and "q" not in alarm: + add("missing_field", "alarm_threshold.q", "quantile 阈值缺少 q") + if atype == "absolute" and "value" not in alarm: + add("missing_field", "alarm_threshold.value", "absolute 阈值缺少 value") + + # 8. drift_check（可选，对象，method + limit） + drift = pack.get("drift_check") + if drift is not None: + if not isinstance(drift, dict): + add("bad_type", "drift_check", f"drift_check 必须为对象，实际为 {type(drift).__name__}") + else: + method = drift.get("method") + known_methods = ("psi", "ks", "chi2") + if not isinstance(method, str) or method not in known_methods: + add( + "unknown_drift_method", + "drift_check.method", + f"drift_check.method 必须为 {known_methods} 之一", + ) + limit = drift.get("limit") + if limit is None: + add("missing_field", "drift_check.limit", "drift_check 缺少 limit") + elif not isinstance(limit, (int, float)) or isinstance(limit, bool): + add("bad_type", "drift_check.limit", "drift_check.limit 必须为数值") + elif not (0 < limit <= 1): + add("bad_range", "drift_check.limit", "drift_check.limit 必须在 (0, 1] 范围内") + + return report + + +def load_pack(pack: Dict[str, Any]) -> HyperparamPack: + """校验并把 dict 装配为 :class:`HyperparamPack`；校验失败抛 ``ValueError``。 + + 供训练 / 推理流水线在加载超参包时 fail-fast 使用。 + """ + report = validate_pack(pack) + if not report.ok: + raise ValueError(report.summary()) + + return HyperparamPack( + model_id=pack["model_id"], + template=pack["template"], + algorithm=pack["algorithm"], + features=list(pack["features"]), + target=pack["target"], + objective=pack.get("objective"), + hyperparams=pack.get("hyperparams"), + train_window=pack.get("train_window"), + alarm_threshold=pack.get("alarm_threshold"), + drift_check=pack.get("drift_check"), + raw=dict(pack), + ) + + +def validate_pack_file(path: str) -> ValidationReport: + """读取 JSON 文件并校验；文件 / JSON 解析错误也记入报告（不抛异常）。""" + report = ValidationReport() + + def add(code: str, msg: str) -> None: + report.issues.append(ValidationIssue(code, "$", msg)) + + if not os.path.exists(path): + add("file_not_found", f"超参包文件不存在：{path}") + return report + try: + with open(path, "r", encoding="utf-8") as fh: + text = fh.read() + except OSError as exc: + add("file_read_error", f"读取超参包失败：{exc}") + return report + try: + pack = json.loads(text) + except json.JSONDecodeError as exc: + add("json_parse_error", f"超参包不是合法 JSON：{exc}") + return report + + inner = validate_pack(pack) + report.issues.extend(inner.issues) + return report diff --git a/core/model-framework/tests/_bootstrap.py b/core/model-framework/tests/_bootstrap.py new file mode 100644 index 0000000..2fcad90 --- /dev/null +++ b/core/model-framework/tests/_bootstrap.py @@ -0,0 +1,16 @@ +# -*- coding: utf-8 -*- +"""测试引导：把 `core/model-framework` 以包名 `model_framework` 挂载到 sys.modules。 + +目录名 `model-framework` 含连字符，无法直接以包名 import；挂载后模块内相对导入 +（`from .hyperparam import ...`）在 unittest 发现机制下可正常解析。 +""" +import os +import sys +import types + +MF_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, MF_DIR) +if "model_framework" not in sys.modules: + pkg = types.ModuleType("model_framework") + pkg.__path__ = [MF_DIR] + sys.modules["model_framework"] = pkg diff --git a/core/model-framework/tests/test_hyperparam.py b/core/model-framework/tests/test_hyperparam.py new file mode 100644 index 0000000..94eb631 --- /dev/null +++ b/core/model-framework/tests/test_hyperparam.py @@ -0,0 +1,239 @@ +# -*- coding: utf-8 -*- +"""超参包 JSON Schema 校验器测试（issue #39）。 + +覆盖：合法包通过 / 必填缺失 / 未知算法 / 特征重复与缺 spec / 目标函数 / +train_window 格式 / alarm_threshold / drift_check / 文件加载与 JSON 解析错误。 +""" +import os +import tempfile +import unittest + +import _bootstrap # noqa: F401 挂载包名 + +from model_framework.hyperparam import ( + HyperparamPack, + load_pack, + validate_pack, + validate_pack_file, +) + + +def _good_pack(): + """PRD 5.3 示例的合法超参包。""" + return { + "model_id": "quality_predict_ti", + "template": "iAOP-Template-Ti", + "algorithm": "xgboost", + "features": [ + {"name": "EMA_CLF_TEMP_5m", "spec": "EMA(CLF-01.TEMP, 5m)"}, + {"name": "RollingStd_CL2_10", "spec": "RollingStd(CLF-01.CL2, 10)"}, + ], + "target": "Ti_purity", + "objective": "reg:squarederror", + "hyperparams": {"max_depth": 6, "eta": 0.1, "n_estimators": 300}, + "train_window": "180d", + "alarm_threshold": {"type": "zscore", "k": 3.0}, + "drift_check": {"method": "psi", "limit": 0.2}, + } + + +class ValidatePackTest(unittest.TestCase): + def test_good_pack_passes(self): + report = validate_pack(_good_pack()) + self.assertTrue(report.ok, report.summary()) + self.assertEqual(report.issues, []) + + def test_minimal_pack_passes(self): + report = validate_pack( + { + "model_id": "anomaly_resin", + "template": "iAOP-Template-Resin", + "algorithm": "isolation_forest", + "features": [{"name": "F1", "spec": "RateOfChange(炉压)"}], + "target": "anomaly_score", + } + ) + self.assertTrue(report.ok, report.summary()) + + def test_missing_required_field(self): + pack = _good_pack() + del pack["algorithm"] + report = validate_pack(pack) + self.assertFalse(report.ok) + codes = [i.code for i in report.issues] + self.assertIn("missing_field", codes) + self.assertTrue(any(i.path == "algorithm" for i in report.issues)) + + def test_missing_all_required(self): + report = validate_pack({}) + self.assertFalse(report.ok) + codes = [i.code for i in report.issues] + # 5 个必填字段全部缺失 + self.assertEqual(codes.count("missing_field"), 5) + + def test_unknown_algorithm_lists_known(self): + pack = _good_pack() + pack["algorithm"] = "magic_boost" + report = validate_pack(pack) + self.assertFalse(report.ok) + issue = next(i for i in report.issues if i.code == "unknown_algorithm") + self.assertIn("xgboost", issue.message) + self.assertIn("magic_boost", issue.message) + + def test_empty_features_rejected(self): + pack = _good_pack() + pack["features"] = [] + report = validate_pack(pack) + codes = [i.code for i in report.issues] + self.assertIn("empty_features", codes) + + def test_duplicate_feature_name(self): + pack = _good_pack() + pack["features"] = [ + {"name": "F1", "spec": "EMA(A, 5m)"}, + {"name": "F1", "spec": "EMA(B, 5m)"}, + ] + report = validate_pack(pack) + codes = [i.code for i in report.issues] + self.assertIn("dup_feature", codes) + + def test_feature_missing_spec(self): + pack = _good_pack() + pack["features"] = [{"name": "F1", "spec": ""}] + report = validate_pack(pack) + codes = [i.code for i in report.issues] + self.assertIn("missing_field", codes) + + def test_feature_bad_name_format(self): + pack = _good_pack() + pack["features"] = [{"name": "1bad", "spec": "EMA(A,5m)"}] + report = validate_pack(pack) + codes = [i.code for i in report.issues] + self.assertIn("bad_format", codes) + + def test_unknown_objective(self): + pack = _good_pack() + pack["objective"] = "reg:magic" + report = validate_pack(pack) + self.assertTrue(any(i.code == "unknown_objective" for i in report.issues)) + + def test_bad_train_window_format(self): + for bad in ["180", "180days", "abc", "", "1y"]: + pack = _good_pack() + pack["train_window"] = bad + report = validate_pack(pack) + self.assertTrue( + any(i.code == "bad_format" and i.path == "train_window" for i in report.issues), + f"应拒绝非法 train_window: {bad!r}", + ) + + def test_good_train_window_units(self): + for good in ["180d", "4w", "72h", "30m"]: + pack = _good_pack() + pack["train_window"] = good + report = validate_pack(pack) + self.assertTrue(report.ok, f"应接受合法 train_window: {good!r} -> {report.summary()}") + + def test_alarm_threshold_zscore_missing_k(self): + pack = _good_pack() + pack["alarm_threshold"] = {"type": "zscore"} + report = validate_pack(pack) + self.assertTrue(any(i.code == "missing_field" and i.path == "alarm_threshold.k" for i in report.issues)) + + def test_alarm_threshold_unknown_type(self): + pack = _good_pack() + pack["alarm_threshold"] = {"type": "voodoo"} + report = validate_pack(pack) + self.assertTrue(any(i.code == "unknown_alarm_type" for i in report.issues)) + + def test_drift_check_limit_range(self): + for bad_limit in [0, 1.5, -0.1]: + pack = _good_pack() + pack["drift_check"] = {"method": "psi", "limit": bad_limit} + report = validate_pack(pack) + self.assertTrue( + any(i.code == "bad_range" for i in report.issues), + f"应拒绝非法 drift_check.limit: {bad_limit}", + ) + + def test_drift_check_unknown_method(self): + pack = _good_pack() + pack["drift_check"] = {"method": "magic", "limit": 0.2} + report = validate_pack(pack) + self.assertTrue(any(i.code == "unknown_drift_method" for i in report.issues)) + + def test_root_not_object(self): + report = validate_pack([1, 2, 3]) # type: ignore[arg-type] + self.assertFalse(report.ok) + self.assertEqual(report.issues[0].code, "bad_type") + + def test_report_summary_ok(self): + self.assertEqual(validate_pack(_good_pack()).summary(), "超参包校验通过") + + def test_report_summary_aggregates(self): + pack = _good_pack() + del pack["algorithm"] + del pack["target"] + summary = validate_pack(pack).summary() + self.assertIn("2 个问题", summary) + self.assertIn("missing_field", summary) + + +class LoadPackTest(unittest.TestCase): + def test_load_good_returns_pack(self): + pack = load_pack(_good_pack()) + self.assertIsInstance(pack, HyperparamPack) + self.assertEqual(pack.model_id, "quality_predict_ti") + self.assertEqual(pack.algorithm, "xgboost") + self.assertEqual(pack.feature_names, ["EMA_CLF_TEMP_5m", "RollingStd_CL2_10"]) + self.assertEqual(pack.hyperparams["max_depth"], 6) + + def test_load_bad_raises(self): + pack = _good_pack() + del pack["features"] + with self.assertRaises(ValueError) as cm: + load_pack(pack) + self.assertIn("超参包校验未通过", str(cm.exception)) + + +class ValidatePackFileTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.mkdtemp() + self.path = os.path.join(self._tmp, "pack.json") + + def _write(self, text): + with open(self.path, "w", encoding="utf-8") as fh: + fh.write(text) + return self.path + + def test_valid_file_passes(self): + import json + + self._write(json.dumps(_good_pack())) + report = validate_pack_file(self.path) + self.assertTrue(report.ok, report.summary()) + + def test_missing_file(self): + report = validate_pack_file(os.path.join(self._tmp, "nope.json")) + self.assertFalse(report.ok) + self.assertEqual(report.issues[0].code, "file_not_found") + + def test_bad_json(self): + self._write("{ not json ") + report = validate_pack_file(self.path) + self.assertFalse(report.ok) + self.assertEqual(report.issues[0].code, "json_parse_error") + + def test_file_with_schema_errors(self): + import json + + bad = _good_pack() + bad["algorithm"] = "unknown_thing" + self._write(json.dumps(bad)) + report = validate_pack_file(self.path) + self.assertFalse(report.ok) + self.assertEqual(report.issues[0].code, "unknown_algorithm") + + +if __name__ == "__main__": + unittest.main()