feat(#55/#62/#63/#64/#65/#66/#67): 模板配置台全链路 + Ti 布局模板(7 子任务,105 测试通过) #130

Closed
bot_dev1 wants to merge 9 commits from feature/issue-62 into main
4 changed files with 565 additions and 0 deletions
Showing only changes of commit 4adfbb78fa - Show all commits
+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"
+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],
}
+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)
+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()