"""手套检查项 —— 每一项都必须**被证明能抓到失败**。 真实数据上 5 项全绿（6 集、左右手各 5-6 条流、零误报）。但"全绿"证明不了什么： **检查项从来没触发过，和检查项根本跑不起来，报告长得一模一样。** 所以这里每条 都构造一份明确的坏数据，断言它被抓到；再造一份好数据，断言不误报。 阈值全部来自实测： * 阵列 16×16 = 256 维 * 取值门限 500（0 到 500 之间没有值） * 设备更新 ~7.5Hz → 受压时静止**恰好 4 帧**（正常，不是卡死） * 整集最低有 11.4% 的帧有压力 pytest tests/test_cleaning_glove.py python3 tests/test_cleaning_glove.py """ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # noqa: E402 from app.processing.cleaning.checks import ( # noqa: E402 CheckContext, StreamEvidence, applicable_checks, get_check, PASS, WARN, FAIL, ) SIZE = 256 # ── 构造 ──────────────────────────────────────────────────── def _frame(value: float = 0.0, size: int = SIZE) -> tuple: return tuple([value] * size) def _context(series: dict, *, channel="tactile", key="left_glove", fps=30.0) -> CheckContext: stream = StreamEvidence(key=key, channel=channel, series=series) return CheckContext(batch_dir=Path("/nonexistent"), fps=fps, streams=(stream,)) def _run(slug: str, ctx: CheckContext, params: dict | None = None): ctx.params = params or {} findings = get_check(slug).run(ctx) assert findings, f"{slug} 一条结论都没产出 —— 报告会显示为『没问题』" return findings def _only(slug, ctx, params=None): findings = _run(slug, ctx, params) statuses = {f.status for f in findings} assert len(statuses) == 1, f"期望单一结论，得到 {statuses}" return findings[0] # ── 可达性：检查项必须真的会被选中 ─────────────────────────── def test_glove_checks_are_applicable(): """★ 防"检查项静默不跑"—— 声明与闸门对不上时它们一条都不会执行。""" slugs = {c.slug for c in applicable_checks( channels={"tactile", "device_status", "time"}, modalities={"glove_sensor"})} for expected in ("glove.array_shape", "glove.device_status", "glove.frozen_array", "glove.gate_floor", "glove.no_contact"): assert expected in slugs, f"{expected} 在手套数据上不可达" def test_glove_checks_skip_other_devices(): """UMI 夹爪上不该跑手套检查 —— 设备卡片闸门要生效。""" slugs = {c.slug for c in applicable_checks( channels={"tactile", "slam", "time"}, modalities={"gripper_device"})} assert not any(s.startswith("glove.") for s in slugs) def test_device_status_check_needs_its_channel(): """没有状态列（老批次）时不该跑 —— 跑起来只会报"没数据"。""" slugs = {c.slug for c in applicable_checks( channels={"tactile", "time"}, modalities={"glove_sensor"})} assert "glove.device_status" not in slugs assert "glove.frozen_array" in slugs # ── 1. 连接状态 ───────────────────────────────────────────── def test_status_constant_connected_passes(): ctx = _context({"device_status": tuple(["connected"] * 100)}, channel="device_status") assert _only("glove.device_status", ctx).status is PASS def test_status_change_mid_recording_fails(): """★ 从"连着"变成别的 = 掉线，与状态词表无关，见到即拦。""" values = tuple(["connected"] * 50 + ["disconnected"] * 50) ctx = _context({"device_status": values}, channel="device_status") finding = _only("glove.device_status", ctx) assert finding.status is FAIL assert finding.metrics["first_change_frame"] == 50 def test_status_unknown_constant_only_warns(): """词表只见过 connected；别的取值不拦批次（拦错了比不拦更糟）。""" ctx = _context({"device_status": tuple(["ok"] * 100)}, channel="device_status") assert _only("glove.device_status", ctx).status is WARN def test_status_all_blank_warns(): ctx = _context({"device_status": tuple([""] * 100)}, channel="device_status") assert _only("glove.device_status", ctx).status is WARN # ── 2. 阵列完整性 ─────────────────────────────────────────── def test_shape_full_array_passes(): ctx = _context({"tactile": tuple([_frame()] * 50)}) assert _only("glove.array_shape", ctx).status is PASS def test_shape_catches_truncation(): """★ 正是那个真实踩过的 bug：256 维被静默截成 3 维。""" ctx = _context({"tactile": tuple([_frame(size=3)] * 500)}) finding = _only("glove.array_shape", ctx) assert finding.status is FAIL assert finding.metrics["bad_ratio"] == 1.0 assert finding.metrics["observed_sizes"] == [3] def test_shape_single_null_frame_only_warns(): """偶发一帧缺失不该拦住整批。""" values = [_frame(600.0)] * 999 + [()] ctx = _context({"tactile": tuple(values)}) assert _only("glove.array_shape", ctx).status is WARN # ── 3. 取值下限 ───────────────────────────────────────────── def test_gate_floor_normal_values_pass(): ctx = _context({"tactile": (_frame(0.0), _frame(508.09), _frame(1500.0))}) assert _only("glove.gate_floor", ctx).status is PASS def test_gate_floor_catches_encoding_change(): """★ 固件改成直接输出原始 ADC 时，值会掉到门限以下 —— 而别的检查全绿。""" values = [_frame(v) for v in (5.0, 20.0, 137.0, 400.0)] + [_frame(900.0)] ctx = _context({"tactile": tuple(values)}) finding = _only("glove.gate_floor", ctx) assert finding.status is FAIL assert finding.metrics["below_ratio"] == 0.8 def test_gate_floor_all_zero_is_not_its_business(): """整集全零是 glove.no_contact 的事，不在这里重复报。""" findings = get_check("glove.gate_floor").run(_context({"tactile": (_frame(),)})) assert findings == [] # ── 4. 阵列卡死 ───────────────────────────────────────────── def test_frozen_normal_device_rate_passes(): """★ 设备本来就每 4 帧才更新一次 —— 静止 4 帧是**正常**，不是卡死。""" values = [] for block in range(100): values.extend([_frame(600.0 + block)] * 4) ctx = _context({"tactile": tuple(values)}) assert _only("glove.frozen_array", ctx).status is PASS def test_frozen_detects_stall(): values = [_frame(600.0)] * 500 + [_frame(900.0)] * 10 ctx = _context({"tactile": tuple(values)}) finding = _only("glove.frozen_array", ctx) assert finding.status is FAIL # 500 帧 ≥ fail_frames(300) assert finding.metrics["longest_run_frames"] == 500 # 区间要能定位到哪儿卡了 assert finding.ranges and finding.ranges[0].start_frame == 0 def test_frozen_ignores_all_zero_stretches(): """★ 实测踩过的坑：全零段天然"完全相同"，算进去会报出大片假区间。 真实数据里一集有 88% 的帧全零，最长"静止"575 帧 —— 全部来自全零段。 """ values = [_frame(0.0)] * 800 + [_frame(600.0)] * 10 ctx = _context({"tactile": tuple(values)}) assert _only("glove.frozen_array", ctx).status is PASS def test_frozen_warn_threshold_between_normal_and_stall(): """40~300 帧是警告区（正常 4 帧，卡死 300+）。""" values = [_frame(600.0)] * 100 + [_frame(700.0)] * 10 ctx = _context({"tactile": tuple(values)}) assert _only("glove.frozen_array", ctx).status is WARN # ── 5. 无接触 ─────────────────────────────────────────────── def test_no_contact_with_pressure_passes(): ctx = _context({"tactile": (_frame(0.0), _frame(620.0))}) assert _only("glove.no_contact", ctx).status is PASS def test_no_contact_all_zero_warns(): ctx = _context({"tactile": tuple([_frame(0.0)] * 500)}) finding = _only("glove.no_contact", ctx) assert finding.status is WARN assert finding.metrics["active_frames"] == 0 if __name__ == "__main__": import pytest raise SystemExit(pytest.main([__file__, "-q"]))