Merge PR #101-#109 (EPIC #5 模型框架 8 子任务：recipe/feature/quality/anomaly/cross-process/pipeline/registry/PoC，命名空间化整合)
This commit is contained in:
@@ -0,0 +1 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
@@ -1,16 +1,28 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""测试引导：把 `core/model-framework` 以包名 `model_framework` 挂载到 sys.modules。
|
||||
"""测试引导：把连字符目录 ``core/model-framework`` 加载为可导入包
|
||||
``model_framework``，使测试可 ``from model_framework import ...``。
|
||||
|
||||
目录名 `model-framework` 含连字符，无法直接以包名 import；挂载后模块内相对导入
|
||||
（`from .hyperparam import ...`）在 unittest 发现机制下可正常解析。
|
||||
与仓库内各 core 模块的测试引导同款模式（importlib 完整加载包，执行
|
||||
``__init__.py``，保持顶层导出可用）。
|
||||
"""
|
||||
import importlib.util
|
||||
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
|
||||
PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
REPO_ROOT = os.path.dirname(os.path.dirname(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||||
|
||||
|
||||
def _load_package(name: str, path: str) -> None:
|
||||
if name in sys.modules:
|
||||
return
|
||||
init_py = os.path.join(path, "__init__.py")
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
name, init_py, submodule_search_locations=[path])
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[name] = module
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
|
||||
_load_package("model_framework", PKG_DIR)
|
||||
|
||||
@@ -0,0 +1,333 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""``anomaly_detection`` 单元测试（issue #37）。
|
||||
|
||||
覆盖：
|
||||
- 配方（Recipe）不可变性 / 序列化往返 / 非法主干、非法阈值策略与越界校验；
|
||||
- 主干工厂注册表 + 自定义主干注册（PRD 5.3「新增结构走插件注册」）；
|
||||
- stub / iforest / lof 三类主干的 fit/decision_function 契约；
|
||||
- 固定主干 + 配方加载：同框架加载 Ti / 树脂两套配方均跑通（PRD 5.3
|
||||
验收口径）；
|
||||
- Metrics 验收口径（PRD 5.3 / 里程碑：检出率 ≥ 95%、误报率 ≤ 5%）；
|
||||
- 阈值策略（contamination 高分位 / sigma Nσ 法则）；
|
||||
- 零外部强依赖：无 sklearn 时 stub 退化仍可加载与校验。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, HERE)
|
||||
import _bootstrap # noqa: E402 注册 model_framework 包
|
||||
|
||||
from model_framework.anomaly_detection import ( # noqa: E402
|
||||
AnomalyDetectionError,
|
||||
AnomalyDetectionModel,
|
||||
BACKBONES,
|
||||
Metrics,
|
||||
ModelHandle,
|
||||
Recipe,
|
||||
build_from_recipe,
|
||||
iforest_backbone,
|
||||
list_sample_recipes,
|
||||
load_recipe,
|
||||
lof_backbone,
|
||||
register_backbone,
|
||||
sample_recipe_path,
|
||||
stub_backbone,
|
||||
)
|
||||
|
||||
|
||||
def _normal_dataset(n=40, n_feat=2, seed=0):
|
||||
"""构造一组「正常」样本（围绕均值的确定性点）。"""
|
||||
X = []
|
||||
for i in range(n):
|
||||
row = []
|
||||
for j in range(n_feat):
|
||||
base = float(i % 7) + 1.0 + 0.1 * j
|
||||
row.append(base)
|
||||
X.append(row)
|
||||
return X
|
||||
|
||||
|
||||
def _labeled_dataset(n_normal=40, n_anomaly=5, n_feat=2):
|
||||
"""构造正常 + 离群点数据集，返回 (X, y_true)，1=异常。"""
|
||||
X = _normal_dataset(n_normal, n_feat)
|
||||
y = [0] * n_normal
|
||||
for k in range(n_anomaly):
|
||||
# 明显远离正常区的离群点
|
||||
X.append([100.0 + k for _ in range(n_feat)])
|
||||
y.append(1)
|
||||
return X, y
|
||||
|
||||
|
||||
class TestRecipe(unittest.TestCase):
|
||||
"""配方数据对象与校验。"""
|
||||
|
||||
def test_defaults_and_immutability(self):
|
||||
r = Recipe(name="t")
|
||||
self.assertEqual(r.backbone, "iforest")
|
||||
self.assertEqual(r.threshold_policy, "contamination")
|
||||
self.assertAlmostEqual(r.recall_floor, 0.95)
|
||||
self.assertAlmostEqual(r.false_alarm_ceil, 0.05)
|
||||
with self.assertRaises(Exception):
|
||||
r.name = "other" # frozen
|
||||
|
||||
def test_roundtrip(self):
|
||||
r = Recipe(name="t", backbone="lof",
|
||||
hyperparams={"n_neighbors": 15},
|
||||
feature_columns=("a", "b"),
|
||||
threshold_policy="sigma",
|
||||
contamination=0.1, sigma=2.5,
|
||||
recall_floor=0.9, false_alarm_ceil=0.1,
|
||||
industry="树脂", notes="n")
|
||||
d = r.to_dict()
|
||||
r2 = Recipe.from_dict(d)
|
||||
self.assertEqual(r, r2)
|
||||
# JSON 往返
|
||||
r3 = Recipe.from_dict(json.loads(json.dumps(d)))
|
||||
self.assertEqual(r, r3)
|
||||
|
||||
def test_invalid_backbone_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", backbone="svm")
|
||||
|
||||
def test_invalid_threshold_policy_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", threshold_policy="quantile")
|
||||
|
||||
def test_contamination_out_of_range(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", contamination=0.0)
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", contamination=1.0)
|
||||
|
||||
def test_sigma_nonpositive_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", sigma=0)
|
||||
|
||||
def test_recall_floor_out_of_range(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", recall_floor=1.5)
|
||||
|
||||
def test_missing_name(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="")
|
||||
|
||||
def test_load_recipe_from_file(self):
|
||||
path = sample_recipe_path("recipe.ti.json")
|
||||
r = load_recipe(path)
|
||||
self.assertEqual(r.name, "ti-cl4-furnace-impurity-anomaly")
|
||||
self.assertEqual(r.backbone, "iforest")
|
||||
self.assertIn("furnace_temp", r.feature_columns)
|
||||
|
||||
|
||||
class TestBackbones(unittest.TestCase):
|
||||
"""主干工厂与注册表。"""
|
||||
|
||||
def test_builtin_backbones_registered(self):
|
||||
for name in ("iforest", "lof", "stub"):
|
||||
self.assertIn(name, BACKBONES)
|
||||
|
||||
def test_register_custom_backbone(self):
|
||||
class _Custom(ModelHandle):
|
||||
def __init__(self, p):
|
||||
super().__init__("custom", p)
|
||||
self._v = 1.0
|
||||
|
||||
def _fit_impl(self, X):
|
||||
self._v = sum(sum(r) for r in X) / (len(X) * len(X[0]))
|
||||
|
||||
def _score_one(self, row):
|
||||
# 离均值越远分数越高
|
||||
return abs(sum(float(v) for v in row) - self._v)
|
||||
|
||||
register_backbone("custom_test", lambda p: _Custom(p))
|
||||
m = AnomalyDetectionModel(backbone="custom_test")
|
||||
X = _normal_dataset()
|
||||
m.fit(X)
|
||||
self.assertEqual(len(m.predict(X)), len(X))
|
||||
# 清理避免污染其它用例
|
||||
BACKBONES.pop("custom_test", None)
|
||||
|
||||
def test_unknown_backbone_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
AnomalyDetectionModel(backbone="not_a_backbone")
|
||||
|
||||
def test_stub_score_is_deterministic_and_nonneg(self):
|
||||
h = stub_backbone({})
|
||||
X = _normal_dataset()
|
||||
h.fit(X)
|
||||
s1 = h.decision_function(X)
|
||||
s2 = h.decision_function(X)
|
||||
self.assertEqual(s1, s2)
|
||||
self.assertTrue(all(isinstance(v, float) for v in s1))
|
||||
self.assertTrue(all(v >= 0 for v in s1))
|
||||
|
||||
def test_decision_before_fit_raises(self):
|
||||
h = stub_backbone({})
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
h.decision_function([[1.0, 2.0]])
|
||||
|
||||
def test_fit_empty_raises(self):
|
||||
h = stub_backbone({})
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
h.fit([])
|
||||
|
||||
def test_iforest_factory_runs_with_or_without_sklearn(self):
|
||||
# 无论 sklearn 是否存在都不应报错
|
||||
h = iforest_backbone({"n_estimators": 20})
|
||||
X = _normal_dataset()
|
||||
h.fit(X)
|
||||
scores = h.decision_function(X)
|
||||
self.assertEqual(len(scores), len(X))
|
||||
|
||||
|
||||
class TestModelContract(unittest.TestCase):
|
||||
"""模型 fit/decision_function/predict 契约。"""
|
||||
|
||||
def test_fit_predict_shapes(self):
|
||||
m = AnomalyDetectionModel(backbone="stub")
|
||||
X = _normal_dataset(20)
|
||||
m.fit(X)
|
||||
self.assertTrue(m.fitted)
|
||||
self.assertIsNotNone(m.threshold)
|
||||
preds = m.predict(X)
|
||||
self.assertEqual(len(preds), len(X))
|
||||
self.assertTrue(all(p in (0, 1) for p in preds))
|
||||
|
||||
def test_predict_before_fit_raises(self):
|
||||
m = AnomalyDetectionModel(backbone="stub")
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
m.predict([[1.0, 2.0]])
|
||||
|
||||
def test_decision_before_fit_raises(self):
|
||||
m = AnomalyDetectionModel(backbone="stub")
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
m.decision_function([[1.0, 2.0]])
|
||||
|
||||
def test_fit_empty_raises(self):
|
||||
m = AnomalyDetectionModel(backbone="stub")
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
m.fit([])
|
||||
|
||||
def test_to_dict_roundtrip_meta(self):
|
||||
m = AnomalyDetectionModel(backbone="iforest",
|
||||
hyperparams={"n_estimators": 5},
|
||||
feature_columns=["a"],
|
||||
threshold_policy="sigma", sigma=2.0)
|
||||
d = m.to_dict()
|
||||
self.assertEqual(d["recipe_meta"]["backbone"], "iforest")
|
||||
self.assertEqual(d["recipe_meta"]["threshold_policy"], "sigma")
|
||||
self.assertIn("handle", d)
|
||||
|
||||
def test_threshold_contamination_isolate_outliers(self):
|
||||
"""contamination 阈值应把注入的离群点判为异常。"""
|
||||
m = AnomalyDetectionModel(
|
||||
backbone="stub", threshold_policy="contamination",
|
||||
contamination=0.10)
|
||||
X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5)
|
||||
m.fit(X)
|
||||
preds = m.predict(X)
|
||||
# 注入的 5 个离群点应被全部判异常
|
||||
self.assertEqual(sum(preds[40:]), 5)
|
||||
|
||||
def test_threshold_sigma_isolate_outliers(self):
|
||||
"""sigma 阈值也应把注入的极端离群点判为异常。"""
|
||||
m = AnomalyDetectionModel(
|
||||
backbone="stub", threshold_policy="sigma", sigma=2.0)
|
||||
X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5)
|
||||
m.fit(X)
|
||||
preds = m.predict(X)
|
||||
self.assertEqual(sum(preds[40:]), 5)
|
||||
|
||||
|
||||
class TestMetrics(unittest.TestCase):
|
||||
"""验收口径（PRD 5.3：检出率 ≥ 95%、误报率 ≤ 5%）。"""
|
||||
|
||||
def test_perfect_predictions_pass(self):
|
||||
y = [1, 1, 0, 0, 0]
|
||||
met = Metrics.compute(y, y, recall_floor=0.95, false_alarm_ceil=0.05)
|
||||
self.assertAlmostEqual(met.recall, 1.0)
|
||||
self.assertAlmostEqual(met.false_alarm_rate, 0.0)
|
||||
self.assertAlmostEqual(met.f1, 1.0)
|
||||
self.assertTrue(met.passed)
|
||||
|
||||
def test_all_miss_fails(self):
|
||||
y_true = [1, 1, 0, 0]
|
||||
y_pred = [0, 0, 0, 0] # 漏检全部异常
|
||||
met = Metrics.compute(y_true, y_pred)
|
||||
self.assertAlmostEqual(met.recall, 0.0)
|
||||
self.assertFalse(met.passed)
|
||||
|
||||
def test_high_false_alarm_fails(self):
|
||||
y_true = [1, 0, 0, 0, 0]
|
||||
y_pred = [1, 1, 1, 1, 1] # 全判异常：检出但误报爆表
|
||||
met = Metrics.compute(y_true, y_pred, false_alarm_ceil=0.05)
|
||||
self.assertAlmostEqual(met.recall, 1.0)
|
||||
self.assertGreater(met.false_alarm_rate, 0.05)
|
||||
self.assertFalse(met.passed)
|
||||
|
||||
def test_length_mismatch_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Metrics.compute([1, 0], [1])
|
||||
|
||||
def test_empty_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Metrics.compute([], [])
|
||||
|
||||
def test_no_anomaly_in_true_recall_zero_div_safe(self):
|
||||
# 无真实异常时 recall 定义为 0，不应抛 ZeroDivision
|
||||
met = Metrics.compute([0, 0, 0], [0, 0, 0])
|
||||
self.assertEqual(met.recall, 0.0)
|
||||
self.assertEqual(met.n_anomaly_true, 0)
|
||||
|
||||
def test_evaluate_end_to_end(self):
|
||||
m = AnomalyDetectionModel(backbone="stub", threshold_policy="sigma",
|
||||
sigma=2.0)
|
||||
X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5)
|
||||
m.fit(X)
|
||||
met = m.evaluate(X, y_true)
|
||||
self.assertIsInstance(met, Metrics)
|
||||
# 离群点应被检出（stub 在极端离群点上召回=1）
|
||||
self.assertEqual(met.recall, 1.0)
|
||||
|
||||
|
||||
class TestSampleRecipes(unittest.TestCase):
|
||||
"""样例协议：同框架加载 Ti / 树脂两套配方均跑通（PRD 5.3 验收口径）。"""
|
||||
|
||||
def test_samples_present(self):
|
||||
names = list_sample_recipes()
|
||||
self.assertIn("recipe.ti.json", names)
|
||||
self.assertIn("recipe.resin.json", names)
|
||||
|
||||
def test_build_from_each_sample_runs(self):
|
||||
for name in ("recipe.ti.json", "recipe.resin.json"):
|
||||
m = build_from_recipe(sample_recipe_path(name))
|
||||
self.assertIn(
|
||||
m.recipe_meta["backbone"], ("iforest", "lof", "stub"))
|
||||
feat = m.recipe_meta["feature_columns"]
|
||||
n_feat = len(feat)
|
||||
self.assertGreater(n_feat, 0)
|
||||
X = [[float(i + j) for j in range(n_feat)] for i in range(30)]
|
||||
# 注入离群点
|
||||
for k in range(3):
|
||||
X.append([100.0 + k for _ in range(n_feat)])
|
||||
y_true = [0] * 30 + [1] * 3
|
||||
m.fit(X)
|
||||
preds = m.predict(X)
|
||||
self.assertEqual(len(preds), len(y_true))
|
||||
met = m.evaluate(X, y_true)
|
||||
self.assertIsInstance(met, Metrics)
|
||||
|
||||
def test_two_recipes_share_same_code(self):
|
||||
"""切换模板仅改配方，模型代码零改动（PRD 5.3）。"""
|
||||
m1 = build_from_recipe(sample_recipe_path("recipe.ti.json"))
|
||||
m2 = build_from_recipe(sample_recipe_path("recipe.resin.json"))
|
||||
self.assertEqual(type(m1), type(m2))
|
||||
self.assertNotEqual(m1.recipe_meta.get("recipe_name"),
|
||||
m2.recipe_meta.get("recipe_name"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,340 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""跨工序寻优模型模板化单元测试（issue #38）。
|
||||
|
||||
覆盖：
|
||||
- 数据对象（DecisionVariable / Stage / Constraint / Objective / Recipe）的
|
||||
构造、校验、序列化往返；
|
||||
- 受限表达式求值 ``_safe_eval``（拒绝危险内建/属性访问）；
|
||||
- 四种求解器（grid / random / analytic / stub）的可行解搜索与目标最大化；
|
||||
- 主干 ``CrossProcessOptimizer.optimize`` + ``build_from_recipe``；
|
||||
- 采纳率口径（PRD 5.3 ≥ 60%）与可解释建议（StageSuggestion 方向）；
|
||||
- 样例配方（Ti / 树脂）均能加载并寻优跑通（验收口径）。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
if HERE not in sys.path:
|
||||
sys.path.insert(0, HERE)
|
||||
|
||||
import _bootstrap # noqa: E402 加载 model_framework 包
|
||||
|
||||
from model_framework.cross_process_optimizer import ( # noqa: E402
|
||||
Constraint,
|
||||
CrossProcessOptError,
|
||||
CrossProcessOptimizer,
|
||||
DecisionVariable,
|
||||
Objective,
|
||||
OptimizationResult,
|
||||
Recipe,
|
||||
Stage,
|
||||
StageSuggestion,
|
||||
SOLVERS,
|
||||
build_from_recipe,
|
||||
list_sample_recipes,
|
||||
load_recipe,
|
||||
register_solver,
|
||||
sample_recipe_path,
|
||||
stub_solver,
|
||||
)
|
||||
|
||||
|
||||
def _two_stage_recipe(solver: str = "grid") -> Recipe:
|
||||
"""构造一个简单的两工序寻优配方用于测试。"""
|
||||
s1 = Stage(
|
||||
name="upstream",
|
||||
decision_vars=(
|
||||
DecisionVariable("u_temp", 100, 200, step=20, default=120),
|
||||
),
|
||||
transfer_vars=("u_yield",),
|
||||
proxy="(u_temp - 100) / 100",
|
||||
)
|
||||
s2 = Stage(
|
||||
name="downstream",
|
||||
decision_vars=(
|
||||
DecisionVariable("d_pressure", 1, 5, step=1, default=2),
|
||||
),
|
||||
transfer_vars=("quality",),
|
||||
proxy="u_yield * 0.5 + d_pressure * 0.1",
|
||||
)
|
||||
return Recipe(
|
||||
name="test-recipe",
|
||||
stages=(s1, s2),
|
||||
constraints=(
|
||||
Constraint("u_temp", "<=", 200, label="安全上限"),
|
||||
Constraint("d_pressure", ">=", 1, label="压力下限"),
|
||||
),
|
||||
objective=Objective("quality", "max", label="质量"),
|
||||
solver=solver,
|
||||
acceptance_floor=0.6,
|
||||
)
|
||||
|
||||
|
||||
class TestDataObjects(unittest.TestCase):
|
||||
"""数据对象构造、校验、序列化往返。"""
|
||||
|
||||
def test_decision_variable_grid_points(self):
|
||||
v = DecisionVariable("x", 0, 10, step=2)
|
||||
self.assertEqual(v.grid_points(), [0, 2, 4, 6, 8, 10])
|
||||
|
||||
def test_decision_variable_rejects_invalid_range(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
DecisionVariable("x", 10, 0)
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
DecisionVariable("x", 0, 10, step=0)
|
||||
|
||||
def test_decision_variable_roundtrip(self):
|
||||
v = DecisionVariable("x", 1.5, 3.5, step=0.5, unit="MPa", default=2.0)
|
||||
v2 = DecisionVariable.from_dict(v.to_dict())
|
||||
self.assertEqual(v, v2)
|
||||
|
||||
def test_constraint_operators(self):
|
||||
ns = {"x": 5}
|
||||
self.assertTrue(Constraint("x", "<=", 5).satisfied(ns))
|
||||
self.assertTrue(Constraint("x", ">=", 5).satisfied(ns))
|
||||
self.assertTrue(Constraint("x", "==", 5).satisfied(ns))
|
||||
self.assertFalse(Constraint("x", "<=", 4).satisfied(ns))
|
||||
self.assertFalse(Constraint("x", ">=", 6).satisfied(ns))
|
||||
|
||||
def test_constraint_rejects_bad_op(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Constraint("x", "!=", 0)
|
||||
|
||||
def test_objective_score_min_inverts(self):
|
||||
obj = Objective("x", "min")
|
||||
# 最小化：x=5 的标准化分数应为 -5（越大越好 = 越小原值）
|
||||
self.assertAlmostEqual(obj.score({"x": 5}), -5.0)
|
||||
|
||||
def test_objective_rejects_bad_sense(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Objective("x", "avg")
|
||||
|
||||
def test_recipe_requires_stages(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Recipe(name="x", stages=())
|
||||
|
||||
def test_recipe_rejects_bad_solver(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Recipe(name="x", stages=(Stage(name="s"),), solver="magic")
|
||||
|
||||
def test_recipe_rejects_bad_acceptance(self):
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
Recipe(name="x", stages=(Stage(name="s"),), acceptance_floor=1.5)
|
||||
|
||||
def test_recipe_roundtrip(self):
|
||||
r = _two_stage_recipe()
|
||||
r2 = Recipe.from_dict(r.to_dict())
|
||||
self.assertEqual(r, r2)
|
||||
self.assertEqual(r2.stages[0].decision_vars[0].name, "u_temp")
|
||||
|
||||
|
||||
class TestSafeEval(unittest.TestCase):
|
||||
"""受限表达式求值安全性。"""
|
||||
|
||||
def test_safe_eval_basic(self):
|
||||
from model_framework.cross_process_optimizer import _safe_eval
|
||||
self.assertAlmostEqual(_safe_eval("1 + 2 * 3", {}), 7.0)
|
||||
self.assertAlmostEqual(_safe_eval("x + y", {"x": 1, "y": 2}), 3.0)
|
||||
self.assertAlmostEqual(_safe_eval("min(x, y)", {"x": 1, "y": 2}), 1.0)
|
||||
|
||||
def test_safe_eval_rejects_empty(self):
|
||||
from model_framework.cross_process_optimizer import _safe_eval
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
_safe_eval("", {})
|
||||
|
||||
def test_safe_eval_rejects_builtins(self):
|
||||
"""禁止访问 __import__ / open / 任意内建（沙箱保护）。"""
|
||||
from model_framework.cross_process_optimizer import _safe_eval
|
||||
with self.assertRaises(Exception):
|
||||
_safe_eval("__import__('os')", {})
|
||||
with self.assertRaises(Exception):
|
||||
_safe_eval("open('x')", {})
|
||||
|
||||
|
||||
class TestSolvers(unittest.TestCase):
|
||||
"""四种求解器的可行解搜索与目标最大化。"""
|
||||
|
||||
def test_grid_solver_finds_feasible(self):
|
||||
r = _two_stage_recipe("grid")
|
||||
opt = CrossProcessOptimizer(r)
|
||||
res = opt.optimize()
|
||||
self.assertIsInstance(res, OptimizationResult)
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
self.assertGreaterEqual(res.objective_score, res.baseline_score)
|
||||
|
||||
def test_grid_solver_no_feasible_raises(self):
|
||||
# 矛盾约束：温度必须同时 <= 100 且 >= 200
|
||||
r = Recipe(
|
||||
name="infeasible",
|
||||
stages=(Stage(name="s",
|
||||
decision_vars=(DecisionVariable("x", 100, 300, step=50, default=150),)),),
|
||||
constraints=(Constraint("x", "<=", 100), Constraint("x", ">=", 200)),
|
||||
objective=Objective("x", "max"),
|
||||
solver="grid",
|
||||
)
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
CrossProcessOptimizer(r).optimize()
|
||||
|
||||
def test_random_solver_finds_feasible(self):
|
||||
r = _two_stage_recipe("random")
|
||||
res = CrossProcessOptimizer(r).optimize(seed=42)
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
self.assertEqual(res.solver, "random")
|
||||
|
||||
def test_random_solver_uses_solver_params(self):
|
||||
r = _two_stage_recipe("random")
|
||||
r = Recipe.from_dict({**r.to_dict(),
|
||||
"solver_params": {"n_samples": 50, "seed": 7}})
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
|
||||
def test_analytic_solver_single_var(self):
|
||||
# 单变量线性最大化目标：应在 high 边界取得最优
|
||||
r = Recipe(
|
||||
name="single",
|
||||
stages=(Stage(name="s",
|
||||
decision_vars=(DecisionVariable("x", 0, 10, step=1, default=2),)),),
|
||||
objective=Objective("x", "max", label="越大越好"),
|
||||
solver="analytic",
|
||||
)
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
self.assertEqual(res.objective_score, 10.0)
|
||||
# 建议把 x 从默认 2 上调到 10
|
||||
sug = res.suggestions[0]
|
||||
self.assertEqual(sug.new_value, 10.0)
|
||||
self.assertEqual(sug.direction, "上调")
|
||||
|
||||
def test_analytic_falls_back_to_grid_for_multi_var(self):
|
||||
r = _two_stage_recipe("analytic")
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
# 多变量时 analytic 退化为 grid，仍能跑通
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
|
||||
def test_analytic_no_feasible_raises(self):
|
||||
r = Recipe(
|
||||
name="bad",
|
||||
stages=(Stage(name="s",
|
||||
decision_vars=(DecisionVariable("x", 0, 10, step=1, default=5),)),),
|
||||
constraints=(Constraint("x", ">=", 100),),
|
||||
objective=Objective("x", "max"),
|
||||
solver="analytic",
|
||||
)
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
CrossProcessOptimizer(r).optimize()
|
||||
|
||||
def test_stub_solver_returns_default(self):
|
||||
r = _two_stage_recipe("stub")
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
# stub 直接取默认值，改善为 0
|
||||
self.assertEqual(res.improvement, 0.0)
|
||||
self.assertEqual(res.solver, "stub")
|
||||
|
||||
def test_unknown_solver_raises(self):
|
||||
r = Recipe.from_dict({**_two_stage_recipe().to_dict(), "solver": "grid"})
|
||||
# 临时篡改 recipe.solver 为非法值（绕过校验）测主干分支
|
||||
object.__setattr__(r, "solver", "voodoo")
|
||||
with self.assertRaises(CrossProcessOptError):
|
||||
CrossProcessOptimizer(r).optimize()
|
||||
|
||||
|
||||
class TestAcceptanceAndSuggestions(unittest.TestCase):
|
||||
"""采纳率口径（PRD 5.3 ≥ 60%）与可解释建议。"""
|
||||
|
||||
def test_grid_improvement_marks_accepted(self):
|
||||
r = _two_stage_recipe("grid")
|
||||
# 默认值非最优，grid 应能找到更优解 → accepted
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
if res.improvement > 1e-9:
|
||||
self.assertTrue(res.accepted)
|
||||
self.assertGreaterEqual(res.acceptance, res.acceptance_floor)
|
||||
|
||||
def test_suggestion_direction(self):
|
||||
s_up = StageSuggestion("s", "x", 1.0, 3.0, 2.0)
|
||||
self.assertEqual(s_up.direction, "上调")
|
||||
s_down = StageSuggestion("s", "x", 3.0, 1.0, -2.0)
|
||||
self.assertEqual(s_down.direction, "下调")
|
||||
s_keep = StageSuggestion("s", "x", 2.0, 2.0, 0.0)
|
||||
self.assertEqual(s_keep.direction, "保持")
|
||||
|
||||
def test_result_to_dict_serializable(self):
|
||||
r = _two_stage_recipe("stub")
|
||||
res = CrossProcessOptimizer(r).optimize()
|
||||
d = res.to_dict()
|
||||
# 可 JSON 序列化
|
||||
json.dumps(d)
|
||||
self.assertIn("suggestions", d)
|
||||
self.assertIn("accepted", d)
|
||||
|
||||
|
||||
class TestSampleRecipes(unittest.TestCase):
|
||||
"""样例配方（Ti / 树脂）加载与寻优（验收口径）。"""
|
||||
|
||||
def test_sample_recipes_listed(self):
|
||||
names = list_sample_recipes()
|
||||
self.assertIn("recipe.ti.json", names)
|
||||
self.assertIn("recipe.resin.json", names)
|
||||
|
||||
def test_ti_recipe_loads_and_optimizes(self):
|
||||
opt = build_from_recipe(sample_recipe_path("recipe.ti.json"))
|
||||
res = opt.optimize()
|
||||
self.assertEqual(res.solver, "grid")
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
self.assertGreaterEqual(res.objective_score, res.baseline_score)
|
||||
# 工序建议覆盖三道工序
|
||||
stages_covered = {s.stage for s in res.suggestions}
|
||||
self.assertEqual(stages_covered, {"氯化", "精制", "还原"})
|
||||
|
||||
def test_resin_recipe_loads_and_optimizes(self):
|
||||
opt = build_from_recipe(sample_recipe_path("recipe.resin.json"))
|
||||
res = opt.optimize()
|
||||
self.assertEqual(res.solver, "random")
|
||||
self.assertGreater(res.feasible_count, 0)
|
||||
stages_covered = {s.stage for s in res.suggestions}
|
||||
self.assertEqual(stages_covered, {"反应", "水洗", "干燥"})
|
||||
|
||||
def test_two_recipes_same_engine_class(self):
|
||||
"""验收口径：同框架加载两套配方，寻优主干类零改动。"""
|
||||
opt_ti = build_from_recipe(sample_recipe_path("recipe.ti.json"))
|
||||
opt_resin = build_from_recipe(sample_recipe_path("recipe.resin.json"))
|
||||
self.assertIs(type(opt_ti), type(opt_resin))
|
||||
# 两套配方的工序拓扑确实不同
|
||||
self.assertNotEqual(opt_ti.recipe_meta["stages"],
|
||||
opt_resin.recipe_meta["stages"])
|
||||
|
||||
def test_load_recipe_from_temp_file(self):
|
||||
r = _two_stage_recipe()
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".json", delete=False, encoding="utf-8") as fh:
|
||||
json.dump(r.to_dict(), fh, ensure_ascii=False)
|
||||
path = fh.name
|
||||
try:
|
||||
r2 = load_recipe(path)
|
||||
self.assertEqual(r, r2)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
class TestRegisterSolver(unittest.TestCase):
|
||||
"""插件式求解器注册。"""
|
||||
|
||||
def test_register_custom_solver(self):
|
||||
called = {"n": 0}
|
||||
|
||||
def my_solver(recipe, **kw):
|
||||
called["n"] += 1
|
||||
return stub_solver(recipe, **kw)
|
||||
|
||||
register_solver("my", my_solver)
|
||||
self.assertIn("my", SOLVERS)
|
||||
# 直接构造主干并替换 recipe.solver 为已注册的自定义求解器
|
||||
r = _two_stage_recipe()
|
||||
object.__setattr__(r, "solver", "my")
|
||||
CrossProcessOptimizer(r).optimize()
|
||||
self.assertEqual(called["n"], 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,334 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""FeatureSpec 声明式特征定义引擎测试（issue #35）。
|
||||
|
||||
覆盖：
|
||||
1. 解析：算子调用、裸点位、数值/窗口字面量、嵌套、中文点位、带符号数值；
|
||||
2. 解析错误：空 spec、非法字符、括号不匹配、多余内容、参数缺失；
|
||||
3. 语义校验：未知算子、arity 不匹配、参数 kind 错误；
|
||||
4. 依赖分析：resolve_inputs 去重与顺序、嵌套算子依赖汇总；
|
||||
5. 执行：EMA/SMA/RollingStd/RateOfChange/Diff/Lag/Log/Scale/Clip/Combine 的
|
||||
数值正确性，缺失点位 fail-fast；
|
||||
6. 插件注册：register_operator 扩展新算子；
|
||||
7. 往返：to_dict/repr 稳定。
|
||||
"""
|
||||
import math
|
||||
import unittest
|
||||
|
||||
import _bootstrap # noqa: F401 挂载包名
|
||||
|
||||
from model_framework.feature_spec import (
|
||||
FeatureAST,
|
||||
Number,
|
||||
OpCall,
|
||||
OPERATORS,
|
||||
ParseError,
|
||||
SpecIssue,
|
||||
TagRef,
|
||||
Window,
|
||||
describe,
|
||||
materialize,
|
||||
parse,
|
||||
register_operator,
|
||||
resolve_inputs,
|
||||
validate,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 解析
|
||||
# ---------------------------------------------------------------------------
|
||||
class ParseTest(unittest.TestCase):
|
||||
def test_simple_op_with_window(self):
|
||||
ast = parse("EMA(CLF-01.TEMP, 5m)")
|
||||
self.assertEqual(
|
||||
ast, OpCall("EMA", (TagRef("CLF-01.TEMP"), Window(5.0, "m")))
|
||||
)
|
||||
|
||||
def test_simple_op_with_number_window(self):
|
||||
ast = parse("RollingStd(CLF-01.CL2, 10)")
|
||||
self.assertEqual(ast, OpCall("RollingStd", (TagRef("CLF-01.CL2"), Number(10))))
|
||||
|
||||
def test_bare_tag(self):
|
||||
self.assertEqual(parse("炉压"), TagRef("炉压"))
|
||||
|
||||
def test_tag_with_dots_and_dash(self):
|
||||
self.assertEqual(parse("A.B-C_01"), TagRef("A.B-C_01"))
|
||||
|
||||
def test_signed_and_scientific_number(self):
|
||||
ast = parse("Scale(A, -0.5)")
|
||||
self.assertEqual(ast, OpCall("Scale", (TagRef("A"), Number(-0.5))))
|
||||
ast2 = parse("Scale(A, 1e-3)")
|
||||
self.assertAlmostEqual(ast2.args[1].value, 0.001)
|
||||
|
||||
def test_nested_op(self):
|
||||
# 嵌套：外层 Scale，内层 EMA 作为第一个参数点位位置（语法合法，语义由算子判定）
|
||||
ast = parse("Combine(EMA(A, 5m), B)")
|
||||
self.assertEqual(ast.name, "Combine")
|
||||
self.assertEqual(len(ast.args), 2)
|
||||
self.assertEqual(ast.args[0].name, "EMA")
|
||||
|
||||
def test_no_arg_op(self):
|
||||
ast = parse("Diff()")
|
||||
self.assertEqual(ast, OpCall("Diff", ()))
|
||||
|
||||
def test_integer_window_vs_number(self):
|
||||
self.assertEqual(parse("Lag(A, 3)").args[1], Number(3))
|
||||
self.assertEqual(parse("Lag(A, 3m)").args[1], Window(3.0, "m"))
|
||||
|
||||
def test_repr_roundtrip(self):
|
||||
for spec in ["EMA(CLF-01.TEMP, 5m)", "RateOfChange(炉压)", "Clip(P, -1, 1)"]:
|
||||
self.assertEqual(repr(parse(spec)).replace(" ", ""), spec.replace(" ", ""))
|
||||
|
||||
# ---- 解析错误 ----
|
||||
def test_empty_raises(self):
|
||||
with self.assertRaises((ValueError, ParseError)):
|
||||
parse("")
|
||||
with self.assertRaises((ValueError, ParseError)):
|
||||
parse(" ")
|
||||
|
||||
def test_non_string_raises(self):
|
||||
with self.assertRaises(ValueError):
|
||||
parse(123) # type: ignore[arg-type]
|
||||
|
||||
def test_unrecognized_char(self):
|
||||
with self.assertRaises(ParseError) as cm:
|
||||
parse("EMA(A, 5m) @")
|
||||
self.assertIsNotNone(cm.exception.position)
|
||||
|
||||
def test_missing_rparen(self):
|
||||
with self.assertRaises(ParseError):
|
||||
parse("EMA(A, 5m")
|
||||
|
||||
def test_missing_rparen_inner(self):
|
||||
with self.assertRaises(ParseError):
|
||||
parse("EMA(A, (5m)")
|
||||
|
||||
def test_trailing_garbage(self):
|
||||
with self.assertRaises(ParseError):
|
||||
parse("EMA(A, 5m) B")
|
||||
|
||||
def test_missing_arg_after_comma(self):
|
||||
with self.assertRaises(ParseError):
|
||||
parse("EMA(A, )")
|
||||
|
||||
def test_starts_with_paren(self):
|
||||
with self.assertRaises(ParseError):
|
||||
parse("(A)")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 语义校验
|
||||
# ---------------------------------------------------------------------------
|
||||
class ValidateTest(unittest.TestCase):
|
||||
def test_known_op_valid(self):
|
||||
self.assertEqual(validate(parse("EMA(A, 5m)")), [])
|
||||
|
||||
def test_unknown_operator(self):
|
||||
issues = validate(parse("FooBar(A, 5m)"))
|
||||
self.assertEqual(len(issues), 1)
|
||||
self.assertEqual(issues[0].code, "unknown_operator")
|
||||
|
||||
def test_arity_too_few(self):
|
||||
issues = validate(parse("EMA(A)"))
|
||||
self.assertTrue(any(i.code == "arity" for i in issues))
|
||||
|
||||
def test_arity_too_many(self):
|
||||
issues = validate(parse("EMA(A, 5m, 7)"))
|
||||
self.assertTrue(any(i.code == "arity" for i in issues))
|
||||
|
||||
def test_bad_arg_kind_number_where_window(self):
|
||||
# EMA 第二参数允许 window/number，故合法
|
||||
self.assertEqual(validate(parse("EMA(A, 7)")), [])
|
||||
# 但 tag 位置传 number 非法
|
||||
issues = validate(parse("EMA(5, 7)"))
|
||||
self.assertTrue(any(i.code == "bad_arg" for i in issues))
|
||||
|
||||
def test_combine_varargs(self):
|
||||
self.assertEqual(validate(parse("Combine(A, B, C)")), [])
|
||||
issues = validate(parse("Combine(A)"))
|
||||
self.assertTrue(any(i.code == "arity" for i in issues))
|
||||
|
||||
def test_nested_unknown(self):
|
||||
issues = validate(parse("Combine(Foo(A), B)"))
|
||||
self.assertTrue(any(i.code == "unknown_operator" for i in issues))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 依赖分析
|
||||
# ---------------------------------------------------------------------------
|
||||
class ResolveInputsTest(unittest.TestCase):
|
||||
def test_single_tag(self):
|
||||
self.assertEqual(resolve_inputs(parse("炉压")), ["炉压"])
|
||||
|
||||
def test_dedup_order(self):
|
||||
# 同一点位重复出现，去重且保持首次出现顺序
|
||||
self.assertEqual(resolve_inputs(parse("Combine(A, A)")), ["A"])
|
||||
|
||||
def test_multiple_tags(self):
|
||||
self.assertEqual(resolve_inputs(parse("Combine(A.tank1, A.tank2)")), ["A.tank1", "A.tank2"])
|
||||
|
||||
def test_op_collects_input(self):
|
||||
self.assertEqual(resolve_inputs(parse("EMA(CLF-01.TEMP, 5m)")), ["CLF-01.TEMP"])
|
||||
|
||||
def test_number_window_no_inputs(self):
|
||||
# 裸数值/窗口虽不是合法特征根，但 resolve_inputs 不报错
|
||||
self.assertEqual(resolve_inputs(Number(3)), [])
|
||||
self.assertEqual(resolve_inputs(Window(5.0, "m")), [])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 执行
|
||||
# ---------------------------------------------------------------------------
|
||||
class MaterializeTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# 一个稳定的伪时序：1..10
|
||||
self.series = {"A": [float(i) for i in range(1, 11)]} # 1..10
|
||||
|
||||
def test_bare_tag(self):
|
||||
self.assertEqual(materialize(parse("A"), self.series), self.series["A"])
|
||||
|
||||
def test_number(self):
|
||||
self.assertEqual(materialize(Number(3), {}), 3)
|
||||
|
||||
def test_sma_window3(self):
|
||||
out = materialize(parse("SMA(A, 3)"), self.series)
|
||||
# 前 2 个 NaN，第 3 个 = (1+2+3)/3 = 2.0
|
||||
self.assertTrue(math.isnan(out[0]) and math.isnan(out[1]))
|
||||
self.assertAlmostEqual(out[2], 2.0)
|
||||
self.assertAlmostEqual(out[9], (8 + 9 + 10) / 3)
|
||||
|
||||
def test_ema_decreasing_weight(self):
|
||||
out = materialize(parse("EMA(A, 5)"), self.series)
|
||||
# EMA 单调（输入单调增），首值 = 首个观测
|
||||
self.assertAlmostEqual(out[0], 1.0)
|
||||
self.assertTrue(all(out[i] <= out[i + 1] for i in range(len(out) - 1)))
|
||||
|
||||
def test_rolling_std(self):
|
||||
out = materialize(parse("RollingStd(A, 2)"), self.series)
|
||||
self.assertTrue(math.isnan(out[0]))
|
||||
# std(1,2) 无偏 = 0.7071...
|
||||
self.assertAlmostEqual(out[1], math.sqrt(0.5))
|
||||
|
||||
def test_rolling_max_min(self):
|
||||
mx = materialize(parse("RollingMax(A, 3)"), self.series)
|
||||
mn = materialize(parse("RollingMin(A, 3)"), self.series)
|
||||
self.assertEqual(mx[2], 3.0)
|
||||
self.assertEqual(mn[2], 1.0)
|
||||
|
||||
def test_diff(self):
|
||||
out = materialize(parse("Diff(A)"), self.series)
|
||||
self.assertTrue(math.isnan(out[0]))
|
||||
self.assertTrue(all(out[i] == 1.0 for i in range(1, len(out))))
|
||||
|
||||
def test_lag(self):
|
||||
out = materialize(parse("Lag(A, 2)"), self.series)
|
||||
self.assertTrue(math.isnan(out[0]) and math.isnan(out[1]))
|
||||
self.assertEqual(out[2], 1.0)
|
||||
|
||||
def test_rate_of_change(self):
|
||||
# 常数序列 → 变化率为 0（非 NaN；NaN 仅出现在前 window 步预热）
|
||||
const = {"C": [5.0] * 6}
|
||||
out = materialize(parse("RateOfChange(C)"), const)
|
||||
self.assertTrue(math.isnan(out[0])) # 预热步 NaN
|
||||
self.assertEqual(out[1], 0.0)
|
||||
# 含 0 的序列 → 分母为 0 → NaN
|
||||
zero_denom = {"Z": [0.0, 1.0, 2.0]}
|
||||
outz = materialize(parse("RateOfChange(Z)"), zero_denom)
|
||||
self.assertTrue(math.isnan(outz[1]))
|
||||
# 线性序列 ROC 步长1 = 1/prev
|
||||
out2 = materialize(parse("RateOfChange(A)"), self.series)
|
||||
self.assertAlmostEqual(out2[1], 1.0 / 1.0)
|
||||
self.assertAlmostEqual(out2[5], 1.0 / 5.0)
|
||||
|
||||
def test_log_negative_nan(self):
|
||||
data = {"P": [1.0, -2.0, math.e]}
|
||||
out = materialize(parse("Log(P)"), data)
|
||||
self.assertAlmostEqual(out[0], 0.0)
|
||||
self.assertTrue(math.isnan(out[1]))
|
||||
self.assertAlmostEqual(out[2], 1.0)
|
||||
|
||||
def test_scale(self):
|
||||
out = materialize(parse("Scale(A, 10)"), self.series)
|
||||
self.assertEqual(out[0], 10.0)
|
||||
self.assertEqual(out[9], 100.0)
|
||||
|
||||
def test_clip(self):
|
||||
out = materialize(parse("Clip(A, 3, 7)"), self.series)
|
||||
self.assertEqual(out, [3.0, 3.0, 3.0, 4.0, 5.0, 6.0, 7.0, 7.0, 7.0, 7.0])
|
||||
|
||||
def test_combine(self):
|
||||
data = {"A": [1.0, 2.0, 3.0], "B": [10.0, 20.0, 30.0]}
|
||||
self.assertEqual(materialize(parse("Combine(A, B)"), data), [11.0, 22.0, 33.0])
|
||||
|
||||
def test_missing_input_fails_fast(self):
|
||||
with self.assertRaises(KeyError):
|
||||
materialize(parse("EMA(Missing, 3)"), {"A": [1.0, 2.0, 3.0]})
|
||||
|
||||
def test_unknown_op_fails_fast(self):
|
||||
with self.assertRaises(ValueError):
|
||||
materialize(OpCall("NoSuchOp", (TagRef("A"),)), self.series)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 插件注册
|
||||
# ---------------------------------------------------------------------------
|
||||
class RegisterOperatorTest(unittest.TestCase):
|
||||
def test_register_then_parse_and_run(self):
|
||||
def _double(series_map, args):
|
||||
tag = args[0]
|
||||
return [x * 2 for x in series_map[tag.name]]
|
||||
|
||||
register_operator(
|
||||
"Double",
|
||||
min_arity=1,
|
||||
max_arity=1,
|
||||
arg_kinds=(("tag",),),
|
||||
func=_double,
|
||||
doc="示例自定义算子：翻倍",
|
||||
)
|
||||
try:
|
||||
self.assertIn("Double", OPERATORS)
|
||||
self.assertEqual(validate(parse("Double(A)")), [])
|
||||
self.assertEqual(
|
||||
materialize(parse("Double(A)"), {"A": [1.0, 2.0]}), [2.0, 4.0]
|
||||
)
|
||||
finally:
|
||||
OPERATORS.pop("Double", None)
|
||||
|
||||
def test_register_overrides(self):
|
||||
register_operator(
|
||||
"Stub", min_arity=0, max_arity=0, arg_kinds=(), func=lambda s, a: 1, doc="v1"
|
||||
)
|
||||
register_operator(
|
||||
"Stub", min_arity=0, max_arity=0, arg_kinds=(), func=lambda s, a: 2, doc="v2"
|
||||
)
|
||||
try:
|
||||
self.assertEqual(OPERATORS["Stub"].doc, "v2")
|
||||
finally:
|
||||
OPERATORS.pop("Stub", None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 描述 / 往返
|
||||
# ---------------------------------------------------------------------------
|
||||
class DescribeAndSerializeTest(unittest.TestCase):
|
||||
def test_describe_contains_inputs(self):
|
||||
d = describe(parse("EMA(CLF-01.TEMP, 5m)"))
|
||||
self.assertIn("CLF-01.TEMP", d)
|
||||
self.assertIn("EMA", d)
|
||||
|
||||
def test_to_dict_roundtrip_shape(self):
|
||||
ast = parse("RateOfChange(炉压)")
|
||||
d = ast.to_dict()
|
||||
self.assertEqual(d["kind"], "op")
|
||||
self.assertEqual(d["name"], "RateOfChange")
|
||||
self.assertEqual(d["args"][0], {"kind": "tag", "name": "炉压"})
|
||||
|
||||
def test_window_seconds(self):
|
||||
w = Window(5.0, "m")
|
||||
self.assertEqual(w.seconds, 300)
|
||||
self.assertEqual(Window(2.0, "h").seconds, 7200)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,303 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""issue #34 Model Recipe 插件接口与样例协议 单元测试。
|
||||
|
||||
覆盖：
|
||||
* 内置四类 Recipe 已注册、字段合法；
|
||||
* build_model 跨主干（gbdt/dnn/lstm/gnn/stub）可构造、fit/predict 契约；
|
||||
* 插件注册（register_recipe / register_backbone）零改码扩展；
|
||||
* ModelRecipe 不可变 + to_dict/from_dict 往返；
|
||||
* 超参包校验（recipe_id / 必需特征 / 主干可构造性）；
|
||||
* 样例协议：树脂 + Ti 两套 Recipe 同框架均跑通（EPIC #5 验收口径）。
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 引导：挂载 model_framework 包（目录含连字符）
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
import _bootstrap # noqa: F401,E402
|
||||
|
||||
import unittest
|
||||
|
||||
from model_framework.model_recipe import ( # noqa: E402
|
||||
BACKBONES,
|
||||
RECIPE_KINDS,
|
||||
ModelRecipe,
|
||||
RecipeError,
|
||||
build_model,
|
||||
get_recipe,
|
||||
list_recipes,
|
||||
load_sample_recipe,
|
||||
register_backbone,
|
||||
register_recipe,
|
||||
validate_hyperparam_pack,
|
||||
)
|
||||
|
||||
|
||||
class TestBuiltinRecipes(unittest.TestCase):
|
||||
"""内置四类 Recipe 注册与字段合法性。"""
|
||||
|
||||
def test_four_builtin_recipes_registered(self):
|
||||
ids = {r["id"] for r in list_recipes()}
|
||||
for rid in (
|
||||
"quality_predict.default",
|
||||
"process_optimize.default",
|
||||
"anomaly_detect.default",
|
||||
"cross_process.default",
|
||||
):
|
||||
self.assertIn(rid, ids, f"缺少内置 Recipe {rid}")
|
||||
|
||||
def test_each_builtin_kind_covered(self):
|
||||
kinds = {get_recipe(rid).kind for rid in (
|
||||
"quality_predict.default",
|
||||
"process_optimize.default",
|
||||
"anomaly_detect.default",
|
||||
"cross_process.default",
|
||||
)}
|
||||
self.assertEqual(kinds, set(RECIPE_KINDS))
|
||||
|
||||
def test_backbone_registered(self):
|
||||
for name in ("gbdt", "dnn", "lstm", "gnn", "stub"):
|
||||
self.assertIn(name, BACKBONES, f"缺少内置主干 {name}")
|
||||
|
||||
|
||||
class TestModelRecipeDataclass(unittest.TestCase):
|
||||
"""ModelRecipe 不可变 + 序列化往返 + 校验。"""
|
||||
|
||||
def test_immutable(self):
|
||||
r = get_recipe("quality_predict.default")
|
||||
with self.assertRaises(Exception):
|
||||
r.id = "x" # type: ignore[misc]
|
||||
|
||||
def test_to_from_dict_roundtrip(self):
|
||||
r = get_recipe("quality_predict.default")
|
||||
d = r.to_dict()
|
||||
r2 = ModelRecipe.from_dict(d)
|
||||
self.assertEqual(r2.to_dict(), d)
|
||||
self.assertEqual(r2.id, r.id)
|
||||
self.assertEqual(r2.backbone, r.backbone)
|
||||
|
||||
def test_invalid_kind_rejected(self):
|
||||
with self.assertRaises(RecipeError):
|
||||
ModelRecipe(id="x.bad", kind="bogus", backbone="gbdt")
|
||||
|
||||
def test_unregistered_backbone_rejected(self):
|
||||
with self.assertRaises(RecipeError):
|
||||
ModelRecipe(id="x.nobackbone", kind="quality_predict", backbone="no-such")
|
||||
|
||||
def test_merged_hyperparams_override_wins(self):
|
||||
r = get_recipe("quality_predict.default")
|
||||
base = r.default_hyperparams
|
||||
merged = r.merged_hyperparams({"max_depth": 99})
|
||||
self.assertEqual(merged["max_depth"], 99)
|
||||
# 默认值未被污染
|
||||
self.assertEqual(base["max_depth"], 6)
|
||||
self.assertIn("eta", merged)
|
||||
|
||||
|
||||
class TestBuildModel(unittest.TestCase):
|
||||
"""build_model 跨主干构造 + fit/predict 契约。"""
|
||||
|
||||
def test_build_each_backbone(self):
|
||||
for rid, bb in (
|
||||
("quality_predict.default", "gbdt"),
|
||||
("process_optimize.default", "gbdt"),
|
||||
("anomaly_detect.default", "dnn"),
|
||||
("cross_process.default", "gnn"),
|
||||
):
|
||||
m = build_model(rid)
|
||||
self.assertEqual(m.backbone, bb)
|
||||
self.assertFalse(m.fitted)
|
||||
|
||||
def test_fit_then_predict_returns_correct_length(self):
|
||||
m = build_model("quality_predict.default")
|
||||
X = [[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]]
|
||||
y = [1.0, 2.0, 3.0]
|
||||
m.fit(X, y)
|
||||
self.assertTrue(m.fitted)
|
||||
pred = m.predict([[2.0, 3.0], [4.0, 5.0]])
|
||||
self.assertEqual(len(pred), 2)
|
||||
for v in pred:
|
||||
self.assertIsInstance(v, float)
|
||||
|
||||
def test_predict_before_fit_fails_closed(self):
|
||||
m = build_model("anomaly_detect.default")
|
||||
with self.assertRaises(RecipeError):
|
||||
m.predict([[1.0, 2.0]])
|
||||
|
||||
def test_unsupervised_fit_without_y(self):
|
||||
# anomaly_detect 主干应允许无 y 拟合
|
||||
m = build_model("anomaly_detect.default")
|
||||
m.fit([[1.0, 2.0], [3.0, 4.0]])
|
||||
self.assertTrue(m.fitted)
|
||||
out = m.predict([[1.0, 2.0]])
|
||||
self.assertEqual(len(out), 1)
|
||||
|
||||
def test_X_width_mismatch_rejected(self):
|
||||
m = build_model("quality_predict.default")
|
||||
with self.assertRaises(ValueError):
|
||||
m.fit([[1.0, 2.0], [3.0]], [1.0, 2.0])
|
||||
|
||||
def test_Xy_length_mismatch_rejected(self):
|
||||
m = build_model("quality_predict.default")
|
||||
with self.assertRaises(ValueError):
|
||||
m.fit([[1.0, 2.0], [3.0, 4.0]], [1.0])
|
||||
|
||||
def test_empty_X_rejected(self):
|
||||
m = build_model("quality_predict.default")
|
||||
with self.assertRaises(ValueError):
|
||||
m.fit([], [])
|
||||
|
||||
def test_handle_to_dict(self):
|
||||
m = build_model("quality_predict.default", {"max_depth": 7})
|
||||
d = m.to_dict()
|
||||
self.assertEqual(d["recipe_id"], "quality_predict.default")
|
||||
self.assertEqual(d["backbone"], "gbdt")
|
||||
self.assertEqual(d["hyperparams"]["max_depth"], 7)
|
||||
self.assertFalse(d["fitted"])
|
||||
|
||||
def test_unknown_recipe_raises(self):
|
||||
with self.assertRaises(RecipeError):
|
||||
build_model("no.such.recipe")
|
||||
|
||||
|
||||
class TestPluginRegistration(unittest.TestCase):
|
||||
"""register_recipe / register_backbone 零改码扩展（PRD「新增结构走插件注册」）。"""
|
||||
|
||||
def test_register_custom_backbone_and_recipe(self):
|
||||
seen = {}
|
||||
|
||||
def my_bb(hp):
|
||||
class _Impl:
|
||||
def iaop_fit(self, rows, y):
|
||||
seen["fit_called"] = True
|
||||
|
||||
def iaop_predict(self, rows):
|
||||
return [42.0 for _ in rows]
|
||||
return _Impl()
|
||||
|
||||
register_backbone("my-gnn", my_bb)
|
||||
self.assertIn("my-gnn", BACKBONES)
|
||||
|
||||
register_recipe(ModelRecipe(
|
||||
id="cross_process.custom_gnn",
|
||||
kind="cross_process",
|
||||
backbone="my-gnn",
|
||||
description="自研 GNN 主干，验证插件扩展",
|
||||
))
|
||||
m = build_model("cross_process.custom_gnn")
|
||||
m.fit([[1.0, 2.0]], [1.0])
|
||||
self.assertTrue(seen.get("fit_called"))
|
||||
self.assertEqual(m.predict([[9.0, 9.0]]), [42.0])
|
||||
|
||||
def test_register_recipe_overwrites(self):
|
||||
# 用独立的临时 recipe 验证"重复注册同 id 覆盖"，不污染内置表
|
||||
register_recipe(ModelRecipe(
|
||||
id="quality_predict.temp",
|
||||
kind="quality_predict",
|
||||
backbone="gbdt",
|
||||
description="第一版",
|
||||
))
|
||||
self.assertEqual(get_recipe("quality_predict.temp").description, "第一版")
|
||||
register_recipe(ModelRecipe(
|
||||
id="quality_predict.temp",
|
||||
kind="quality_predict",
|
||||
backbone="stub",
|
||||
description="第二版覆盖",
|
||||
))
|
||||
self.assertEqual(get_recipe("quality_predict.temp").backbone, "stub")
|
||||
self.assertEqual(get_recipe("quality_predict.temp").description, "第二版覆盖")
|
||||
|
||||
def test_register_invalid_backbone_name_rejected(self):
|
||||
with self.assertRaises(RecipeError):
|
||||
register_backbone("bad name!", lambda hp: None)
|
||||
|
||||
def test_register_non_callable_factory_rejected(self):
|
||||
with self.assertRaises(RecipeError):
|
||||
register_backbone("oops", "not callable") # type: ignore[arg-type]
|
||||
|
||||
def test_register_non_recipe_rejected(self):
|
||||
with self.assertRaises(RecipeError):
|
||||
register_recipe("not a recipe") # type: ignore[arg-type]
|
||||
|
||||
|
||||
class TestHyperparamPackValidation(unittest.TestCase):
|
||||
"""超参包校验（Recipe 视角）。"""
|
||||
|
||||
def test_valid_pack_no_issues(self):
|
||||
pack = load_sample_recipe("ti")
|
||||
self.assertEqual(validate_hyperparam_pack(pack), [])
|
||||
|
||||
def test_missing_required_field(self):
|
||||
issues = validate_hyperparam_pack({"recipe_id": "quality_predict.default"})
|
||||
msgs = " ".join(issues)
|
||||
self.assertIn("model_id", msgs)
|
||||
self.assertIn("features", msgs)
|
||||
|
||||
def test_unknown_recipe_id(self):
|
||||
issues = validate_hyperparam_pack({
|
||||
"model_id": "x", "recipe_id": "no.such", "features": [],
|
||||
})
|
||||
self.assertTrue(any("未注册" in i for i in issues))
|
||||
|
||||
def test_missing_required_feature(self):
|
||||
# quality_predict.default 要求 'target' 特征
|
||||
issues = validate_hyperparam_pack({
|
||||
"model_id": "x",
|
||||
"recipe_id": "quality_predict.default",
|
||||
"features": [{"name": "only_a"}],
|
||||
})
|
||||
self.assertTrue(any("target" in i for i in issues))
|
||||
|
||||
|
||||
class TestSampleRecipesAcceptance(unittest.TestCase):
|
||||
"""EPIC #5 / PRD 5.3 验收口径：同框架加载树脂与 Ti 两套 Recipe 均跑通。"""
|
||||
|
||||
def test_both_samples_build_fit_predict(self):
|
||||
for name in ("resin", "ti"):
|
||||
pack = load_sample_recipe(name)
|
||||
self.assertEqual(validate_hyperparam_pack(pack), [],
|
||||
f"样例 {name} 校验未通过")
|
||||
m = build_model(pack["recipe_id"], pack.get("hyperparams"))
|
||||
# 构造与目标维度无关的训练样本（2 特征列）
|
||||
X = [[float(i), float(i + 1)] for i in range(6)]
|
||||
y = [float(i) for i in range(6)]
|
||||
m.fit(X, y)
|
||||
self.assertTrue(m.fitted)
|
||||
pred = m.predict([[1.0, 2.0]])
|
||||
self.assertEqual(len(pred), 1)
|
||||
|
||||
def test_samples_share_same_framework(self):
|
||||
# 关键：两套样例用同一个 recipe_id（quality_predict.default），
|
||||
# 仅超参不同——证明「切换模板仅改超参包，模型代码零改动」
|
||||
r1 = load_sample_recipe("resin")
|
||||
r2 = load_sample_recipe("ti")
|
||||
self.assertEqual(r1["recipe_id"], r2["recipe_id"])
|
||||
# 但超参不同（max_depth 4 vs 6）
|
||||
self.assertNotEqual(
|
||||
r1["hyperparams"]["max_depth"],
|
||||
r2["hyperparams"]["max_depth"],
|
||||
)
|
||||
# 各自 build 得到不同超参的句柄
|
||||
m1 = build_model(r1["recipe_id"], r1["hyperparams"])
|
||||
m2 = build_model(r2["recipe_id"], r2["hyperparams"])
|
||||
self.assertEqual(m1.hyperparams["max_depth"], 4)
|
||||
self.assertEqual(m2.hyperparams["max_depth"], 6)
|
||||
|
||||
def test_load_unknown_sample_raises(self):
|
||||
with self.assertRaises(RecipeError):
|
||||
load_sample_recipe("bogus")
|
||||
|
||||
|
||||
class TestBackboneFallback(unittest.TestCase):
|
||||
"""主干在无第三方依赖时退化为 stub，接口契约不变。"""
|
||||
|
||||
def test_lstm_gnn_fallback_to_stub_contract(self):
|
||||
# 无论是否有 torch，lstm/gnn 主干都应能构造并 fit/predict
|
||||
for rid in ("cross_process.default",):
|
||||
m = build_model(rid)
|
||||
m.fit([[1.0, 2.0]], [1.0])
|
||||
self.assertEqual(len(m.predict([[1.0, 2.0]])), 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,301 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""训练 / 推理流水线编排单元测试（issue #40）。
|
||||
|
||||
覆盖：
|
||||
- Context 读写与快照；
|
||||
- ModelRegistry 注册 / 版本 / 别名（latest / stable）；
|
||||
- 估计器（MeanRegressor / MajorityClassifier）训练与预测；
|
||||
- 各 Step（LoadData / Train / Evaluate / Register / LoadModel / Predict / Custom）
|
||||
的执行与产物传递；
|
||||
- Pipeline 顺序编排、失败短路、dry_run；
|
||||
- PipelineConfig 声明式配置往返与 from_config 构建。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
if HERE not in sys.path:
|
||||
sys.path.insert(0, HERE)
|
||||
|
||||
import _bootstrap # noqa: E402 加载 model_framework 包
|
||||
|
||||
from model_framework.pipeline import ( # noqa: E402
|
||||
Context,
|
||||
CustomStep,
|
||||
ESTIMATORS,
|
||||
EvaluateStep,
|
||||
LoadDataStep,
|
||||
LoadModelStep,
|
||||
MajorityClassifier,
|
||||
MeanRegressor,
|
||||
ModelArtifact,
|
||||
ModelRegistry,
|
||||
Pipeline,
|
||||
PipelineConfig,
|
||||
PipelineError,
|
||||
PredictStep,
|
||||
RegisterStep,
|
||||
StepResult,
|
||||
TrainStep,
|
||||
register_estimator,
|
||||
register_step_type,
|
||||
)
|
||||
|
||||
|
||||
class TestContext(unittest.TestCase):
|
||||
def test_get_set(self):
|
||||
ctx = Context(params={"lr": 0.1})
|
||||
ctx.set("x", 1)
|
||||
self.assertEqual(ctx.get("x"), 1)
|
||||
self.assertEqual(ctx.get("missing", "d"), "d")
|
||||
self.assertEqual(ctx.params["lr"], 0.1)
|
||||
|
||||
def test_snapshot(self):
|
||||
ctx = Context()
|
||||
ctx.set("a", 1)
|
||||
ctx.set("b", 2)
|
||||
snap = ctx.snapshot()
|
||||
self.assertEqual(snap["artifacts_keys"], ["a", "b"])
|
||||
|
||||
|
||||
class TestModelRegistry(unittest.TestCase):
|
||||
def test_register_and_latest(self):
|
||||
reg = ModelRegistry()
|
||||
a1 = ModelArtifact("m", "v1", object())
|
||||
a2 = ModelArtifact("m", "v2", object())
|
||||
reg.register(a1)
|
||||
reg.register(a2)
|
||||
self.assertEqual(reg.get("m").version, "v2") # latest
|
||||
self.assertEqual(reg.get("m", "v1").version, "v1")
|
||||
self.assertEqual(reg.list_versions("m"), ["v1", "v2"])
|
||||
|
||||
def test_alias(self):
|
||||
reg = ModelRegistry()
|
||||
reg.register(ModelArtifact("m", "v1", object()))
|
||||
reg.register(ModelArtifact("m", "v2", object()))
|
||||
reg.set_alias("m", "stable", "v1")
|
||||
self.assertEqual(reg.get("m", "stable").version, "v1")
|
||||
self.assertEqual(reg.get("m", "latest").version, "v2")
|
||||
|
||||
def test_missing_raises(self):
|
||||
reg = ModelRegistry()
|
||||
with self.assertRaises(PipelineError):
|
||||
reg.get("nope")
|
||||
reg.register(ModelArtifact("m", "v1", object()))
|
||||
with self.assertRaises(PipelineError):
|
||||
reg.get("m", "v99")
|
||||
|
||||
def test_register_requires_name_version(self):
|
||||
reg = ModelRegistry()
|
||||
with self.assertRaises(PipelineError):
|
||||
reg.register(ModelArtifact("", "v1", object()))
|
||||
|
||||
|
||||
class TestEstimators(unittest.TestCase):
|
||||
def test_mean_regressor(self):
|
||||
est = MeanRegressor()
|
||||
est.fit([[1], [2], [3]], [10, 20, 30])
|
||||
self.assertEqual(est.predict([[99], [100]]), [20.0, 20.0])
|
||||
|
||||
def test_majority_classifier(self):
|
||||
est = MajorityClassifier()
|
||||
est.fit([[1], [2], [3]], [0, 1, 1])
|
||||
self.assertEqual(est.predict([[9], [10]]), [1.0, 1.0])
|
||||
|
||||
def test_empty_fit_raises(self):
|
||||
with self.assertRaises(PipelineError):
|
||||
MeanRegressor().fit([], [])
|
||||
|
||||
def test_register_estimator(self):
|
||||
class MyEst(MeanRegressor):
|
||||
name = "my_est"
|
||||
register_estimator("my_est", MyEst)
|
||||
self.assertIn("my_est", ESTIMATORS)
|
||||
|
||||
|
||||
class TestSteps(unittest.TestCase):
|
||||
def test_load_data_from_list(self):
|
||||
ctx = Context()
|
||||
r = LoadDataStep("load", {"source": [[1, 2], [3, 4]]}).execute(ctx)
|
||||
self.assertTrue(r.success)
|
||||
self.assertEqual(ctx.get("dataset"), [[1.0, 2.0], [3.0, 4.0]])
|
||||
|
||||
def test_load_data_from_csv(self):
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".csv", delete=False, encoding="utf-8") as fh:
|
||||
fh.write("a,b,y\n1,2,3\n4,5,6\n")
|
||||
path = fh.name
|
||||
try:
|
||||
ctx = Context()
|
||||
r = LoadDataStep("load", {"source": path}).execute(ctx)
|
||||
self.assertTrue(r.success)
|
||||
self.assertEqual(ctx.get("dataset"), [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_load_data_missing_source(self):
|
||||
ctx = Context()
|
||||
r = LoadDataStep("load", {}).execute(ctx)
|
||||
self.assertFalse(r.success)
|
||||
self.assertIn("source", r.error or "")
|
||||
|
||||
def test_train_step(self):
|
||||
ctx = Context()
|
||||
ctx.set("dataset", [[1, 10], [2, 20], [3, 30]]) # 最后一列 target
|
||||
r = TrainStep("train", {"estimator": "mean_regressor"}).execute(ctx)
|
||||
self.assertTrue(r.success)
|
||||
est = ctx.get("model")
|
||||
self.assertEqual(est.predict([[9]]), [20.0])
|
||||
|
||||
def test_train_unknown_estimator(self):
|
||||
ctx = Context()
|
||||
ctx.set("dataset", [[1, 10]])
|
||||
r = TrainStep("train", {"estimator": "voodoo"}).execute(ctx)
|
||||
self.assertFalse(r.success)
|
||||
|
||||
def test_evaluate_step_regression(self):
|
||||
ctx = Context()
|
||||
ctx.set("dataset", [[1, 10], [2, 20], [3, 30]])
|
||||
TrainStep("train", {}).execute(ctx)
|
||||
r = EvaluateStep("eval", {}).execute(ctx)
|
||||
self.assertTrue(r.success)
|
||||
m = ctx.get("metrics")
|
||||
# 均值预测：mae 为各 |y-20| 的均值
|
||||
self.assertAlmostEqual(m["mae"], (10 + 0 + 10) / 3)
|
||||
self.assertGreaterEqual(m["rmse"], 0)
|
||||
|
||||
def test_evaluate_step_classification(self):
|
||||
ctx = Context()
|
||||
ctx.set("dataset", [[1, 0], [2, 1], [3, 1]])
|
||||
TrainStep("train", {"estimator": "majority_classifier"}).execute(ctx)
|
||||
EvaluateStep("eval", {}).execute(ctx)
|
||||
m = ctx.get("metrics")
|
||||
self.assertIn("accuracy", m)
|
||||
self.assertGreaterEqual(m["accuracy"], 0.0)
|
||||
|
||||
def test_register_and_load_model(self):
|
||||
ctx = Context()
|
||||
ctx.set("dataset", [[1, 10], [2, 20]])
|
||||
TrainStep("train", {}).execute(ctx)
|
||||
reg_r = RegisterStep("reg", {"model_name": "demo", "version": "v1"}).execute(ctx)
|
||||
self.assertTrue(reg_r.success)
|
||||
registry = ctx.get("registry")
|
||||
self.assertIsInstance(registry, ModelRegistry)
|
||||
self.assertEqual(registry.list_versions("demo"), ["v1"])
|
||||
|
||||
load_r = LoadModelStep("load", {"model_name": "demo", "version": "v1"}).execute(ctx)
|
||||
self.assertTrue(load_r.success)
|
||||
serving = ctx.get("serving_model")
|
||||
self.assertEqual(serving.predict([[9]]), [15.0])
|
||||
|
||||
def test_load_model_missing_registry(self):
|
||||
ctx = Context()
|
||||
r = LoadModelStep("load", {"model_name": "x"}).execute(ctx)
|
||||
self.assertFalse(r.success)
|
||||
|
||||
def test_predict_step(self):
|
||||
ctx = Context()
|
||||
ctx.set("dataset", [[1, 10], [2, 20]])
|
||||
TrainStep("train", {}).execute(ctx)
|
||||
RegisterStep("reg", {"model_name": "demo", "version": "v1"}).execute(ctx)
|
||||
LoadModelStep("load", {"model_name": "demo"}).execute(ctx)
|
||||
ctx.set("input", [[5], [6]])
|
||||
r = PredictStep("predict", {}).execute(ctx)
|
||||
self.assertTrue(r.success)
|
||||
self.assertEqual(ctx.get("predictions"), [15.0, 15.0])
|
||||
|
||||
def test_custom_step(self):
|
||||
ctx = Context()
|
||||
r = CustomStep("c", {"handler": lambda c: {"out": 42}}).execute(ctx)
|
||||
self.assertTrue(r.success)
|
||||
self.assertEqual(ctx.get("out"), 42)
|
||||
|
||||
def test_custom_step_bad_handler(self):
|
||||
ctx = Context()
|
||||
r = CustomStep("c", {"handler": "not_callable"}).execute(ctx)
|
||||
self.assertFalse(r.success)
|
||||
|
||||
def test_step_requires_name(self):
|
||||
with self.assertRaises(PipelineError):
|
||||
TrainStep("", {})
|
||||
|
||||
|
||||
class TestPipeline(unittest.TestCase):
|
||||
def _full_pipeline(self):
|
||||
return Pipeline("demo", [
|
||||
LoadDataStep("load", {"source": [[1, 10], [2, 20], [3, 30]]}),
|
||||
TrainStep("train", {"estimator": "mean_regressor"}),
|
||||
EvaluateStep("eval", {}),
|
||||
RegisterStep("register", {"model_name": "demo", "version": "v1"}),
|
||||
LoadModelStep("load_model", {"model_name": "demo", "version": "v1"}),
|
||||
PredictStep("predict", {"input_key": "dataset"}),
|
||||
])
|
||||
|
||||
def test_full_pipeline_success(self):
|
||||
result = self._full_pipeline().run()
|
||||
self.assertTrue(result.success)
|
||||
self.assertEqual(len(result.step_results), 6)
|
||||
self.assertIsNone(result.failed_step)
|
||||
|
||||
def test_pipeline_context_shared(self):
|
||||
pipe = Pipeline("p", [
|
||||
CustomStep("a", {"handler": lambda c: {"v": 7}}),
|
||||
CustomStep("b", {"handler": lambda c: {"v2": c.get("v") * 2}}),
|
||||
])
|
||||
r = pipe.run()
|
||||
self.assertTrue(r.success)
|
||||
self.assertEqual(pipe is not None, True)
|
||||
|
||||
def test_pipeline_failure_short_circuits(self):
|
||||
# 第二步失败（无 dataset），应短路不执行后续
|
||||
pipe = Pipeline("p", [
|
||||
CustomStep("a", {"handler": lambda c: {}}),
|
||||
EvaluateStep("bad_eval", {}), # 缺 model → 失败
|
||||
CustomStep("c", {"handler": lambda c: {"never": 1}}),
|
||||
])
|
||||
r = pipe.run()
|
||||
self.assertFalse(r.success)
|
||||
self.assertEqual(r.failed_step, "bad_eval")
|
||||
self.assertEqual(len(r.step_results), 2) # a + bad_eval
|
||||
|
||||
def test_dry_run(self):
|
||||
r = self._full_pipeline().run(dry_run=True)
|
||||
self.assertTrue(r.success)
|
||||
# dry_run 不真正 run，predictions 不存在
|
||||
# （dry_run 不产出 artifacts）
|
||||
|
||||
def test_pipeline_requires_name(self):
|
||||
with self.assertRaises(PipelineError):
|
||||
Pipeline("", [])
|
||||
|
||||
def test_register_step_type_and_from_config(self):
|
||||
register_step_type("double", lambda n, p: CustomStep(
|
||||
n, {"handler": lambda c: {"doubled": c.params.get("x", 0) * 2}}))
|
||||
cfg = PipelineConfig("cfg", steps=[
|
||||
{"type": "double", "name": "d", "params": {}},
|
||||
], params={"x": 21})
|
||||
pipe = Pipeline.from_config(cfg)
|
||||
ctx = Context()
|
||||
r = pipe.run(ctx)
|
||||
self.assertTrue(r.success)
|
||||
self.assertEqual(ctx.get("doubled"), 42)
|
||||
|
||||
def test_from_config_unknown_type(self):
|
||||
cfg = PipelineConfig("cfg", steps=[{"type": "voodoo"}])
|
||||
with self.assertRaises(PipelineError):
|
||||
Pipeline.from_config(cfg)
|
||||
|
||||
def test_config_roundtrip(self):
|
||||
cfg = PipelineConfig("c", steps=[{"type": "train", "name": "t", "params": {}}],
|
||||
params={"k": 1})
|
||||
d = cfg.to_dict()
|
||||
cfg2 = PipelineConfig.from_dict(json.loads(json.dumps(d)))
|
||||
self.assertEqual(cfg2.name, "c")
|
||||
self.assertEqual(cfg2.params, {"k": 1})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,255 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""``quality_forecast`` 单元测试（issue #36）。
|
||||
|
||||
覆盖：
|
||||
- 配方（Recipe）不可变性 / 序列化往返 / 非法主干与越界校验；
|
||||
- 主干工厂注册表 + 自定义主干注册（PRD 5.3「新增结构走插件注册」）；
|
||||
- stub / gbdt / dnn 三类主干的 fit/predict/evaluate 契约；
|
||||
- 固定主干 + 配方加载：同框架加载 Ti / 树脂两套配方均跑通（PRD 5.3
|
||||
验收口径）；
|
||||
- Accuracy 验收口径（PRD 5.3 / 里程碑：准确率 ≥ 90%）；
|
||||
- 零外部强依赖：无 sklearn 时 stub 退化仍可加载与校验。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, HERE)
|
||||
import _bootstrap # noqa: E402 注册 model_framework 包
|
||||
|
||||
from model_framework.quality_forecast import ( # noqa: E402
|
||||
Accuracy,
|
||||
BACKBONES,
|
||||
ModelHandle,
|
||||
QualityForecastError,
|
||||
QualityForecastModel,
|
||||
Recipe,
|
||||
build_from_recipe,
|
||||
dnn_backbone,
|
||||
gbdt_backbone,
|
||||
list_sample_recipes,
|
||||
load_recipe,
|
||||
register_backbone,
|
||||
sample_recipe_path,
|
||||
stub_backbone,
|
||||
)
|
||||
|
||||
|
||||
def _linear_dataset(n=40, noise=0.0):
|
||||
"""构造一个 y ≈ 2*x0 + x1 的可学习数据集（带可选噪声）。"""
|
||||
X, y = [], []
|
||||
for i in range(n):
|
||||
x0 = float(i % 7) + 1.0
|
||||
x1 = float(i % 5) * 0.5 + 0.5
|
||||
yv = 2.0 * x0 + x1 + noise * (i % 3 - 1)
|
||||
X.append([x0, x1])
|
||||
y.append(yv)
|
||||
return X, y
|
||||
|
||||
|
||||
class TestRecipe(unittest.TestCase):
|
||||
"""配方数据对象与校验。"""
|
||||
|
||||
def test_defaults_and_immutability(self):
|
||||
r = Recipe(name="t")
|
||||
self.assertEqual(r.backbone, "gbdt")
|
||||
self.assertEqual(r.target_column, "quality_index")
|
||||
self.assertAlmostEqual(r.accuracy_floor, 0.90)
|
||||
with self.assertRaises(Exception):
|
||||
r.name = "other" # frozen
|
||||
|
||||
def test_roundtrip(self):
|
||||
r = Recipe(name="t", backbone="dnn",
|
||||
hyperparams={"max_iter": 50},
|
||||
feature_columns=("a", "b"),
|
||||
target_column="y",
|
||||
accuracy_floor=0.8, industry="树脂", notes="n")
|
||||
d = r.to_dict()
|
||||
r2 = Recipe.from_dict(d)
|
||||
self.assertEqual(r, r2)
|
||||
# JSON 往返
|
||||
r3 = Recipe.from_dict(json.loads(json.dumps(d)))
|
||||
self.assertEqual(r, r3)
|
||||
|
||||
def test_invalid_backbone_raises(self):
|
||||
with self.assertRaises(QualityForecastError):
|
||||
Recipe(name="t", backbone="svm")
|
||||
|
||||
def test_accuracy_floor_out_of_range(self):
|
||||
with self.assertRaises(QualityForecastError):
|
||||
Recipe(name="t", accuracy_floor=1.5)
|
||||
with self.assertRaises(QualityForecastError):
|
||||
Recipe(name="t", accuracy_floor=-0.1)
|
||||
|
||||
def test_missing_name(self):
|
||||
with self.assertRaises(QualityForecastError):
|
||||
Recipe(name="")
|
||||
|
||||
def test_load_recipe_from_file(self, ):
|
||||
path = sample_recipe_path("recipe.ti.json")
|
||||
r = load_recipe(path)
|
||||
self.assertEqual(r.name, "ti-cl4-quality")
|
||||
self.assertEqual(r.backbone, "gbdt")
|
||||
self.assertIn("furnace_temp", r.feature_columns)
|
||||
|
||||
|
||||
class TestBackbones(unittest.TestCase):
|
||||
"""主干工厂与注册表。"""
|
||||
|
||||
def test_builtin_backbones_registered(self):
|
||||
for name in ("gbdt", "dnn", "stub"):
|
||||
self.assertIn(name, BACKBONES)
|
||||
|
||||
def test_register_custom_backbone(self):
|
||||
class _Custom(ModelHandle):
|
||||
def __init__(self, p):
|
||||
super().__init__("custom", p)
|
||||
self._v = 1.0
|
||||
|
||||
def _fit_impl(self, X, y):
|
||||
self._v = sum(y) / len(y)
|
||||
|
||||
def _predict_one(self, row):
|
||||
return self._v
|
||||
|
||||
register_backbone("custom_test", lambda p: _Custom(p))
|
||||
m = QualityForecastModel(backbone="custom_test")
|
||||
X, y = _linear_dataset()
|
||||
m.fit(X, y)
|
||||
self.assertEqual(len(m.predict(X)), len(X))
|
||||
# 清理避免污染其它用例
|
||||
BACKBONES.pop("custom_test", None)
|
||||
|
||||
def test_unknown_backbone_raises(self):
|
||||
with self.assertRaises(QualityForecastError):
|
||||
QualityForecastModel(backbone="not_a_backbone")
|
||||
|
||||
def test_stub_predict_is_deterministic(self):
|
||||
h = stub_backbone({})
|
||||
X, y = _linear_dataset()
|
||||
h.fit(X, y)
|
||||
p1 = h.predict(X)
|
||||
p2 = h.predict(X)
|
||||
self.assertEqual(p1, p2)
|
||||
self.assertTrue(all(isinstance(v, float) for v in p1))
|
||||
|
||||
def test_gbdt_factory_runs_with_or_without_sklearn(self):
|
||||
# 无论 sklearn 是否存在都不应报错
|
||||
h = gbdt_backbone({"n_estimators": 20, "max_depth": 2})
|
||||
X, y = _linear_dataset()
|
||||
h.fit(X, y)
|
||||
preds = h.predict(X)
|
||||
self.assertEqual(len(preds), len(y))
|
||||
|
||||
|
||||
class TestModelContract(unittest.TestCase):
|
||||
"""模型 fit/predict/evaluate 契约。"""
|
||||
|
||||
def test_fit_predict_shapes(self):
|
||||
m = QualityForecastModel(backbone="stub")
|
||||
X, y = _linear_dataset(20)
|
||||
m.fit(X, y)
|
||||
self.assertTrue(m.fitted)
|
||||
preds = m.predict(X)
|
||||
self.assertEqual(len(preds), len(y))
|
||||
|
||||
def test_predict_before_fit_raises(self):
|
||||
m = QualityForecastModel(backbone="stub")
|
||||
with self.assertRaises(QualityForecastError):
|
||||
m.predict([[1.0, 2.0]])
|
||||
|
||||
def test_fit_mismatched_lengths_raises(self):
|
||||
m = QualityForecastModel(backbone="stub")
|
||||
with self.assertRaises(QualityForecastError):
|
||||
m.fit([[1.0], [2.0]], [1.0])
|
||||
|
||||
def test_fit_empty_raises(self):
|
||||
m = QualityForecastModel(backbone="stub")
|
||||
with self.assertRaises(QualityForecastError):
|
||||
m.fit([], [])
|
||||
|
||||
def test_to_dict_roundtrip_meta(self):
|
||||
m = QualityForecastModel(backbone="gbdt",
|
||||
hyperparams={"n_estimators": 5},
|
||||
feature_columns=["a"],
|
||||
target_column="y")
|
||||
d = m.to_dict()
|
||||
self.assertEqual(d["recipe_meta"]["backbone"], "gbdt")
|
||||
self.assertIn("handle", d)
|
||||
|
||||
|
||||
class TestAccuracy(unittest.TestCase):
|
||||
"""验收口径（PRD 5.3：准确率 ≥ 90%）。"""
|
||||
|
||||
def test_perfect_predictions_pass(self):
|
||||
y = [10.0, 20.0, 30.0, 40.0]
|
||||
acc = Accuracy.compute(y, y, accuracy_floor=0.9)
|
||||
self.assertAlmostEqual(acc.accuracy, 1.0)
|
||||
self.assertAlmostEqual(acc.mae, 0.0)
|
||||
self.assertAlmostEqual(acc.rmse, 0.0)
|
||||
self.assertTrue(acc.passed)
|
||||
|
||||
def test_bad_predictions_fail(self):
|
||||
y_true = [10.0, 20.0, 30.0, 40.0]
|
||||
y_pred = [11.0, 50.0, 5.0, 80.0] # 大偏差
|
||||
acc = Accuracy.compute(y_true, y_pred, accuracy_floor=0.9)
|
||||
self.assertLess(acc.accuracy, 0.9)
|
||||
self.assertFalse(acc.passed)
|
||||
self.assertGreater(acc.mae, 0.0)
|
||||
self.assertGreater(acc.rmse, 0.0)
|
||||
|
||||
def test_length_mismatch_raises(self):
|
||||
with self.assertRaises(QualityForecastError):
|
||||
Accuracy.compute([1.0, 2.0], [1.0])
|
||||
|
||||
def test_empty_raises(self):
|
||||
with self.assertRaises(QualityForecastError):
|
||||
Accuracy.compute([], [])
|
||||
|
||||
def test_evaluate_end_to_end(self):
|
||||
# stub 主干在确定性、低噪声线性数据上应能给出确定性的验收结果
|
||||
m = QualityForecastModel(backbone="stub", accuracy_floor=0.0)
|
||||
X, y = _linear_dataset(30)
|
||||
m.fit(X, y)
|
||||
acc = m.evaluate(X, y)
|
||||
self.assertIsInstance(acc, Accuracy)
|
||||
self.assertEqual(acc.to_dict()["accuracy_floor"], 0.0)
|
||||
|
||||
|
||||
class TestSampleRecipes(unittest.TestCase):
|
||||
"""样例协议：同框架加载 Ti / 树脂两套配方均跑通（PRD 5.3 验收口径）。"""
|
||||
|
||||
def test_samples_present(self):
|
||||
names = list_sample_recipes()
|
||||
self.assertIn("recipe.ti.json", names)
|
||||
self.assertIn("recipe.resin.json", names)
|
||||
|
||||
def test_build_from_each_sample_runs(self):
|
||||
for name in ("recipe.ti.json", "recipe.resin.json"):
|
||||
m = build_from_recipe(sample_recipe_path(name))
|
||||
self.assertIn(m.recipe_meta["backbone"], ("gbdt", "dnn", "stub"))
|
||||
# 用配方里声明的特征数构造一份演示数据跑通完整链路
|
||||
feat = m.recipe_meta["feature_columns"]
|
||||
n_feat = len(feat)
|
||||
self.assertGreater(n_feat, 0)
|
||||
X = [[float(i + j) for j in range(n_feat)] for i in range(12)]
|
||||
y = [float(i % 4) + 1.0 for i in range(12)]
|
||||
m.fit(X, y)
|
||||
preds = m.predict(X)
|
||||
self.assertEqual(len(preds), len(y))
|
||||
acc = m.evaluate(X, y)
|
||||
self.assertIsInstance(acc, Accuracy)
|
||||
|
||||
def test_two_recipes_share_same_code(self):
|
||||
"""切换模板仅改配方，模型代码零改动（PRD 5.3）。"""
|
||||
m1 = build_from_recipe(sample_recipe_path("recipe.ti.json"))
|
||||
m2 = build_from_recipe(sample_recipe_path("recipe.resin.json"))
|
||||
self.assertEqual(type(m1), type(m2))
|
||||
self.assertNotEqual(m1.recipe_meta.get("recipe_name"),
|
||||
m2.recipe_meta.get("recipe_name"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -0,0 +1,170 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""模型框架模板化 PoC 单元测试（issue #42）。
|
||||
|
||||
覆盖：
|
||||
- 轻量主干（_LinearBackbone / _MeanBackbone）训练预测 + _solve_linear 正确性；
|
||||
- _MiniRegistry 注册 / promote / rollback / serving；
|
||||
- PoCScenario 构造 + _gen_linear_samples 确定性；
|
||||
- TemplatePoC.run 端到端链路 + PoCReport 三条 RISK 验收口径（R1/R2/R3）；
|
||||
- 内置 Ti + 树脂场景跑通且 all_passed。
|
||||
"""
|
||||
import math
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
if HERE not in sys.path:
|
||||
sys.path.insert(0, HERE)
|
||||
|
||||
import _bootstrap # noqa: E402 加载 model_framework 包
|
||||
|
||||
from model_framework.template_poc import ( # noqa: E402
|
||||
PoCError,
|
||||
PoCReport,
|
||||
PoCScenario,
|
||||
TemplatePoC,
|
||||
resin_quality_scenario,
|
||||
run_poc,
|
||||
ti_quality_scenario,
|
||||
)
|
||||
from model_framework.template_poc import ( # noqa: E402
|
||||
BACKBONES,
|
||||
_Artifact,
|
||||
_gen_linear_samples,
|
||||
_LinearBackbone,
|
||||
_MeanBackbone,
|
||||
_MiniRegistry,
|
||||
_solve_linear,
|
||||
)
|
||||
|
||||
|
||||
class TestSolveLinear(unittest.TestCase):
|
||||
def test_simple(self):
|
||||
# 2x + 3y = 8; x - y = 1 => x=2.2, y=1.2
|
||||
w = _solve_linear([[2, 3], [1, -1]], [8, 1])
|
||||
self.assertAlmostEqual(w[0], 2.2, places=6)
|
||||
self.assertAlmostEqual(w[1], 1.2, places=6)
|
||||
|
||||
def test_identity(self):
|
||||
w = _solve_linear([[1, 0], [0, 1]], [3, 5])
|
||||
self.assertEqual(w, [3.0, 5.0])
|
||||
|
||||
|
||||
class TestBackbones(unittest.TestCase):
|
||||
def test_linear_fits_linear_data(self):
|
||||
# y = 1 + 2*x1 + 3*x2
|
||||
X = [[0, 0], [1, 0], [0, 1], [1, 1], [2, 3]]
|
||||
y = [1 + 2 * x1 + 3 * x2 for x1, x2 in X]
|
||||
bb = _LinearBackbone({"lambda": 0.0})
|
||||
bb.fit(X, y)
|
||||
preds = bb.predict([[1, 1], [2, 2]])
|
||||
self.assertAlmostEqual(preds[0], 6.0, places=4)
|
||||
self.assertAlmostEqual(preds[1], 11.0, places=4)
|
||||
|
||||
def test_mean_backbone(self):
|
||||
bb = _MeanBackbone({})
|
||||
bb.fit([[1], [2], [3]], [10, 20, 30])
|
||||
self.assertEqual(bb.predict([[9]]), [20.0])
|
||||
|
||||
def test_backbones_registered(self):
|
||||
self.assertIn("linear", BACKBONES)
|
||||
self.assertIn("mean", BACKBONES)
|
||||
|
||||
|
||||
class TestMiniRegistry(unittest.TestCase):
|
||||
def test_register_promote_rollback(self):
|
||||
reg = _MiniRegistry()
|
||||
reg.register(_Artifact("m", "v1", "linear", {"mae": 1.0}))
|
||||
self.assertEqual(reg.serving("m", "dev"), "v1")
|
||||
reg.promote("m", "v1") # dev->staging
|
||||
reg.promote("m", "v1") # staging->prod
|
||||
self.assertEqual(reg.serving("m", "prod"), "v1")
|
||||
reg.register(_Artifact("m", "v2", "linear", {"mae": 0.8}))
|
||||
reg.promote("m", "v2")
|
||||
reg.promote("m", "v2")
|
||||
reg.rollback("m", "prod", "v1")
|
||||
self.assertEqual(reg.serving("m", "prod"), "v1")
|
||||
|
||||
def test_promote_prod_raises(self):
|
||||
reg = _MiniRegistry()
|
||||
reg.register(_Artifact("m", "v1", "linear", {}))
|
||||
reg.promote("m", "v1")
|
||||
reg.promote("m", "v1")
|
||||
with self.assertRaises(PoCError):
|
||||
reg.promote("m", "v1")
|
||||
|
||||
|
||||
class TestGenSamples(unittest.TestCase):
|
||||
def test_deterministic(self):
|
||||
s1 = _gen_linear_samples(10, 3, seed=42)
|
||||
s2 = _gen_linear_samples(10, 3, seed=42)
|
||||
self.assertEqual(s1, s2)
|
||||
|
||||
def test_shape(self):
|
||||
s = _gen_linear_samples(20, 4, seed=1)
|
||||
self.assertEqual(len(s), 20)
|
||||
self.assertEqual(len(s[0]), 5) # 4 feat + 1 target
|
||||
|
||||
|
||||
class TestTemplatePoC(unittest.TestCase):
|
||||
def test_run_two_scenarios_all_passed(self):
|
||||
report = run_poc()
|
||||
self.assertIsInstance(report, PoCReport)
|
||||
self.assertEqual(len(report.scenario_results), 2)
|
||||
self.assertTrue(report.all_passed, report.summary())
|
||||
self.assertTrue(report.r1_precision_ok)
|
||||
self.assertTrue(report.r2_recipe_switch_ok)
|
||||
self.assertTrue(report.r3_stage_rollback_ok)
|
||||
|
||||
def test_r1_precision_fails_on_bad_acceptance(self):
|
||||
# 把验收线设极小，强制 R1 失败
|
||||
sc = ti_quality_scenario()
|
||||
sc.acceptance_mae = 0.0001 # 不可能达到
|
||||
report = TemplatePoC([sc]).run()
|
||||
self.assertFalse(report.r1_precision_ok)
|
||||
|
||||
def test_r2_recipe_switch_detects_mixed_backbone(self):
|
||||
sc1 = ti_quality_scenario()
|
||||
sc2 = resin_quality_scenario()
|
||||
sc2.backbone = "mean" # 故意用不同主干
|
||||
report = TemplatePoC([sc1, sc2]).run()
|
||||
self.assertFalse(report.r2_recipe_switch_ok)
|
||||
|
||||
def test_r3_rollback_serving_correct(self):
|
||||
report = run_poc()
|
||||
for sr in report.scenario_results:
|
||||
self.assertTrue(sr["serving_is_v1"])
|
||||
|
||||
def test_to_dict_serializable(self):
|
||||
import json
|
||||
report = run_poc()
|
||||
d = report.to_dict()
|
||||
json.dumps(d) # 可序列化
|
||||
self.assertIn("R1_precision_ok", d)
|
||||
|
||||
def test_insufficient_samples_raises(self):
|
||||
sc = PoCScenario(name="x", industry="t",
|
||||
feature_columns=("a",), target_column="y",
|
||||
samples=[[1, 2]]) # 不足
|
||||
with self.assertRaises(PoCError):
|
||||
TemplatePoC([sc]).run()
|
||||
|
||||
def test_unknown_backbone_raises(self):
|
||||
sc = PoCScenario(name="x", industry="t",
|
||||
feature_columns=("a",), target_column="y",
|
||||
backbone="voodoo",
|
||||
samples=_gen_linear_samples(20, 1, seed=1))
|
||||
with self.assertRaises(PoCError):
|
||||
TemplatePoC([sc]).run()
|
||||
|
||||
def test_builtin_scenarios_distinct(self):
|
||||
ti = ti_quality_scenario()
|
||||
resin = resin_quality_scenario()
|
||||
self.assertNotEqual(ti.feature_columns, resin.feature_columns)
|
||||
self.assertEqual(ti.backbone, resin.backbone) # 共用主干（R2）
|
||||
self.assertNotEqual(ti.hyperparams, resin.hyperparams) # 配方不同
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,235 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""模型模板注册 / 加载 / 版本机制单元测试（issue #41）。
|
||||
|
||||
覆盖：
|
||||
- 版本号语义校验 ``is_valid_version``；
|
||||
- Stage 枚举与 ``next_stage`` 阶段提升顺序；
|
||||
- ModelTemplate 构造校验（name/version/backbone/stage）+ 序列化往返；
|
||||
- TemplateRegistry：注册（拒重复 / force 覆盖）、加载（version/stage/默认）、
|
||||
阶段提升 promote、回滚 rollback、set_stage、查询（list_*）、审计日志、
|
||||
JSON 持久化 save/load 往返一致性。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
if HERE not in sys.path:
|
||||
sys.path.insert(0, HERE)
|
||||
|
||||
import _bootstrap # noqa: E402 加载 model_framework 包
|
||||
|
||||
from model_framework.template_registry import ( # noqa: E402
|
||||
ModelTemplate,
|
||||
Stage,
|
||||
TemplateRegistry,
|
||||
TemplateRegistryError,
|
||||
is_valid_version,
|
||||
next_stage,
|
||||
)
|
||||
|
||||
|
||||
class TestVersionValidation(unittest.TestCase):
|
||||
def test_valid_versions(self):
|
||||
for v in ["v1", "1.0", "v1.2.3", "1.2.3", "v1.0-rc1", "v2.0.0+build5"]:
|
||||
self.assertTrue(is_valid_version(v), f"应合法：{v}")
|
||||
|
||||
def test_invalid_versions(self):
|
||||
for v in ["", "v", "abc", "v1.x", "1..2", None, "v 1"]:
|
||||
self.assertFalse(is_valid_version(v), f"应非法：{v!r}")
|
||||
|
||||
|
||||
class TestStage(unittest.TestCase):
|
||||
def test_from_str(self):
|
||||
self.assertEqual(Stage.from_str("dev"), Stage.DEV)
|
||||
self.assertEqual(Stage.from_str("PROD"), Stage.PROD)
|
||||
|
||||
def test_from_str_invalid(self):
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
Stage.from_str("qa")
|
||||
|
||||
def test_next_stage(self):
|
||||
self.assertEqual(next_stage(Stage.DEV), Stage.STAGING)
|
||||
self.assertEqual(next_stage(Stage.STAGING), Stage.PROD)
|
||||
self.assertIsNone(next_stage(Stage.PROD))
|
||||
|
||||
|
||||
class TestModelTemplate(unittest.TestCase):
|
||||
def test_construct_minimal(self):
|
||||
t = ModelTemplate(name="m", version="v1")
|
||||
self.assertEqual(t.backbone, "generic")
|
||||
self.assertEqual(t.stage, Stage.DEV)
|
||||
|
||||
def test_rejects_empty_name(self):
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
ModelTemplate(name="", version="v1")
|
||||
|
||||
def test_rejects_bad_version(self):
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
ModelTemplate(name="m", version="abc")
|
||||
|
||||
def test_rejects_bad_backbone(self):
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
ModelTemplate(name="m", version="v1", backbone="magic")
|
||||
|
||||
def test_accepts_known_backbones(self):
|
||||
for b in ("quality_forecast", "anomaly_detection",
|
||||
"cross_process_opt", "recipe_opt", "generic"):
|
||||
ModelTemplate(name="m", version="v1", backbone=b)
|
||||
|
||||
def test_roundtrip(self):
|
||||
t = ModelTemplate(
|
||||
name="qa-model", version="v1.2.0", backbone="quality_forecast",
|
||||
hyperparams={"lr": 0.1}, feature_columns=("a", "b"),
|
||||
target_column="y", metrics={"accuracy": 0.93},
|
||||
stage="prod", description="d", extra={"k": "v"})
|
||||
t2 = ModelTemplate.from_dict(json.loads(json.dumps(t.to_dict())))
|
||||
self.assertEqual(t, t2)
|
||||
self.assertEqual(t2.stage, Stage.PROD)
|
||||
self.assertEqual(t2.metrics["accuracy"], 0.93)
|
||||
|
||||
|
||||
class TestRegistryRegisterLoad(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.reg = TemplateRegistry()
|
||||
self.t1 = ModelTemplate(name="m", version="v1", backbone="quality_forecast")
|
||||
self.t2 = ModelTemplate(name="m", version="v2", backbone="quality_forecast")
|
||||
|
||||
def test_register_and_get_by_version(self):
|
||||
self.reg.register(self.t1)
|
||||
self.assertEqual(self.reg.get("m", "v1").version, "v1")
|
||||
|
||||
def test_register_duplicate_rejected(self):
|
||||
self.reg.register(self.t1)
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
self.reg.register(self.t1)
|
||||
|
||||
def test_register_force_overwrites(self):
|
||||
self.reg.register(self.t1)
|
||||
t1_updated = ModelTemplate(
|
||||
name="m", version="v1", description="updated")
|
||||
self.reg.register(t1_updated, force=True)
|
||||
self.assertEqual(self.reg.get("m", "v1").description, "updated")
|
||||
|
||||
def test_get_missing_name(self):
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
self.reg.get("nope")
|
||||
|
||||
def test_get_missing_version(self):
|
||||
self.reg.register(self.t1)
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
self.reg.get("m", "v99")
|
||||
|
||||
def test_get_default_latest(self):
|
||||
self.reg.register(self.t1)
|
||||
time.sleep(0.01)
|
||||
self.reg.register(self.t2)
|
||||
self.assertEqual(self.reg.get("m").version, "v2")
|
||||
|
||||
def test_get_by_stage_pointer(self):
|
||||
self.reg.register(self.t1)
|
||||
# 新注册默认进 dev
|
||||
self.assertEqual(self.reg.get("m", stage=Stage.DEV).version, "v1")
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
self.reg.get("m", stage=Stage.PROD)
|
||||
|
||||
|
||||
class TestPromoteRollback(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.reg = TemplateRegistry()
|
||||
self.reg.register(ModelTemplate(name="m", version="v1"))
|
||||
self.reg.register(ModelTemplate(name="m", version="v2"))
|
||||
|
||||
def test_promote_chain(self):
|
||||
self.reg.promote("m", "v1") # dev -> staging
|
||||
self.assertEqual(self.reg.stage_pointer("m", Stage.STAGING), "v1")
|
||||
self.reg.promote("m", "v1") # staging -> prod
|
||||
self.assertEqual(self.reg.stage_pointer("m", Stage.PROD), "v1")
|
||||
|
||||
def test_promote_prod_raises(self):
|
||||
self.reg.promote("m", "v1")
|
||||
self.reg.promote("m", "v1") # 到 prod
|
||||
with self.assertRaises(TemplateRegistryError):
|
||||
self.reg.promote("m", "v1") # prod 无法继续
|
||||
|
||||
def test_rollback_stage_pointer(self):
|
||||
self.reg.promote("m", "v2") # v2 -> staging
|
||||
self.reg.promote("m", "v2") # v2 -> prod
|
||||
# 回滚 prod 到 v1
|
||||
self.reg.rollback("m", Stage.PROD, "v1")
|
||||
self.assertEqual(self.reg.stage_pointer("m", Stage.PROD), "v1")
|
||||
# v2 版本本身仍在（可审计）
|
||||
self.assertIn("v2", self.reg.list_versions("m"))
|
||||
|
||||
def test_set_stage_direct(self):
|
||||
self.reg.set_stage("m", "v1", Stage.PROD)
|
||||
self.assertEqual(self.reg.get("m", "v1").stage, Stage.PROD)
|
||||
self.assertEqual(self.reg.stage_pointer("m", Stage.PROD), "v1")
|
||||
|
||||
|
||||
class TestQueries(unittest.TestCase):
|
||||
def test_list_names_and_versions(self):
|
||||
reg = TemplateRegistry()
|
||||
reg.register(ModelTemplate(name="a", version="v1"))
|
||||
reg.register(ModelTemplate(name="a", version="v2"))
|
||||
reg.register(ModelTemplate(name="b", version="v1"))
|
||||
self.assertEqual(reg.list_names(), ["a", "b"])
|
||||
self.assertEqual(reg.list_versions("a"), ["v1", "v2"])
|
||||
self.assertIn("a", reg)
|
||||
self.assertNotIn("c", reg)
|
||||
self.assertEqual(len(reg), 3)
|
||||
|
||||
def test_list_by_stage(self):
|
||||
reg = TemplateRegistry()
|
||||
reg.register(ModelTemplate(name="m", version="v1"))
|
||||
reg.register(ModelTemplate(name="m", version="v2"))
|
||||
reg.promote("m", "v2") # v2 -> staging
|
||||
self.assertEqual(reg.list_by_stage("m", Stage.DEV), ["v1"])
|
||||
self.assertEqual(reg.list_by_stage("m", Stage.STAGING), ["v2"])
|
||||
|
||||
|
||||
class TestHistoryAndPersist(unittest.TestCase):
|
||||
def test_history_logged(self):
|
||||
reg = TemplateRegistry()
|
||||
reg.register(ModelTemplate(name="m", version="v1"))
|
||||
reg.promote("m", "v1")
|
||||
h = reg.history("m")
|
||||
actions = [e["action"] for e in h]
|
||||
self.assertIn("register", actions)
|
||||
self.assertIn("promote", actions)
|
||||
|
||||
def test_history_filter_by_name(self):
|
||||
reg = TemplateRegistry()
|
||||
reg.register(ModelTemplate(name="a", version="v1"))
|
||||
reg.register(ModelTemplate(name="b", version="v1"))
|
||||
self.assertEqual(len(reg.history("a")), 1)
|
||||
self.assertEqual(len(reg.history("b")), 1)
|
||||
|
||||
def test_save_load_roundtrip(self):
|
||||
reg = TemplateRegistry()
|
||||
reg.register(ModelTemplate(
|
||||
name="m", version="v1", backbone="quality_forecast",
|
||||
metrics={"accuracy": 0.9}, stage="dev"))
|
||||
reg.promote("m", "v1")
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".json", delete=False, encoding="utf-8") as fh:
|
||||
path = fh.name
|
||||
try:
|
||||
reg.save(path)
|
||||
reg2 = TemplateRegistry.load(path)
|
||||
self.assertEqual(reg2.list_names(), ["m"])
|
||||
self.assertEqual(reg2.get("m", "v1").backbone, "quality_forecast")
|
||||
self.assertEqual(reg2.get("m", "v1").metrics["accuracy"], 0.9)
|
||||
# 阶段指针恢复
|
||||
self.assertEqual(reg2.stage_pointer("m", Stage.STAGING), "v1")
|
||||
# 审计日志恢复
|
||||
self.assertTrue(len(reg2.history("m")) >= 2)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user