Blame for processor/tests/test_device_task_progress.py

"""采集端任务进度账本单测(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)