140 lines
5.0 KiB
Python
140 lines
5.0 KiB
Python
# -*- 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())
|