189 lines
7.2 KiB
Python
189 lines
7.2 KiB
Python
|
# -*- 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()
|