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