Spaces:
Running
Running
stouchinglhh
Claude
sync processor from GitLab c697b6e: 版本 1.19.4 / 前端 1.12.4 —— 服务端 i18n(首屏不再闪英文)、推理模块并入模型页、前端抽出 shared/、缓存键改用前端号
bb59211 | """「数据不对 → 卡在人工审核」的行为锁。 | |
| 盯住三件事，每一件坏掉都是**静默**的（不报错、流程照常走完）： | |
| 1. 判 FAIL/ERROR 的批次被推回 ``to_review``，且清掉 ``approved_at`` | |
| 2. WARN 只记录、不拦 —— 否则警告阈值一响，整批数据全卡住 | |
| 3. **自动批准路径主动让路** —— AI 标注 / 视频门禁各看各的报告，不看质检 | |
| 结论；不查那个粘性标记的话，它们会在质检推回 to_review 之后又把状态 | |
| 置成 reviewed，坏数据照样被放行 | |
| 全部在临时存储根上跑（monkeypatch ``localstore`` 的路径常量），不碰真实数据。 | |
| pytest tests/test_cleaning_gate.py | |
| python3 tests/test_cleaning_gate.py | |
| """ | |
| import sys | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # noqa: E402 | |
| import pytest # noqa: E402 | |
| from app import localstore # noqa: E402 | |
| from app.processing.cleaning import store # noqa: E402 | |
| from app.processing.cleaning import contract as C # noqa: E402 | |
| def sandbox(tmp_path, monkeypatch): | |
| """把 episode state 落到临时目录 —— 别碰 ~/.cache/egodata-data。""" | |
| states = tmp_path / "state" / "episode_states" | |
| states.mkdir(parents=True, exist_ok=True) | |
| monkeypatch.setattr(localstore, "STATE_ROOT", tmp_path / "state") | |
| monkeypatch.setattr(localstore, "EPISODE_STATES_DIR", states) | |
| localstore._episode_state_cache.clear() | |
| yield states | |
| localstore._episode_state_cache.clear() | |
| def _report(status, *, ranges=0): | |
| """造一份最小报告 —— 只关心 ``episode.status``，那是拦截判据。""" | |
| return { | |
| "schema_version": 2, | |
| "episode_id": "E1", | |
| "computed_at": "2026-09-24T00:00:00+00:00", | |
| "passed": status == C.PASS, | |
| "streams": [], | |
| "ranges": [{}] * ranges, | |
| "episode": { | |
| "status": status, | |
| "status_label": C.SEVERITY_LABELS.get(status, status), | |
| "failed_streams": [], "warn_streams": [], "pending_streams": [], | |
| "summary": "test", | |
| }, | |
| "findings": [], | |
| } | |
| def _seed(states, status="to_review", approved_at=None): | |
| import json | |
| eid = "E1" | |
| (states / f"{eid}.json").write_text(json.dumps({ | |
| "id": eid, "name": eid, "project": "P", "status": status, | |
| "approved_at": approved_at, | |
| }, ensure_ascii=False), encoding="utf-8") | |
| localstore._episode_state_cache.pop(eid, None) | |
| return eid | |
| # ── 拦截 ──────────────────────────────────────────────────── | |
| def test_failure_pushes_back_to_review(sandbox, status): | |
| eid = _seed(sandbox, status="reviewed", approved_at="2026-01-01T00:00:00Z") | |
| assert store.save_report(eid, _report(status)) is True | |
| state = localstore.read_episode_state(eid) | |
| assert state["status"] == "to_review" # 自动批准被撤销 | |
| assert state["approved_at"] is None # 批准时间一起清掉 | |
| assert store.is_approval_blocked(state) is True | |
| def test_non_failure_does_not_block(sandbox, status): | |
| """WARN/PENDING 只是警告 —— 拦的话整批数据都会卡住。""" | |
| eid = _seed(sandbox, status="reviewed") | |
| store.save_report(eid, _report(status)) | |
| state = localstore.read_episode_state(eid) | |
| assert state["status"] == "reviewed" # 状态没被动过 | |
| assert store.is_approval_blocked(state) is False | |
| def test_does_not_resurrect_deleted_episode(sandbox): | |
| """质检不能把用户删掉/否决过的批次复活。""" | |
| for status in ("deleted", "rejected"): | |
| eid = _seed(sandbox, status=status) | |
| store.save_report(eid, _report(C.FAIL)) | |
| assert localstore.read_episode_state(eid)["status"] == status | |
| def test_report_still_written_when_not_blocking(sandbox): | |
| """不拦也得把报告写进去 —— 拦截不是写报告的前提。""" | |
| eid = _seed(sandbox) | |
| store.save_report(eid, _report(C.WARN, ranges=3)) | |
| summary = store.load_summary(eid) | |
| assert summary["status"] == C.WARN and summary["ranges"] == 3 | |
| # ── 自动批准让路 ──────────────────────────────────────────── | |
| def _quality_gate(state): | |
| """转调 ai_annotation 的真实写入路径，避免测试复制一份逻辑。""" | |
| from app.ai_annotation import _set_ai_quality_state | |
| return _set_ai_quality_state | |
| def test_ai_gate_refuses_to_auto_approve_when_blocked(sandbox): | |
| """★ 核心用例：质检判 FAIL 后，AI 标注门禁即使自己通过也不得自动批准。""" | |
| eid = _seed(sandbox, status="to_review") | |
| store.save_report(eid, _report(C.FAIL)) | |
| _quality_gate(None)(eid, {"passed": True}, auto_approve=True) | |
| state = localstore.read_episode_state(eid) | |
| assert state["status"] == "to_review", "坏数据被 AI 门禁放行了" | |
| assert state["ai_quality_status"] == "passed" # 它自己的结论照常记录 | |
| def test_ai_gate_still_auto_approves_when_not_blocked(sandbox): | |
| """别把拦截修成"自动批准彻底失效" —— 干净数据该自动过。""" | |
| eid = _seed(sandbox, status="to_review") | |
| store.save_report(eid, _report(C.PASS)) | |
| _quality_gate(None)(eid, {"passed": True}, auto_approve=True) | |
| assert localstore.read_episode_state(eid)["status"] == "reviewed" | |
| def test_video_gate_refuses_to_auto_approve_when_blocked(sandbox): | |
| """视频干净 ≠ 数据干净 —— 视频门禁同样要让路。""" | |
| from app.video_quality import _apply_video_quality_result | |
| eid = _seed(sandbox, status="to_review") | |
| store.save_report(eid, _report(C.FAIL)) | |
| _apply_video_quality_result(eid, {"passed": True}) | |
| state = localstore.read_episode_state(eid) | |
| assert state["status"] == "to_review", "坏数据被视频门禁放行了" | |
| assert state["video_quality_status"] == "passed" | |
| # ── 标记的粘性 ────────────────────────────────────────────── | |
| def test_block_clears_once_data_is_fixed(sandbox): | |
| """数据修好后重跑要能自动批准 —— 否则卡住就再也出不来了。 | |
| 注意 ``save_report`` **只拦不批**：它解封后不会自己把状态改回 reviewed， | |
| 批准是 AI/视频门禁或人工的事。所以这里走一遍真实的"修好 → 重跑"顺序： | |
| worker 先置 to_review，质检复核（清除标记），随后门禁自动批准。 | |
| """ | |
| eid = _seed(sandbox, status="to_review") | |
| store.save_report(eid, _report(C.FAIL)) | |
| assert store.is_approval_blocked(localstore.read_episode_state(eid)) is True | |
| # —— 重跑 —— | |
| localstore.set_episode_status(eid, "to_review") # worker 完成回调做的 | |
| store.save_report(eid, _report(C.PASS)) # 质检复核：数据已修好 | |
| state = localstore.read_episode_state(eid) | |
| assert store.is_approval_blocked(state) is False | |
| _quality_gate(None)(eid, {"passed": True}, auto_approve=True) | |
| assert localstore.read_episode_state(eid)["status"] == "reviewed" | |
| def test_save_report_never_approves(sandbox): | |
| """质检**只拦不批** —— 它不该越过门禁自己把批次标成通过。""" | |
| eid = _seed(sandbox, status="to_review") | |
| store.save_report(eid, _report(C.PASS)) | |
| assert localstore.read_episode_state(eid)["status"] == "to_review" | |
| if __name__ == "__main__": | |
| raise SystemExit(pytest.main([__file__, "-q"])) | |