verification/verify_service_errors_summary.py

from __future__ import annotations

import os
import sys
import tempfile
from pathlib import Path

from fastapi.testclient import TestClient

_repo_root = Path(__file__).resolve().parents[2]
if str(_repo_root) not in sys.path:
    sys.path.insert(0, str(_repo_root))

from Spider_XHS.service.app import create_app
from Spider_XHS.service.errors import ERROR_CODE_SUCCESS
from Spider_XHS.service.tasks import TaskRecord, TaskStatus


def _assert(cond: bool, msg: str) -> None:
    if not cond:
        raise AssertionError(msg)


def _extract(body: dict) -> dict:
    _assert(body.get("code") == ERROR_CODE_SUCCESS, f"unexpected code={body.get('code')}, body={body}")
    data = body.get("data") or {}
    _assert(isinstance(data, dict), "data_not_dict")
    return data


def verify_errors_summary_aggregate_and_filters() -> None:
    os.environ["ENABLE_LEGACY_ROUTES"] = "0"
    with tempfile.TemporaryDirectory() as td:
        os.environ["STORAGE_ROOT"] = td
        app = create_app()
        with TestClient(app) as client:
            storage = app.state.storage

            t1 = TaskRecord(
                id="t1",
                status=TaskStatus.failed,
                task_type="note_url",
                target="u1",
                payload={"operator": "tester"},
                engine="api",
                created=100.0,
                error={"kind": "auth", "message": "m"},
            )
            t2 = TaskRecord(
                id="t2",
                status=TaskStatus.failed,
                task_type="search",
                target="k",
                payload={"operator": "tester"},
                engine="browser",
                created=101.0,
                error={"kind": "rate", "message": "m"},
            )
            t3 = TaskRecord(
                id="t3",
                status=TaskStatus.succeeded,
                task_type="selftest",
                target="",
                payload={"operator": "tester"},
                engine="selftest",
                created=102.0,
            )
            t4 = TaskRecord(
                id="t4",
                status=TaskStatus.rpa_failed,
                task_type="note_url",
                target="u4",
                payload={"operator": "tester"},
                engine="api",
                created=103.0,
                error={"kind": "risk", "message": "m"},
            )
            t5 = TaskRecord(
                id="t5",
                status=TaskStatus.failed,
                task_type="note_url",
                target="u5",
                payload={"operator": "tester"},
                engine="api",
                created=104.0,
                error={"kind": "auth", "message": "m"},
            )

            for t in (t1, t2, t3, t4, t5):
                storage.create_task(t)

            mtimes = {"t1": 50.0, "t2": 51.0, "t3": 52.0, "t4": 53.0, "t5": 40.0}
            for tid, ts in mtimes.items():
                p = storage.tasks_dir / f"{tid}.json"
                os.utime(p, (ts, ts))

            r0 = client.get("/api/v1/errors/summary", params={"scan_limit": 3})
            _assert(r0.status_code == 200, f"expected 200, got={r0.status_code}, body={r0.text}")
            d0 = _extract(r0.json())
            _assert(d0.get("scan_limit") == 3, f"scan_limit_mismatch={d0.get('scan_limit')}")
            _assert(d0.get("scanned") == 3, f"scanned_mismatch={d0.get('scanned')}")
            _assert(d0.get("total") == 2, f"total_mismatch={d0.get('total')}")
            _assert(
                d0.get("error_kind_counts") == {"risk": 1, "rate": 1},
                f"counts_mismatch={d0.get('error_kind_counts')}",
            )
            tasks0 = d0.get("tasks") or []
            _assert([t.get("id") for t in tasks0] == ["t4", "t2"], f"tasks_mismatch={tasks0}")

            r1 = client.get("/api/v1/errors/summary", params={"scan_limit": 5})
            d1 = _extract(r1.json())
            _assert(d1.get("total") == 4, f"total_mismatch={d1.get('total')}")
            counts1 = d1.get("error_kind_counts") or {}
            _assert(counts1.get("auth") == 2, f"auth_count_mismatch={counts1}")
            _assert(counts1.get("rate") == 1, f"rate_count_mismatch={counts1}")
            _assert(counts1.get("risk") == 1, f"risk_count_mismatch={counts1}")

            r2 = client.get("/api/v1/errors/summary", params={"scan_limit": 5, "error_kind": "auth"})
            d2 = _extract(r2.json())
            _assert(d2.get("total") == 2, f"auth_total_mismatch={d2.get('total')}")
            _assert(d2.get("error_kind_counts") == {"auth": 2}, f"auth_counts_mismatch={d2.get('error_kind_counts')}")

            r3 = client.get("/api/v1/errors/summary", params={"scan_limit": 5, "error_kind": "auth", "limit": 1, "offset": 1})
            d3 = _extract(r3.json())
            tasks3 = d3.get("tasks") or []
            _assert([t.get("id") for t in tasks3] == ["t5"], f"auth_page_mismatch={tasks3}")

            r4 = client.get("/api/v1/errors/summary", params={"scan_limit": 5, "status": "succeeded"})
            d4 = _extract(r4.json())
            _assert(d4.get("total") == 1, f"succeeded_total_mismatch={d4.get('total')}")
            _assert(d4.get("error_kind_counts") == {"unknown": 1}, f"succeeded_counts_mismatch={d4.get('error_kind_counts')}")
            tasks4 = d4.get("tasks") or []
            _assert([t.get("id") for t in tasks4] == ["t3"], f"succeeded_tasks_mismatch={tasks4}")


def main() -> None:
    verify_errors_summary_aggregate_and_filters()
    print("OK")


if __name__ == "__main__":
    main()