m2-mcp/scripts/smoke_test.py

from __future__ import annotations

import argparse
import json
from typing import Any

from m2_mcp.config_api import (
    get_command_log,
    get_ai_online_v2,
    search_ai_rcmd_operations,
    search_ai_systems,
    set_multi_values,
)


def _parse_bool(raw: str) -> bool:
    text = str(raw).strip().lower()
    if text in {"1", "true", "yes", "y", "on"}:
        return True
    if text in {"0", "false", "no", "n", "off"}:
        return False
    raise argparse.ArgumentTypeError(f"invalid boolean value: {raw}")


def _print(payload: Any) -> None:
    print(json.dumps(payload, ensure_ascii=False, indent=2))


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Smoke test for instrument config MCP upstream APIs")
    subparsers = parser.add_subparsers(dest="command", required=True)

    p = subparsers.add_parser("search-ai-systems")
    p.add_argument("--project-key", required=True)
    p.add_argument("--keyword", default="")
    p.add_argument("--page-size", type=int, default=20)
    p.add_argument("--page-num", type=int, default=1)
    p.add_argument("--order-by", nargs="*", default=["-id"])

    p = subparsers.add_parser("search-ai-rcmd-operations")
    p.add_argument("--project-key", required=True)
    p.add_argument("--codes", nargs="+", required=True)
    p.add_argument("--end", required=True)
    p.add_argument("--page-size", type=int, default=20)
    p.add_argument("--page-num", type=int, default=1)
    p.add_argument("--order", default="-create_time")

    p = subparsers.add_parser("get-ai-online-v2")
    p.add_argument("--project-key", required=True)
    p.add_argument("--codes", nargs="+", required=True)

    p = subparsers.add_parser("get-command-log")
    p.add_argument("--project-key", required=True)
    p.add_argument("--point-id", required=True)
    p.add_argument("--begin", type=int, required=True)
    p.add_argument("--end", type=int, required=True)
    p.add_argument("--page-size", type=int, default=20)
    p.add_argument("--page-num", type=int, default=1)
    p.add_argument("--export", type=_parse_bool, default=False)

    p = subparsers.add_parser("set-multi-values")
    p.add_argument("--project-key", required=True)
    p.add_argument("--points-json", required=True)
    p.add_argument("--from", dest="from_", default="M2_BACKEND")

    return parser


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()

    if args.command == "search-ai-systems":
        _print(
            search_ai_systems(
                args.project_key,
                keyword=args.keyword,
                page_size=args.page_size,
                page_num=args.page_num,
                order_by=args.order_by,
            )
        )
        return

    if args.command == "search-ai-rcmd-operations":
        _print(
            search_ai_rcmd_operations(
                args.project_key,
                codes=args.codes,
                end=args.end,
                page_size=args.page_size,
                page_num=args.page_num,
                order=args.order,
            )
        )
        return

    if args.command == "get-ai-online-v2":
        _print(get_ai_online_v2(args.project_key, codes=args.codes))
        return

    if args.command == "get-command-log":
        _print(
            get_command_log(
                args.project_key,
                point_id=args.point_id,
                begin=args.begin,
                end=args.end,
                page_size=args.page_size,
                page_num=args.page_num,
                export=args.export,
            )
        )
        return

    if args.command == "set-multi-values":
        _print(set_multi_values(args.project_key, points=json.loads(args.points_json), from_=args.from_))
        return

    raise ValueError(f"unsupported command: {args.command}")


if __name__ == "__main__":
    main()