compliance/run_api_tests.py
2025-06-05 15:17:51 +08:00

531 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
API测试工具
此工具使用DDMS测试编排器从YAPI或Swagger定义执行API测试。
支持使用规则库进行高级验证。
"""
import os
import sys
import json
import logging
import argparse
from pathlib import Path
from typing import List, Optional
from ddms_compliance_suite.api_caller.caller import APICallDetail
from ddms_compliance_suite.test_orchestrator import APITestOrchestrator, TestSummary
from ddms_compliance_suite.input_parser.parser import ParsedAPISpec
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def parse_args():
"""解析命令行参数"""
parser = argparse.ArgumentParser(description='DDMS API测试工具')
# 基本参数
parser.add_argument('--base-url', required=True, help='API基础URL')
parser.add_argument('--verbose', '-v', action='store_true', help='启用详细日志')
parser.add_argument('--output', '-o', help='输出目录或主报告文件路径 (例如 ./test_reports/ 或 ./test_reports/summary.json)')
parser.add_argument('--format', choices=['json', 'html'], default='json', help='主测试摘要报告的输出格式')
parser.add_argument('--api-calls-output', help='API 调用详情的 Markdown 输出文件路径 (例如 ./test_reports/api_calls.md)。如果未提供,将尝试使用 --output 目录和默认文件名 api_call_details.md。始终会额外生成一个同名的 .txt 文件包含纯 cURL 命令。')
# API定义参数
api_group = parser.add_argument_group('API定义源')
api_group.add_argument('--yapi', help='YAPI定义文件路径')
api_group.add_argument('--swagger', help='Swagger定义文件路径')
# 过滤参数
filter_group = parser.add_argument_group('过滤选项')
filter_group.add_argument('--categories', help='YAPI分类逗号分隔')
filter_group.add_argument('--tags', help='Swagger标签逗号分隔')
filter_group.add_argument('--list-categories', action='store_true', help='列出YAPI分类')
filter_group.add_argument('--list-tags', action='store_true', help='列出Swagger标签')
# 新增:自定义测试用例参数组
custom_tc_group = parser.add_argument_group('自定义测试用例选项')
custom_tc_group.add_argument('--custom-test-cases-dir',
default=None, # 或者 './custom_testcases' 如果想设为默认
help='存放自定义APITestCase Python文件的目录路径。如果未提供则不加载自定义测试。')
# 新增LLM 配置选项
llm_group = parser.add_argument_group('LLM 配置选项 (可选)')
llm_group.add_argument('--llm-api-key',
default=os.environ.get("OPENAI_API_KEY"), # 尝试从环境变量获取
help='LLM服务的API密钥 (例如 OpenAI API Key)。默认从环境变量 OPENAI_API_KEY 读取。')
llm_group.add_argument('--llm-base-url',
default="https://dashscope.aliyuncs.com/compatible-mode/v1",
help='LLM服务的自定义基础URL (例如 OpenAI API代理)。')
llm_group.add_argument('--llm-model-name',
default="qwen-plus", # 设置一个常用的默认模型
help='要使用的LLM模型名称 (例如 "gpt-3.5-turbo", "gpt-4")。')
llm_group.add_argument('--use-llm-for-request-body',
action='store_true',
default=False, # 默认不使用LLM生成请求体
help='是否启用LLM为API请求生成请求体数据。')
llm_group.add_argument('--use-llm-for-path-params',
action='store_true',
default=False,
help='是否启用LLM为API请求生成路径参数。')
llm_group.add_argument('--use-llm-for-query-params',
action='store_true',
default=False,
help='是否启用LLM为API请求生成查询参数。')
llm_group.add_argument('--use-llm-for-headers',
action='store_true',
default=False,
help='是否启用LLM为API请求生成头部参数。')
# 新增:场景测试参数组
scenario_group = parser.add_argument_group('API测试阶段 (Stage) 选项 (可选)')
scenario_group.add_argument('--stages-dir',
default=None,
help='存放自定义APIStage Python文件的目录路径。如果未提供则不执行测试阶段。')
return parser.parse_args()
def list_yapi_categories(yapi_file: str):
"""列出YAPI分类"""
from ddms_compliance_suite.input_parser.parser import InputParser
logger.info(f"从YAPI文件解析分类: {yapi_file}")
parser = InputParser()
parsed_yapi = parser.parse_yapi_spec(yapi_file)
if not parsed_yapi:
logger.error(f"解析YAPI文件失败: {yapi_file}")
return
print("\nYAPI分类:")
for i, category in enumerate(parsed_yapi.categories, 1):
print(f"{i}. {category.get('name', '未命名')} - {category.get('desc', '无描述')}")
def list_swagger_tags(swagger_file: str):
"""列出Swagger标签"""
from ddms_compliance_suite.input_parser.parser import InputParser
logger.info(f"从Swagger文件解析标签: {swagger_file}")
parser = InputParser()
parsed_swagger = parser.parse_swagger_spec(swagger_file)
if not parsed_swagger:
logger.error(f"解析Swagger文件失败: {swagger_file}")
return
print("\nSwagger标签:")
for i, tag in enumerate(parsed_swagger.tags, 1):
print(f"{i}. {tag.get('name', '未命名')} - {tag.get('description', '无描述')}")
def save_results(summary: TestSummary, output_file_path: str, format_type: str):
"""保存主测试摘要结果"""
output_path = Path(output_file_path)
# Ensure the directory for the output file exists
try:
output_path.parent.mkdir(parents=True, exist_ok=True)
except OSError as e:
logger.error(f"Error creating directory for output file {output_path.parent}: {e}")
return
if format_type == 'json':
with open(output_path, 'w', encoding='utf-8') as f:
f.write(summary.to_json(pretty=True))
logger.info(f"测试结果已保存为JSON: {output_path}")
elif format_type == 'html':
# Creating simple HTML report
html_content = f"""
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>API测试报告</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
.summary {{ background-color: #f5f5f5; padding: 15px; border-radius: 5px; }}
.pass {{ color: green; }}
.fail {{ color: red; }}
.error {{ color: orange; }}
.skip {{ color: gray; }}
table {{ border-collapse: collapse; width: 100%; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #f2f2f2; }}
tr:nth-child(even) {{ background-color: #f9f9f9; }}
</style>
</head>
<body>
<h1>API测试报告</h1>
<div class="summary">
<h2>测试结果摘要</h2>
<p>总测试数: {summary.total_test_cases_executed}</p>
<p class="pass">通过: {summary.test_cases_passed}</p>
<p class="fail">失败: {summary.test_cases_failed}</p>
<p class="error">错误: {summary.test_cases_error}</p>
<p class="skip">跳过: {summary.test_cases_skipped_in_endpoint}</p>
<p>成功率: {summary.test_case_success_rate:.2f}%</p>
<p>总耗时: {summary.duration:.2f}秒</p>
<p>开始时间: {summary.start_time.isoformat()}</p>
<p>结束时间: {summary.end_time.isoformat() if summary.end_time else 'N/A'}</p>
</div>
<h2>详细测试结果</h2>
<table>
<tr>
<th>端点</th>
<th>测试用例ID</th>
<th>测试用例名称</th>
<th>状态</th>
<th>消息</th>
<th>耗时(秒)</th>
</tr>
"""
for endpoint_result in summary.detailed_results:
for tc_result in endpoint_result.executed_test_cases:
status_class = "pass" if tc_result.status == ExecutedTestCaseResult.Status.PASSED else \
"fail" if tc_result.status == ExecutedTestCaseResult.Status.FAILED else \
"error" if tc_result.status == ExecutedTestCaseResult.Status.ERROR else "skip"
html_content += f"""
<tr>
<td>{endpoint_result.endpoint_name} ({endpoint_result.endpoint_id})</td>
<td>{tc_result.test_case_id}</td>
<td>{tc_result.test_case_name}</td>
<td class="{status_class}">{tc_result.status.value}</td>
<td>{tc_result.message}</td>
<td>{tc_result.duration:.4f}</td>
</tr>
"""
html_content += """
</table>
</body>
</html>
"""
with open(output_path, 'w', encoding='utf-8') as f:
f.write(html_content)
logger.info(f"测试结果已保存为HTML: {output_path}")
def save_api_call_details_to_file(api_call_details: List[APICallDetail], output_dir_path: str, filename: str = "api_call_details.md"):
"""
将API调用详情列表保存到指定目录下的 Markdown 文件中。
同时,额外生成一个纯文本文件 (.txt),每行包含一个 cURL 命令。
"""
if not api_call_details:
logger.info("没有API调用详情可供保存。")
return
output_dir = Path(output_dir_path)
try:
output_dir.mkdir(parents=True, exist_ok=True)
except OSError as e:
logger.error(f"创建API调用详情输出目录 {output_dir} 失败: {e}")
return
# 主文件是 Markdown 文件
md_output_file = output_dir / filename
# 确保它是 .md尽管 main 函数应该已经处理了
if md_output_file.suffix.lower() not in ['.md', '.markdown']:
md_output_file = md_output_file.with_suffix('.md')
markdown_content = []
for detail in api_call_details:
# Request URL with params (if any)
url_to_display = detail.request_url
if detail.request_params:
try:
# Ensure urllib is available for this formatting step
import urllib.parse
query_string = urllib.parse.urlencode(detail.request_params)
url_to_display = f"{detail.request_url}?{query_string}"
except Exception as e:
logger.warning(f"Error formatting URL with params for display: {e}")
# Fallback to just the base URL if params formatting fails
markdown_content.append(f"## `{detail.request_method} {url_to_display}`")
markdown_content.append("**cURL Command:**")
markdown_content.append("```sh")
markdown_content.append(detail.curl_command)
markdown_content.append("```")
markdown_content.append("### Request Details")
markdown_content.append(f"- **Method:** `{detail.request_method}`")
markdown_content.append(f"- **Full URL:** `{url_to_display}`")
markdown_content.append("- **Headers:**")
markdown_content.append("```json")
markdown_content.append(json.dumps(detail.request_headers, indent=2, ensure_ascii=False))
markdown_content.append(" ```")
if detail.request_params:
markdown_content.append("- **Query Parameters:**")
markdown_content.append("```json")
markdown_content.append(json.dumps(detail.request_params, indent=2, ensure_ascii=False))
markdown_content.append(" ```")
if detail.request_body is not None:
markdown_content.append("- **Body:**")
body_lang = "text"
formatted_body = str(detail.request_body)
try:
# Try to parse as JSON for pretty printing
if isinstance(detail.request_body, str):
try:
parsed_json = json.loads(detail.request_body)
formatted_body = json.dumps(parsed_json, indent=2, ensure_ascii=False)
body_lang = "json"
except json.JSONDecodeError:
pass # Keep as text
elif isinstance(detail.request_body, (dict, list)):
formatted_body = json.dumps(detail.request_body, indent=2, ensure_ascii=False)
body_lang = "json"
except Exception as e:
logger.warning(f"Error formatting request body for Markdown: {e}")
markdown_content.append(f"```{body_lang}")
markdown_content.append(formatted_body)
markdown_content.append(" ```")
markdown_content.append("### Response Details")
markdown_content.append(f"- **Status Code:** `{detail.response_status_code}`")
markdown_content.append(f"- **Elapsed Time:** `{detail.response_elapsed_time:.4f}s`")
markdown_content.append("- **Headers:**")
markdown_content.append("```json")
markdown_content.append(json.dumps(detail.response_headers, indent=2, ensure_ascii=False))
markdown_content.append(" ```")
if detail.response_body is not None:
markdown_content.append("- **Body:**")
resp_body_lang = "text"
formatted_resp_body = str(detail.response_body)
try:
# Try to parse as JSON for pretty printing
if isinstance(detail.response_body, str):
try:
# If it's already a string that might be JSON, try parsing and re-dumping
parsed_json_resp = json.loads(detail.response_body)
formatted_resp_body = json.dumps(parsed_json_resp, indent=2, ensure_ascii=False)
resp_body_lang = "json"
except json.JSONDecodeError:
# It's a string, but not valid JSON, keep as text
pass
elif isinstance(detail.response_body, (dict, list)):
# It's already a dict/list, dump it as JSON
formatted_resp_body = json.dumps(detail.response_body, indent=2, ensure_ascii=False)
resp_body_lang = "json"
# If it's neither string nor dict/list (e.g. int, bool from parsed json), str() is fine.
except Exception as e:
logger.warning(f"Error formatting response body for Markdown: {e}")
markdown_content.append(f"```{resp_body_lang}")
markdown_content.append(formatted_resp_body)
markdown_content.append(" ```")
markdown_content.append("") # Add a blank line for spacing before next --- or EOF
markdown_content.append("---") # Separator
try:
with open(md_output_file, 'w', encoding='utf-8') as f_md:
f_md.write("\n".join(markdown_content))
logger.info(f"API调用详情已保存为 Markdown: {md_output_file}")
except Exception as e:
logger.error(f"保存API调用详情到 Markdown 文件 {md_output_file} 失败: {e}", exc_info=True)
# 额外生成 .txt 文件,只包含 cURL 命令
# txt_output_filename = md_output_file.with_suffix('.txt').name
# txt_output_file_path = output_dir / txt_output_filename
# try:
# with open(txt_output_file_path, 'w', encoding='utf-8') as f_txt:
# for detail in api_call_details:
# f_txt.write(detail.curl_command + '\n')
# logger.info(f"可直接执行的cURL命令已保存到纯文本文件: {txt_output_file_path}")
# except Exception as e:
# logger.error(f"保存cURL命令到文本文件 {txt_output_file_path} 失败: {e}", exc_info=True)
def main():
"""主函数"""
args = parse_args()
if args.verbose:
logging.getLogger('ddms_compliance_suite').setLevel(logging.DEBUG)
logger.setLevel(logging.DEBUG)
logger.debug("已启用详细日志模式")
if not args.yapi and not args.swagger:
logger.error("请提供API定义源--yapi 或 --swagger")
sys.exit(1)
if args.list_categories and args.yapi:
list_yapi_categories(args.yapi)
sys.exit(0)
if args.list_tags and args.swagger:
list_swagger_tags(args.swagger)
sys.exit(0)
categories = args.categories.split(',') if args.categories else None
tags = args.tags.split(',') if args.tags else None
DEFAULT_OUTPUT_DIR = Path("./test_reports")
output_directory: Path
main_report_file_path: Path
if args.output:
output_arg_path = Path(args.output)
if output_arg_path.suffix and output_arg_path.name: # Check if it looks like a file
output_directory = output_arg_path.parent
main_report_file_path = output_arg_path
else:
output_directory = output_arg_path
main_report_file_path = output_directory / f"summary.{args.format}"
else:
output_directory = DEFAULT_OUTPUT_DIR
main_report_file_path = output_directory / f"summary.{args.format}"
try:
output_directory.mkdir(parents=True, exist_ok=True)
logger.info(f"主输出目录设置为: {output_directory.resolve()}")
except OSError as e:
logger.error(f"创建主输出目录失败 {output_directory}: {e}")
sys.exit(1)
orchestrator = APITestOrchestrator(
base_url=args.base_url,
custom_test_cases_dir=args.custom_test_cases_dir,
llm_api_key=args.llm_api_key,
llm_base_url=args.llm_base_url,
llm_model_name=args.llm_model_name,
use_llm_for_request_body=args.use_llm_for_request_body,
use_llm_for_path_params=args.use_llm_for_path_params,
use_llm_for_query_params=args.use_llm_for_query_params,
use_llm_for_headers=args.use_llm_for_headers,
output_dir=str(output_directory),
stages_dir=args.stages_dir # 将 stages_dir 传递给编排器
)
test_summary: Optional[TestSummary] = None
parsed_spec_for_scenarios: Optional[ParsedAPISpec] = None # 用于存储已解析的规范,供场景使用
try:
if args.yapi:
logger.info(f"从YAPI文件运行测试: {args.yapi}")
# orchestrator.run_tests_from_yapi 现在返回一个元组
test_summary, parsed_spec_for_scenarios = orchestrator.run_tests_from_yapi(
yapi_file_path=args.yapi,
categories=categories,
custom_test_cases_dir=args.custom_test_cases_dir
)
if not parsed_spec_for_scenarios: # 检查解析是否成功
# orchestrator 内部应该已经记录了具体的解析错误
logger.error(f"YAPI文件 '{args.yapi}' 解析失败 (由编排器报告)。程序将退出。")
sys.exit(1)
elif args.swagger:
logger.info(f"从Swagger文件运行测试: {args.swagger}")
# orchestrator.run_tests_from_swagger 现在返回一个元组
test_summary, parsed_spec_for_scenarios = orchestrator.run_tests_from_swagger(
swagger_file_path=args.swagger,
tags=tags,
custom_test_cases_dir=args.custom_test_cases_dir
)
if not parsed_spec_for_scenarios: # 检查解析是否成功
logger.error(f"Swagger文件 '{args.swagger}' 解析失败 (由编排器报告)。程序将退出。")
sys.exit(1)
# Deliberately not having an else here, as the initial check for yapi/swagger presence handles it.
# If test_summary remains None here, it implies neither --yapi nor --swagger was processed correctly
# or an unexpected path was taken, which should be caught by later checks or an error.
except Exception as e:
logger.error(f"执行测试用例时发生意外错误: {e}", exc_info=True)
sys.exit(1)
if test_summary:
# 在保存单个测试用例结果之后运行API测试阶段 (如果指定了目录)
if args.stages_dir and parsed_spec_for_scenarios:
logger.info(f"开始执行API测试阶段 (Stages),目录: {args.stages_dir}")
# 注意:这里假设 test_orchestrator.py 中已经有了 run_stages_from_spec 方法
# 并且 APITestOrchestrator 的 __init__ 也接受 stages_dir
orchestrator.run_stages_from_spec( # 调用 run_stages_from_spec
# stages_dir is managed by orchestrator's __init__
parsed_spec=parsed_spec_for_scenarios, # 使用之前解析的规范
summary=test_summary # 将阶段结果添加到同一个摘要对象
)
logger.info("API测试阶段 (Stages) 执行完毕。")
# 阶段执行后,摘要已更新,重新最终确定和打印摘要
test_summary.finalize_summary() # 重新计算总时长等
test_summary.print_summary_to_console() # 打印包含阶段结果的更新摘要
# 保存主测试摘要 (现在可能包含测试阶段结果)
save_results(test_summary, str(main_report_file_path), args.format)
api_calls_output_path_str: Optional[str] = None
# 默认文件名现在是 .md
api_calls_filename: str = "api_call_details.md"
if args.api_calls_output:
api_calls_output_file = Path(args.api_calls_output)
# 确保后缀是 .md如果用户提供了其他后缀或没有后缀
if api_calls_output_file.suffix.lower() not in ['.md', '.markdown']:
api_calls_output_file = api_calls_output_file.with_suffix('.md')
logger.info(f"API调用详情输出文件名已调整为 Markdown 格式: {api_calls_output_file.name}")
api_calls_output_path_str = str(api_calls_output_file.parent)
api_calls_filename = api_calls_output_file.name
logger.info(f"API调用详情将以 Markdown 格式保存到: {api_calls_output_file}")
elif args.output:
output_arg_path = Path(args.output)
if output_arg_path.is_dir():
api_calls_output_path_str = str(output_arg_path)
else:
api_calls_output_path_str = str(output_arg_path.parent)
logger.info(f"API调用详情将以 Markdown 格式保存到目录 '{api_calls_output_path_str}' (使用默认文件名 '{api_calls_filename}')")
else:
api_calls_output_path_str = "."
logger.info(f"API调用详情将以 Markdown 格式保存到当前目录 '.' (使用默认文件名 '{api_calls_filename}')")
# 保存API调用详情
if orchestrator and api_calls_output_path_str:
save_api_call_details_to_file(
orchestrator.get_api_call_details(),
api_calls_output_path_str,
filename=api_calls_filename
)
# Improved HTML report summary access
failed_count = getattr(test_summary, 'endpoints_failed', 0) + getattr(test_summary, 'test_cases_failed', 0)
error_count = getattr(test_summary, 'endpoints_error', 0) + getattr(test_summary, 'test_cases_error', 0)
if failed_count > 0 or error_count > 0:
logger.info("部分测试失败或出错,请检查报告。")
# sys.exit(1) # Keep this commented if a report is always desired regardless of outcome
else:
logger.info("所有测试完成。")
else:
logger.error("未能生成测试摘要。")
sys.exit(1)
sys.exit(0)
if __name__ == '__main__':
main()
# python run_api_tests.py --base-url http://127.0.0.1:4523/m1/6389742-6086420-default --swagger assets/doc/井筒API示例swagger.json --custom-test-cases-dir ./custom_testcases \
# --verbose \
# --output test_report.json
# python run_api_tests.py --base-url https://127.0.0.1:4523/m1/6389742-6086420-default --yapi assets/doc/井筒API示例_simple.json --custom-test-cases-dir ./custom_testcases \
# --verbose \
# --output test_report.json
# 示例:同时运行测试用例和场景
# python run_api_tests.py --base-url http://127.0.0.1:8000 --swagger ./assets/doc/petstore_swagger.json --custom-test-cases-dir ./custom_testcases --stages-dir ./custom_stages -v -o reports/