Spaces:

Stouching-Robotics
/

DataFlowing

Running

DataFlowing / processor /tests /test_cleaning_report.py

stouchinglhh

Claude

sync processor from GitLab c697b6e: 版本 1.19.4 / 前端 1.12.4 —— 服务端 i18n(首屏不再闪英文)、推理模块并入模型页、前端抽出 shared/、缓存键改用前端号

bb59211 about 1 month ago

7.49 kB

"""清洗报告组装单测(app/processing/cleaning/report.py)。
覆盖:四层结果的每一层、时间轴拍平与去重、训练用途判定、fail-closed 语义。
全是纯函数 —— 不需要真视频、不需要 cv2、不需要任何 mock。
    pytest tests/test_cleaning_report.py
    python3 tests/test_cleaning_report.py
"""
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from app.processing.cleaning import contract as C  # noqa: E402
from app.processing.cleaning import report as R  # noqa: E402


def _range(start, end, code="video.freeze", severity=C.WARN, stream="s1"):
    return C.Range(start_sec=start, end_sec=end, code=code,
                   severity=severity, stream=stream)


def _stream(name="s1", status=C.PASS, ranges=(), channels=("rgb", "time")):
    return C.StreamResult(stream=name, status=status, ranges=tuple(ranges),
                          channels=tuple(channels))


# ── 第二层:时间轴 ───────────────────────────────────────────
def test_timeline_sorted_by_time():
    streams = [
        _stream("s1", ranges=[_range(5, 6), _range(1, 2)]),
        _stream("s2", ranges=[_range(3, 4)]),
    ]
    axis = R.build_timeline(streams)
    assert [row["start_sec"] for row in axis] == [1, 3, 5]


def test_timeline_dedupes_identical_rows():
    dup = _range(1, 2)
    axis = R.build_timeline([_stream("s1", ranges=[dup, dup])])
    assert len(axis) == 1


def test_timeline_fills_stream_from_parent():
    """range 自己没写 stream 时,从所属的 stream 补上。"""
    bare = C.Range(start_sec=1, end_sec=2, code="x", severity=C.WARN)
    axis = R.build_timeline([_stream("cam_a", ranges=[bare])])
    assert axis[0]["stream"] == "cam_a"


def test_timeline_accepts_dict_streams():
    """报告从磁盘读回来时 streams 是 dict,不是 StreamResult。"""
    axis = R.build_timeline([{
        "stream": "s1",
        "ranges": [{"start_sec": 1, "end_sec": 2, "code": "x", "severity": C.WARN}],
    }])
    assert axis[0]["code"] == "x"


def test_timeline_empty():
    assert R.build_timeline([]) == []


# ── 第三层:整集结论 ─────────────────────────────────────────
def test_episode_takes_worst_stream():
    episode = R.rollup_episode([_stream("a", C.PASS), _stream("b", C.FAIL)])
    assert episode["status"] == C.FAIL
    assert episode["failed_streams"] == ["b"]


def test_episode_includes_findings_in_rollup():
    """跨模态检查没有对应的 stream,它的结论也要参与整集判定。"""
    finding = C.Finding(rule="cross.sync", version="1.0", status=C.FAIL,
                        scope="cross_modal")
    episode = R.rollup_episode([_stream("a", C.PASS)], [finding])
    assert episode["status"] == C.FAIL


def test_episode_all_pass():
    episode = R.rollup_episode([_stream("a", C.PASS), _stream("b", C.PASS)])
    assert episode["status"] == C.PASS
    assert episode["summary"] == "All checks passed"


def test_episode_summary_counts_each_bucket():
    episode = R.rollup_episode([
        _stream("a", C.FAIL), _stream("b", C.WARN), _stream("c", C.PENDING),
    ])
    assert "1 stream(s) failed" in episode["summary"]
    assert "1 stream(s) with warnings" in episode["summary"]
    assert "1 stream(s) pending review" in episode["summary"]


