AiDOPWarehouse/doc/plan/sql/uat-data/run_uat_dashboard_verify.py

#!/usr/bin/env python3
"""Verify UAT home/detail dashboards, diagnosis, workbench, and S8 APIs."""

from __future__ import annotations

import json

import run_wp_wb8_api_loop as api


MODULES = ("S1", "S2", "S3", "S4", "S5", "S6", "S7", "S9")
PASSWORD = "1234567890dop"


def item_coverage(body: object) -> dict[str, int]:
    items = body.get("items", []) if isinstance(body, dict) else []
    return {
        "items": len(items),
        "valued": sum(item.get("currentValue") is not None for item in items),
    }


def main() -> int:
    api.BASE = "http://127.0.0.1:5005"
    output: dict[str, object] = {}
    failed = False
    for tenant in (api.A, api.B, api.DEMO):
        token = api.login(tenant["admin"]["account"], tenant["tenant_id"], PASSWORD)["token"]
        modules: dict[str, object] = {}
        for module in MODULES:
            home_status, home_body = api.request(
                "GET", f"/api/AidopKanban/home-grid/{module}?factoryId=1", token=token
            )
            detail_status, detail_body = api.request(
                "GET", f"/api/AidopKanban/detail-kpis/{module}?factoryId=1", token=token
            )
            diagnosis_status, _ = api.request(
                "GET", f"/api/AidopKanban/smart-diagnosis/{module}?factoryId=1", token=token
            )
            home = item_coverage(home_body)
            detail = item_coverage(detail_body)
            modules[module] = {
                "homeStatus": home_status,
                "home": home,
                "detailStatus": detail_status,
                "detail": detail,
                "diagnosisStatus": diagnosis_status,
            }
            failed |= home_status != 200 or detail_status != 200 or diagnosis_status != 200
            failed |= home["items"] != home["valued"]

        workbench_status, _ = api.request("GET", "/api/AidopWorkbench/summary?top=6", token=token)
        s8_status, s8_body = api.request(
            "GET",
            f"/api/AidopKanban/s8-alerts?tenantId={tenant['tenant_id']}&factoryId={tenant['factory_id']}",
            token=token,
        )
        output[tenant["code"]] = {
            "modules": modules,
            "workbenchStatus": workbench_status,
            "s8Status": s8_status,
            "s8Rows": len(s8_body) if isinstance(s8_body, list) else None,
        }
        failed |= workbench_status != 200 or s8_status != 200

    print(json.dumps(output, ensure_ascii=False, indent=2))
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())