bot_dev1
78bfc5d73d
把已交付树脂驾驶舱布局资产（cockpit.resin.yaml）提取为可复用模板：
- dashboard/extractor.py：extract_template（变量化 R-801.TEMP→{device}.{point}、
metric→{metric}）、instantiate（点位字典回填）、diff_layouts（差异比对）；
- PointDict.from_csv 加载 9 列点位字典；零依赖 _parse_yaml_subset 解析 YAML；
- tests/test_extractor.py 19 用例全部通过。
279 lines
11 KiB
Python
279 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""树脂驾驶舱布局提取器测试（issue #84）。
|
||
|
||
覆盖（>=8 用例）：
|
||
1. load_layout：驾驶舱布局资产解析（schema/title/theme/widgets 9 个）；
|
||
2. extract_template：trend bind 变量化（R-801.TEMP → {device}.{point}）；
|
||
3. extract_template：kpi metric 变量化（→ {metric}）；
|
||
4. extract_template：占位符语义（placeholder_binding）+ 提取来源记录；
|
||
5. PointDict.from_csv：21 条点位、device 去重；
|
||
6. PointDict.lookup：按设备/测点查询；
|
||
7. instantiate：用点位字典回填 bind（具体点位）；
|
||
8. instantiate：metric 回填为模板原指标名（业务语义键不变）；
|
||
9. instantiate：去除内部字段（extracted_from/placeholder_binding）；
|
||
10. diff_layouts：一致布局 → is_identical；
|
||
11. diff_layouts：bind 变更 → binding_changes；
|
||
12. diff_layouts：几何变更 → geometry_changes；
|
||
13. WidgetType.is_allowed / COCKPIT_SCHEMA 常量。
|
||
14. 端到端：extract → instantiate → 与原布局一致（回填幂等性）。
|
||
"""
|
||
import copy
|
||
import os
|
||
import sys
|
||
import unittest
|
||
|
||
# 把 templates/resin/dashboard 挂到 sys.path 以便 `from extractor import ...`
|
||
_RESIN_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
_DASHBOARD_DIR = os.path.join(_RESIN_DIR, "dashboard")
|
||
sys.path.insert(0, _DASHBOARD_DIR)
|
||
|
||
from extractor import ( # noqa: E402
|
||
BIND_TEMPLATE,
|
||
COCKPIT_SCHEMA,
|
||
DEVICE_PLACEHOLDER,
|
||
METRIC_PLACEHOLDER,
|
||
POINT_PLACEHOLDER,
|
||
DEFAULT_LAYOUT_PATH,
|
||
DEFAULT_POINT_DICT_PATH,
|
||
ALLOWED_WIDGETS,
|
||
LayoutDiff,
|
||
PointDict,
|
||
PointEntry,
|
||
WidgetType,
|
||
diff_layouts,
|
||
extract_template,
|
||
instantiate,
|
||
load_layout,
|
||
)
|
||
|
||
|
||
class TestYAMLSubsetParse(unittest.TestCase):
|
||
"""零依赖 YAML 子集解析（驾驶舱布局资产）。"""
|
||
|
||
def test_load_layout_basic(self):
|
||
"""#1 驾驶舱资产解析：schema/title/theme/widgets 9 个。"""
|
||
layout = load_layout()
|
||
self.assertEqual(layout["$schema"], COCKPIT_SCHEMA)
|
||
self.assertEqual(layout["title"], "吸附树脂车间驾驶舱")
|
||
self.assertEqual(layout["theme"], "dark")
|
||
self.assertEqual(len(layout["widgets"]), 9)
|
||
|
||
def test_widget_types_all_allowed(self):
|
||
"""#13 所有 widget 类型在 PRD 5.5 schema 允许集合内。"""
|
||
layout = load_layout()
|
||
for w in layout["widgets"]:
|
||
self.assertIn(w["type"], ALLOWED_WIDGETS)
|
||
self.assertTrue(WidgetType.is_allowed(w["type"]))
|
||
|
||
def test_parse_inline_scalar_types(self):
|
||
"""YAML 行内标量解析：int/float/bool/None/带引号字符串。"""
|
||
from extractor import _parse_yaml_subset, _parse_value
|
||
self.assertEqual(_parse_value("12"), 12)
|
||
self.assertEqual(_parse_value("4.5"), 4.5)
|
||
self.assertIs(_parse_value("true"), True)
|
||
self.assertIs(_parse_value("false"), False)
|
||
self.assertIsNone(_parse_value("null"))
|
||
self.assertEqual(_parse_value("'x'"), "x")
|
||
# 缩进嵌套 + 列表项 mapping
|
||
doc = _parse_yaml_subset(
|
||
"title: T\nwidgets:\n - type: trend\n bind: R1.X\n x: 1\n")
|
||
self.assertEqual(doc["title"], "T")
|
||
self.assertEqual(doc["widgets"][0]["type"], "trend")
|
||
self.assertEqual(doc["widgets"][0]["bind"], "R1.X")
|
||
self.assertEqual(doc["widgets"][0]["x"], 1)
|
||
|
||
|
||
class TestExtractTemplate(unittest.TestCase):
|
||
"""布局资产 → 通用模板（变量化）。"""
|
||
|
||
def setUp(self):
|
||
self.layout = load_layout()
|
||
self.tmpl = extract_template(self.layout)
|
||
|
||
def test_trend_bind_placeholderized(self):
|
||
"""#2 trend bind：R-801.TEMP → {device}.{point}。"""
|
||
trend_binds = [w.get("bind") for w in self.tmpl["widgets"]
|
||
if w.get("type") == "trend"]
|
||
self.assertTrue(trend_binds)
|
||
for b in trend_binds:
|
||
self.assertEqual(b, BIND_TEMPLATE)
|
||
|
||
def test_kpi_metric_placeholderized(self):
|
||
"""#3 kpi metric：resin_exchange_capacity → {metric}。"""
|
||
kpi_metrics = [w.get("metric") for w in self.tmpl["widgets"]
|
||
if w.get("type") == "kpi_card"]
|
||
self.assertEqual(len(kpi_metrics), 4)
|
||
for m in kpi_metrics:
|
||
self.assertEqual(m, METRIC_PLACEHOLDER)
|
||
|
||
def test_placeholder_semantics_and_source(self):
|
||
"""#4 占位符语义（placeholder_binding）+ 提取来源记录。"""
|
||
# 占位符语义说明
|
||
pb = self.tmpl["placeholder_binding"]
|
||
self.assertIn(DEVICE_PLACEHOLDER, pb)
|
||
self.assertIn(POINT_PLACEHOLDER, pb)
|
||
self.assertIn(METRIC_PLACEHOLDER, pb)
|
||
# trend widget 记录提取来源（device/point 二元组）
|
||
trend = next(w for w in self.tmpl["widgets"]
|
||
if w.get("type") == "trend")
|
||
self.assertEqual(trend["extracted_from"]["bind"], ["R-801", "TEMP"])
|
||
# kpi widget 记录原指标名
|
||
kpi = next(w for w in self.tmpl["widgets"]
|
||
if w.get("type") == "kpi_card")
|
||
self.assertEqual(kpi["extracted_from"]["metric"],
|
||
"resin_exchange_capacity")
|
||
|
||
def test_extract_from_yaml_text_and_path(self):
|
||
"""extract_template 接受路径 / dict / YAML 文本三种入参。"""
|
||
tmpl_from_path = extract_template(DEFAULT_LAYOUT_PATH)
|
||
self.assertEqual(tmpl_from_path["$schema"], COCKPIT_SCHEMA)
|
||
tmpl_from_dict = extract_template(self.layout)
|
||
self.assertEqual(tmpl_from_dict["title"], self.tmpl["title"])
|
||
|
||
|
||
class TestPointDict(unittest.TestCase):
|
||
"""点位字典加载与查询。"""
|
||
|
||
def setUp(self):
|
||
self.pd = PointDict.from_csv()
|
||
|
||
def test_csv_load_count_and_devices(self):
|
||
"""#5 21 条点位 + device 去重保序。"""
|
||
self.assertEqual(len(self.pd.entries), 21)
|
||
self.assertEqual(self.pd.entries[0].device_id, "R-801")
|
||
self.assertEqual(self.pd.devices()[0], "R-801")
|
||
# device 去重（R-801 出现 6 次，去重后 1）
|
||
self.assertEqual(self.pd.devices().count("R-801"), 1)
|
||
|
||
def test_lookup_and_point_short_name(self):
|
||
"""#6 查询某设备测点 + point 短名（去设备前缀）。"""
|
||
r801 = self.pd.lookup("R-801")
|
||
self.assertTrue(r801)
|
||
# 短测点名剥离设备前缀
|
||
temps = [e.point for e in r801 if e.point == "TEMP"]
|
||
self.assertEqual(temps, ["TEMP"])
|
||
# 精确测点过滤
|
||
temp_only = self.pd.lookup("R-801", "TEMP")
|
||
self.assertEqual(len(temp_only), 1)
|
||
self.assertEqual(temp_only[0].unit, "℃")
|
||
# 未知设备 → 空列表
|
||
self.assertEqual(self.pd.lookup("NOPE"), [])
|
||
|
||
|
||
class TestInstantiate(unittest.TestCase):
|
||
"""模板 + 点位字典 → 具体布局（回填占位符）。"""
|
||
|
||
def setUp(self):
|
||
self.pd = PointDict.from_csv()
|
||
self.tmpl = extract_template(load_layout())
|
||
self.concrete = instantiate(self.tmpl, self.pd)
|
||
|
||
def test_backfill_trend_bind(self):
|
||
"""#7 trend bind 回填为具体点位（R-801.TEMP）。"""
|
||
binds = [w.get("bind") for w in self.concrete["widgets"]
|
||
if w.get("type") == "trend"]
|
||
self.assertEqual(binds, ["R-801.TEMP", "R-801.AGIT"])
|
||
# 不残留占位符
|
||
for b in binds:
|
||
self.assertNotIn(DEVICE_PLACEHOLDER, b)
|
||
|
||
def test_backfill_metric_keeps_business_key(self):
|
||
"""#8 metric 回填为模板原指标名（业务语义键不变）。"""
|
||
metrics = [w.get("metric") for w in self.concrete["widgets"]
|
||
if w.get("type") == "kpi_card"]
|
||
self.assertEqual(metrics[0], "resin_exchange_capacity")
|
||
|
||
def test_instantiate_strips_internal_fields(self):
|
||
"""#9 实例化输出去除 extracted_from / placeholder_binding 内部字段。"""
|
||
self.assertNotIn("placeholder_binding", self.concrete)
|
||
for w in self.concrete["widgets"]:
|
||
self.assertNotIn("extracted_from", w)
|
||
# 输出仍是合法布局（schema/widgets 齐全）
|
||
self.assertEqual(self.concrete["$schema"], COCKPIT_SCHEMA)
|
||
|
||
def test_instantiate_fallback_unknown_device(self):
|
||
"""占位符回填：来源设备不在点位字典 → 回退首个设备。"""
|
||
tmpl = copy.deepcopy(self.tmpl)
|
||
# 篡改 trend 来源为字典中不存在的设备
|
||
for w in tmpl["widgets"]:
|
||
if w.get("type") == "trend":
|
||
w["extracted_from"]["bind"] = ["ZZZ-999", "TEMP"]
|
||
break
|
||
conc = instantiate(tmpl, self.pd)
|
||
bind0 = next(w.get("bind") for w in conc["widgets"]
|
||
if w.get("type") == "trend")
|
||
self.assertTrue(bind0.startswith(self.pd.devices()[0] + "."))
|
||
|
||
def test_instantiate_rejects_empty_pointdict(self):
|
||
"""空点位字典 → ValueError（无法回填）。"""
|
||
empty = PointDict(entries=[])
|
||
with self.assertRaises(ValueError):
|
||
instantiate(self.tmpl, empty)
|
||
|
||
|
||
class TestDiffLayouts(unittest.TestCase):
|
||
"""布局差异比对。"""
|
||
|
||
def setUp(self):
|
||
self.layout = load_layout()
|
||
|
||
def test_identical_layout(self):
|
||
"""#10 相同布局 → is_identical。"""
|
||
d = diff_layouts(self.layout, self.layout)
|
||
self.assertTrue(d.is_identical)
|
||
self.assertIn("完全一致", d.summary()[0])
|
||
|
||
def test_binding_change_detected(self):
|
||
"""#11 bind 变更 → binding_changes（换行业回归比对）。"""
|
||
b = copy.deepcopy(self.layout)
|
||
for w in b["widgets"]:
|
||
if w.get("type") == "trend" and w.get("bind") == "R-801.TEMP":
|
||
w["bind"] = "R-802.TEMP"
|
||
d = diff_layouts(self.layout, b)
|
||
self.assertFalse(d.is_identical)
|
||
self.assertTrue(any("bind" in s for s in d.binding_changes))
|
||
self.assertTrue(any("R-802.TEMP" in s for s in d.binding_changes))
|
||
|
||
def test_geometry_change_detected(self):
|
||
"""#12 几何位置变更 → geometry_changes。"""
|
||
b = copy.deepcopy(self.layout)
|
||
b["widgets"][0]["x"] = 99
|
||
b["title"] = "改了"
|
||
d = diff_layouts(self.layout, b)
|
||
self.assertFalse(d.is_identical)
|
||
self.assertTrue(d.title_changed)
|
||
self.assertTrue(any("widget[0].x" in s for s in d.geometry_changes))
|
||
|
||
def test_widget_count_diff(self):
|
||
"""widget 数量不同 → 摘要体现。"""
|
||
a = {"$schema": COCKPIT_SCHEMA, "title": "t", "widgets": [
|
||
{"type": "trend", "bind": "R-801.TEMP"}]}
|
||
b = {"$schema": COCKPIT_SCHEMA, "title": "t", "widgets": []}
|
||
d = diff_layouts(a, b)
|
||
self.assertEqual(d.widget_count_a, 1)
|
||
self.assertEqual(d.widget_count_b, 0)
|
||
self.assertFalse(d.is_identical)
|
||
|
||
|
||
class TestEndToEndRoundTrip(unittest.TestCase):
|
||
"""端到端：extract → instantiate 与原布局一致（回填幂等性）。"""
|
||
|
||
def test_roundtrip_identical_to_source(self):
|
||
"""extract 后用同一点位字典 instantiate，结果与原布局一致。
|
||
|
||
验证「布局资产 ↔ 模板 ↔ 点位字典」三者解耦后可无损还原——
|
||
这是模板可复用的关键：换行业只需换点位字典即可重排驾驶舱。
|
||
"""
|
||
layout = load_layout()
|
||
pd = PointDict.from_csv()
|
||
tmpl = extract_template(layout)
|
||
concrete = instantiate(tmpl, pd)
|
||
d = diff_layouts(layout, concrete)
|
||
self.assertTrue(
|
||
d.is_identical,
|
||
f"回填结果与原布局不一致：{d.summary()}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|