Spaces:
Running
Running
| """采集端任务进度账本单测(app/device_progress.py)。 | |
| 覆盖:幂等重放、多设备聚合、增量水位、参数校验、未知任务拒绝(400 而非 | |
| 404 —— 采集端把 404 当"后端未实现"永久降级)、持久化与坏文件容错。 | |
| pytest tests/test_device_task_progress.py | |
| python3 tests/test_device_task_progress.py # 无 pytest 也能跑 | |
| """ | |
| import json | |
| import sys | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[1])) | |
| from app import device_progress as dp # noqa: E402 | |
| TASK = "task-aaa" | |
| OTHER = "task-bbb" | |
| KNOWN = {TASK, OTHER} | |
| def _report(tmp, increment, session_id, task=TASK, device="EGO_001"): | |
| return dp.apply_report( | |
| {"task_id": task, "device_name": device, | |
| "session_id": session_id, "increment": increment}, | |
| KNOWN, path=tmp) | |
| def test_first_report_sets_total(tmp_path): | |
| f = tmp_path / "p.json" | |
| out = _report(f, 1, "EGO_001:t:0") | |
| assert out["applied"] is True | |
| assert out["completed_count"] == 1 | |
| assert dp.reported_total(TASK, path=f) == 1 | |
| def test_replayed_watermark_is_idempotent(tmp_path): | |
| """采集端崩溃重启后会重发同一水位 —— 不能重复计数。""" | |
| f = tmp_path / "p.json" | |
| first = _report(f, 12, "EGO_001:t:0") | |
| again = _report(f, 12, "EGO_001:t:0") | |
| assert first["completed_count"] == 12 | |
| assert again["applied"] is False | |
| assert again["completed_count"] == 12 | |
| # 重放仍要回当前聚合数(采集端拿它当后端权威数) | |
| assert isinstance(again["completed_count"], int) | |
| assert not isinstance(again["completed_count"], bool) | |
| def test_incremental_watermarks_accumulate(tmp_path): | |
| f = tmp_path / "p.json" | |
| _report(f, 12, "EGO_001:t:0") | |
| out = _report(f, 1, "EGO_001:t:12") | |
| assert out["completed_count"] == 13 | |
| def test_multiple_devices_aggregate(tmp_path): | |
| f = tmp_path / "p.json" | |
| _report(f, 3, "EGO_001:t:0", device="EGO_001") | |
| _report(f, 2, "EGO_002:t:0", device="EGO_002") | |
| out = _report(f, 1, "EGO_003:t:0", device="EGO_003") | |
| assert out["completed_count"] == 6 | |
| # 同一水位键跨设备不冲突 | |
| assert dp.reported_total(TASK, path=f) == 6 | |
| def test_zero_increment_is_recorded_without_growth(tmp_path): | |
| f = tmp_path / "p.json" | |
| _report(f, 5, "EGO_001:t:0") | |
| out = _report(f, 0, "EGO_001:t:5") | |
| assert out["applied"] is True and out["completed_count"] == 5 | |
| def test_totals_only_contain_reported_tasks(tmp_path): | |
| """没上报过的项目不能出现在这里 —— 调用方据此回退 session 计数。""" | |
| f = tmp_path / "p.json" | |
| _report(f, 1, "EGO_001:t:0") | |
| totals = dp.reported_totals(path=f) | |
| assert totals == {TASK: 1} | |
| assert dp.reported_total(OTHER, path=f) is None | |
| def test_validation_errors(tmp_path): | |
| f = tmp_path / "p.json" | |
| bad = [ | |
| ({}, "task_id"), # 空 body | |
| ({"task_id": TASK, "session_id": "s"}, "device_name"), | |
| ({"task_id": TASK, "device_name": "d"}, "session_id"), | |
| ({"task_id": TASK, "device_name": "d", "session_id": "s", | |
| "increment": -1}, "increment"), | |
| ({"task_id": TASK, "device_name": "d", "session_id": "s", | |
| "increment": "x"}, "increment"), | |
| ({"task_id": TASK, "device_name": "d", "session_id": "s", | |
| "increment": True}, "increment"), | |
| ({"task_id": TASK, "device_name": "d", "session_id": "s", | |
| "increment": dp.MAX_INCREMENT + 1}, "increment"), | |
| ] | |
| for body, needle in bad: | |
| try: | |
| dp.apply_report(body, KNOWN, path=f) | |
| except ValueError as exc: | |
| assert needle in str(exc), (body, str(exc)) | |
| else: | |
| raise AssertionError(f"应当拒绝: {body}") | |
| assert dp.reported_totals(path=f) == {} # 拒绝的请求不落账 | |
| def test_unknown_task_rejected_not_silently_accepted(tmp_path): | |
| """未知任务要报错(路由转 400)。绝不能 404:采集端会永久降级上报。""" | |
| f = tmp_path / "p.json" | |
| try: | |
| _report(f, 1, "EGO_001:t:0", task="no-such-task") | |
| except ValueError as exc: | |
| assert "unknown task_id" in str(exc) | |
| else: | |
| raise AssertionError("未知任务应当被拒绝") | |
| # known_task_ids=None(不校验)时照常落账 —— 路由必须传已知集合 | |
| out = dp.apply_report({"task_id": "no-such-task", "device_name": "d", | |
| "session_id": "s", "increment": 1}, None, path=f) | |
| assert out["completed_count"] == 1 | |
| def test_count_prefers_report_over_sessions(tmp_path): | |
| """核心口径:有上报就用录制数 —— 之后再上传多少条都不动它。""" | |
| f = tmp_path / "p.json" | |
| proj = {"id": TASK, "name": "UMIGripper_AI"} | |
| _report(f, 12, "EGO_001:t:0") | |
| reported = dp.reported_totals(path=f) | |
| assert dp.resolve_count(proj, reported, sessions=16) == (12, "reported") | |
| # 又上传了 3 条(项目下 session 变 19)→ 进度不变 | |
| assert dp.resolve_count(proj, reported, sessions=19) == (12, "reported") | |
| # 录了一条 → 上报后 +1 | |
| _report(f, 1, "EGO_001:t:12") | |
| assert dp.resolve_count(proj, dp.reported_totals(path=f), | |
| sessions=19) == (13, "reported") | |
| def test_count_falls_back_to_sessions_before_any_report(tmp_path): | |
| """升级前就有数据的历史项目不归零:没有上报时仍按 session 数。""" | |
| proj = {"id": TASK, "name": "UMIGripper_AI"} | |
| assert dp.resolve_count(proj, {}, sessions=16) == (16, "sessions") | |
| assert dp.needs_session_fallback([proj], {}) is True | |
| assert dp.needs_session_fallback([proj], {TASK: 12}) is False | |
| def test_persisted_across_reload(tmp_path): | |
| f = tmp_path / "p.json" | |
| _report(f, 2, "EGO_001:t:0") | |
| raw = json.loads(f.read_text(encoding="utf-8")) | |
| assert raw["tasks"][TASK]["devices"]["EGO_001"]["count"] == 2 | |
| # 重新读文件(模拟进程重启)后水位仍生效 | |
| again = _report(f, 2, "EGO_001:t:0") | |
| assert again["applied"] is False and again["completed_count"] == 2 | |
| def test_corrupt_file_degrades_to_empty(tmp_path): | |
| f = tmp_path / "p.json" | |
| f.write_text("{ not json", encoding="utf-8") | |
| assert dp.reported_totals(path=f) == {} | |
| out = _report(f, 1, "EGO_001:t:0") | |
| assert out["completed_count"] == 1 | |
| if __name__ == "__main__": | |
| import tempfile | |
| import traceback | |
| failed = 0 | |
| for name, fn in sorted(globals().items()): | |
| if not name.startswith("test_") or not callable(fn): | |
| continue | |
| with tempfile.TemporaryDirectory() as d: | |
| try: | |
| fn(Path(d)) | |
| print(f" PASS: {name}") | |
| except Exception: | |
| failed += 1 | |
| print(f" FAIL: {name}") | |
| traceback.print_exc() | |
| print("FAIL" if failed else "PASS: device_task_progress 全部通过") | |
| sys.exit(1 if failed else 0) | |