smoke_test.py 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100
  1. from __future__ import annotations
  2. import argparse
  3. import json
  4. from typing import Any
  5. from m2_mcp.config_api import (
  6. get_ai_online_v2,
  7. search_ai_rcmd_operations,
  8. search_ai_systems,
  9. set_multi_values,
  10. )
  11. def _parse_bool(raw: str) -> bool:
  12. text = str(raw).strip().lower()
  13. if text in {"1", "true", "yes", "y", "on"}:
  14. return True
  15. if text in {"0", "false", "no", "n", "off"}:
  16. return False
  17. raise argparse.ArgumentTypeError(f"invalid boolean value: {raw}")
  18. def _print(payload: Any) -> None:
  19. print(json.dumps(payload, ensure_ascii=False, indent=2))
  20. def build_parser() -> argparse.ArgumentParser:
  21. parser = argparse.ArgumentParser(description="Smoke test for instrument config MCP upstream APIs")
  22. subparsers = parser.add_subparsers(dest="command", required=True)
  23. p = subparsers.add_parser("search-ai-systems")
  24. p.add_argument("--project-key", required=True)
  25. p.add_argument("--keyword", default="")
  26. p.add_argument("--page-size", type=int, default=20)
  27. p.add_argument("--page-num", type=int, default=1)
  28. p.add_argument("--order-by", nargs="*", default=["-id"])
  29. p = subparsers.add_parser("search-ai-rcmd-operations")
  30. p.add_argument("--project-key", required=True)
  31. p.add_argument("--codes", nargs="+", required=True)
  32. p.add_argument("--end", required=True)
  33. p.add_argument("--page-size", type=int, default=20)
  34. p.add_argument("--page-num", type=int, default=1)
  35. p.add_argument("--order", default="-create_time")
  36. p = subparsers.add_parser("get-ai-online-v2")
  37. p.add_argument("--project-key", required=True)
  38. p.add_argument("--codes", nargs="+", required=True)
  39. p = subparsers.add_parser("set-multi-values")
  40. p.add_argument("--project-key", required=True)
  41. p.add_argument("--points-json", required=True)
  42. p.add_argument("--from", dest="from_", default="M2_BACKEND")
  43. return parser
  44. def main() -> None:
  45. parser = build_parser()
  46. args = parser.parse_args()
  47. if args.command == "search-ai-systems":
  48. _print(
  49. search_ai_systems(
  50. args.project_key,
  51. keyword=args.keyword,
  52. page_size=args.page_size,
  53. page_num=args.page_num,
  54. order_by=args.order_by,
  55. )
  56. )
  57. return
  58. if args.command == "search-ai-rcmd-operations":
  59. _print(
  60. search_ai_rcmd_operations(
  61. args.project_key,
  62. codes=args.codes,
  63. end=args.end,
  64. page_size=args.page_size,
  65. page_num=args.page_num,
  66. order=args.order,
  67. )
  68. )
  69. return
  70. if args.command == "get-ai-online-v2":
  71. _print(get_ai_online_v2(args.project_key, codes=args.codes))
  72. return
  73. if args.command == "set-multi-values":
  74. _print(set_multi_values(args.project_key, points=json.loads(args.points_json), from_=args.from_))
  75. return
  76. raise ValueError(f"unsupported command: {args.command}")
  77. if __name__ == "__main__":
  78. main()