387 lines
14 KiB
Python
387 lines
14 KiB
Python
#!/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 ddms_compliance_suite.test_orchestrator import APITestOrchestrator, TestSummary
|
||
from ddms_compliance_suite.models.rule_models import (
|
||
PerformanceRule, SecurityRule, RESTfulDesignRule, ErrorHandlingRule,
|
||
RuleCategory, TargetType, RuleLifecycle, RuleScope, SeverityLevel
|
||
)
|
||
|
||
# 配置日志
|
||
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='输出文件路径')
|
||
parser.add_argument('--format', choices=['json', 'html'], default='json', help='输出格式')
|
||
|
||
# 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标签')
|
||
|
||
# 规则库参数
|
||
rule_group = parser.add_argument_group('规则库选项')
|
||
rule_group.add_argument('--rules-path', default='./rules', help='规则库路径')
|
||
# rule_group.add_argument('--disable-rules', action='store_true', help='禁用规则验证',default=False)
|
||
rule_group.add_argument('--disable-rules', action='store_true', help='禁用规则验证',default=True)
|
||
rule_group.add_argument('--list-rules', action='store_true', help='列出可用规则',default=False)
|
||
rule_group.add_argument('--create-basic-rules', action='store_true',
|
||
help='创建基本规则集（性能、安全、RESTful设计、错误处理）',default=False)
|
||
|
||
# 新增：自定义测试用例参数组
|
||
custom_tc_group = parser.add_argument_group('自定义测试用例选项')
|
||
custom_tc_group.add_argument('--custom-test-cases-dir',
|
||
default=None, # 或者 './custom_testcases' 如果想设为默认
|
||
help='存放自定义APITestCase 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 create_basic_rules(orchestrator):
|
||
"""创建基本规则集"""
|
||
logger.info("创建基本规则集...")
|
||
|
||
# 1. 性能规则 - 响应时间不超过500毫秒
|
||
performance_rule = PerformanceRule(
|
||
id="response-time-max-500ms",
|
||
name="响应时间不超过500毫秒",
|
||
description="验证API响应时间不超过500毫秒",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=SeverityLevel.WARNING,
|
||
target_type=TargetType.API_RESPONSE,
|
||
lifecycle=RuleLifecycle.RESPONSE_VALIDATION,
|
||
scope=RuleScope.RESPONSE_TIME,
|
||
threshold=500.0,
|
||
metric="response_time",
|
||
unit="ms"
|
||
)
|
||
|
||
# 2. 安全规则 - HTTPS必须使用
|
||
security_rule = SecurityRule(
|
||
id="https-only-rule",
|
||
name="HTTPS强制使用规则",
|
||
description="验证API请求是否使用了HTTPS协议",
|
||
category=RuleCategory.SECURITY,
|
||
severity=SeverityLevel.ERROR,
|
||
target_type=TargetType.API_REQUEST,
|
||
lifecycle=RuleLifecycle.REQUEST_PREPARATION,
|
||
scope=RuleScope.SECURITY,
|
||
check_type="transport_security",
|
||
expected_value="https"
|
||
)
|
||
|
||
# 3. RESTful设计规则 - URL路径格式
|
||
restful_rule = RESTfulDesignRule(
|
||
id="restful-url-pattern",
|
||
name="RESTful URL设计规则",
|
||
description="验证API URL是否符合RESTful设计规范",
|
||
category=RuleCategory.API_DESIGN,
|
||
severity=SeverityLevel.WARNING,
|
||
target_type=TargetType.API_REQUEST,
|
||
lifecycle=RuleLifecycle.REQUEST_PREPARATION,
|
||
scope=RuleScope.REQUEST_URL,
|
||
design_aspect="URL设计",
|
||
pattern=r"^/api/v\d+/[a-z0-9-]+(/[a-z0-9-]+)*$"
|
||
)
|
||
|
||
# 4. 错误处理规则 - 错误响应格式
|
||
error_rule = ErrorHandlingRule(
|
||
id="standard-error-response",
|
||
name="标准错误响应格式规则",
|
||
description="验证API错误响应是否符合标准格式",
|
||
category=RuleCategory.ERROR_HANDLING,
|
||
severity=SeverityLevel.WARNING,
|
||
target_type=TargetType.API_RESPONSE,
|
||
lifecycle=RuleLifecycle.RESPONSE_VALIDATION,
|
||
scope=RuleScope.RESPONSE_BODY,
|
||
error_code="*",
|
||
expected_status=400
|
||
)
|
||
|
||
# 保存规则到规则库
|
||
orchestrator.rule_repo.save_rule(performance_rule)
|
||
orchestrator.rule_repo.save_rule(security_rule)
|
||
orchestrator.rule_repo.save_rule(restful_rule)
|
||
orchestrator.rule_repo.save_rule(error_rule)
|
||
|
||
logger.info("已创建基本规则集")
|
||
|
||
def list_rules(rule_repo_path: str):
|
||
"""列出可用规则"""
|
||
from ddms_compliance_suite.models.config_models import RuleRepositoryConfig, RuleStorageConfig
|
||
from ddms_compliance_suite.rule_repository.repository import RuleRepository
|
||
|
||
# 初始化规则库
|
||
rule_config = RuleRepositoryConfig(
|
||
storage=RuleStorageConfig(path=rule_repo_path)
|
||
)
|
||
repo = RuleRepository(rule_config)
|
||
|
||
# 查询所有规则
|
||
rules = repo.query_rules()
|
||
|
||
if not rules:
|
||
print("未找到规则。使用 --create-basic-rules 创建基本规则集。")
|
||
return
|
||
|
||
print(f"\n找到 {len(rules)} 条规则:")
|
||
|
||
# 按类别分组
|
||
rules_by_category = {}
|
||
for rule in rules:
|
||
category = str(rule.category)
|
||
if category not in rules_by_category:
|
||
rules_by_category[category] = []
|
||
rules_by_category[category].append(rule)
|
||
|
||
# 打印规则
|
||
for category, category_rules in rules_by_category.items():
|
||
print(f"\n● {category}:")
|
||
for rule in category_rules:
|
||
print(f" - {rule.id}: {rule.name} (严重性: {rule.severity}, 版本: {rule.version})")
|
||
print(f" {rule.description}")
|
||
print(f" 生命周期: {rule.lifecycle}, 作用域: {rule.scope}")
|
||
|
||
def save_results(summary: TestSummary, output_file: str, format_type: str):
|
||
"""保存测试结果"""
|
||
if format_type == 'json':
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
f.write(summary.to_json(pretty=True))
|
||
logger.info(f"测试结果已保存为JSON: {output_file}")
|
||
elif format_type == 'html':
|
||
# 创建简单的HTML报告
|
||
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}</p>
|
||
<p class="pass">通过: {summary.passed}</p>
|
||
<p class="fail">失败: {summary.failed}</p>
|
||
<p class="error">错误: {summary.error}</p>
|
||
<p class="skip">跳过: {summary.skipped}</p>
|
||
<p>成功率: {summary.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>状态</th>
|
||
<th>消息</th>
|
||
<th>响应码</th>
|
||
<th>耗时(秒)</th>
|
||
</tr>
|
||
"""
|
||
|
||
for result in summary.results:
|
||
status_class = "pass" if result.status == "通过" else "fail" if result.status == "失败" else "error" if result.status == "错误" else "skip"
|
||
response_code = result.api_response.status_code if result.api_response else "N/A"
|
||
|
||
html_content += f"""
|
||
<tr>
|
||
<td>{result.endpoint_id}</td>
|
||
<td class="{status_class}">{result.status}</td>
|
||
<td>{result.message}</td>
|
||
<td>{response_code}</td>
|
||
<td>{result.elapsed_time:.4f}</td>
|
||
</tr>
|
||
"""
|
||
|
||
html_content += """
|
||
</table>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
f.write(html_content)
|
||
logger.info(f"测试结果已保存为HTML: {output_file}")
|
||
|
||
def main():
|
||
"""主函数"""
|
||
args = parse_args()
|
||
|
||
if args.verbose:
|
||
logging.getLogger('ddms_compliance_suite').setLevel(logging.DEBUG)
|
||
logger.setLevel(logging.DEBUG)
|
||
logger.debug("已启用详细日志模式")
|
||
|
||
# 列出规则
|
||
if args.list_rules:
|
||
list_rules(args.rules_path)
|
||
return 0
|
||
|
||
# 检查是否提供了API定义源
|
||
if not args.yapi and not args.swagger:
|
||
logger.error("请提供API定义源：--yapi 或 --swagger")
|
||
return 1
|
||
|
||
# 列出分类/标签
|
||
if args.list_categories and args.yapi:
|
||
list_yapi_categories(args.yapi)
|
||
return 0
|
||
|
||
if args.list_tags and args.swagger:
|
||
list_swagger_tags(args.swagger)
|
||
return 0
|
||
|
||
# 解析分类/标签过滤器
|
||
categories = args.categories.split(',') if args.categories else None
|
||
tags = args.tags.split(',') if args.tags else None
|
||
|
||
# 实例化测试编排器
|
||
# 将 custom_test_cases_dir 参数传递给 APITestOrchestrator 的构造函数
|
||
orchestrator = APITestOrchestrator(
|
||
base_url=args.base_url,
|
||
rule_repo_path=args.rules_path,
|
||
custom_test_cases_dir=args.custom_test_cases_dir # 新增参数
|
||
)
|
||
|
||
# 创建基本规则集
|
||
if args.create_basic_rules:
|
||
create_basic_rules(orchestrator)
|
||
|
||
# 如果禁用规则，替换规则执行器的方法
|
||
if args.disable_rules:
|
||
logger.info("规则验证已禁用")
|
||
# 替换规则执行器的方法为空实现
|
||
orchestrator.rule_executor.execute_rules_for_lifecycle = lambda *args, **kwargs: []
|
||
orchestrator.rule_executor.execute_rules_for_target = lambda *args, **kwargs: []
|
||
|
||
# 运行测试
|
||
summary = None
|
||
|
||
if args.yapi:
|
||
logger.info(f"从YAPI文件运行测试: {args.yapi}")
|
||
summary = orchestrator.run_tests_from_yapi(
|
||
yapi_file_path=args.yapi,
|
||
categories=categories,
|
||
custom_test_cases_dir=args.custom_test_cases_dir # 也传递给具体执行方法，以支持运行时覆盖
|
||
)
|
||
elif args.swagger:
|
||
logger.info(f"从Swagger文件运行测试: {args.swagger}")
|
||
summary = orchestrator.run_tests_from_swagger(
|
||
swagger_file_path=args.swagger,
|
||
tags=tags,
|
||
custom_test_cases_dir=args.custom_test_cases_dir # 也传递给具体执行方法
|
||
)
|
||
else:
|
||
logger.error("必须提供YAPI或Swagger文件路径")
|
||
|
||
if not summary:
|
||
logger.error("测试执行失败")
|
||
return 1
|
||
|
||
# 打印结果摘要
|
||
summary.print_summary_to_console()
|
||
|
||
# 保存结果
|
||
if args.output:
|
||
save_results(summary, args.output, args.format)
|
||
|
||
# 根据测试结果设置退出码
|
||
# 直接访问 TestSummary 的属性
|
||
has_endpoint_errors = summary.endpoints_error > 0
|
||
has_endpoint_failures = summary.endpoints_failed > 0
|
||
|
||
# 或者，也可以关注测试用例级别的失败/错误
|
||
# has_test_case_errors = summary.test_cases_error > 0
|
||
# has_test_case_failures = summary.test_cases_failed > 0
|
||
|
||
if has_endpoint_errors or has_endpoint_failures:
|
||
return 1 # 表示测试运行中存在失败或错误
|
||
else:
|
||
return 0 # 所有端点测试通过或部分成功（无错误或关键失败）
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(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 |