Merge PR #130 (feat #62-67 模板配置台全链路:RBAC/点位导入/配置/预览/发布推送 + #55 Ti 布局模板改进)

This commit is contained in:
2026-08-05 08:27:01 +08:00
parent 52d986e6f0
commit 6036e5e151
17 changed files with 3243 additions and 0 deletions
+82
View File
@@ -0,0 +1,82 @@
# ⑤.7 模板配置台(Template Console)内核引擎
> 父 EPIC:#9「⑤.7 模板配置台 Template Console」
> 子 issue:#62 / #63 / #64 / #65 / #66 / #67(同一 feature 分支 `feature/issue-62`,单 PR 关联全部 6 个 issue)
配置台是一个**无代码、配置驱动**的内核能力,让实施工程师(而非开发者)按现场调
模板:点位字典、模型超参、RAG、驾驶舱布局全部在配置台编排,预览确认后发布版本,
再把版本推送给内核(edge-gateway / rag-kb / model-framework)生效。本目录是配置台
的**纯标准库核心引擎**(不是 Web 前端——前端由 cockpit 渲染本引擎产出的结构化输出)。
## 为什么放在 `core/`?
配置台是**跨模板通用的内核能力**(RBAC / 配置存储 / 版本 / 推送契约服务于所有行业
模板:氯化 ti-cl4、树脂 resin、…),与 `core/edge-gateway`、`core/rag-kb`、
`core/model-framework` 同级,而非属于某个具体模板,故置于 `core/template-console/`。
## 6 个子任务映射
| issue | 模块 | 职责 |
|-------|------|------|
| #62 | `rbac.py` | 三级 RBAC(管理员 admin / 行业工程师 engineer / 只读 readonly),角色继承、`has_permission(resource, action)` 带理由判定、细粒度收窄 |
| #63 | `point_importer.py` | 点位字典 CSV 导入 + 自动校验页面。**复用** `core/edge-gateway/point_dict` 校验器(量纲/数据类型/采样率/重复点号/协议),增加 OPC 节点格式校验、表头列序校验、模板级量纲收窄(resin/ti)、行级结果聚合 |
| #64 | `config_store.py` | 配置项 CRUD(模型超参 / RAG / 布局三类),文件系统版本化 JSON 存储,list/get/upsert/delete + 按类别校验,原子写,快照 snapshot/restore |
| #65 | `preview.py` | 预览渲染引擎:布局(widget 卡片 + 网格占用率/越界/重叠检测)/ 告警(规则渲染 + 模拟触发评估)/ NL 查询(模板 → 示例问句)。对齐 `iAOP-cockpit-layout-v1` widget 类型 |
| #66 | `release.py` | 版本发布 + 回滚点。基于 `config_store` 快照的 Release,semver 单调递增校验,publish 固化快照、rollback 恢复快照(不删历史、回滚事件可追溯) |
| #67 | `push_channel.py` | 配置台↔内核配置推送契约。PushManifest(版本/快照/SHA256 校验和),PushChannel 模拟推送(写 manifest 到内核 inbox)、幂等(同版本不重复推送)、retract 撤回、verify 完整性校验 |
## 设计原则(对齐 PRD「可解释可溯源」与既有内核范式)
- **纯标准库零运行时依赖**:不 import pyyaml/numpy/pandas。需要哈希用 `hashlib`,
JSON 用 `json`,CSV 用 `csv`。
- **dataclass + Enum + 类型注解 + 中文 docstring**,与 `core/data-bus`、
`core/edge-gateway` 风格一致。
- **可解释性**:关键决策都带 `meaning` / `reason` 字段(RBAC 判定理据、配置项修改
原因、发布 changelog、回滚事件、推送日志),便于审计与配置台展示。
- **复用而非重造**:#63 直接复用 `core/edge-gateway/point_dict`(schema/loader/validator),
只增加配置台专属校验维度,避免与内核点位字典机制漂移。
## 目录结构
```
core/template-console/
├── __init__.py # 包入口(导出 RBAC 公共 API)
├── rbac.py # #62 三级 RBAC
├── point_importer.py # #63 点位字典 CSV 导入+校验
├── config_store.py # #64 配置项 CRUD 存储
├── preview.py # #65 预览渲染引擎
├── release.py # #66 版本发布+回滚
├── push_channel.py # #67 配置推送契约
├── _sanity_check.py # 离线基本校验(跑全部测试 + 冒烟)
├── README.md # 本文件
└── tests/
├── _bootstrap.py # 挂载 template_console 包 + 暴露 edge-gateway/point_dict
├── test_rbac.py
├── test_point_importer.py
├── test_config_store.py
├── test_preview.py
├── test_release.py
└── test_push_channel.py
```
## 运行测试
```bash
# 嵌入式 Python(无 pip/pyyaml)
/c/gitea/python312/python.exe -m unittest discover \
-s core/template-console/tests -p "test_*.py" -v
# 离线基本校验(跑全部测试 + 冒烟)
/c/gitea/python312/python.exe core/template-console/_sanity_check.py
```
## 数据流(配置台典型用例)
```
实施工程师导入点位字典(#63) ─┐
行业工程师调模型超参/RAG/布局(#64) ─┼─▶ 预览确认(#65) ─▶ 管理员发布版本(#66)
│ │
│ ▼
└──────────────────── 配置推送内核(#67) ─▶ edge-gateway/rag-kb/...
全程受三级 RBAC(#62) 权限管控;每次变更可解释、可溯源、可回滚。
```
+48
View File
@@ -0,0 +1,48 @@
# -*- coding: utf-8 -*-
"""⑤.7 模板配置台(Template Console)内核引擎 —— EPIC #9。
配置台是**跨模板通用的内核能力**:为实施工程师提供一个无代码的配置驱动
界面,把"模型超参 / RAG / 布局"三类配置 + 点位字典 + 版本发布统一编排,
并将发布的配置**推送给内核**(edge-gateway / rag-kb / model-framework)。
本包拆为 6 个子模块,对应 6 个 issue(同一 feature 分支承载,单 PR 关联):
- ``rbac`` (#62) 三级 RBAC 权限(admin / engineer / readonly);
- ``point_importer`` (#63) 点位字典 CSV 导入 + 自动校验页面(复用
``core/edge-gateway/point_dict`` 校验器,增加配置台
级结果聚合 + OPC 节点格式校验 + 模板选择);
- ``config_store`` (#64) 配置项 CRUD(模型超参 / RAG / 布局三类,文件系统
版本化 JSON 存储);
- ``preview`` (#65) 预览渲染引擎(布局/告警/查询 → 可预览结构化输出,
对齐 iAOP-cockpit-layout-v1 widget 类型);
- ``release`` (#66) 版本发布 + 回滚点(基于 config_store 快照,semver);
- ``push_channel`` (#67) 配置台↔内核配置推送契约(JSON manifest + 校验和 +
幂等性)。
设计原则(对齐 PRD「可解释可溯源」与既有内核范式):
- 纯标准库零运行时依赖(无 pyyaml/numpy/pandas),YAML 子集用内置解析器;
- dataclass + Enum + 类型注解 + 中文 docstring;
- 关键决策均带 ``meaning`` / ``reason`` 字段,便于审计与可解释性。
"""
from __future__ import annotations
from .rbac import (
Action,
Permission,
Role,
RoleKind,
User,
has_permission,
)
__all__ = [
"Action",
"Permission",
"Role",
"RoleKind",
"User",
"has_permission",
]
#: 本包版本(对齐 EPIC #9 模板配置台交付节奏)
__version__ = "1.0.0"
+139
View File
@@ -0,0 +1,139 @@
# -*- coding: utf-8 -*-
"""⑤.7 模板配置台离线基本校验(无构建环境下的离线验证)。
检查项:
1. 全部单元测试通过(unittest discover);
2. 冒烟:6 个子模块可导入;
3. 冒烟:端到端数据流跑通——
RBAC 判定(#62) → 点位导入(#63) → 配置 CRUD(#64) → 预览(#65)
→ 发布版本(#66) → 推送内核(#67) → 完整性校验通过。
用法:python _sanity_check.py
退出码:0 成功 / 1 失败。
"""
from __future__ import annotations
import os
import sys
import tempfile
import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
def _run_unit_tests() -> tuple[int, int]:
"""跑 tests/ 下全部测试,返回 (run, failures+errors)。"""
# 挂载 template_console 包 + edge-gateway/point_dict(同 tests/_bootstrap.py)
sys.path.insert(0, HERE)
import types
if "template_console" not in sys.modules:
pkg = types.ModuleType("template_console")
pkg.__path__ = [HERE]
sys.modules["template_console"] = pkg
edge_gw = os.path.join(os.path.dirname(HERE), "edge-gateway")
if os.path.isdir(edge_gw) and edge_gw not in sys.path:
sys.path.insert(0, edge_gw)
loader = unittest.TestLoader()
suite = loader.discover(os.path.join(HERE, "tests"), pattern="test_*.py")
runner = unittest.TextTestRunner(verbosity=1, stream=sys.stdout)
result = runner.run(suite)
return result.testsRun, len(result.failures) + len(result.errors)
def _smoke_flow() -> list[str]:
"""端到端冒烟:返回问题列表(空=通过)。"""
problems: list[str] = []
try:
from template_console.rbac import ( # type: ignore
Action, Resource, RoleKind, User, has_permission,
)
from template_console.point_importer import ( # type: ignore
TemplateKind, import_csv_string,
)
from template_console.config_store import ConfigKind, ConfigStore # type: ignore
from template_console.preview import preview_from_store, PreviewKind # type: ignore
from template_console.release import ReleaseManager # type: ignore
from template_console.push_channel import PushChannel # type: ignore
except Exception as exc: # noqa: BLE001
problems.append(f"模块导入失败:{exc}")
return problems
# 1) RBAC:admin 可发布,readonly 不可
admin = User("a", RoleKind.ADMIN)
viewer = User("v", RoleKind.READONLY)
if not has_permission(admin, Resource.RELEASE, Action.PUBLISH).allow:
problems.append("RBAC:admin 应能发布")
if has_permission(viewer, Resource.RELEASE, Action.PUBLISH).allow:
problems.append("RBAC:readonly 不应能发布")
# 2) 点位字典导入
csv_text = (
"device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\n"
"CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua\n"
)
_, rep = import_csv_string(csv_text, template=TemplateKind.TI)
if not rep.ok:
problems.append(f"点位导入应通过:{rep.summary()}")
# 3) 配置 CRUD + 4) 预览 + 5) 发布 + 6) 推送
tmp = tempfile.mkdtemp()
try:
store = ConfigStore(tmp)
store.upsert(ConfigKind.LAYOUT, "dashboard",
[{"type": "trend", "bind": "CLF-01.TEMP",
"x": 0, "y": 0, "w": 6, "h": 2}],
updated_by="li", reason="冒烟")
# 预览
prev = preview_from_store(store, PreviewKind.LAYOUT)
if not prev.items:
problems.append("预览:布局应渲染出 widget")
# 发布
rm = ReleaseManager(store)
rel = rm.publish("1.0.0", released_by="admin", changelog="冒烟发布")
if rm.latest().version != "1.0.0":
problems.append("发布:最新版本应为 1.0.0")
# 推送 + 完整性校验
inbox = os.path.join(tmp, "inbox")
ch = PushChannel(inbox=inbox)
manifest = ch.push(rel, pushed_by="admin")
if not PushChannel.verify(manifest):
problems.append("推送:manifest 完整性校验失败")
if not os.path.isfile(os.path.join(inbox, "manifest-1.0.0.json")):
problems.append("推送:manifest 文件未写入 inbox")
finally:
import shutil
shutil.rmtree(tmp, ignore_errors=True)
return problems
def main() -> int:
print("=" * 60)
print("⑤.7 模板配置台 离线基本校验")
print("=" * 60)
# 1) 单元测试
print("\n[1/2] 单元测试")
run_count, fail_count = _run_unit_tests()
if fail_count:
print(f"\nFAIL: 单元测试 {fail_count} 项失败(共 {run_count} 项)")
return 1
# 2) 冒烟
print("\n[2/2] 端到端冒烟")
problems = _smoke_flow()
if problems:
print("FAIL")
for p in problems:
print(" -", p)
return 1
print(f"\nOK: 单元测试 {run_count} 项全过;端到端冒烟通过(#62→#67 数据流正常)")
return 0
if __name__ == "__main__":
sys.exit(main())
+292
View File
@@ -0,0 +1,292 @@
# -*- coding: utf-8 -*-
"""⑤.7 配置项 CRUD 存储引擎 —— issue #64 / PRD ⑤.7。
配置台要管理三类业务配置:**模型超参 / RAG / 布局**。这些配置是模板交付物的
"活"部分——实施工程师按现场调参,每次改动都要**可解释、可校验、可版本化**
(为 #66 发布/回滚提供快照源)。本模块提供基于文件系统的版本化 JSON 存储:
- 三类配置各对应一个 JSON 文件(``model_params.json`` / ``rag_configs.json`` /
``layout.json``),存放在一个 store 根目录下;
- 每条配置项是一个 ``ConfigItem``(key + value + 含义 + 校验规则);
- 提供 ``list / get / upsert / delete`` CRUD,所有写操作都先**校验**再落盘,
并记录 ``updated_by`` / ``reason``(对齐 PRD「可解释可溯源」);
- 校验规则按类别内置(模型超参的范围/类型、RAG 的来源数、布局的 widget 类型),
非法值在 upsert 阶段即被拒绝,避免坏数据进入版本快照。
存储格式(每类一个 JSON,内容为 ``{items: [ConfigItem, ...], schema_version}``)
刻意简单、人可读,便于实施工程师直接查看/备份。
零运行时依赖:仅用 json / dataclass / Enum / 标准库。
"""
from __future__ import annotations
import json
import os
import re
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
# ---------------------------------------------------------------------------
# 配置类别
# ---------------------------------------------------------------------------
class ConfigKind(str, Enum):
"""三类业务配置(对齐 #64 需求)。"""
MODEL_PARAM = "model_param" # 模型超参(学习率/迭代数/特征开关…)
RAG_CONFIG = "rag_config" # RAG 知识库配置(top_k/相似度阈值/来源…)
LAYOUT = "layout" # 驾驶舱布局(widget 列表)
#: 各类别对应的存储文件名
KIND_FILENAME: Dict[ConfigKind, str] = {
ConfigKind.MODEL_PARAM: "model_params.json",
ConfigKind.RAG_CONFIG: "rag_configs.json",
ConfigKind.LAYOUT: "layout.json",
}
#: 存储结构版本(schema 演进时升级,发布快照会带上)
STORE_SCHEMA_VERSION = 1
#: 驾驶舱布局允许的 widget 类型(对齐 iAOP-cockpit-layout-v1 / resin cockpit)
ALLOWED_WIDGET_TYPES = {"process_view", "trend", "kpi_card", "alarm_panel", "nl_query"}
# ---------------------------------------------------------------------------
# 配置项数据模型
# ---------------------------------------------------------------------------
@dataclass
class ConfigItem:
"""一条配置项(可解释:带含义、更新人、原因)。"""
key: str # 配置键(类别内唯一,如 learning_rate)
value: Any # 配置值(标量或结构化)
kind: ConfigKind # 所属类别
meaning: str = "" # 业务含义(供配置台展示与审计)
updated_by: str = "system" # 最后修改人(对接 RBAC 用户名)
reason: str = "" # 本次修改原因(可解释可溯源)
updated_at: str = "" # ISO8601 时间戳
def to_dict(self) -> dict:
d = asdict(self)
d["kind"] = self.kind.value # 枚举序列化为字符串
return d
@classmethod
def from_dict(cls, raw: dict) -> "ConfigItem":
return cls(
key=raw["key"],
value=raw.get("value"),
kind=ConfigKind(raw.get("kind")),
meaning=raw.get("meaning", ""),
updated_by=raw.get("updated_by", "system"),
reason=raw.get("reason", ""),
updated_at=raw.get("updated_at", ""),
)
def _now_iso() -> str:
"""当前 UTC 时间 ISO8601(无时区歧义)。"""
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
# ---------------------------------------------------------------------------
# 校验(按类别内置规则)
# ---------------------------------------------------------------------------
@dataclass
class ValidationResult:
"""配置项校验结果。"""
ok: bool
errors: List[str] = field(default_factory=list)
def __bool__(self) -> bool:
return self.ok
def validate_item(kind: ConfigKind, key: str, value: Any) -> ValidationResult:
"""按类别校验配置项的 key/value 合法性。
校验规则(配置台 upsert 前置门禁,防止坏数据进快照):
- 通用:key 非空、匹配 ``[a-z0-9_.-]+``;
- model_param:value 为标量(int/float/bool/str)、标量列表,或结构化 dict
(如 optimizer 配置 / 告警规则等复合超参);
- rag_config:top_k 为 1~50 的正整数、similarity_threshold 为 0~1 浮点、
sources 为非空字符串列表;
- layout:value 为 widget 列表,每个 widget 有合法 type 与 x/y/w/h。
"""
errors: List[str] = []
if not key or not isinstance(key, str):
errors.append("key 不能为空")
elif not re.match(r"^[a-z0-9_.\-]+$", key):
errors.append(f"key '{key}' 仅允许小写字母/数字/._-")
if kind == ConfigKind.MODEL_PARAM:
if not isinstance(value, (int, float, bool, str, list, dict)):
errors.append("model_param 的 value 必须为标量/标量列表/结构化对象")
elif isinstance(value, list) and any(
not isinstance(v, (int, float, bool, str, dict)) for v in value):
errors.append("model_param 列表 value 仅允许标量或对象元素")
# 常见超参范围提示(软约束,仅对已知键)
if key == "learning_rate" and isinstance(value, (int, float)):
if not (0 < value < 1):
errors.append("learning_rate 应在 (0, 1) 区间")
if key == "iterations" and isinstance(value, int):
if value <= 0:
errors.append("iterations 必须为正整数")
elif kind == ConfigKind.RAG_CONFIG:
if key == "top_k":
if not (isinstance(value, int) and 1 <= value <= 50):
errors.append("top_k 必须为 1~50 的整数")
elif key == "similarity_threshold":
if not (isinstance(value, (int, float)) and 0 <= value <= 1):
errors.append("similarity_threshold 必须为 0~1 的数")
elif key == "sources":
if not (isinstance(value, list) and value
and all(isinstance(s, str) and s for s in value)):
errors.append("sources 必须为非空字符串列表")
elif kind == ConfigKind.LAYOUT:
if not isinstance(value, list):
errors.append("layout 的 value 必须为 widget 列表")
else:
for i, w in enumerate(value):
if not isinstance(w, dict):
errors.append(f"widget[{i}] 必须为对象")
continue
wt = w.get("type")
if wt not in ALLOWED_WIDGET_TYPES:
errors.append(
f"widget[{i}] 非法 type '{wt}'(合法:{sorted(ALLOWED_WIDGET_TYPES)})")
for coord in ("x", "y", "w", "h"):
if not isinstance(w.get(coord), int) or w.get(coord) < 0:
errors.append(f"widget[{i}] {coord} 必须为非负整数")
return ValidationResult(ok=not errors, errors=errors)
# ---------------------------------------------------------------------------
# 存储引擎
# ---------------------------------------------------------------------------
class ConfigStore:
"""基于文件系统的版本化配置存储(三类配置各一 JSON)。
用法:
store = ConfigStore("/path/to/store")
store.upsert(ConfigKind.MODEL_PARAM, "learning_rate", 0.001,
meaning="学习率", updated_by="li", reason="首次标定")
items = store.list(ConfigKind.MODEL_PARAM)
"""
def __init__(self, root: str) -> None:
self.root = root
os.makedirs(root, exist_ok=True)
# -- 路径 --
def _path(self, kind: ConfigKind) -> str:
return os.path.join(self.root, KIND_FILENAME[kind])
def _read(self, kind: ConfigKind) -> List[ConfigItem]:
path = self._path(kind)
if not os.path.isfile(path):
return []
with open(path, "r", encoding="utf-8") as fh:
blob = json.load(fh)
return [ConfigItem.from_dict(r) for r in blob.get("items", [])]
def _write(self, kind: ConfigKind, items: List[ConfigItem]) -> None:
blob = {
"schema_version": STORE_SCHEMA_VERSION,
"kind": kind.value,
"items": [it.to_dict() for it in items],
}
path = self._path(kind)
# 先写临时文件再替换,避免写一半被读到(原子写)
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(blob, fh, ensure_ascii=False, indent=2)
os.replace(tmp, path)
# -- 查询 --
def list(self, kind: ConfigKind) -> List[ConfigItem]:
"""列出某类全部配置项。"""
return self._read(kind)
def get(self, kind: ConfigKind, key: str) -> Optional[ConfigItem]:
"""取单条配置项(不存在返回 None)。"""
for it in self._read(kind):
if it.key == key:
return it
return None
# -- 写 --
def upsert(
self,
kind: ConfigKind,
key: str,
value: Any,
meaning: str = "",
updated_by: str = "system",
reason: str = "",
) -> ConfigItem:
"""新增或更新一条配置项(先校验,再落盘)。
Raises:
ValueError: 校验失败(带全部错误明细)。
"""
vr = validate_item(kind, key, value)
if not vr:
raise ValueError(f"配置项校验失败 [{kind.value}:{key}]:{'; '.join(vr.errors)}")
items = self._read(kind)
now = _now_iso()
existing_idx = next((i for i, it in enumerate(items) if it.key == key), None)
item = ConfigItem(
key=key, value=value, kind=kind, meaning=meaning,
updated_by=updated_by, reason=reason, updated_at=now,
)
if existing_idx is None:
items.append(item)
else:
items[existing_idx] = item
self._write(kind, items)
return item
def delete(self, kind: ConfigKind, key: str) -> bool:
"""删除一条配置项。返回是否实际删除。"""
items = self._read(kind)
new_items = [it for it in items if it.key != key]
if len(new_items) == len(items):
return False
self._write(kind, new_items)
return True
# -- 快照(供 #66 release 使用) --
def snapshot(self) -> Dict[str, Any]:
"""全量配置快照(三类聚合,供发布版本固化)。"""
return {
"schema_version": STORE_SCHEMA_VERSION,
"captured_at": _now_iso(),
"kinds": {
kind.value: [it.to_dict() for it in self._read(kind)]
for kind in ConfigKind
},
}
def restore(self, snapshot: Dict[str, Any]) -> None:
"""从快照恢复全部配置(#66 回滚入口)。"""
kinds = snapshot.get("kinds", {})
for kind in ConfigKind:
raw_items = kinds.get(kind.value, [])
items = [ConfigItem.from_dict(r) for r in raw_items]
self._write(kind, items)
def item_counts(self) -> Dict[str, int]:
"""各类配置项数量(配置台仪表盘用)。"""
return {kind.value: len(self._read(kind)) for kind in ConfigKind}
+318
View File
@@ -0,0 +1,318 @@
# -*- coding: utf-8 -*-
"""⑤.7 点位字典 CSV 导入 + 自动校验页面 —— issue #63 / PRD ⑤.7。
配置台的"导入页面"要解决:实施工程师拿着 DCS 点表(Excel 转 CSV)粘进配置台,
**一次性看到所有问题**(表头错/量纲错/重复点号/采样率非正/协议非法/OPC 节点
格式错),而不是改一条报一条。本模块是导入页面的后端引擎。
**复用而非重造**:点位字典的 schema/加载/校验(量纲/数据类型/采样率/重复点号/
协议)已由内核 ``core/edge-gateway/point_dict``(loader/validator/schema)实现
并被边缘网关正式使用。本模块在其基础上增加**配置台专属**校验维度:
1. OPC 节点格式校验(OPC UA 节点须形如 ``ns=<数字>;s=<名>`` 或 PLC 寄存器
``holding:<数字>`` / ``coil:<数字>``,与 simulator/opcua 驱动约定一致);
2. 表头列顺序严格对齐(实施工程师照表填列,列序错位是高频错误);
3. 行级结果聚合为 ``ImportRowIssue``(行号 + 严重级别 + 问题 + 修复建议),
供配置台前端逐行渲染、按严重级别过滤;
4. 模板选择(resin/ti):不同行业模板的合法量纲集合不同(如树脂含 rpm/mmol·g⁻¹),
导入时按模板切换校验基线。
零运行时依赖:复用 ``point_dict`` 子包(纯标准库 csv/dataclass)。
"""
from __future__ import annotations
import os
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Tuple
# 复用内核 edge-gateway 的点位字典加载/校验(schema/validator/loader)
# 在 tests/_bootstrap.py 中已把 core/edge-gateway 加入 sys.path,
# 故此处以顶层包 point_dict 引用(与 edge-gateway 自身测试一致)。
from point_dict import ( # noqa: E402
CSV_HEADERS,
VALID_PROTOCOLS,
VALID_UNITS,
Point,
PointDict,
load_point_dict_csv,
)
from point_dict import schema as _pd_schema # noqa: E402
class Severity(str, Enum):
"""问题严重级别(配置台前端据此着色/过滤)。"""
ERROR = "error" # 阻断:不修复无法入库
WARN = "warn" # 警告:可入库但建议复核(如 OPC 节点为空)
class TemplateKind(str, Enum):
"""行业模板(决定合法量纲等校验基线)。"""
TI = "ti" # 氯化/化工通用(templates/ti-cl4)
RESIN = "resin" # 吸附树脂(templates/resin,含 rpm / mmol/g)
# OPC 节点格式(与 simulator/opcua/s7 驱动约定一致):
# ns=2;s=CLF.Temp —— OPC UA 节点(namespace + 字符串 id)
# holding:40010 / coil:1 —— Modbus 寄存器(保持/线圈 + 地址)
_OPC_UA_RE = re.compile(r"^ns=\d+;s=[^\s,]+$")
_MODBUS_RE = re.compile(r"^(holding|coil|input|discrete):(\d+)$")
def _unit_set(template: TemplateKind) -> set:
"""按模板返回合法量纲集合(树脂含 rpm/mmol·g⁻¹ 等扩展)。"""
base = set(VALID_UNITS)
if template == TemplateKind.RESIN:
# VALID_UNITS 已含树脂扩展(rpm/mmol/g),直接复用
return base
# ti 模板:移除树脂专属量纲,避免化工模板误用树脂量纲
base.discard("rpm")
base.discard("mmol/g")
return base
@dataclass
class ImportRowIssue:
"""导入页一行的问题(行号 + 严重级别 + 问题 + 修复建议,可解释)。"""
row: int # CSV 行号(表头=1,数据从 2 起)
severity: Severity
code: str # 错误码(对齐 point_dict.validator 的 code + 本模块扩展)
message: str # 问题描述
suggestion: str = "" # 修复建议(供配置台"一键修复"提示)
def to_dict(self) -> dict:
return {
"row": self.row, "severity": self.severity.value,
"code": self.code, "message": self.message,
"suggestion": self.suggestion,
}
@dataclass
class ImportReport:
"""导入校验报告(配置台导入页面数据模型)。"""
template: TemplateKind
total_rows: int = 0 # 数据行数
issues: List[ImportRowIssue] = field(default_factory=list)
loaded_points: int = 0 # 成功加载的点数
file_path: str = ""
@property
def ok(self) -> bool:
"""无 ERROR 级问题即可入库(WARN 不阻断)。"""
return not any(i.severity == Severity.ERROR for i in self.issues)
@property
def error_count(self) -> int:
return sum(1 for i in self.issues if i.severity == Severity.ERROR)
@property
def warn_count(self) -> int:
return sum(1 for i in self.issues if i.severity == Severity.WARN)
def summary(self) -> str:
"""人类可读汇总(配置台导入结果横幅)。"""
status = "通过" if self.ok else "未通过"
return (f"导入校验{status}:{self.loaded_points} 点 / "
f"{self.total_rows} 行,错误 {self.error_count},警告 {self.warn_count}")
def to_dict(self) -> dict:
return {
"template": self.template.value,
"total_rows": self.total_rows,
"loaded_points": self.loaded_points,
"ok": self.ok,
"error_count": self.error_count,
"warn_count": self.warn_count,
"summary": self.summary(),
"issues": [i.to_dict() for i in self.issues],
}
# ---------------------------------------------------------------------------
# 校验扩展
# ---------------------------------------------------------------------------
def _validate_opc_node(point: Point) -> List[ImportRowIssue]:
"""OPC 节点格式校验(配置台扩展维度)。
约定(与驱动注册表对齐):
- 协议 opcua:节点须匹配 ``ns=<数字>;s=<名>``;
- 协议 modbus:节点须匹配 ``holding/coil/input/discrete:<数字>``;
- 协议 simulator:节点可空,或任意上述格式(演示用,宽松);
- 节点为空:WARN(可入库但运行时无法采集,建议补全)。
"""
out: List[ImportRowIssue] = []
node = (point.opc_node or "").strip()
if not node:
out.append(ImportRowIssue(
row=point.row_number, severity=Severity.WARN, code="empty_opc_node",
message=f"第{point.row_number}行 opcNode 为空",
suggestion="运行时无法采集,建议补全 OPC UA 节点或 PLC 寄存器地址",
))
return out
proto = (point.protocol or "").lower()
ok_ua = bool(_OPC_UA_RE.match(node))
ok_mb = bool(_MODBUS_RE.match(node))
if proto == "opcua" and not ok_ua:
out.append(ImportRowIssue(
row=point.row_number, severity=Severity.ERROR, code="bad_opc_node",
message=(f"第{point.row_number}行 opcNode '{node}' 不符合 OPC UA "
f"格式 ns=<ns>;s=<name>"),
suggestion="示例:ns=2;s=CLF.Temp",
))
elif proto == "modbus" and not ok_mb:
out.append(ImportRowIssue(
row=point.row_number, severity=Severity.ERROR, code="bad_opc_node",
message=(f"第{point.row_number}行 opcNode '{node}' 不符合 Modbus "
f"格式 holding/coil/input/discrete:<addr>"),
suggestion="示例:holding:40010",
))
elif proto not in ("opcua", "modbus") and not (ok_ua or ok_mb):
# simulator/s7/... 节点为空已 WARN;非空但格式都不符则 WARN(宽松)
out.append(ImportRowIssue(
row=point.row_number, severity=Severity.WARN, code="bad_opc_node",
message=(f"第{point.row_number}行 opcNode '{node}' 既非 OPC UA 也非 "
f"Modbus 格式"),
suggestion="确认节点格式或清空(演示协议可空)",
))
return out
def _validate_header_order(headers: List[str]) -> List[ImportRowIssue]:
"""表头列顺序严格对齐(列序错位是实施工程师高频错误)。"""
out: List[ImportRowIssue] = []
if not headers:
out.append(ImportRowIssue(
row=1, severity=Severity.ERROR, code="empty_header",
message="CSV 缺少表头行",
suggestion=f"表头应为:{','.join(CSV_HEADERS)}",
))
return out
missing = [h for h in CSV_HEADERS if h not in headers]
for h in missing:
out.append(ImportRowIssue(
row=1, severity=Severity.ERROR, code="missing_column",
message=f"表头缺少必填列:{h}",
suggestion=f"补列 {h}(完整表头:{','.join(CSV_HEADERS)})",
))
if headers[: len(CSV_HEADERS)] != CSV_HEADERS and not missing:
out.append(ImportRowIssue(
row=1, severity=Severity.WARN, code="bad_column_order",
message=f"表头列顺序与标准不一致:{headers}",
suggestion=f"标准顺序:{','.join(CSV_HEADERS)}",
))
return out
def _convert_validator_issues(report: "object", severity_for: Dict[str, Severity]) -> List[ImportRowIssue]:
"""把内核 validator.ValidationReport.issues 转成 ImportRowIssue。"""
out: List[ImportRowIssue] = []
for it in getattr(report, "issues", []):
sev = severity_for.get(it.code, Severity.ERROR)
out.append(ImportRowIssue(
row=it.row, severity=sev, code=it.code, message=it.message,
))
return out
# 内核 validator 错误码 → 严重级别映射
_SEVERITY_MAP: Dict[str, Severity] = {
"missing_column": Severity.ERROR,
"missing_field": Severity.ERROR,
"bad_unit": Severity.ERROR,
"bad_data_type": Severity.ERROR,
"bad_sample_rate": Severity.ERROR,
"bad_protocol": Severity.ERROR,
"dup_point": Severity.ERROR,
}
# ---------------------------------------------------------------------------
# 导入入口
# ---------------------------------------------------------------------------
def import_csv(
path: str,
template: TemplateKind = TemplateKind.TI,
extra_unit_check: bool = True,
) -> Tuple[PointDict, ImportReport]:
"""导入并校验点位字典 CSV(配置台导入页面后端入口)。
Args:
path: CSV 文件路径(UTF-8,9 列表头);
template: 行业模板(决定合法量纲集合,resin/ti);
extra_unit_check: 是否按模板收窄量纲集合做额外校验。
Returns:
(PointDict, ImportReport):加载的点位模型 + 校验报告。
报告 ``ok`` 为 True 即可入库;WARN 不阻断。
"""
report = ImportReport(template=template, file_path=path)
# 1) 表头校验(先读表头行)
import csv as _csv
with open(path, "r", encoding="utf-8-sig") as fh:
reader = _csv.reader(fh)
rows = list(reader)
headers = [c.strip() for c in rows[0]] if rows else []
report.issues.extend(_validate_header_order(headers))
# 2) 加载 + 内核校验(量纲/数据类型/采样率/重复点号/协议)
point_dict = load_point_dict_csv(path)
report.loaded_points = len(point_dict)
report.total_rows = len(point_dict.points)
from point_dict.validator import validate_point_dict
kernel_report = validate_point_dict(point_dict, headers)
report.issues.extend(_convert_validator_issues(kernel_report, _SEVERITY_MAP))
# 3) 模板级量纲收窄(resin 才允许 rpm/mmol·g⁻¹)
if extra_unit_check:
allowed_units = _unit_set(template)
for p in point_dict.points:
if p.unit and p.unit not in allowed_units:
# 内核 validator 已按全集校验过;这里只补充模板级差异提示
if p.unit in ("rpm", "mmol/g") and template == TemplateKind.TI:
report.issues.append(ImportRowIssue(
row=p.row_number, severity=Severity.ERROR,
code="template_unit_mismatch",
message=(f"第{p.row_number}行 量纲 '{p.unit}' 为树脂模板专属,"
f"当前导入的是 {template.value} 模板"),
suggestion="切换模板为 resin,或修正量纲",
))
# 4) OPC 节点格式校验(配置台扩展)
for p in point_dict.points:
report.issues.extend(_validate_opc_node(p))
# 行号排序,便于配置台逐行展示
report.issues.sort(key=lambda i: (i.row, i.code))
return point_dict, report
def import_csv_string(
content: str,
template: TemplateKind = TemplateKind.TI,
encoding: str = "utf-8",
) -> Tuple[PointDict, ImportReport]:
"""从 CSV 文本导入(配置台粘贴框场景,落临时文件后复用 import_csv)。"""
import tempfile
tmp = tempfile.NamedTemporaryFile(
mode="w", encoding=encoding, suffix=".csv", delete=False)
try:
tmp.write(content)
tmp.flush()
tmp.close()
return import_csv(tmp.name, template=template)
finally:
try:
os.unlink(tmp.name)
except OSError:
pass
+329
View File
@@ -0,0 +1,329 @@
# -*- coding: utf-8 -*-
"""⑤.7 配置预览渲染引擎 —— issue #65 / PRD ⑤.7。
配置台让实施工程师"边配边看":改完布局/告警/查询配置后,立即在预览区看到
驾驶舱会变成什么样、告警会怎么触发、NL 查询会怎么响应——**不必发布到生产
就能确认效果**。本模块是预览区的渲染后端,把 ``ConfigStore`` 里的配置渲染
为**结构化的预览片段**(dict/JSON),对齐 ``iAOP-cockpit-layout-v1`` 的
widget 类型与既有驾驶舱资产(resin cockpit)。
三类预览:
- **布局预览**(``render_layout_preview``):把 layout widget 列表渲染为
带占位网格坐标的 widget 描述(type/bind/metric/description + x/y/w/h),
计算网格占用率(发现越界/重叠);
- **告警预览**(``render_alarm_preview``):把告警规则(point + 阈值 + 级别)
渲染为"当 X 超过 Y 时,触发 级别 告警"的可读条目 + 模拟评估(给定当前值
是否触发);
- **NL 查询预览**(``render_nl_query_preview``):把 NL 查询模板渲染为示例
问答对(模板 × 示例槽位 → 渲染后的问句 + 预期数据来源)。
预览是**只读、无副作用**的——只读配置、产出结构化输出,不改任何状态,
对齐 PRD「预演不污染生产」。
零运行时依赖:仅用 dataclass / Enum / 标准库。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
from .config_store import ALLOWED_WIDGET_TYPES, ConfigKind, ConfigStore
# ---------------------------------------------------------------------------
# 预览模型
# ---------------------------------------------------------------------------
class PreviewKind(str, Enum):
"""三类预览。"""
LAYOUT = "layout"
ALARM = "alarm"
NL_QUERY = "nl_query"
@dataclass
class PreviewResult:
"""一次预览渲染的结果(结构化片段 + 说明 + 问题提示)。"""
kind: PreviewKind
title: str
items: List[Dict[str, Any]] = field(default_factory=list) # 渲染后的条目
notes: List[str] = field(default_factory=list) # 说明 / 渲染提示
warnings: List[str] = field(default_factory=list) # 布局越界/重叠等
reason: str = "" # 本次预览的来源说明
@property
def ok(self) -> bool:
return not self.warnings
def to_dict(self) -> dict:
return {
"kind": self.kind.value,
"title": self.title,
"items": self.items,
"notes": self.notes,
"warnings": self.warnings,
"reason": self.reason,
"ok": self.ok,
}
# ---------------------------------------------------------------------------
# 布局预览
# ---------------------------------------------------------------------------
#: 驾驶舱网格规格(对齐 resin cockpit:12 列 × 若干行,w/h 以网格单元计)
GRID_COLUMNS = 12
def render_layout_preview(
widgets: List[Dict[str, Any]],
title: str = "驾驶舱布局预览",
grid_columns: int = GRID_COLUMNS,
) -> PreviewResult:
"""渲染布局 widget 列表为预览片段。
每个 widget 渲染为带 type/描述/坐标的卡片;同时做**布局体检**:
- 越界(x+w 超出列数 / y+h 超出合理行数);
- 重叠(两个 widget 矩形相交);
- 非法类型(不在 ``ALLOWED_WIDGET_TYPES``)。
"""
result = PreviewResult(kind=PreviewKind.LAYOUT, title=title,
reason=f"渲染 {len(widgets)} 个 widget")
seen_rects: List[Dict[str, int]] = []
for i, w in enumerate(widgets):
wt = w.get("type")
x, y = w.get("x", 0), w.get("y", 0)
ww, hh = w.get("w", 0), w.get("h", 0)
card: Dict[str, Any] = {
"index": i,
"type": wt,
"x": x, "y": y, "w": ww, "h": hh,
"description": w.get("description", ""),
}
# 携带业务绑定(trend.bind / kpi_card.metric)
if wt == "trend":
card["bind"] = w.get("bind", "")
elif wt == "kpi_card":
card["metric"] = w.get("metric", "")
card["label"] = w.get("label", "")
elif wt == "process_view":
card["src"] = w.get("src", "")
result.items.append(card)
# 体检:类型合法
if wt not in ALLOWED_WIDGET_TYPES:
result.warnings.append(f"widget[{i}] 非法类型 '{wt}'")
# 越界
if x < 0 or y < 0 or ww <= 0 or hh <= 0:
result.warnings.append(f"widget[{i}] 坐标/尺寸非法 ({x},{y},{ww},{hh})")
elif x + ww > grid_columns:
result.warnings.append(
f"widget[{i}] 越界:x+w={x + ww} > {grid_columns} 列")
else:
# 重叠检测(矩形相交)
rect = {"x": x, "y": y, "w": ww, "h": hh}
for j, prev in enumerate(seen_rects):
if _rects_overlap(rect, prev):
result.warnings.append(f"widget[{i}] 与 widget[{j}] 重叠")
seen_rects.append(rect)
# 网格占用率
total_area = sum(r["w"] * r["h"] for r in seen_rects)
max_row = max((r["y"] + r["h"] for r in seen_rects), default=0)
grid_area = grid_columns * max(max_row, 1)
usage = round(total_area / grid_area * 100, 1) if grid_area else 0.0
result.notes.append(f"网格占用率 {usage}%({grid_columns} 列,最大 {max_row} 行)")
return result
def _rects_overlap(a: Dict[str, int], b: Dict[str, int]) -> bool:
"""两个网格矩形是否相交(不含边界共享视为不重叠)。"""
ax2, ay2 = a["x"] + a["w"], a["y"] + a["h"]
bx2, by2 = b["x"] + b["w"], b["y"] + b["h"]
return not (ax2 <= b["x"] or bx2 <= a["x"] or ay2 <= b["y"] or by2 <= a["y"])
# ---------------------------------------------------------------------------
# 告警预览
# ---------------------------------------------------------------------------
#: 合法的告警级别(对齐 cockpit alarm_panel)
ALARM_LEVELS = {"info", "warn", "critical"}
@dataclass
class AlarmRule:
"""一条告警规则(供告警预览渲染与模拟评估)。"""
point_id: str # 关联测点
metric: str # 指标名(展示用)
operator: str # 比较运算符 > / >= / < / <= / ==
threshold: float # 阈值
level: str = "warn" # 告警级别 info/warn/critical
message: str = "" # 告警文案模板(可含 {value})
def evaluate(self, value: float) -> bool:
"""给定当前值,判断是否触发告警。"""
ops = {
">": value > self.threshold,
">=": value >= self.threshold,
"<": value < self.threshold,
"<=": value <= self.threshold,
"==": value == self.threshold,
}
return ops.get(self.operator, False)
def render_alarm_preview(
rules: List[AlarmRule],
current_values: Optional[Dict[str, float]] = None,
title: str = "告警规则预览",
) -> PreviewResult:
"""渲染告警规则为可读条目,并用当前值模拟触发评估。
Args:
rules: 告警规则列表;
current_values: 当前测点值(point_id → value),用于模拟评估;
不提供则只渲染规则、不做触发评估。
"""
result = PreviewResult(kind=PreviewKind.ALARM, title=title,
reason=f"渲染 {len(rules)} 条告警规则")
for r in rules:
if r.level not in ALARM_LEVELS:
result.warnings.append(f"告警 '{r.point_id}' 非法级别 '{r.level}'")
if r.operator not in (">", ">=", "<", "<=", "=="):
result.warnings.append(f"告警 '{r.point_id}' 非法运算符 '{r.operator}'")
text = (f"当 {r.metric}({r.point_id}) {r.operator} {r.threshold} 时,"
f"触发 [{r.level}] 告警")
entry: Dict[str, Any] = {
"point_id": r.point_id, "metric": r.metric,
"operator": r.operator, "threshold": r.threshold,
"level": r.level, "text": text,
}
if current_values is not None and r.point_id in current_values:
val = current_values[r.point_id]
triggered = r.evaluate(val)
entry["current_value"] = val
entry["triggered"] = triggered
entry["state"] = "触发" if triggered else "正常"
result.items.append(entry)
if current_values is not None:
triggered_count = sum(1 for e in result.items if e.get("triggered"))
result.notes.append(f"模拟评估:{triggered_count}/{len(rules)} 条触发")
return result
# ---------------------------------------------------------------------------
# NL 查询预览
# ---------------------------------------------------------------------------
@dataclass
class NLQueryTemplate:
"""一条 NL 查询模板(供 NL 查询预览渲染)。"""
name: str # 模板名
question_template: str # 问句模板(含 {slot} 占位)
slots: Dict[str, List[str]] # 槽位 → 候选取值(用于生成示例问句)
data_source: str = "" # 预期数据来源(如 tdengine/rag)
answer_hint: str = "" # 预期答案提示
def render_examples(self, max_per_slot: int = 2) -> List[str]:
"""用槽位候选值生成示例问句(笛卡尔积,限量)。"""
if not self.slots:
return [self.question_template]
examples: List[str] = []
# 取每个槽位前 N 个候选,做限量笛卡尔积
first_slot = next(iter(self.slots))
for val in self.slots[first_slot][:max_per_slot]:
examples.append(self.question_template.replace("{" + first_slot + "}", val))
if not examples:
examples.append(self.question_template)
return examples
def render_nl_query_preview(
templates: List[NLQueryTemplate],
title: str = "NL 查询模板预览",
) -> PreviewResult:
"""渲染 NL 查询模板为示例问答对。"""
result = PreviewResult(kind=PreviewKind.NL_QUERY, title=title,
reason=f"渲染 {len(templates)} 个查询模板")
for t in templates:
examples = t.render_examples()
entry: Dict[str, Any] = {
"name": t.name,
"data_source": t.data_source,
"answer_hint": t.answer_hint,
"examples": examples,
}
result.items.append(entry)
if not t.question_template:
result.warnings.append(f"模板 '{t.name}' 问句模板为空")
result.notes.append(f"共生成 {sum(len(e['examples']) for e in result.items)} 条示例问句")
return result
# ---------------------------------------------------------------------------
# 从 ConfigStore 一键预览
# ---------------------------------------------------------------------------
def preview_from_store(
store: ConfigStore,
kind: PreviewKind = PreviewKind.LAYOUT,
current_values: Optional[Dict[str, float]] = None,
) -> PreviewResult:
"""从 ConfigStore 读取配置并渲染对应预览(配置台预览区入口)。
- LAYOUT:读 ``layout`` 类目下 key 含 'dashboard' 的 widget 列表;
- ALARM:读 ``model_param`` 类目下 key 以 'alarm_' 开头的规则;
- NL_QUERY:读 ``rag_config`` 类目下 key 以 'nl_' 开头的模板。
配置缺失时返回空结果(含提示),不报错——预览是只读的、宽容的。
"""
if kind == PreviewKind.LAYOUT:
widgets: List[Dict[str, Any]] = []
for it in store.list(ConfigKind.LAYOUT):
if isinstance(it.value, list):
widgets.extend(it.value)
if not widgets:
return PreviewResult(kind=kind, title="布局预览(空)",
notes=["未配置布局 widget,请在布局编辑页添加"])
return render_layout_preview(widgets)
if kind == PreviewKind.ALARM:
rules: List[AlarmRule] = []
for it in store.list(ConfigKind.MODEL_PARAM):
if it.key.startswith("alarm_") and isinstance(it.value, dict):
rules.append(AlarmRule(
point_id=it.value.get("point_id", ""),
metric=it.value.get("metric", ""),
operator=it.value.get("operator", ">"),
threshold=float(it.value.get("threshold", 0)),
level=it.value.get("level", "warn"),
message=it.value.get("message", ""),
))
if not rules:
return PreviewResult(kind=kind, title="告警预览(空)",
notes=["未配置告警规则,请在告警编辑页添加"])
return render_alarm_preview(rules, current_values=current_values)
# NL_QUERY
templates: List[NLQueryTemplate] = []
for it in store.list(ConfigKind.RAG_CONFIG):
if it.key.startswith("nl_") and isinstance(it.value, dict):
templates.append(NLQueryTemplate(
name=it.value.get("name", it.key),
question_template=it.value.get("question_template", ""),
slots=it.value.get("slots", {}),
data_source=it.value.get("data_source", ""),
answer_hint=it.value.get("answer_hint", ""),
))
if not templates:
return PreviewResult(kind=kind, title="NL 查询预览(空)",
notes=["未配置 NL 查询模板,请在查询编辑页添加"])
return render_nl_query_preview(templates)
+263
View File
@@ -0,0 +1,263 @@
# -*- coding: utf-8 -*-
"""⑤.7 配置台 ↔ 内核配置推送契约 —— issue #67 / PRD ⑤.7。
发布(#66)之后的下一步是**把配置真正送进内核**让 edge-gateway / rag-kb /
model-framework 生效。配置台与内核是两个独立部署单元,二者通过**配置推送契约**
解耦:配置台把一份已发布版本打包为内核可消费的 **JSON manifest**(带校验和),
内核侧拉取/接收后先验完整性再加载。本模块实现这个契约的"配置台侧":
- ``PushManifest``:推送给内核的清单(版本 / 快照 / 校验和 / 生成时间 / 来源);
- ``PushChannel``:推送通道。
- ``build_manifest(release)``:把 Release 打包成 manifest,计算 SHA256 校验和
(对快照做规范 JSON 序列化后哈希,确保内核侧可复算验证);
- ``push(release)``:模拟推送——把 manifest 写到内核预期的接收目录
(``<inbox>/manifest-<version>.json``),并记录推送日志(幂等:同版本不重复推送);
- ``pushed_versions()``:已成功推送的版本清单;
- ``verify(manifest)``:校验 manifest 的校验和是否一致(内核侧或配置台侧复用)。
幂等性:同一版本重复 push 返回已推送的旧记录(不覆盖、不重复写文件),避免内核
重复加载;要重推需先 ``retract``(撤回)该版本。
零运行时依赖:仅用 json / hashlib / dataclass / 标准库。
"""
from __future__ import annotations
import hashlib
import json
import os
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from .release import Release
MANIFEST_SCHEMA_VERSION = 1
MANIFEST_FILENAME_FMT = "manifest-{version}.json"
PUSH_LOG_FILENAME = "push_log.json"
PUSH_LOG_SCHEMA_VERSION = 1
def _now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _canonical_json(obj: Any) -> str:
"""规范 JSON 序列化(排序键、无空白),用于稳定哈希。"""
return json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
def checksum(snapshot: Dict[str, Any]) -> str:
"""计算配置快照的 SHA256 校验和(规范序列化后哈希)。
内核侧收到 manifest 后,对 ``snapshot`` 用同样算法复算,比对 ``checksum``
即可确认传输无损/未篡改。
"""
return hashlib.sha256(_canonical_json(snapshot).encode("utf-8")).hexdigest()
# ---------------------------------------------------------------------------
# 推送清单
# ---------------------------------------------------------------------------
@dataclass
class PushManifest:
"""推送给内核的配置清单(自描述:版本/快照/校验和/来源)。"""
schema_version: int = MANIFEST_SCHEMA_VERSION
version: str = "" # 对应 Release 的 semver
snapshot: Dict[str, Any] = field(default_factory=dict)
checksum: str = "" # snapshot 的 SHA256
generated_at: str = "" # manifest 生成时间
source: str = "template-console" # 来源标识(内核侧据此识别推送方)
description: str = "" # 推送说明(可解释)
def to_dict(self) -> dict:
return asdict(self)
def to_json(self) -> str:
"""manifest 序列化为 JSON 文本(推送载荷)。"""
return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)
@classmethod
def from_dict(cls, raw: dict) -> "PushManifest":
return cls(
schema_version=raw.get("schema_version", MANIFEST_SCHEMA_VERSION),
version=raw["version"],
snapshot=raw.get("snapshot", {}),
checksum=raw.get("checksum", ""),
generated_at=raw.get("generated_at", ""),
source=raw.get("source", "template-console"),
description=raw.get("description", ""),
)
# ---------------------------------------------------------------------------
# 推送通道
# ---------------------------------------------------------------------------
@dataclass
class PushRecord:
"""一次推送的记录(幂等判定与审计依据)。"""
version: str
checksum: str
pushed_at: str
pushed_by: str
manifest_path: str
status: str = "pushed" # pushed / retracted
reason: str = ""
class PushChannel:
"""配置台 → 内核的配置推送通道(基于文件系统的模拟推送)。
用法:
rm = ReleaseManager(store)
rel = rm.publish("1.0.0", ...)
ch = PushChannel(inbox="/path/to/kernel/inbox")
manifest = ch.push(rel, pushed_by="admin")
# 内核侧:读 manifest,复算 checksum 比对,加载 snapshot
"""
def __init__(self, inbox: str, push_log_path: Optional[str] = None) -> None:
"""``inbox`` 是内核侧接收目录(模拟推送就是把 manifest 写到此处)。
``push_log_path`` 推送日志路径(默认与 inbox 同目录的 push_log.json),
记录每个版本的推送状态,支撑幂等与撤回。
"""
self.inbox = inbox
os.makedirs(inbox, exist_ok=True)
self.push_log_path = push_log_path or os.path.join(inbox, PUSH_LOG_FILENAME)
# -- manifest 构建 --
def build_manifest(
self, release: Release, description: str = "",
) -> PushManifest:
"""把 Release 打包为 PushManifest(含校验和)。"""
snap = release.snapshot
return PushManifest(
schema_version=MANIFEST_SCHEMA_VERSION,
version=release.version,
snapshot=snap,
checksum=checksum(snap),
generated_at=_now_iso(),
source="template-console",
description=description or f"推送版本 {release.version}",
)
# -- 推送日志 --
def _read_log(self) -> List[PushRecord]:
if not os.path.isfile(self.push_log_path):
return []
with open(self.push_log_path, "r", encoding="utf-8") as fh:
blob = json.load(fh)
return [PushRecord(**r) for r in blob.get("records", [])]
def _write_log(self, records: List[PushRecord]) -> None:
blob = {
"schema_version": PUSH_LOG_SCHEMA_VERSION,
"records": [asdict(r) for r in records],
}
tmp = self.push_log_path + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(blob, fh, ensure_ascii=False, indent=2)
os.replace(tmp, self.push_log_path)
def _find_record(self, version: str) -> Optional[PushRecord]:
for r in self._read_log():
if r.version == version:
return r
return None
# -- 推送 / 撤回 --
def push(
self, release: Release, pushed_by: str = "system",
description: str = "", force: bool = False,
) -> PushManifest:
"""推送一个已发布版本到内核接收目录(幂等:同版本不重复推送)。
幂等性:若该版本已成功推送且未撤回,直接返回原 manifest(不重复写文件、
不重复触发内核加载)。要强制重推,先 ``retract`` 或传 ``force=True``。
Args:
release: 已发布的版本(含快照);
pushed_by: 推送人(对接 RBAC);
description: 推送说明;
force: 强制重推(覆盖既有 manifest)。
Returns:
推送的 PushManifest。
"""
existing = self._find_record(release.version)
if existing and existing.status == "pushed" and not force:
# 幂等:返回已推送的 manifest(从 inbox 读回)
if os.path.isfile(existing.manifest_path):
with open(existing.manifest_path, "r", encoding="utf-8") as fh:
return PushManifest.from_dict(json.load(fh))
manifest = self.build_manifest(release, description=description)
manifest_path = os.path.join(
self.inbox, MANIFEST_FILENAME_FMT.format(version=release.version))
tmp = manifest_path + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
fh.write(manifest.to_json())
os.replace(tmp, manifest_path)
# 更新推送日志(覆盖同版本旧记录)
records = [r for r in self._read_log() if r.version != release.version]
records.append(PushRecord(
version=release.version, checksum=manifest.checksum,
pushed_at=_now_iso(), pushed_by=pushed_by,
manifest_path=manifest_path, status="pushed",
reason=description or f"推送 {release.version}",
))
self._write_log(records)
return manifest
def retract(self, version: str, by: str = "system", reason: str = "") -> bool:
"""撤回一个已推送版本(标记为 retracted,不删 manifest 文件,可追溯)。
撤回后该版本可重新 push(幂等解除)。返回是否实际撤回。
"""
rec = self._find_record(version)
if rec is None or rec.status != "pushed":
return False
records = self._read_log()
for i, r in enumerate(records):
if r.version == version:
records[i] = PushRecord(
version=r.version, checksum=r.checksum,
pushed_at=r.pushed_at, pushed_by=r.pushed_by,
manifest_path=r.manifest_path, status="retracted",
reason=f"撤回 by {by}:{reason or '未说明'}",
)
self._write_log(records)
return True
# -- 查询 / 校验 --
def pushed_versions(self) -> List[Dict[str, Any]]:
"""已推送版本摘要(配置台推送状态列表用)。"""
return [
{"version": r.version, "checksum": r.checksum,
"pushed_at": r.pushed_at, "pushed_by": r.pushed_by,
"status": r.status}
for r in self._read_log()
]
@staticmethod
def verify(manifest: PushManifest) -> bool:
"""校验 manifest 的 checksum 与其 snapshot 是否一致。
内核侧收到 manifest 后调用此方法,确认传输无损;配置台侧也可在推送前自检。
"""
return manifest.checksum == checksum(manifest.snapshot)
@staticmethod
def verify_payload(payload: Dict[str, Any]) -> bool:
"""从原始 payload(dict)校验:用同算法复算 checksum 比对。"""
try:
manifest = PushManifest.from_dict(payload)
except (KeyError, TypeError):
return False
return PushChannel.verify(manifest)
+310
View File
@@ -0,0 +1,310 @@
# -*- coding: utf-8 -*-
"""⑤.7 配置台三级 RBAC 权限模型 —— issue #62 / PRD ⑤.7。
配置台面向**多角色协作**:实施工程师配模板,行业工程师调参数,运维/管理者
发布上线。直接对所有人开放写权限会带来误改与不可溯源风险。本模块用三级
RBAC(基于角色的访问控制)锁定"谁能对哪类配置做什么",并把每次权限判定
的**理由**一并返回,对齐 PRD「可解释可溯源」。
三级角色(由低到高,后者继承前者全部权限):
- ``readonly`` (只读):查看配置 / 预览 / 历史版本,不可写;
- ``engineer`` (行业工程师):只读权限 + 编辑/校验/导入配置(模型超参 /
RAG / 布局 / 点位字典),但**不能发布与回滚**;
- ``admin`` (管理员):工程师权限 + 发布 / 回滚 / 推送内核 / 用户管理。
权限判定核心为 ``has_permission(user, resource, action)``,返回
``PermissionDecision``(allow + reason),便于配置台前端把"为什么拒绝"
直接展示给操作者,而不是一个干瘪的 403。
零运行时依赖:仅用 dataclass / Enum / 标准库。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Set
# ---------------------------------------------------------------------------
# 权限维度:资源 × 动作
# ---------------------------------------------------------------------------
class Resource(str, Enum):
"""配置台可管控的资源(对齐 #63~#67 子模块)。"""
POINT_DICT = "point_dict" # 点位字典(#63)
MODEL_PARAM = "model_param" # 模型超参配置(#64)
RAG_CONFIG = "rag_config" # RAG 知识库配置(#64)
LAYOUT = "layout" # 驾驶舱布局配置(#64/#65)
PREVIEW = "preview" # 预览(#65)
RELEASE = "release" # 版本发布/回滚(#66)
PUSH = "push" # 配置推送内核(#67)
USER = "user" # 用户/角色管理
class Action(str, Enum):
"""对资源可执行的动作。"""
VIEW = "view" # 查看 / 预览 / 列表
EDIT = "edit" # 新增 / 修改 / 删除 / 导入 / 校验
PUBLISH = "publish" # 发布版本 / 回滚 / 推送内核
MANAGE = "manage" # 用户与角色管理
class RoleKind(str, Enum):
"""三级角色枚举(值即配置资产中的角色标识)。"""
READONLY = "readonly"
ENGINEER = "engineer"
ADMIN = "admin"
# 各资源的「写」动作等价集合:EDIT 含新增/修改/删除/导入/校验。
# PUBLISH 含发布/回滚/推送。这样配置台前端只需关心粗粒度动作。
_WRITE_ACTIONS: Set[Action] = {Action.EDIT, Action.PUBLISH, Action.MANAGE}
# ---------------------------------------------------------------------------
# 权限模型
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Permission:
"""一条权限授予(角色 → 资源 → 动作)。
``meaning`` 解释该权限的业务含义,用于审计日志与配置台权限矩阵展示。
注意:权限**匹配**基于 ``resource:action``(资源×动作),与授予角色无关——
这正是角色继承能生效的关键(admin 继承 engineer 的 edit,匹配键相同)。
``role`` 仅作为审计元数据,记录"是谁授予的"。
"""
role: RoleKind
resource: Resource
action: Action
meaning: str = ""
def key(self) -> str:
"""权限匹配键(资源:动作)—— 角色继承据此累计。"""
return f"{self.resource.value}:{self.action.value}"
def audit_key(self) -> str:
"""审计唯一键(角色/资源/动作三元组,含授予者)。"""
return f"{self.role.value}:{self.resource.value}:{self.action.value}"
@dataclass
class Role:
"""一个角色:权限集合 + 继承的父角色。"""
kind: RoleKind
label: str # 中文展示名
permissions: List[Permission] = field(default_factory=list)
inherits: Optional[RoleKind] = None # 继承的低一级角色
description: str = "" # 角色职责说明(可解释性)
def permission_keys(self) -> Set[str]:
"""本角色直接授予的权限键集合。"""
return {p.key() for p in self.permissions}
@dataclass
class User:
"""配置台用户。"""
username: str
role: RoleKind
display_name: str = ""
# 可选资源级收窄:即便角色允许,列表中的资源也会被额外限制为只读。
# 用于"只允许工程师改某几类配置"的细粒度场景。
restricted_to_readonly: List[Resource] = field(default_factory=list)
@dataclass
class PermissionDecision:
"""``has_permission`` 的判定结果(带理由,可解释)。"""
allow: bool
reason: str # 人类可读的判定理由(允许/拒绝原因)
role: RoleKind
resource: Resource
action: Action
source: str = "explicit" # explicit(本角色直接授予)/ inherited(继承自父角色)
# ---------------------------------------------------------------------------
# 角色注册表:三级权限矩阵(对齐 PRD ⑤.7「三级 RBAC」)
# ---------------------------------------------------------------------------
def _build_role_registry() -> Dict[RoleKind, Role]:
"""构建三级角色及其权限矩阵。
权限设计依据(PRD ⑤.7):
- readonly:可查看所有配置/预览/历史,但不能改、不能发;
- engineer:在 readonly 基础上,可编辑/校验/导入四类业务配置,
但**发布/回滚/推送/用户管理仍归 admin**(避免未经评审上线);
- admin:在 engineer 基础上,可发布/回滚/推送 + 管理用户角色。
"""
ro = Role(
kind=RoleKind.READONLY,
label="只读",
description="实施/运维只读角色:查看配置、预览、历史版本,不可写。",
permissions=[
Permission(RoleKind.READONLY, Resource.POINT_DICT, Action.VIEW,
"查看点位字典与校验报告"),
Permission(RoleKind.READONLY, Resource.MODEL_PARAM, Action.VIEW,
"查看模型超参配置"),
Permission(RoleKind.READONLY, Resource.RAG_CONFIG, Action.VIEW,
"查看 RAG 知识库配置"),
Permission(RoleKind.READONLY, Resource.LAYOUT, Action.VIEW,
"查看驾驶舱布局配置"),
Permission(RoleKind.READONLY, Resource.PREVIEW, Action.VIEW,
"查看配置预览"),
Permission(RoleKind.READONLY, Resource.RELEASE, Action.VIEW,
"查看历史发布版本"),
],
)
engineer = Role(
kind=RoleKind.ENGINEER,
label="行业工程师",
inherits=RoleKind.READONLY,
description="行业工程师:编辑/校验/导入业务配置,但不能发布与推送。",
permissions=[
Permission(RoleKind.ENGINEER, Resource.POINT_DICT, Action.EDIT,
"导入/编辑/校验点位字典 CSV"),
Permission(RoleKind.ENGINEER, Resource.MODEL_PARAM, Action.EDIT,
"调整模型超参配置"),
Permission(RoleKind.ENGINEER, Resource.RAG_CONFIG, Action.EDIT,
"编辑 RAG 知识库配置"),
Permission(RoleKind.ENGINEER, Resource.LAYOUT, Action.EDIT,
"编辑驾驶舱布局配置"),
Permission(RoleKind.ENGINEER, Resource.PREVIEW, Action.VIEW,
"预览配置效果(编辑后必看)"),
],
)
admin = Role(
kind=RoleKind.ADMIN,
label="管理员",
inherits=RoleKind.ENGINEER,
description="管理员:在工程师基础上负责发布/回滚/推送与用户管理。",
permissions=[
Permission(RoleKind.ADMIN, Resource.RELEASE, Action.PUBLISH,
"发布新版本与回滚到历史版本"),
Permission(RoleKind.ADMIN, Resource.PUSH, Action.PUBLISH,
"把已发布配置推送给内核"),
Permission(RoleKind.ADMIN, Resource.USER, Action.MANAGE,
"管理用户与角色分配"),
Permission(RoleKind.ADMIN, Resource.POINT_DICT, Action.PUBLISH,
"确认点位字典上线(审批环节)"),
Permission(RoleKind.ADMIN, Resource.MODEL_PARAM, Action.PUBLISH,
"确认模型超参上线"),
Permission(RoleKind.ADMIN, Resource.LAYOUT, Action.PUBLISH,
"确认布局上线"),
],
)
return {RoleKind.READONLY: ro, RoleKind.ENGINEER: engineer, RoleKind.ADMIN: admin}
_ROLES: Dict[RoleKind, Role] = _build_role_registry()
def get_role(kind: RoleKind) -> Role:
"""获取角色定义。"""
return _ROLES[kind]
def all_roles() -> List[Role]:
"""全部角色(按权限由低到高)。"""
return [_ROLES[RoleKind.READONLY], _ROLES[RoleKind.ENGINEER], _ROLES[RoleKind.ADMIN]]
def effective_permissions(kind: RoleKind) -> Set[str]:
"""角色有效权限键(含继承链)。
继承解析:admin 继承 engineer 继承 readonly,递归向上累计权限键。
"""
role = _ROLES[kind]
keys: Set[str] = set(role.permission_keys())
if role.inherits is not None:
keys |= effective_permissions(role.inherits)
return keys
# ---------------------------------------------------------------------------
# 判定 API
# ---------------------------------------------------------------------------
def has_permission(
user: User,
resource: Resource,
action: Action,
) -> PermissionDecision:
"""判定用户对某资源执行某动作是否被允许(带理由)。
判定顺序:
1. 计算角色有效权限(含继承),命中即允许并标注来源(本角色/继承);
2. 命中后若该资源在用户 ``restricted_to_readonly`` 列表且动作是写动作,
则降级拒绝(细粒度收窄);
3. 未命中则拒绝,理由标注缺失的权限三元组。
Args:
user: 配置台用户;
resource: 目标资源;
action: 目标动作。
Returns:
PermissionDecision:allow + reason(可直接展示给操作者)。
"""
target = f"{resource.value}:{action.value}"
eff = effective_permissions(user.role)
# 细粒度收窄:即便角色允许,特定资源也被限制为只读
if resource in user.restricted_to_readonly and action in _WRITE_ACTIONS:
return PermissionDecision(
allow=False,
reason=(f"用户 '{user.username}' 对资源 '{resource.value}' 被收窄为只读,"
f"禁止执行 '{action.value}' 动作"),
role=user.role, resource=resource, action=action, source="restricted",
)
if target in eff:
# 判定来源:本角色直接授予 or 继承自父角色
own = get_role(user.role).permission_keys()
source = "explicit" if target in own else "inherited"
src_label = "本角色直接授予" if source == "explicit" else "继承自低级角色"
return PermissionDecision(
allow=True,
reason=(f"用户 '{user.username}'({get_role(user.role).label})"
f"允许对 '{resource.value}' 执行 '{action.value}'({src_label})"),
role=user.role, resource=resource, action=action, source=source,
)
return PermissionDecision(
allow=False,
reason=(f"用户 '{user.username}'({get_role(user.role).label})缺少权限 "
f"{user.role.value}:{resource.value}:{action.value};"
f"该动作需更高角色或审批"),
role=user.role, resource=resource, action=action, source="denied",
)
def can_publish(user: User) -> bool:
"""便捷判定:用户是否具备发布(发布/回滚/推送)能力。"""
return has_permission(user, Resource.RELEASE, Action.PUBLISH).allow
def user_summary(user: User) -> Dict[str, object]:
"""用户权限概览(供配置台用户卡片/审计日志展示)。"""
role = get_role(user.role)
return {
"username": user.username,
"display_name": user.display_name or user.username,
"role": user.role.value,
"role_label": role.label,
"description": role.description,
"effective_permission_count": len(effective_permissions(user.role)),
"restricted_to_readonly": [r.value for r in user.restricted_to_readonly],
}
+238
View File
@@ -0,0 +1,238 @@
# -*- coding: utf-8 -*-
"""⑤.7 版本发布 + 回滚点 —— issue #66 / PRD ⑤.7。
配置台的每次发布都应是一个**可回滚的版本**:实施工程师改了配置 → 预览确认 →
管理员发布;发布即固化当时全量配置快照为一个带 semver 的 Release;若线上出问题,
一键回滚到上一个版本,把 ``ConfigStore`` 恢复成那份快照。这样配置变更"可追溯、
可逆转",对齐 PRD「版本化发布与回滚点」。
本模块提供:
- ``Release`` 数据类(semver 版本号 / 时间戳 / 快照 / 发布人 / 变更说明);
- ``ReleaseManager``:list / publish / rollback;
- ``publish``:固化 ConfigStore 快照为新版本,semver 单调递增校验
(新版本必须严格大于当前最新版),拒绝重复发布空快照;
- ``rollback``:把 ConfigStore 恢复为指定历史版本的快照,并记一条"回滚事件"
(不删除任何历史版本——回滚本身也是一次可追溯的变更);
- semver 校验(``MAJOR.MINOR.PATCH``,单调递增)。
发布记录持久化为 ``releases.json``(与 ConfigStore 同根目录),人可读、可备份。
零运行时依赖:仅用 json / dataclass / 标准库。
"""
from __future__ import annotations
import json
import os
import re
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
from .config_store import ConfigStore
# ---------------------------------------------------------------------------
# semver
# ---------------------------------------------------------------------------
_SEMVER_RE = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$")
RELEASES_FILENAME = "releases.json"
RELEASES_SCHEMA_VERSION = 1
def is_valid_semver(version: str) -> bool:
"""是否合法 semver(MAJOR.MINOR.PATCH,无预发布后缀)。"""
return bool(_SEMVER_RE.match(version))
def semver_tuple(version: str) -> Tuple[int, int, int]:
"""semver → (major, minor, patch) 元组(用于比较)。"""
m = _SEMVER_RE.match(version)
if not m:
raise ValueError(f"非法 semver:{version}")
return tuple(int(x) for x in m.groups()) # type: ignore[return-value]
def semver_gt(a: str, b: str) -> bool:
"""a 是否严格大于 b。"""
return semver_tuple(a) > semver_tuple(b)
def bump_patch(version: str) -> str:
"""patch 位 +1(默认递增策略,发布时若用户未指定版本号则用此)。"""
major, minor, patch = semver_tuple(version)
return f"{major}.{minor}.{patch + 1}"
# ---------------------------------------------------------------------------
# Release 数据模型
# ---------------------------------------------------------------------------
@dataclass
class Release:
"""一次发布版本(可解释:含发布人、变更说明、来源)。"""
version: str # semver,如 1.2.0
created_at: str # ISO8601 发布时间
snapshot: Dict[str, Any] # 全量配置快照(ConfigStore.snapshot())
released_by: str = "system" # 发布人(对接 RBAC 用户名)
changelog: str = "" # 变更说明(本次发布改了什么、为什么)
reason: str = "" # 发布理由(可解释可溯源)
def to_dict(self) -> dict:
return asdict(self)
@classmethod
def from_dict(cls, raw: dict) -> "Release":
return cls(
version=raw["version"],
created_at=raw.get("created_at", ""),
snapshot=raw.get("snapshot", {}),
released_by=raw.get("released_by", "system"),
changelog=raw.get("changelog", ""),
reason=raw.get("reason", ""),
)
def _now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
# ---------------------------------------------------------------------------
# 发布管理器
# ---------------------------------------------------------------------------
class ReleaseManager:
"""版本发布 + 回滚管理器。
用法:
store = ConfigStore("/path/to/store")
rm = ReleaseManager(store) # releases 落在 store 同目录
rel = rm.publish("1.0.0", released_by="admin", changelog="首次发布")
rm.rollback("0.9.9", released_by="admin") # 回滚到 0.9.9 的快照
"""
def __init__(self, store: ConfigStore, releases_path: Optional[str] = None) -> None:
self.store = store
self.releases_path = releases_path or os.path.join(store.root, RELEASES_FILENAME)
# -- 持久化 --
def _read_all(self) -> List[Release]:
if not os.path.isfile(self.releases_path):
return []
with open(self.releases_path, "r", encoding="utf-8") as fh:
blob = json.load(fh)
return [Release.from_dict(r) for r in blob.get("releases", [])]
def _write_all(self, releases: List[Release]) -> None:
blob = {
"schema_version": RELEASES_SCHEMA_VERSION,
"releases": [r.to_dict() for r in releases],
}
tmp = self.releases_path + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(blob, fh, ensure_ascii=False, indent=2)
os.replace(tmp, self.releases_path)
# -- 查询 --
def list(self) -> List[Release]:
"""全部发布版本(按版本号升序)。"""
rels = self._read_all()
return sorted(rels, key=lambda r: semver_tuple(r.version))
def latest(self) -> Optional[Release]:
"""最新发布版本(无则 None)。"""
rels = self.list()
return rels[-1] if rels else None
def get(self, version: str) -> Optional[Release]:
"""取指定版本(不存在返回 None)。"""
for r in self._read_all():
if r.version == version:
return r
return None
# -- 发布 --
def publish(
self,
version: str,
released_by: str = "system",
changelog: str = "",
reason: str = "",
) -> Release:
"""发布新版本(固化当前 ConfigStore 快照)。
Raises:
ValueError: semver 非法 / 版本号非单调递增 / 版本号已存在 /
快照为空(无任何配置不允许发布)。
"""
if not is_valid_semver(version):
raise ValueError(f"版本号 '{version}' 非法(须为 MAJOR.MINOR.PATCH)")
releases = self._read_all()
existing = {r.version for r in releases}
if version in existing:
raise ValueError(f"版本号 '{version}' 已存在,不可重复发布")
# 单调递增:新版本必须严格大于当前最新
if releases:
current_latest = max((r.version for r in releases), key=semver_tuple)
if not semver_gt(version, current_latest):
raise ValueError(
f"新版本 '{version}' 必须大于当前最新 '{current_latest}'(单调递增)")
snapshot = self.store.snapshot()
total_items = sum(len(v) for v in snapshot.get("kinds", {}).values())
if total_items == 0:
raise ValueError("配置快照为空,不允许发布(先在配置台录入配置)")
release = Release(
version=version, created_at=_now_iso(), snapshot=snapshot,
released_by=released_by, changelog=changelog, reason=reason,
)
releases.append(release)
self._write_all(releases)
return release
# -- 回滚 --
def rollback(
self,
target_version: str,
released_by: str = "system",
reason: str = "",
) -> Release:
"""回滚到指定历史版本的快照(把 ConfigStore 恢复成该版本快照)。
回滚**不删除**任何历史版本,而是:恢复快照 + 记一条回滚说明。返回
目标版本(便于调用方确认恢复到哪)。
Raises:
ValueError: 目标版本不存在 / 回滚到当前已是的状态。
"""
target = self.get(target_version)
if target is None:
raise ValueError(f"回滚目标版本 '{target_version}' 不存在")
self.store.restore(target.snapshot)
# 记录回滚事件(作为一条带 changelog 的元信息,不新增版本号)
rollback_note = (
f"[回滚] 已把配置恢复到 {target_version}(发布于 {target.created_at});"
f"操作人={released_by};原因={reason or '未说明'}")
# 把回滚事件追加到目标版本的 reason 字段(可追溯,不污染版本号序列)
target.reason = (target.reason + " | " + rollback_note).strip(" |") if target.reason else rollback_note
releases = self._read_all()
for i, r in enumerate(releases):
if r.version == target_version:
releases[i] = target
self._write_all(releases)
return target
def history(self) -> List[Dict[str, Any]]:
"""发布历史摘要(配置台版本列表展示用)。"""
return [
{
"version": r.version,
"created_at": r.created_at,
"released_by": r.released_by,
"changelog": r.changelog,
"item_count": sum(len(v) for v in r.snapshot.get("kinds", {}).values()),
"reason": r.reason,
}
for r in self.list()
]
+26
View File
@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""测试引导:把 `core/template-console` 以包名 `template_console` 挂载到 sys.modules。
目录名 `template-console` 含连字符,无法直接以包名 import;挂载后模块内
相对导入(`from .rbac import ...`)在 unittest 发现机制下可正常解析。
同时把兄弟内核目录 `core/edge-gateway` 加入 sys.path,使 point_importer
可复用其 `point_dict` 子包(loader/validator/schema),避免重复造轮子。
"""
import os
import sys
import types
# 1) 挂载 core/template-console 为 template_console 包
CONSOLE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, CONSOLE_DIR)
if "template_console" not in sys.modules:
pkg = types.ModuleType("template_console")
pkg.__path__ = [CONSOLE_DIR]
sys.modules["template_console"] = pkg
# 2) 暴露兄弟内核 edge-gateway/point_dict(#63 复用其校验器)
CORE_DIR = os.path.dirname(CONSOLE_DIR)
EDGE_GW_DIR = os.path.join(CORE_DIR, "edge-gateway")
if os.path.isdir(EDGE_GW_DIR) and EDGE_GW_DIR not in sys.path:
sys.path.insert(0, EDGE_GW_DIR)
@@ -0,0 +1,191 @@
# -*- coding: utf-8 -*-
"""配置项 CRUD 存储引擎测试(issue #64)。
覆盖:
1. 三类配置 CRUD(list/get/upsert/delete);
2. 原子写 + 持久化(重开 store 仍在);
3. 校验规则(model_param/rag_config/layout,非法值拒绝);
4. 快照 snapshot/restore(为 #66 提供基础);
5. 可解释字段(meaning/reason/updated_by/updated_at 落盘)。
"""
import json
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401
from template_console.config_store import ( # noqa: E402
ALLOWED_WIDGET_TYPES,
ConfigItem,
ConfigKind,
ConfigStore,
ValidationResult,
validate_item,
)
class _TmpStore:
def __init__(self):
self._tmp = tempfile.mkdtemp()
self.store = ConfigStore(self._tmp)
def cleanup(self):
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
class ValidationTest(unittest.TestCase):
"""校验规则。"""
def test_model_param_scalar_ok(self):
self.assertTrue(validate_item(ConfigKind.MODEL_PARAM, "learning_rate", 0.001))
def test_model_param_learning_rate_range(self):
vr = validate_item(ConfigKind.MODEL_PARAM, "learning_rate", 1.5)
self.assertFalse(vr)
self.assertTrue(any("learning_rate" in e for e in vr.errors))
def test_model_param_bad_key(self):
vr = validate_item(ConfigKind.MODEL_PARAM, "Bad Key!", 1)
self.assertFalse(vr)
def test_rag_top_k_bounds(self):
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "top_k", 0))
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "top_k", 51))
self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "top_k", 10))
def test_rag_similarity_threshold(self):
self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "similarity_threshold", 0.5))
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "similarity_threshold", 1.5))
def test_rag_sources_must_be_nonempty_list(self):
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "sources", []))
self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "sources", ["", "x"]))
self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "sources", ["sop", "gb"]))
def test_layout_widget_type(self):
bad = [{"type": "unknown", "x": 0, "y": 0, "w": 1, "h": 1}]
self.assertFalse(validate_item(ConfigKind.LAYOUT, "dashboard", bad))
good = [{"type": "trend", "x": 0, "y": 0, "w": 6, "h": 2}]
self.assertTrue(validate_item(ConfigKind.LAYOUT, "dashboard", good))
def test_layout_widget_coords_nonneg_int(self):
bad = [{"type": "trend", "x": -1, "y": 0, "w": 1, "h": 1}]
vr = validate_item(ConfigKind.LAYOUT, "dashboard", bad)
self.assertFalse(vr)
class CrudTest(unittest.TestCase):
"""CRUD + 持久化。"""
def setUp(self):
self.ctx = _TmpStore()
self.store = self.ctx.store
def tearDown(self):
self.ctx.cleanup()
def test_upsert_and_get(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "iterations", 100,
meaning="迭代数", updated_by="li", reason="标定")
it = self.store.get(ConfigKind.MODEL_PARAM, "iterations")
self.assertIsNotNone(it)
self.assertEqual(it.value, 100)
self.assertEqual(it.updated_by, "li")
self.assertEqual(it.reason, "标定")
self.assertTrue(it.updated_at) # 时间戳已写
def test_upsert_rejects_invalid(self):
with self.assertRaises(ValueError):
self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 999)
def test_upsert_overwrites(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.1)
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.01, reason="调小")
it = self.store.get(ConfigKind.MODEL_PARAM, "lr")
self.assertEqual(it.value, 0.01)
self.assertEqual(it.reason, "调小")
def test_list_and_delete(self):
self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 5)
self.store.upsert(ConfigKind.RAG_CONFIG, "similarity_threshold", 0.6)
self.assertEqual(len(self.store.list(ConfigKind.RAG_CONFIG)), 2)
self.assertTrue(self.store.delete(ConfigKind.RAG_CONFIG, "top_k"))
self.assertIsNone(self.store.get(ConfigKind.RAG_CONFIG, "top_k"))
self.assertFalse(self.store.delete(ConfigKind.RAG_CONFIG, "nope"))
def test_persistence_across_reopen(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
# 重开一个指向同一目录的 store
store2 = ConfigStore(self.ctx._tmp)
it = store2.get(ConfigKind.MODEL_PARAM, "lr")
self.assertIsNotNone(it)
self.assertEqual(it.value, 0.001)
def test_json_file_is_human_readable(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001, meaning="学习率")
path = os.path.join(self.ctx._tmp, "model_params.json")
with open(path, encoding="utf-8") as fh:
blob = json.load(fh)
self.assertEqual(blob["schema_version"], 1)
self.assertEqual(blob["kind"], "model_param")
self.assertEqual(blob["items"][0]["meaning"], "学习率")
class SnapshotTest(unittest.TestCase):
"""快照与恢复(#66 基础)。"""
def setUp(self):
self.ctx = _TmpStore()
self.store = self.ctx.store
def tearDown(self):
self.ctx.cleanup()
def test_snapshot_captures_all_kinds(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 8)
snap = self.store.snapshot()
self.assertIn("captured_at", snap)
self.assertEqual(set(snap["kinds"].keys()),
{"model_param", "rag_config", "layout"})
self.assertEqual(len(snap["kinds"]["model_param"]), 1)
def test_restore_replicates_state(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 8)
snap = self.store.snapshot()
# 清空再恢复
self.store.delete(ConfigKind.MODEL_PARAM, "lr")
self.store.delete(ConfigKind.RAG_CONFIG, "top_k")
self.store.restore(snap)
self.assertEqual(self.store.get(ConfigKind.MODEL_PARAM, "lr").value, 0.001)
self.assertEqual(self.store.get(ConfigKind.RAG_CONFIG, "top_k").value, 8)
def test_item_counts(self):
self.store.upsert(ConfigKind.LAYOUT, "dashboard",
[{"type": "trend", "x": 0, "y": 0, "w": 6, "h": 2}])
counts = self.store.item_counts()
self.assertEqual(counts["layout"], 1)
self.assertEqual(counts["model_param"], 0)
class ConfigItemSerializationTest(unittest.TestCase):
"""ConfigItem 序列化往返。"""
def test_roundtrip(self):
it = ConfigItem(key="lr", value=0.1, kind=ConfigKind.MODEL_PARAM,
meaning="学习率", updated_by="li", reason="init",
updated_at="2026-01-01T00:00:00Z")
d = it.to_dict()
self.assertEqual(d["kind"], "model_param")
it2 = ConfigItem.from_dict(d)
self.assertEqual(it2.value, 0.1)
self.assertEqual(it2.kind, ConfigKind.MODEL_PARAM)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,228 @@
# -*- coding: utf-8 -*-
"""点位字典 CSV 导入 + 自动校验测试(issue #63)。
覆盖:
1. 合法 CSV 导入通过(ti / resin 两套模板);
2. 表头校验(缺失列 / 列序错位);
3. 内核校验复用(量纲/数据类型/采样率/重复点号/协议);
4. OPC 节点格式校验(opcua/modbus/空);
5. 模板级量纲收窄(rpm 仅 resin 允许);
6. 报告 ok/汇总/字典化 + 粘贴框入口。
"""
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401
from template_console.point_importer import ( # noqa: E402
ImportReport,
ImportRowIssue,
Severity,
TemplateKind,
import_csv,
import_csv_string,
)
GOOD_TI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua
CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres,opcua
"""
GOOD_RESIN = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
R-801,R-801.TEMP,反应釜温度,℃,float,1000,true,ns=2;s=R801.Temp,opcua
R-801,R-801.AGIT,搅拌转速,rpm,float,1000,true,ns=2;s=R801.Agit,opcua
"""
BAD_MULTI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,badnode,opcua
CLF-01,CLF-01.TEMP,炉压,kPa,badtype,0,true,ns=2;s=CLF.Pres,opcua
CLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus
"""
class _TmpCsv:
"""临时 CSV 文件助手。"""
def __init__(self, content):
self._tmp = tempfile.mkdtemp()
self.path = os.path.join(self._tmp, "points.csv")
with open(self.path, "w", encoding="utf-8") as fh:
fh.write(content)
def cleanup(self):
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
class GoodImportTest(unittest.TestCase):
"""合法 CSV 导入。"""
def test_good_ti_imports_ok(self):
f = _TmpCsv(GOOD_TI)
try:
pd, rep = import_csv(f.path, template=TemplateKind.TI)
self.assertTrue(rep.ok, rep.summary())
self.assertEqual(rep.loaded_points, 2)
self.assertEqual(rep.error_count, 0)
finally:
f.cleanup()
def test_good_resin_imports_ok_with_rpm(self):
f = _TmpCsv(GOOD_RESIN)
try:
pd, rep = import_csv(f.path, template=TemplateKind.RESIN)
self.assertTrue(rep.ok, rep.summary())
# rpm 在 resin 模板合法
self.assertEqual(rep.error_count, 0)
finally:
f.cleanup()
def test_report_summary_and_dict(self):
f = _TmpCsv(GOOD_TI)
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
self.assertIn("通过", rep.summary())
d = rep.to_dict()
self.assertTrue(d["ok"])
self.assertEqual(d["template"], "ti")
self.assertEqual(d["loaded_points"], 2)
finally:
f.cleanup()
class HeaderValidationTest(unittest.TestCase):
"""表头校验。"""
def test_missing_column_is_error(self):
bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp\n"
f = _TmpCsv(bad)
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
self.assertFalse(rep.ok)
codes = [i.code for i in rep.issues if i.row == 1]
self.assertIn("missing_column", codes)
finally:
f.cleanup()
def test_wrong_column_order_is_warn(self):
# 列齐全但顺序错(name 提前)→ WARN,不阻断
bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,protocol,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,opcua,ns=2;s=CLF.Temp\n"
f = _TmpCsv(bad)
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
self.assertIn("bad_column_order", [i.code for i in rep.issues])
finally:
f.cleanup()
class KernelValidationTest(unittest.TestCase):
"""复用内核校验(量纲/数据类型/采样率/重复点号)。"""
def test_dup_point_detected(self):
bad = GOOD_TI + "CLF-01,CLF-01.TEMP,炉温2,℃,float,1000,true,ns=2;s=CLF.Temp2,opcua\n"
f = _TmpCsv(bad)
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
self.assertFalse(rep.ok)
self.assertIn("dup_point", [i.code for i in rep.issues])
finally:
f.cleanup()
def test_bad_data_type_and_sample_rate(self):
f = _TmpCsv(BAD_MULTI)
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
codes = [i.code for i in rep.issues]
self.assertIn("bad_data_type", codes)
self.assertIn("bad_sample_rate", codes)
finally:
f.cleanup()
def test_bad_protocol_detected(self):
bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,unknownproto\n"
f = _TmpCsv(bad)
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
self.assertIn("bad_protocol", [i.code for i in rep.issues])
finally:
f.cleanup()
class OpcNodeValidationTest(unittest.TestCase):
"""OPC 节点格式校验(配置台扩展维度)。"""
def test_bad_opcua_node_is_error(self):
# BAD_MULTI 第1行 opcNode=badnode 协议 opcua → ERROR
f = _TmpCsv(BAD_MULTI)
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
opc_issues = [i for i in rep.issues if i.code == "bad_opc_node"]
self.assertTrue(any(i.severity == Severity.ERROR for i in opc_issues))
finally:
f.cleanup()
def test_valid_modbus_node_ok(self):
# BAD_MULTI 第3行 holding:40010 modbus → 不报 bad_opc_node
f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus\n")
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
self.assertNotIn("bad_opc_node", [i.code for i in rep.issues
if i.severity == Severity.ERROR])
finally:
f.cleanup()
def test_empty_opc_node_is_warn(self):
f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,,simulator\n")
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
empties = [i for i in rep.issues if i.code == "empty_opc_node"]
self.assertEqual(len(empties), 1)
self.assertEqual(empties[0].severity, Severity.WARN)
# 警告不阻断
self.assertTrue(rep.ok)
finally:
f.cleanup()
class TemplateUnitTest(unittest.TestCase):
"""模板级量纲收窄。"""
def test_rpm_rejected_in_ti_template(self):
f = _TmpCsv(GOOD_RESIN)
try:
_, rep = import_csv(f.path, template=TemplateKind.TI)
# rpm 是树脂专属,ti 模板应报 template_unit_mismatch
self.assertIn("template_unit_mismatch", [i.code for i in rep.issues])
self.assertFalse(rep.ok)
finally:
f.cleanup()
def test_rpm_allowed_in_resin_template(self):
f = _TmpCsv(GOOD_RESIN)
try:
_, rep = import_csv(f.path, template=TemplateKind.RESIN)
self.assertNotIn("template_unit_mismatch", [i.code for i in rep.issues])
self.assertTrue(rep.ok, rep.summary())
finally:
f.cleanup()
class ImportStringTest(unittest.TestCase):
"""粘贴框入口(import_csv_string)。"""
def test_import_from_string(self):
pd, rep = import_csv_string(GOOD_TI, template=TemplateKind.TI)
self.assertTrue(rep.ok)
self.assertEqual(len(pd), 2)
def test_import_string_bad_csv(self):
bad = "device_id,point_id\nCLF-01,CLF-01.TEMP\n" # 缺列
_, rep = import_csv_string(bad, template=TemplateKind.TI)
self.assertFalse(rep.ok)
if __name__ == "__main__":
unittest.main()
+188
View File
@@ -0,0 +1,188 @@
# -*- coding: utf-8 -*-
"""配置预览渲染引擎测试(issue #65)。
覆盖:
1. 布局预览(widget 卡片 + 网格占用率 + 越界/重叠检测);
2. 告警预览(规则渲染 + 模拟触发评估);
3. NL 查询预览(模板 + 示例问句生成);
4. 从 ConfigStore 一键预览(含空配置的宽容处理);
5. PreviewResult 的 ok/to_dict。
"""
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401
from template_console.config_store import ConfigKind, ConfigStore # noqa: E402
from template_console.preview import ( # noqa: E402
AlarmRule,
GRID_COLUMNS,
NLQueryTemplate,
PreviewKind,
PreviewResult,
preview_from_store,
render_alarm_preview,
render_layout_preview,
render_nl_query_preview,
)
class LayoutPreviewTest(unittest.TestCase):
"""布局预览。"""
def test_basic_render(self):
widgets = [
{"type": "process_view", "src": "x.svg", "x": 0, "y": 0, "w": 12, "h": 4,
"description": "工艺流程"},
{"type": "trend", "bind": "R-801.TEMP", "x": 0, "y": 4, "w": 6, "h": 2},
]
r = render_layout_preview(widgets)
self.assertEqual(len(r.items), 2)
self.assertEqual(r.items[0]["type"], "process_view")
self.assertEqual(r.items[1]["bind"], "R-801.TEMP")
self.assertTrue(r.ok) # 无越界/重叠
self.assertTrue(any("网格占用率" in n for n in r.notes))
def test_overflow_warning(self):
# x+w 超过 12 列
widgets = [{"type": "trend", "bind": "p", "x": 8, "y": 0, "w": 6, "h": 2}]
r = render_layout_preview(widgets)
self.assertFalse(r.ok)
self.assertTrue(any("越界" in w for w in r.warnings))
def test_overlap_warning(self):
widgets = [
{"type": "trend", "bind": "a", "x": 0, "y": 0, "w": 6, "h": 2},
{"type": "kpi_card", "metric": "m", "label": "L", "x": 3, "y": 0, "w": 6, "h": 2},
]
r = render_layout_preview(widgets)
self.assertFalse(r.ok)
self.assertTrue(any("重叠" in w for w in r.warnings))
def test_bad_widget_type(self):
widgets = [{"type": "unknown", "x": 0, "y": 0, "w": 1, "h": 1}]
r = render_layout_preview(widgets)
self.assertFalse(r.ok)
self.assertTrue(any("非法类型" in w for w in r.warnings))
def test_kpi_card_carries_metric_and_label(self):
widgets = [{"type": "kpi_card", "metric": "yield", "label": "产率",
"x": 0, "y": 0, "w": 3, "h": 2}]
r = render_layout_preview(widgets)
self.assertEqual(r.items[0]["metric"], "yield")
self.assertEqual(r.items[0]["label"], "产率")
def test_empty_widgets(self):
r = render_layout_preview([])
self.assertEqual(r.items, [])
self.assertTrue(r.ok)
class AlarmPreviewTest(unittest.TestCase):
"""告警预览。"""
def test_render_rules(self):
rules = [AlarmRule("R-801.TEMP", "反应釜温度", ">", 120.0, "critical")]
r = render_alarm_preview(rules)
self.assertEqual(len(r.items), 1)
self.assertIn("critical", r.items[0]["text"])
self.assertTrue(r.ok)
def test_evaluate_triggered(self):
rules = [AlarmRule("R-801.TEMP", "温度", ">", 120.0, "critical")]
r = render_alarm_preview(rules, current_values={"R-801.TEMP": 130.0})
self.assertTrue(r.items[0]["triggered"])
self.assertEqual(r.items[0]["state"], "触发")
self.assertTrue(any("1/1" in n for n in r.notes))
def test_evaluate_not_triggered(self):
rules = [AlarmRule("P1", "温度", ">", 120.0, "warn")]
r = render_alarm_preview(rules, current_values={"P1": 100.0})
self.assertFalse(r.items[0]["triggered"])
self.assertEqual(r.items[0]["state"], "正常")
def test_operators(self):
for op, val, thr in [(">=", 120, 120), ("<", 50, 100), ("<=", 100, 100), ("==", 5, 5)]:
rule = AlarmRule("P", "m", op, thr, "warn")
self.assertTrue(rule.evaluate(val), f"{op} {val} {thr} 应触发")
def test_bad_level_and_operator(self):
rules = [AlarmRule("P", "m", "~", 1.0, level="boom")]
r = render_alarm_preview(rules)
self.assertFalse(r.ok)
self.assertTrue(any("非法级别" in w for w in r.warnings))
self.assertTrue(any("非法运算符" in w for w in r.warnings))
class NLQueryPreviewTest(unittest.TestCase):
"""NL 查询预览。"""
def test_render_with_examples(self):
t = NLQueryTemplate(
name="batch_query",
question_template="最近一批的{metric}是多少?",
slots={"metric": ["产率", "能耗"]},
data_source="tdengine",
answer_hint="返回当批聚合值",
)
r = render_nl_query_preview([t])
self.assertEqual(len(r.items), 1)
self.assertEqual(len(r.items[0]["examples"]), 2)
self.assertIn("产率", r.items[0]["examples"][0])
self.assertEqual(r.items[0]["data_source"], "tdengine")
def test_empty_question_warns(self):
t = NLQueryTemplate(name="x", question_template="", slots={})
r = render_nl_query_preview([t])
self.assertFalse(r.ok)
def test_no_slots_returns_template(self):
t = NLQueryTemplate(name="x", question_template="整体能耗?", slots={})
self.assertEqual(t.render_examples(), ["整体能耗?"])
class StorePreviewTest(unittest.TestCase):
"""从 ConfigStore 一键预览。"""
def setUp(self):
self._tmp = tempfile.mkdtemp()
self.store = ConfigStore(self._tmp)
def tearDown(self):
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
def test_layout_preview_from_store(self):
self.store.upsert(ConfigKind.LAYOUT, "dashboard",
[{"type": "trend", "bind": "p", "x": 0, "y": 0, "w": 6, "h": 2}])
r = preview_from_store(self.store, PreviewKind.LAYOUT)
self.assertEqual(len(r.items), 1)
self.assertEqual(r.items[0]["bind"], "p")
def test_empty_layout_is_graceful(self):
r = preview_from_store(self.store, PreviewKind.LAYOUT)
self.assertEqual(r.items, [])
self.assertTrue(any("未配置" in n for n in r.notes))
def test_alarm_preview_from_store(self):
self.store.upsert(ConfigKind.MODEL_PARAM, "alarm_temp",
{"point_id": "R-801.TEMP", "metric": "温度",
"operator": ">", "threshold": 120, "level": "critical"})
r = preview_from_store(self.store, PreviewKind.ALARM,
current_values={"R-801.TEMP": 130})
self.assertTrue(r.items[0]["triggered"])
def test_nl_query_preview_from_store(self):
self.store.upsert(ConfigKind.RAG_CONFIG, "nl_batch",
{"name": "批次查询", "question_template": "{m}多少?",
"slots": {"m": ["产率"]}, "data_source": "tdengine"})
r = preview_from_store(self.store, PreviewKind.NL_QUERY)
self.assertEqual(len(r.items), 1)
self.assertEqual(r.items[0]["name"], "批次查询")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,215 @@
# -*- coding: utf-8 -*-
"""配置推送契约测试(issue #67)。
覆盖:
1. manifest 构建(版本/快照/校验和/来源);
2. checksum 稳定性 + 完整性校验(verify);
3. 推送幂等(同版本不重复写文件、返回原 manifest);
4. force 强制重推;
5. retract 撤回 + 重新推送;
6. 模拟传输损坏(篡改 snapshot → verify 失败);
7. 推送日志与查询。
"""
import json
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401
from template_console.config_store import ConfigKind, ConfigStore # noqa: E402
from template_console.push_channel import ( # noqa: E402
PushChannel,
PushManifest,
PushRecord,
checksum,
)
from template_console.release import Release, ReleaseManager # noqa: E402
def _make_release(version: str = "1.0.0") -> Release:
"""构造一个带快照的 Release(不走文件系统,直接内存构造)。"""
snap = {
"schema_version": 1,
"captured_at": "2026-01-01T00:00:00Z",
"kinds": {"model_param": [
{"key": "lr", "value": 0.001, "kind": "model_param",
"meaning": "学习率", "updated_by": "li", "reason": "init",
"updated_at": "2026-01-01T00:00:00Z"}]},
}
return Release(version=version, created_at="2026-01-01T00:00:00Z",
snapshot=snap, released_by="admin", changelog="t", reason="r")
class _Tmp:
def __init__(self):
self._tmp = tempfile.mkdtemp()
self.inbox = os.path.join(self._tmp, "inbox")
def cleanup(self):
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
class ChecksumTest(unittest.TestCase):
"""校验和稳定性 + 完整性。"""
def test_checksum_stable(self):
snap = {"kinds": {"a": [1, 2, 3]}}
self.assertEqual(checksum(snap), checksum(snap))
def test_checksum_key_order_independent(self):
# 键序不影响校验和(规范序列化)
a = checksum({"x": 1, "y": 2})
b = checksum({"y": 2, "x": 1})
self.assertEqual(a, b)
def test_checksum_changes_on_value_change(self):
self.assertNotEqual(checksum({"v": 1}), checksum({"v": 2}))
def test_checksum_is_sha256_hex(self):
cs = checksum({"v": 1})
self.assertEqual(len(cs), 64)
self.assertTrue(all(c in "0123456789abcdef" for c in cs))
class ManifestTest(unittest.TestCase):
"""manifest 构建。"""
def test_build_manifest_has_checksum(self):
ch = PushChannel(inbox=tempfile.mkdtemp())
rel = _make_release()
m = ch.build_manifest(rel)
self.assertEqual(m.version, "1.0.0")
self.assertTrue(m.checksum)
self.assertEqual(m.source, "template-console")
self.assertTrue(m.generated_at)
def test_manifest_roundtrip(self):
m = PushManifest(version="1.0.0", snapshot={"a": 1},
checksum=checksum({"a": 1}), generated_at="t")
text = m.to_json()
m2 = PushManifest.from_dict(json.loads(text))
self.assertEqual(m2.version, "1.0.0")
self.assertEqual(m2.checksum, m.checksum)
class PushIdempotencyTest(unittest.TestCase):
"""推送幂等。"""
def setUp(self):
self.ctx = _Tmp()
self.ch = PushChannel(inbox=self.ctx.inbox)
self.rel = _make_release()
def tearDown(self):
self.ctx.cleanup()
def test_push_writes_manifest_file(self):
self.ch.push(self.rel, pushed_by="admin")
path = os.path.join(self.ctx.inbox, "manifest-1.0.0.json")
self.assertTrue(os.path.isfile(path))
def test_push_is_idempotent(self):
m1 = self.ch.push(self.rel, pushed_by="a")
m2 = self.ch.push(self.rel, pushed_by="b") # 重复推送
# 同版本返回同一 manifest(校验和一致)
self.assertEqual(m1.checksum, m2.checksum)
# 推送日志只有一条记录
self.assertEqual(len(self.ch.pushed_versions()), 1)
def test_force_overrides_idempotency(self):
self.ch.push(self.rel, pushed_by="a")
before = self.ch.pushed_versions()[0]["pushed_at"]
# force 重推(时间戳可能更新)
self.ch.push(self.rel, pushed_by="b", force=True)
records = self.ch.pushed_versions()
self.assertEqual(len(records), 1)
self.assertEqual(records[0]["pushed_by"], "b")
def test_push_log_records_pushed_by(self):
self.ch.push(self.rel, pushed_by="admin_zhang")
rec = self.ch.pushed_versions()[0]
self.assertEqual(rec["pushed_by"], "admin_zhang")
self.assertEqual(rec["status"], "pushed")
class RetractTest(unittest.TestCase):
"""撤回 + 重新推送。"""
def setUp(self):
self.ctx = _Tmp()
self.ch = PushChannel(inbox=self.ctx.inbox)
self.rel = _make_release()
def tearDown(self):
self.ctx.cleanup()
def test_retract_marks_status(self):
self.ch.push(self.rel)
self.assertTrue(self.ch.retract("1.0.0", by="admin", reason="有问题"))
rec = self.ch.pushed_versions()[0]
self.assertEqual(rec["status"], "retracted")
def test_retract_unknown_returns_false(self):
self.assertFalse(self.ch.retract("9.9.9"))
def test_retract_allows_repush(self):
self.ch.push(self.rel)
self.ch.retract("1.0.0")
# 撤回后可重新推送(幂等解除)
m = self.ch.push(self.rel, pushed_by="admin2")
rec = self.ch.pushed_versions()[0]
self.assertEqual(rec["status"], "pushed")
self.assertEqual(rec["pushed_by"], "admin2")
class VerifyTest(unittest.TestCase):
"""完整性校验。"""
def test_verify_valid_manifest(self):
ch = PushChannel(inbox=tempfile.mkdtemp())
m = ch.build_manifest(_make_release())
self.assertTrue(PushChannel.verify(m))
def test_verify_tampered_snapshot_fails(self):
ch = PushChannel(inbox=tempfile.mkdtemp())
m = ch.build_manifest(_make_release())
# 篡改 snapshot 但不改 checksum → 校验失败
m.snapshot["kinds"]["model_param"][0]["value"] = 0.999
self.assertFalse(PushChannel.verify(m))
def test_verify_payload_dict(self):
ch = PushChannel(inbox=tempfile.mkdtemp())
m = ch.build_manifest(_make_release())
self.assertTrue(PushChannel.verify_payload(m.to_dict()))
def test_verify_payload_bad_dict(self):
self.assertFalse(PushChannel.verify_payload({"nope": 1}))
class IntegrationTest(unittest.TestCase):
"""端到端:store → publish → push → verify。"""
def test_store_publish_push_flow(self):
tmp = tempfile.mkdtemp()
try:
store = ConfigStore(tmp)
store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
rm = ReleaseManager(store)
rel = rm.publish("1.0.0", released_by="admin", changelog="首发")
inbox = os.path.join(tmp, "inbox")
ch = PushChannel(inbox=inbox)
m = ch.push(rel, pushed_by="admin")
# 内核侧校验通过
self.assertTrue(PushChannel.verify(m))
self.assertEqual(len(ch.pushed_versions()), 1)
finally:
import shutil
shutil.rmtree(tmp, ignore_errors=True)
if __name__ == "__main__":
unittest.main()
+181
View File
@@ -0,0 +1,181 @@
# -*- coding: utf-8 -*-
"""三级 RBAC 权限模型测试(issue #62)。
覆盖:
1. 三级角色权限矩阵正确(readonly/engineer/admin);
2. 角色继承(admin 继承 engineer 继承 readonly);
3. has_permission 允许/拒绝判定 + 理由可解释;
4. 细粒度收窄(restricted_to_readonly 把写动作降级拒绝);
5. 便捷判定 can_publish / 用户概览。
"""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401
from template_console.rbac import ( # noqa: E402
Action,
Permission,
Resource,
Role,
RoleKind,
User,
all_roles,
can_publish,
effective_permissions,
get_role,
has_permission,
user_summary,
)
class RoleRegistryTest(unittest.TestCase):
"""三级角色注册表。"""
def test_three_roles_present(self):
roles = {r.kind for r in all_roles()}
self.assertEqual(roles, {RoleKind.READONLY, RoleKind.ENGINEER, RoleKind.ADMIN})
def test_role_labels_in_chinese(self):
self.assertEqual(get_role(RoleKind.READONLY).label, "只读")
self.assertEqual(get_role(RoleKind.ENGINEER).label, "行业工程师")
self.assertEqual(get_role(RoleKind.ADMIN).label, "管理员")
def test_role_descriptions_explainable(self):
# 可解释性:每个角色都有职责说明
for role in all_roles():
self.assertTrue(role.description, f"{role.kind} 缺少 description")
def test_inheritance_chain(self):
self.assertEqual(get_role(RoleKind.ADMIN).inherits, RoleKind.ENGINEER)
self.assertEqual(get_role(RoleKind.ENGINEER).inherits, RoleKind.READONLY)
self.assertIsNone(get_role(RoleKind.READONLY).inherits)
def test_permission_key_format(self):
p = Permission(RoleKind.ADMIN, Resource.USER, Action.MANAGE)
# 匹配键为资源:动作(角色无关,便于继承);审计键含授予角色
self.assertEqual(p.key(), "user:manage")
self.assertEqual(p.audit_key(), "admin:user:manage")
class EffectivePermissionTest(unittest.TestCase):
"""继承后的有效权限集合。"""
def test_admin_inherits_engineer_and_readonly(self):
eff = effective_permissions(RoleKind.ADMIN)
# 匹配键为 resource:action:admin 拥有自身的 user:manage,
# 也继承 engineer 的 model_param:edit 与 readonly 的 layout:view
self.assertIn("user:manage", eff)
self.assertIn("model_param:edit", eff)
self.assertIn("layout:view", eff)
def test_engineer_cannot_publish(self):
eff = effective_permissions(RoleKind.ENGINEER)
# 工程师不能发布/推送/管用户
self.assertNotIn("release:publish", eff)
self.assertNotIn("push:publish", eff)
self.assertNotIn("user:manage", eff)
def test_readonly_has_no_write(self):
eff = effective_permissions(RoleKind.READONLY)
for key in eff:
# 只读权限只能以 :view 结尾
self.assertTrue(key.endswith(":view"), f"readonly 不应有写/发布权限: {key}")
class HasPermissionTest(unittest.TestCase):
"""has_permission 判定 + 理由。"""
def setUp(self):
self.ro = User("viewer", RoleKind.READONLY, "查看员")
self.eng = User("li_engineer", RoleKind.ENGINEER, "李工")
self.admin = User("root_admin", RoleKind.ADMIN, "管理员甲")
def test_readonly_view_allowed(self):
d = has_permission(self.ro, Resource.LAYOUT, Action.VIEW)
self.assertTrue(d.allow)
self.assertEqual(d.source, "explicit")
def test_readonly_edit_denied(self):
d = has_permission(self.ro, Resource.LAYOUT, Action.EDIT)
self.assertFalse(d.allow)
self.assertIn("缺少权限", d.reason)
def test_engineer_edit_allowed_inherited_view(self):
# 工程师编辑是本角色权限(explicit)
d_edit = has_permission(self.eng, Resource.LAYOUT, Action.EDIT)
self.assertTrue(d_edit.allow)
self.assertEqual(d_edit.source, "explicit")
# 工程师查看布局是继承自 readonly(inherited)
d_view = has_permission(self.eng, Resource.LAYOUT, Action.VIEW)
self.assertTrue(d_view.allow)
self.assertEqual(d_view.source, "inherited")
def test_engineer_publish_denied(self):
d = has_permission(self.eng, Resource.RELEASE, Action.PUBLISH)
self.assertFalse(d.allow)
def test_admin_publish_allowed(self):
d = has_permission(self.admin, Resource.RELEASE, Action.PUBLISH)
self.assertTrue(d.allow)
self.assertEqual(d.source, "explicit")
def test_admin_inherited_engineer_edit(self):
d = has_permission(self.admin, Resource.MODEL_PARAM, Action.EDIT)
self.assertTrue(d.allow)
self.assertEqual(d.source, "inherited")
def test_decision_carries_reason(self):
# 可解释性:无论允许/拒绝,reason 非空且含用户名与资源
for user in (self.ro, self.eng, self.admin):
d = has_permission(user, Resource.PUSH, Action.PUBLISH)
self.assertIn(user.username, d.reason)
self.assertIn(Resource.PUSH.value, d.reason)
class RestrictedUserTest(unittest.TestCase):
"""细粒度收窄:restricted_to_readonly。"""
def test_restricted_engineer_cannot_edit_that_resource(self):
# 工程师本可编辑布局,但被收窄为只读后应拒绝
u = User("limited", RoleKind.ENGINEER, "受限工程师",
restricted_to_readonly=[Resource.LAYOUT])
d = has_permission(u, Resource.LAYOUT, Action.EDIT)
self.assertFalse(d.allow)
self.assertEqual(d.source, "restricted")
def test_restricted_engineer_can_still_view(self):
u = User("limited", RoleKind.ENGINEER, "受限工程师",
restricted_to_readonly=[Resource.LAYOUT])
d = has_permission(u, Resource.LAYOUT, Action.VIEW)
self.assertTrue(d.allow)
def test_restricted_only_affects_named_resource(self):
u = User("limited", RoleKind.ENGINEER, "受限工程师",
restricted_to_readonly=[Resource.LAYOUT])
# 模型超参未被收窄,仍可编辑
d = has_permission(u, Resource.MODEL_PARAM, Action.EDIT)
self.assertTrue(d.allow)
class ConvenienceTest(unittest.TestCase):
"""便捷判定与用户概览。"""
def test_can_publish(self):
self.assertFalse(can_publish(User("v", RoleKind.READONLY)))
self.assertFalse(can_publish(User("e", RoleKind.ENGINEER)))
self.assertTrue(can_publish(User("a", RoleKind.ADMIN)))
def test_user_summary(self):
s = user_summary(User("li", RoleKind.ENGINEER, "李工"))
self.assertEqual(s["username"], "li")
self.assertEqual(s["role"], "engineer")
self.assertEqual(s["role_label"], "行业工程师")
self.assertGreater(s["effective_permission_count"], 0)
self.assertEqual(s["restricted_to_readonly"], [])
if __name__ == "__main__":
unittest.main()
+193
View File
@@ -0,0 +1,193 @@
# -*- coding: utf-8 -*-
"""版本发布 + 回滚点测试(issue #66)。
覆盖:
1. semver 校验 / 比较 / 递增;
2. publish 发布(快照固化、单调递增、重复拒绝、空快照拒绝);
3. rollback 回滚(恢复快照、不删历史、回滚事件可追溯);
4. list/latest/get/history 查询;
5. 持久化(重开 manager 仍在)。
"""
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401
from template_console.config_store import ConfigKind, ConfigStore # noqa: E402
from template_console.release import ( # noqa: E402
Release,
ReleaseManager,
bump_patch,
is_valid_semver,
semver_gt,
semver_tuple,
)
class _Tmp:
def __init__(self):
self._tmp = tempfile.mkdtemp()
self.store = ConfigStore(self._tmp)
self.rm = ReleaseManager(self.store)
def cleanup(self):
import shutil
shutil.rmtree(self._tmp, ignore_errors=True)
class SemverTest(unittest.TestCase):
"""semver 工具。"""
def test_valid(self):
self.assertTrue(is_valid_semver("1.0.0"))
self.assertTrue(is_valid_semver("0.0.1"))
self.assertTrue(is_valid_semver("10.20.30"))
def test_invalid(self):
self.assertFalse(is_valid_semver("1.0"))
self.assertFalse(is_valid_semver("1.0.0.0"))
self.assertFalse(is_valid_semver("v1.0.0"))
self.assertFalse(is_valid_semver("1.0.0-rc"))
def test_tuple_and_gt(self):
self.assertEqual(semver_tuple("1.2.3"), (1, 2, 3))
self.assertTrue(semver_gt("1.0.1", "1.0.0"))
self.assertTrue(semver_gt("2.0.0", "1.9.9"))
self.assertFalse(semver_gt("1.0.0", "1.0.0"))
def test_bump_patch(self):
self.assertEqual(bump_patch("1.0.0"), "1.0.1")
self.assertEqual(bump_patch("0.9.9"), "0.9.10")
class PublishTest(unittest.TestCase):
"""发布。"""
def setUp(self):
self.ctx = _Tmp()
def tearDown(self):
self.ctx.cleanup()
def test_publish_requires_nonempty_store(self):
with self.assertRaises(ValueError):
self.ctx.rm.publish("1.0.0")
def test_publish_first_version(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
rel = self.ctx.rm.publish("1.0.0", released_by="admin", changelog="首次发布")
self.assertEqual(rel.version, "1.0.0")
self.assertEqual(rel.released_by, "admin")
self.assertIn("model_param", rel.snapshot["kinds"])
def test_publish_monotonic_increase(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.ctx.rm.publish("1.0.0")
# 较低版本号应被拒绝
with self.assertRaises(ValueError):
self.ctx.rm.publish("0.9.0")
# 相同版本号应被拒绝
with self.assertRaises(ValueError):
self.ctx.rm.publish("1.0.0")
# 更高版本 OK
self.ctx.rm.publish("1.0.1")
def test_publish_invalid_semver(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
with self.assertRaises(ValueError):
self.ctx.rm.publish("1.0")
def test_snapshot_captures_current_state(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.ctx.rm.publish("1.0.0")
# 发布后改配置,原版本快照不受影响
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.01)
rel1 = self.ctx.rm.get("1.0.0")
self.assertEqual(
rel1.snapshot["kinds"]["model_param"][0]["value"], 0.001)
class RollbackTest(unittest.TestCase):
"""回滚。"""
def setUp(self):
self.ctx = _Tmp()
def tearDown(self):
self.ctx.cleanup()
def test_rollback_restores_snapshot(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.ctx.rm.publish("1.0.0", changelog="v1 lr=0.001")
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.01)
self.ctx.rm.publish("1.1.0", changelog="v2 lr=0.01")
# 当前 store 的 lr 应是 0.01
self.assertEqual(self.ctx.store.get(ConfigKind.MODEL_PARAM, "lr").value, 0.01)
# 回滚到 1.0.0
target = self.ctx.rm.rollback("1.0.0", released_by="admin", reason="线上异常")
# store 恢复成 1.0.0 的快照
self.assertEqual(self.ctx.store.get(ConfigKind.MODEL_PARAM, "lr").value, 0.001)
self.assertEqual(target.version, "1.0.0")
def test_rollback_keeps_history(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.ctx.rm.publish("1.0.0")
self.ctx.rm.rollback("1.0.0")
# 回滚不删除任何版本
self.assertEqual(len(self.ctx.rm.list()), 1)
self.assertIsNotNone(self.ctx.rm.get("1.0.0"))
def test_rollback_records_event(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.ctx.rm.publish("1.0.0")
self.ctx.rm.rollback("1.0.0", released_by="admin", reason="紧急回滚")
rel = self.ctx.rm.get("1.0.0")
self.assertIn("回滚", rel.reason)
self.assertIn("紧急回滚", rel.reason)
def test_rollback_unknown_version(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.ctx.rm.publish("1.0.0")
with self.assertRaises(ValueError):
self.ctx.rm.rollback("9.9.9")
class QueryTest(unittest.TestCase):
"""查询 + 持久化。"""
def setUp(self):
self.ctx = _Tmp()
def tearDown(self):
self.ctx.cleanup()
def test_list_latest_history(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.ctx.rm.publish("1.0.0", changelog="c1")
self.ctx.rm.publish("1.1.0", changelog="c2")
self.assertEqual([r.version for r in self.ctx.rm.list()], ["1.0.0", "1.1.0"])
self.assertEqual(self.ctx.rm.latest().version, "1.1.0")
hist = self.ctx.rm.history()
self.assertEqual(len(hist), 2)
self.assertEqual(hist[1]["changelog"], "c2")
self.assertEqual(hist[1]["item_count"], 1)
def test_persistence_across_reopen(self):
self.ctx.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
self.ctx.rm.publish("1.0.0")
# 重开 manager(同一 store 目录)
store2 = ConfigStore(self.ctx._tmp)
rm2 = ReleaseManager(store2)
self.assertIsNotNone(rm2.get("1.0.0"))
self.assertEqual(rm2.latest().version, "1.0.0")
def test_get_nonexistent(self):
self.assertIsNone(self.ctx.rm.get("9.9.9"))
self.assertIsNone(self.ctx.rm.latest())
if __name__ == "__main__":
unittest.main()
@@ -13,8 +13,10 @@
9. 空布局 / 边界。 9. 空布局 / 边界。
""" """
import os import os
import sys
import unittest import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401 (sys.path 挂载) import _bootstrap # noqa: F401 (sys.path 挂载)
from layout_validator import ( from layout_validator import (