def test_episode_empty_streams_is_pass():
    """没有流可检查 → 这一层判 PASS(是否 fail-closed 由 build_report/passed 决定)。"""
    assert R.rollup_episode([])["status"] == C.PASS


# ── 第四层:训练用途 ─────────────────────────────────────────
def test_training_use_fails_on_missing_channel():
    rows = {row["use"]: row for row in R.rollup_training_use({"rgb", "time"})}
    assert rows["pure_vision"]["status"] == C.PASS
    assert rows["vision_tactile_joint"]["status"] == C.FAIL
    assert "not collected" in rows["vision_tactile_joint"]["reason"]
    assert rows["vision_tactile_joint"]["missing"] == ["tactile"]


def test_training_use_reports_all_missing_channels():
    rows = {row["use"]: row for row in R.rollup_training_use(set())}
    assert set(rows["pure_vision"]["missing"]) == {"rgb", "time"}


def test_training_use_inherits_episode_status():
    """信道齐全但整集有问题 → 该用途跟着判 warn/fail,而不是 pass。"""
    rows = {row["use"]: row for row in R.rollup_training_use({"rgb", "time"}, C.WARN)}
    assert rows["pure_vision"]["status"] == C.WARN
    assert rows["pure_vision"]["reason"] == "data quality problem"


def test_training_use_covers_every_declared_rule():
    rows = R.rollup_training_use({"rgb", "time"})
    assert len(rows) == len(R.TRAINING_USE_RULES)


# ── 完整报告 ────────────────────────────────────────────────
def test_build_report_shape():
    rep = R.build_report(
        [_stream("cam_a", C.WARN, [_range(1, 2)])],
        [],
        episode_id="ep1", project="proj",
        modalities={"actual": ["mono_rgb"]},
        channels={"rgb", "time"},
        ruleset_revision="abc123",
        template_source="workflow",
    )
    assert rep["schema_version"] == R.SCHEMA_VERSION
    assert rep["episode_id"] == "ep1"
    assert rep["ruleset_revision"] == "abc123"
    assert rep["template_source"] == "workflow"
    assert rep["episode"]["status"] == C.WARN
    assert rep["passed"] is False          # 只有 PASS 算通过,WARN 不算
    assert len(rep["ranges"]) == 1
    assert len(rep["streams"]) == 1
    assert rep["training_use"]


def test_build_report_passed_only_when_all_clean():
    rep = R.build_report([_stream("a", C.PASS)], [], channels={"rgb", "time"})
    assert rep["episode"]["status"] == C.PASS
    assert rep["passed"] is True


def test_warn_is_not_passed():
    """WARN 不算通过 —— 这是"卡在人工审核"的判定依据,必须显式锁住。"""
    assert R.passed_for(C.WARN) is False
    assert R.passed_for(C.PASS) is True


def test_build_report_pulls_channels_from_streams():
    """没显式传 channels 时,从各条流自己声明的信道汇总。"""
    rep = R.build_report([_stream("a", C.PASS, channels=("rgb", "time", "tactile"))])
    rows = {row["use"]: row for row in rep["training_use"]}
    assert rows["vision_tactile_joint"]["status"] == C.PASS


def test_empty_report_is_fail_closed():
    """没有数据 ≠ 数据合格。"""
    rep = R.empty_report(episode_id="ep1", reason="no_data")
    assert rep["passed"] is False
    assert rep["episode"]["status"] == C.ERROR
    assert rep["reason"] == "no_data"
    assert rep["streams"] == []


