From 86744d33774af8bdd9e6b00f74f963c4d1cedfda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B7=9D=E6=9E=97?= Date: Fri, 27 Jun 2025 10:48:39 +0800 Subject: [PATCH] unittest --- .../alibabacloud/utils.py | 9 + src/alibaba_cloud_ops_mcp_server/server.py | 3 + .../tools/cms_tools.py | 5 +- .../utils/__init__.py | 1 + .../utils/logging_decorator.py | 203 ++++++++++++++++++ uv.lock | 2 +- 6 files changed, 220 insertions(+), 3 deletions(-) create mode 100644 src/alibaba_cloud_ops_mcp_server/utils/__init__.py create mode 100644 src/alibaba_cloud_ops_mcp_server/utils/logging_decorator.py diff --git a/src/alibaba_cloud_ops_mcp_server/alibabacloud/utils.py b/src/alibaba_cloud_ops_mcp_server/alibabacloud/utils.py index 9a91c93..9195434 100644 --- a/src/alibaba_cloud_ops_mcp_server/alibabacloud/utils.py +++ b/src/alibaba_cloud_ops_mcp_server/alibabacloud/utils.py @@ -44,3 +44,12 @@ def create_config(): config = Config(credential=credentialsClient) config.user_agent = 'alibaba-cloud-ops-mcp-server' return config + + +def log_args_and_return(func): + def wrapper(*args, **kwargs): + logger.info(f"Calling {func.__name__} with args: {args}, kwargs: {kwargs}") + result = func(*args, **kwargs) + logger.info(f"{func.__name__} returned: {result}") + return result + return wrapper diff --git a/src/alibaba_cloud_ops_mcp_server/server.py b/src/alibaba_cloud_ops_mcp_server/server.py index b569f73..8474418 100644 --- a/src/alibaba_cloud_ops_mcp_server/server.py +++ b/src/alibaba_cloud_ops_mcp_server/server.py @@ -6,6 +6,8 @@ from alibaba_cloud_ops_mcp_server.config import config from alibaba_cloud_ops_mcp_server.tools import cms_tools, oos_tools, oss_tools, api_tools, common_api_tools +from alibaba_cloud_ops_mcp_server.alibabacloud.utils import log_args_and_return + logger = logging.getLogger(__name__) SUPPORTED_SERVICES_MAP = { @@ -47,6 +49,7 @@ default=None, help="Comma-separated list of supported services, e.g., 'ecs,vpc,rds'", ) + def main(transport: str, port: int, host: str, services: str): # Create an MCP server mcp = FastMCP( diff --git a/src/alibaba_cloud_ops_mcp_server/tools/cms_tools.py b/src/alibaba_cloud_ops_mcp_server/tools/cms_tools.py index 4cdce72..7efdd90 100644 --- a/src/alibaba_cloud_ops_mcp_server/tools/cms_tools.py +++ b/src/alibaba_cloud_ops_mcp_server/tools/cms_tools.py @@ -6,7 +6,7 @@ from alibabacloud_cms20190101.client import Client as cms20190101Client from alibabacloud_cms20190101 import models as cms_20190101_models from alibaba_cloud_ops_mcp_server.alibabacloud.utils import create_config - +from alibaba_cloud_ops_mcp_server.utils.logging_decorator import tool_log END_STATUSES = ['Success', 'Failed', 'Cancelled'] @@ -36,6 +36,7 @@ def _get_cms_metric_data(region_id: str, instance_ids: List[str], metric_name: s return describe_metric_last_resp.body.datapoints @tools.append +@tool_log def CMS_GetCpuUsageData( InstanceIds: List[str] = Field(description='AlibabaCloud ECS instance ID List'), RegionId: str = Field(description='AlibabaCloud region ID', default='cn-hangzhou') @@ -105,7 +106,7 @@ def CMS_GetDiskTotalData( """获取磁盘分区总容量指标数据""" return _get_cms_metric_data(RegionId, InstanceIds, 'diskusage_total') - +@tool_log @tools.append def CMS_GetDiskUsedData( InstanceIds: List[str] = Field(description='AlibabaCloud ECS instance ID List'), diff --git a/src/alibaba_cloud_ops_mcp_server/utils/__init__.py b/src/alibaba_cloud_ops_mcp_server/utils/__init__.py new file mode 100644 index 0000000..0519ecb --- /dev/null +++ b/src/alibaba_cloud_ops_mcp_server/utils/__init__.py @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/src/alibaba_cloud_ops_mcp_server/utils/logging_decorator.py b/src/alibaba_cloud_ops_mcp_server/utils/logging_decorator.py new file mode 100644 index 0000000..5091601 --- /dev/null +++ b/src/alibaba_cloud_ops_mcp_server/utils/logging_decorator.py @@ -0,0 +1,203 @@ +""" +======================================== +日志装饰器模块 +用于记录函数的输入参数、输出结果和异常信息 +======================================== +""" + +import logging +import json +import functools +from typing import Any, Callable, Dict +import traceback + + +# 配置日志格式 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) + +logger = logging.getLogger(__name__) + + +def safe_json_dumps(obj: Any, max_length: int = 1000) -> str: + """ + 安全地将对象转换为JSON字符串，处理不可序列化的对象 + + Args: + obj: 要序列化的对象 + max_length: 最大字符串长度，超过会被截断 + + Returns: + JSON字符串 + """ + try: + result = json.dumps(obj, ensure_ascii=False, default=str, indent=2) + if len(result) > max_length: + result = result[:max_length] + "... (truncated)" + return result + except Exception: + return str(obj)[:max_length] + ("... (truncated)" if len(str(obj)) > max_length else "") + + +def log_function_calls( + log_input: bool = True, + log_output: bool = True, + log_exceptions: bool = True, + max_log_length: int = 1000, + logger_name: str = None +) -> Callable: + """ + 日志装饰器，用于记录函数的输入、输出和异常 + + Args: + log_input: 是否记录输入参数 + log_output: 是否记录输出结果 + log_exceptions: 是否记录异常信息 + max_log_length: 日志内容最大长度 + logger_name: 自定义logger名称 + + Returns: + 装饰器函数 + """ + def decorator(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + # 使用自定义logger或默认logger + func_logger = logging.getLogger(logger_name or f"{func.__module__}.{func.__name__}") + + function_name = f"{func.__module__}.{func.__name__}" + + # 记录函数开始执行 + func_logger.info(f"🚀 开始执行函数: {function_name}") + + # 记录输入参数 + if log_input: + input_info = { + "args": args, + "kwargs": kwargs + } + func_logger.info(f"📥 输入参数: {safe_json_dumps(input_info, max_log_length)}") + + try: + # 执行原函数 + result = func(*args, **kwargs) + + # 记录输出结果 + if log_output: + func_logger.info(f"📤 输出结果: {safe_json_dumps(result, max_log_length)}") + + func_logger.info(f"✅ 函数执行成功: {function_name}") + return result + + except Exception as e: + # 记录异常信息 + if log_exceptions: + error_info = { + "error_type": type(e).__name__, + "error_message": str(e), + "traceback": traceback.format_exc() + } + func_logger.error(f"❌ 函数执行异常: {function_name}") + func_logger.error(f"🔥 异常详情: {safe_json_dumps(error_info, max_log_length * 2)}") + + # 重新抛出异常 + raise + + return wrapper + return decorator + + +def simple_log(func: Callable = None) -> Callable: + """ + 简化版本的日志装饰器，使用默认配置 + + Args: + func: 要装饰的函数 + + Returns: + 装饰后的函数 + """ + def decorator(f: Callable) -> Callable: + if f is None: + raise ValueError("simple_log装饰器必须应用于一个函数") + return log_function_calls()(f) + + # 如果func不为None，说明是直接使用@simple_log而不是@simple_log() + if func is not None: + return decorator(func) + + # 如果func为None，说明是使用@simple_log()形式，返回装饰器 + return decorator + + +def tool_log(func: Callable = None) -> Callable: + """ + 专门为工具函数设计的日志装饰器 + + Args: + func: 要装饰的工具函数 + + Returns: + 装饰后的函数 + """ + def decorator(f: Callable) -> Callable: + if f is None: + raise ValueError("tool_log装饰器必须应用于一个函数") + + return log_function_calls( + log_input=True, + log_output=True, + log_exceptions=True, + max_log_length=2000, + logger_name=f"tools.{f.__name__}" + )(f) + + # 如果func不为None，说明是直接使用@tool_log而不是@tool_log() + if func is not None: + return decorator(func) + + # 如果func为None，说明是使用@tool_log()形式，返回装饰器 + return decorator + + +# 示例使用方法 +if __name__ == "__main__": + + @log_function_calls() + def test_function(x: int, y: int, name: str = "test") -> Dict[str, Any]: + """测试函数""" + if x < 0: + raise ValueError("x不能为负数") + return {"result": x + y, "name": name} + + @simple_log + def simple_test(message: str) -> str: + """简单测试函数""" + return f"处理消息: {message}" + + @tool_log + def tool_test(config: Dict[str, Any]) -> str: + """工具函数测试""" + return f"工具执行结果: {config}" + + # 测试正常执行 + print("=== 测试正常执行 ===") + result1 = test_function(1, 2, name="测试") + print(f"返回结果: {result1}") + + print("\n=== 测试简单装饰器 ===") + result2 = simple_test("Hello World") + print(f"返回结果: {result2}") + + print("\n=== 测试工具装饰器 ===") + result3 = tool_test({"key": "value", "number": 123}) + print(f"返回结果: {result3}") + + # 测试异常情况 + print("\n=== 测试异常情况 ===") + try: + test_function(-1, 2) + except ValueError as e: + print(f"捕获异常: {e}") \ No newline at end of file diff --git a/uv.lock b/uv.lock index b373562..201190e 100644 --- a/uv.lock +++ b/uv.lock @@ -116,7 +116,7 @@ wheels = [ [[package]] name = "alibaba-cloud-ops-mcp-server" -version = "0.9.0" +version = "0.9.1" source = { editable = "." } dependencies = [ { name = "alibabacloud-cms20190101" },