iAOP/core/template-console/_sanity_check.py

# -*- 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())