if __name__ == "__main__":
    import inspect
    import tempfile

    _failures = 0
    for _name, _fn in sorted(globals().items()):
        if not _name.startswith("test_") or not callable(_fn):
            continue
        try:
            if "tmp_path" in inspect.signature(_fn).parameters:
                with tempfile.TemporaryDirectory() as _tmp:
                    _fn(Path(_tmp))
            else:
                _fn()
            print(f"PASS {_name}")
        except Exception as _exc:  # noqa: BLE001
            _failures += 1
            print(f"FAIL {_name}: {_exc}")
    raise SystemExit(1 if _failures else 0)

	"""清洗报告组装单测(app/processing/cleaning/report.py)。

	覆盖:四层结果的每一层、时间轴拍平与去重、训练用途判定、fail-closed 语义。

	全是纯函数 —— 不需要真视频、不需要 cv2、不需要任何 mock。

	pytest tests/test_cleaning_report.py
	python3 tests/test_cleaning_report.py
	"""
	import sys
	from pathlib import Path

	sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

	from app.processing.cleaning import contract as C # noqa: E402
	from app.processing.cleaning import report as R # noqa: E402


	def _range(start, end, code="video.freeze", severity=C.WARN, stream="s1"):
	return C.Range(start_sec=start, end_sec=end, code=code,
	severity=severity, stream=stream)


	def _stream(name="s1", status=C.PASS, ranges=(), channels=("rgb", "time")):
	return C.StreamResult(stream=name, status=status, ranges=tuple(ranges),
	channels=tuple(channels))


	# ── 第二层:时间轴 ───────────────────────────────────────────
	def test_timeline_sorted_by_time():
	streams = [
	_stream("s1", ranges=[_range(5, 6), _range(1, 2)]),
	_stream("s2", ranges=[_range(3, 4)]),
	]
	axis = R.build_timeline(streams)
	assert [row["start_sec"] for row in axis] == [1, 3, 5]


	def test_timeline_dedupes_identical_rows():
	dup = _range(1, 2)
	axis = R.build_timeline([_stream("s1", ranges=[dup, dup])])
	assert len(axis) == 1


	def test_timeline_fills_stream_from_parent():
	"""range 自己没写 stream 时,从所属的 stream 补上。"""
	bare = C.Range(start_sec=1, end_sec=2, code="x", severity=C.WARN)
	axis = R.build_timeline([_stream("cam_a", ranges=[bare])])
	assert axis[0]["stream"] == "cam_a"


	def test_timeline_accepts_dict_streams():
	"""报告从磁盘读回来时 streams 是 dict,不是 StreamResult。"""
	axis = R.build_timeline([{
	"stream": "s1",
	"ranges": [{"start_sec": 1, "end_sec": 2, "code": "x", "severity": C.WARN}],
	}])
	assert axis[0]["code"] == "x"


	def test_timeline_empty():
	assert R.build_timeline([]) == []


	# ── 第三层:整集结论 ─────────────────────────────────────────
	def test_episode_takes_worst_stream():
	episode = R.rollup_episode([_stream("a", C.PASS), _stream("b", C.FAIL)])
	assert episode["status"] == C.FAIL
	assert episode["failed_streams"] == ["b"]


	def test_episode_includes_findings_in_rollup():
	"""跨模态检查没有对应的 stream,它的结论也要参与整集判定。"""
	finding = C.Finding(rule="cross.sync", version="1.0", status=C.FAIL,
	scope="cross_modal")
	episode = R.rollup_episode([_stream("a", C.PASS)], [finding])
	assert episode["status"] == C.FAIL


	def test_episode_all_pass():
	episode = R.rollup_episode([_stream("a", C.PASS), _stream("b", C.PASS)])
	assert episode["status"] == C.PASS
	assert episode["summary"] == "All checks passed"


	def test_episode_summary_counts_each_bucket():
	episode = R.rollup_episode([
	_stream("a", C.FAIL), _stream("b", C.WARN), _stream("c", C.PENDING),
	])
	assert "1 stream(s) failed" in episode["summary"]
	assert "1 stream(s) with warnings" in episode["summary"]
	assert "1 stream(s) pending review" in episode["summary"]


	def test_episode_empty_streams_is_pass():
	"""没有流可检查 → 这一层判 PASS(是否 fail-closed 由 build_report/passed 决定)。"""
	assert R.rollup_episode([])["status"] == C.PASS


	# ── 第四层:训练用途 ─────────────────────────────────────────
	def test_training_use_fails_on_missing_channel():
	rows = {row["use"]: row for row in R.rollup_training_use({"rgb", "time"})}
	assert rows["pure_vision"]["status"] == C.PASS
	assert rows["vision_tactile_joint"]["status"] == C.FAIL
	assert "not collected" in rows["vision_tactile_joint"]["reason"]
	assert rows["vision_tactile_joint"]["missing"] == ["tactile"]


	def test_training_use_reports_all_missing_channels():
	rows = {row["use"]: row for row in R.rollup_training_use(set())}
	assert set(rows["pure_vision"]["missing"]) == {"rgb", "time"}


	def test_training_use_inherits_episode_status():
	"""信道齐全但整集有问题 → 该用途跟着判 warn/fail,而不是 pass。"""
	rows = {row["use"]: row for row in R.rollup_training_use({"rgb", "time"}, C.WARN)}
	assert rows["pure_vision"]["status"] == C.WARN
	assert rows["pure_vision"]["reason"] == "data quality problem"


	def test_training_use_covers_every_declared_rule():
	rows = R.rollup_training_use({"rgb", "time"})
	assert len(rows) == len(R.TRAINING_USE_RULES)


	# ── 完整报告 ────────────────────────────────────────────────
	def test_build_report_shape():
	rep = R.build_report(
	[_stream("cam_a", C.WARN, [_range(1, 2)])],
	[],
	episode_id="ep1", project="proj",
	modalities={"actual": ["mono_rgb"]},
	channels={"rgb", "time"},
	ruleset_revision="abc123",
	template_source="workflow",
	)
	assert rep["schema_version"] == R.SCHEMA_VERSION
	assert rep["episode_id"] == "ep1"
	assert rep["ruleset_revision"] == "abc123"
	assert rep["template_source"] == "workflow"
	assert rep["episode"]["status"] == C.WARN
	assert rep["passed"] is False # 只有 PASS 算通过,WARN 不算
	assert len(rep["ranges"]) == 1
	assert len(rep["streams"]) == 1
	assert rep["training_use"]


	def test_build_report_passed_only_when_all_clean():
	rep = R.build_report([_stream("a", C.PASS)], [], channels={"rgb", "time"})
	assert rep["episode"]["status"] == C.PASS
	assert rep["passed"] is True


	def test_warn_is_not_passed():
	"""WARN 不算通过 —— 这是"卡在人工审核"的判定依据,必须显式锁住。"""
	assert R.passed_for(C.WARN) is False
	assert R.passed_for(C.PASS) is True


	def test_build_report_pulls_channels_from_streams():
	"""没显式传 channels 时,从各条流自己声明的信道汇总。"""
	rep = R.build_report([_stream("a", C.PASS, channels=("rgb", "time", "tactile"))])
	rows = {row["use"]: row for row in rep["training_use"]}
	assert rows["vision_tactile_joint"]["status"] == C.PASS


	def test_empty_report_is_fail_closed():
	"""没有数据 ≠ 数据合格。"""
	rep = R.empty_report(episode_id="ep1", reason="no_data")
	assert rep["passed"] is False
	assert rep["episode"]["status"] == C.ERROR
	assert rep["reason"] == "no_data"
	assert rep["streams"] == []


	if __name__ == "__main__":
	import inspect
	import tempfile

	_failures = 0
	for _name, _fn in sorted(globals().items()):
	if not _name.startswith("test_") or not callable(_fn):
	continue
	try:
	if "tmp_path" in inspect.signature(_fn).parameters:
	with tempfile.TemporaryDirectory() as _tmp:
	_fn(Path(_tmp))
	else:
	_fn()
	print(f"PASS {_name}")
	except Exception as _exc: # noqa: BLE001
	_failures += 1
	print(f"FAIL {_name}: {_exc}")
	raise SystemExit(1 if _failures else 0)