bot_dev1
1290a8bc54
纯标准库实现，对齐 PRD 5.5「⑤ 配置化驾驶舱」iAOP-cockpit-layout-v1： - cockpit.ti.yaml：海绵钛四状态工艺流程（氯化→精制→还原→蒸馏）， process_view 主视图声明 stages 覆盖，trend/kpi_card bind 对齐 point_dict.default.csv 的 point_id（CLF-01/RF-01/E-01/ST-01）。 - layout_validator.py：LayoutValidator 校验 widget 类型合法、12 列网格 不越界、bind point_id 在点位字典内、四状态覆盖完整（order 单调/id 唯一）； 零依赖 YAML 子集解析（复制 impurity-forecast _parse_yaml_subset）。 - tests：18 用例覆盖真实资产端到端、非法类型、网格越界/负坐标/零宽、 bind 漂移、缺 process_view、缺必需状态、order 非单调、stage 重复、 $schema 头、CSV 加载、空布局。 - _sanity_check.py：冒烟验证 9 widgets 全校验通过。
308 lines
12 KiB
Python
308 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""海绵钛驾驶舱布局校验器测试（Issue #55）。
|
||
|
||
覆盖：
|
||
1. 真实 cockpit.ti.yaml + point_dict.default.csv 全部通过（端到端）；
|
||
2. widget 类型合法集合（非法类型 → ERROR）；
|
||
3. 12 列网格校验（越界/负坐标/非正 w/h）；
|
||
4. bind point_id 在点位字典内（漂移 → ERROR）；
|
||
5. process_view 四状态覆盖（缺 stage id / order 非单调 / 缺必需状态）；
|
||
6. $schema 头校验；
|
||
7. YAML 解析（flow map / 嵌套）；
|
||
8. 点位字典 CSV 加载（缺表头/空文件）；
|
||
9. 空布局 / 边界。
|
||
"""
|
||
import os
|
||
import unittest
|
||
|
||
import _bootstrap # noqa: F401 (sys.path 挂载)
|
||
|
||
from layout_validator import (
|
||
GRID_COLUMNS,
|
||
LayoutError,
|
||
LayoutValidator,
|
||
REQUIRED_STAGES,
|
||
Severity,
|
||
load_point_ids,
|
||
)
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
DASHBOARD_DIR = os.path.dirname(HERE)
|
||
LAYOUT_YAML = os.path.join(DASHBOARD_DIR, "cockpit.ti.yaml")
|
||
POINT_DICT_CSV = os.path.join(
|
||
DASHBOARD_DIR, os.pardir, "point-dict", "point_dict.default.csv")
|
||
|
||
|
||
def _write_layout(tmp_path: str, content: str) -> str:
|
||
"""把布局内容写到临时文件，返回路径。"""
|
||
path = os.path.join(tmp_path, "cockpit.test.yaml")
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
fh.write(content)
|
||
return path
|
||
|
||
|
||
def _write_point_dict(tmp_path: str, ids: list) -> str:
|
||
"""写一个最小点位字典 CSV（仅 point_id 列）。"""
|
||
path = os.path.join(tmp_path, "points.csv")
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
fh.write("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\n")
|
||
for i, pid in enumerate(ids):
|
||
fh.write(f"D{i},{pid},n,u,float,1000,true,n,simulator\n")
|
||
return path
|
||
|
||
|
||
# 最小合法布局模板（便于构造各类变形）
|
||
_VALID_LAYOUT = """\
|
||
$schema: iAOP-cockpit-layout-v1
|
||
title: 测试驾驶舱
|
||
theme: dark
|
||
widgets:
|
||
- type: process_view
|
||
src: ti_four_state.svg
|
||
x: 0
|
||
y: 0
|
||
w: 12
|
||
h: 4
|
||
description: 四状态工艺流程
|
||
stages:
|
||
- id: chlorination
|
||
name: 氯化
|
||
order: 1
|
||
- id: purification
|
||
name: 精制
|
||
order: 2
|
||
- id: reduction
|
||
name: 还原
|
||
order: 3
|
||
- id: distillation
|
||
name: 蒸馏
|
||
order: 4
|
||
- type: trend
|
||
bind: CLF-01.TEMP
|
||
x: 0
|
||
y: 4
|
||
w: 6
|
||
h: 2
|
||
description: 氯化炉温度
|
||
"""
|
||
|
||
|
||
class TestEndToEndRealAssets(unittest.TestCase):
|
||
"""真实 cockpit.ti.yaml + point_dict.default.csv 端到端校验。"""
|
||
|
||
def test_real_layout_passes(self):
|
||
report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
|
||
if not report.passed:
|
||
for issue in report.errors:
|
||
print("ERROR:", issue.widget_id, issue.field, issue.reason)
|
||
self.assertTrue(report.passed, "真实布局应通过全部校验")
|
||
self.assertGreater(report.widget_count, 0)
|
||
|
||
def test_real_layout_has_process_view_with_four_stages(self):
|
||
report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
|
||
# 无 stages 相关 ERROR
|
||
stage_errors = [i for i in report.errors if i.field == "stages"]
|
||
self.assertEqual(stage_errors, [])
|
||
|
||
|
||
class TestWidgetType(unittest.TestCase):
|
||
"""widget 类型合法性。"""
|
||
|
||
def test_invalid_widget_type_error(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
layout = _VALID_LAYOUT.replace("type: trend", "type: radar_chart")
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
errors = [i for i in report.errors if i.field == "type"]
|
||
self.assertEqual(len(errors), 1)
|
||
self.assertIn("非法 widget 类型", errors[0].reason)
|
||
|
||
|
||
class TestGridBounds(unittest.TestCase):
|
||
"""12 列网格校验。"""
|
||
|
||
def test_x_plus_w_exceeds_columns(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# trend x=10 w=6 → 16 > 12
|
||
layout = _VALID_LAYOUT.replace(
|
||
" bind: CLF-01.TEMP\n x: 0\n y: 4\n w: 6\n h: 2",
|
||
" bind: CLF-01.TEMP\n x: 10\n y: 4\n w: 6\n h: 2")
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
grid_errors = [i for i in report.errors if i.field == "grid"
|
||
and "越出" in i.reason]
|
||
self.assertEqual(len(grid_errors), 1)
|
||
|
||
def test_negative_x_rejected(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
layout = _VALID_LAYOUT.replace(" x: 0\n y: 4\n w: 6\n h: 2",
|
||
" x: -1\n y: 4\n w: 6\n h: 2")
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
neg = [i for i in report.errors if "坐标不能为负" in i.reason]
|
||
self.assertEqual(len(neg), 1)
|
||
|
||
def test_zero_width_rejected(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
layout = _VALID_LAYOUT.replace(" x: 0\n y: 4\n w: 6\n h: 2",
|
||
" x: 0\n y: 4\n w: 0\n h: 2")
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
wh = [i for i in report.errors if "w/h 必须为正整数" in i.reason]
|
||
self.assertEqual(len(wh), 1)
|
||
|
||
|
||
class TestBindPointId(unittest.TestCase):
|
||
"""bind point_id 在点位字典内。"""
|
||
|
||
def test_bind_not_in_dict_error(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = _write_layout(td, _VALID_LAYOUT)
|
||
# 点位字典不含 CLF-01.TEMP
|
||
csv_path = _write_point_dict(td, ["OTHER-01.X"])
|
||
report = LayoutValidator(path, csv_path).validate()
|
||
bind_err = [i for i in report.errors if i.field == "bind"]
|
||
self.assertEqual(len(bind_err), 1)
|
||
self.assertIn("不在点位字典内", bind_err[0].reason)
|
||
|
||
def test_bind_in_dict_passes(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = _write_layout(td, _VALID_LAYOUT)
|
||
csv_path = _write_point_dict(td, ["CLF-01.TEMP"])
|
||
report = LayoutValidator(path, csv_path).validate()
|
||
bind_err = [i for i in report.errors if i.field == "bind"]
|
||
self.assertEqual(bind_err, [])
|
||
|
||
|
||
class TestProcessViewStages(unittest.TestCase):
|
||
"""process_view 四状态覆盖。"""
|
||
|
||
def test_missing_process_view_error(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 删除 process_view 块（保留 trend）
|
||
layout = """\
|
||
$schema: iAOP-cockpit-layout-v1
|
||
title: t
|
||
widgets:
|
||
- type: trend
|
||
bind: CLF-01.TEMP
|
||
x: 0
|
||
y: 0
|
||
w: 6
|
||
h: 2
|
||
"""
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
pv_err = [i for i in report.errors if i.field == "process_view"]
|
||
self.assertEqual(len(pv_err), 1)
|
||
|
||
def test_missing_required_stage_error(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 删除 distillation stage
|
||
layout = _VALID_LAYOUT.replace(
|
||
" - id: distillation\n name: 蒸馏\n order: 4\n", "")
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
missing = [i for i in report.errors if "未覆盖必需四状态" in i.reason]
|
||
self.assertEqual(len(missing), 1)
|
||
self.assertIn("distillation", missing[0].reason)
|
||
|
||
def test_non_monotonic_order_error(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 把 reduction order 改为 5（> distillation 的 4）→ 非单调
|
||
layout = _VALID_LAYOUT.replace(" - id: reduction\n name: 还原\n order: 3",
|
||
" - id: reduction\n name: 还原\n order: 5")
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
order_err = [i for i in report.errors if "order 非单调递增" in i.reason]
|
||
self.assertEqual(len(order_err), 1)
|
||
|
||
def test_duplicate_stage_id_warn(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 重复 chlorination（覆盖必需状态校验仍过，但 WARN 重复）
|
||
layout = _VALID_LAYOUT + """\
|
||
"""
|
||
# 构造一个有重复 stage 的 process_view（替换 stages 块）
|
||
dup_layout = _VALID_LAYOUT.replace(
|
||
" - id: distillation\n name: 蒸馏\n order: 4",
|
||
" - id: distillation\n name: 蒸馏\n order: 4\n"
|
||
" - id: chlorination\n name: 氯化2\n order: 5")
|
||
path = _write_layout(td, dup_layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
dup = [i for i in report.issues if "stage id 重复" in i.reason]
|
||
self.assertEqual(len(dup), 1)
|
||
|
||
|
||
class TestSchemaHeader(unittest.TestCase):
|
||
"""$schema 头校验。"""
|
||
|
||
def test_wrong_schema_error(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
layout = _VALID_LAYOUT.replace("iAOP-cockpit-layout-v1", "some-other-schema")
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
schema_err = [i for i in report.errors if i.field == "$schema"]
|
||
self.assertEqual(len(schema_err), 1)
|
||
|
||
|
||
class TestPointDictLoader(unittest.TestCase):
|
||
"""点位字典 CSV 加载。"""
|
||
|
||
def test_load_point_ids(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
csv_path = _write_point_dict(td, ["A.X", "B.Y"])
|
||
ids = load_point_ids(csv_path)
|
||
self.assertEqual(ids, ["A.X", "B.Y"])
|
||
|
||
def test_missing_csv_raises(self):
|
||
with self.assertRaises(LayoutError):
|
||
load_point_ids("/nonexistent/points.csv")
|
||
|
||
def test_csv_missing_point_id_column_raises(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = os.path.join(td, "bad.csv")
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
fh.write("device_id,name\nD1,n\n")
|
||
with self.assertRaises(LayoutError):
|
||
load_point_ids(path)
|
||
|
||
|
||
class TestReportExport(unittest.TestCase):
|
||
"""报告序列化 + 边界。"""
|
||
|
||
def test_empty_layout_error(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
layout = """\
|
||
$schema: iAOP-cockpit-layout-v1
|
||
title: t
|
||
widgets: []
|
||
"""
|
||
path = _write_layout(td, layout)
|
||
report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
|
||
self.assertFalse(report.passed)
|
||
|
||
def test_report_to_dict(self):
|
||
report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
|
||
d = report.to_dict()
|
||
self.assertEqual(d["passed"], True)
|
||
self.assertIn("widget_count", d)
|
||
self.assertEqual(d["error_count"], 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|