289 lines
13 KiB
Python
289 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
AI-Powered API Test Generator, Runner & Coverage Analyzer
|
||
──────────────────────────────────────────────────────────
|
||
Usage:
|
||
# 基础用法
|
||
python main.py --api-desc examples/api_desc.json \\
|
||
--requirements "创建用户,删除用户"
|
||
|
||
# 在需求中自然语言描述参数生成要求
|
||
python main.py --api-desc examples/api_desc.json \\
|
||
--requirements "创建用户（测试参数不少于10组，覆盖边界值和异常值）,删除用户"
|
||
|
||
# 从文件读取需求（每行一条，支持参数生成指令）
|
||
python main.py --api-desc examples/api_desc.json \\
|
||
--req-file examples/requirements.txt \\
|
||
--batch-size 10
|
||
"""
|
||
|
||
import argparse
|
||
import logging
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from config import config
|
||
from core.parser import InterfaceParser, ApiDescriptor
|
||
from core.requirement_parser import RequirementParser, ParamConstraint
|
||
from core.prompt_builder import PromptBuilder
|
||
from core.llm_client import LLMClient
|
||
from core.test_generator import TestGenerator
|
||
from core.test_runner import TestRunner
|
||
from core.analyzer import CoverageAnalyzer, CoverageReport
|
||
from core.report_generator import ReportGenerator
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 日志初始化
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
def setup_logging(debug: bool = False):
|
||
"""
|
||
配置日志级别。
|
||
|
||
⚠️ Bug 规避（openai-python v2.24.0 + pydantic-core）：
|
||
第三方 SDK logger 强制锁定在 WARNING，避免触发
|
||
model_dump(by_alias=None) → pydantic-core Rust 层 TypeError。
|
||
Ref: https://github.com/openai/openai-python/issues/2921
|
||
"""
|
||
root_level = logging.DEBUG if debug else logging.INFO
|
||
logging.basicConfig(
|
||
level=root_level,
|
||
format="%(asctime)s [%(levelname)s] %(name)s — %(message)s",
|
||
handlers=[logging.StreamHandler(sys.stdout)],
|
||
)
|
||
for name in ["openai", "openai._base_client", "anthropic", "httpx", "httpcore"]:
|
||
logging.getLogger(name).setLevel(logging.WARNING)
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# CLI
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
def build_arg_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(
|
||
description="AI-Powered API Test Generator & Coverage Analyzer",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
参数生成指令示例（可直接写在需求文本中）：
|
||
"创建用户，测试参数不少于10组，覆盖边界值和异常值"
|
||
"查询接口（至少8组，边界值）"
|
||
"所有接口参数不少于5组，覆盖等价类"
|
||
""",
|
||
)
|
||
p.add_argument("--api-desc", required=True, help="接口描述 JSON 文件")
|
||
p.add_argument("--requirements", default="", help="测试需求（逗号分隔，支持参数生成指令）")
|
||
p.add_argument("--req-file", default="", help="测试需求文件（每行一条）")
|
||
p.add_argument("--batch-size", type=int, default=0, help="每批接口数量（0=使用 config.LLM_BATCH_SIZE）")
|
||
p.add_argument("--workers", type=int, default=0, help="并行执行线程数（0=使用 config.TEST_MAX_WORKERS）")
|
||
p.add_argument("--skip-run", action="store_true", help="只生成，不执行")
|
||
p.add_argument("--skip-analyze", action="store_true", help="跳过覆盖率分析")
|
||
p.add_argument("--output-dir", default="", help="测试文件根输出目录")
|
||
p.add_argument("--debug", action="store_true", help="开启 DEBUG 日志（第三方 SDK 仍保持 WARNING）")
|
||
return p
|
||
|
||
|
||
def load_requirements(args) -> list[str]:
|
||
reqs: list[str] = []
|
||
if args.requirements:
|
||
reqs += [r.strip() for r in args.requirements.split(",") if r.strip()]
|
||
if args.req_file:
|
||
with open(args.req_file, "r", encoding="utf-8") as f:
|
||
reqs += [line.strip() for line in f if line.strip()]
|
||
if not reqs:
|
||
logger.error("未提供任何测试需求，请使用 --requirements 或 --req-file 指定。")
|
||
sys.exit(1)
|
||
return reqs
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# Main
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
def main():
|
||
args = build_arg_parser().parse_args()
|
||
setup_logging(debug=args.debug)
|
||
|
||
if args.output_dir: config.GENERATED_TESTS_DIR = args.output_dir
|
||
if args.batch_size > 0: config.LLM_BATCH_SIZE = args.batch_size
|
||
if args.workers > 0: config.TEST_MAX_WORKERS = args.workers
|
||
|
||
# ── Step 1: 解析接口描述 ──────────────────────────────────
|
||
logger.info("▶ Step 1: 解析接口描述 …")
|
||
parser = InterfaceParser()
|
||
descriptor = parser.parse_file(args.api_desc)
|
||
|
||
project = descriptor.project
|
||
project_desc = descriptor.description
|
||
interfaces = descriptor.interfaces
|
||
|
||
logger.info(f" 项目名称：{project}")
|
||
logger.info(f" 项目描述：{project_desc}")
|
||
logger.info(f" 接口数量：{len(interfaces)}")
|
||
for iface in interfaces:
|
||
if iface.protocol == "function":
|
||
logger.info(
|
||
f" [function] {iface.name} "
|
||
f"← {iface.source_file} (module: {iface.module_path})"
|
||
)
|
||
else:
|
||
logger.info(f" [http] {iface.name} ← {iface.http_full_url}")
|
||
|
||
# ── Step 2: 加载并解析测试需求（含参数生成指令）──────────
|
||
logger.info("▶ Step 2: 加载并解析测试需求 …")
|
||
raw_requirements = load_requirements(args)
|
||
|
||
req_parser = RequirementParser()
|
||
clean_reqs, per_req_constraints, global_constraint = req_parser.parse(
|
||
raw_requirements
|
||
)
|
||
|
||
# 打印原始需求 & 解析结果
|
||
for i, (raw, clean, c) in enumerate(
|
||
zip(raw_requirements, clean_reqs, per_req_constraints), 1
|
||
):
|
||
if c.has_param_directive:
|
||
logger.info(
|
||
f" {i}. {raw}\n"
|
||
f" └─ 参数约束：{c} | 业务需求：{clean}"
|
||
)
|
||
else:
|
||
logger.info(f" {i}. {raw}")
|
||
|
||
if global_constraint.has_param_directive:
|
||
logger.info(f" 全局参数约束：{global_constraint}")
|
||
|
||
# 使用"清洗后的需求"传给 LLM（去掉参数指令，保留纯业务描述）
|
||
# 同时保留原始需求用于覆盖率分析（保证与用户输入一致）
|
||
effective_requirements = clean_reqs
|
||
|
||
# ── Step 3: 构建 Prompt 并调用 LLM 生成测试用例 ──────────
|
||
logger.info(
|
||
f"▶ Step 3: 调用 LLM 生成测试用例 "
|
||
f"(batch_size={getattr(config, 'LLM_BATCH_SIZE', 10)}) …"
|
||
)
|
||
builder = PromptBuilder()
|
||
iface_summary = parser.to_summary_dict(interfaces)
|
||
project_header = builder.build_project_header(project, project_desc)
|
||
|
||
# System Prompt：注入全局参数约束规则
|
||
system_prompt = builder.get_system_prompt(
|
||
global_constraint=global_constraint
|
||
)
|
||
|
||
test_cases = LLMClient().generate_test_cases(
|
||
system_prompt=system_prompt,
|
||
user_prompt="",
|
||
iface_summaries=iface_summary,
|
||
requirements=effective_requirements,
|
||
project_header=project_header,
|
||
param_constraint=global_constraint, # ← 注入参数约束
|
||
)
|
||
logger.info(f" 共生成测试用例：{len(test_cases)} 个")
|
||
|
||
# ── Step 4: 生成测试文件 ──────────────────────────────────
|
||
logger.info("▶ Step 4: 生成测试文件 …")
|
||
generator = TestGenerator(project=project, project_desc=project_desc)
|
||
test_files = generator.generate(test_cases)
|
||
out = generator.output_dir
|
||
logger.info(f" 输出目录：{out.resolve()}")
|
||
|
||
run_results = []
|
||
|
||
if not args.skip_run:
|
||
# ── Step 5: 并行执行测试 ──────────────────────────────
|
||
logger.info(
|
||
f"▶ Step 5: 执行测试 "
|
||
f"(workers={getattr(config, 'TEST_MAX_WORKERS', 8)}) …"
|
||
)
|
||
runner = TestRunner()
|
||
run_results = runner.run_all(test_files)
|
||
runner.print_summary(run_results)
|
||
runner.save_results(run_results, str(out / "run_results.json"))
|
||
|
||
if not args.skip_analyze:
|
||
# ── Step 6: 覆盖率分析 ────────────────────────────────
|
||
logger.info("▶ Step 6: 覆盖率分析 …")
|
||
report = CoverageAnalyzer(
|
||
interfaces=interfaces,
|
||
requirements=raw_requirements, # 用原始需求做覆盖率分析
|
||
test_cases=test_cases,
|
||
run_results=run_results,
|
||
).analyze()
|
||
|
||
# ── Step 7: 生成报告 ──────────────────────────────────
|
||
logger.info("▶ Step 7: 生成报告 …")
|
||
rg = ReportGenerator()
|
||
rg.save_json(report, str(out / "coverage_report.json"))
|
||
rg.save_html(report, str(out / "coverage_report.html"))
|
||
|
||
_print_terminal_summary(report, out, project, global_constraint)
|
||
|
||
logger.info(f"\n✅ 完成。输出目录：{out.resolve()}")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 终端摘要
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
def _print_terminal_summary(
|
||
report: "CoverageReport",
|
||
out: Path,
|
||
project: str,
|
||
global_constraint: ParamConstraint,
|
||
):
|
||
W = 68
|
||
|
||
def bar(rate: float, w: int = 20) -> str:
|
||
filled = int(rate * w)
|
||
empty = w - filled
|
||
icon = "✅" if rate >= 0.8 else ("⚠️ " if rate >= 0.5 else "❌")
|
||
return f"{'█' * filled}{'░' * empty} {rate * 100:.1f}% {icon}"
|
||
|
||
print(f"\n{'═' * W}")
|
||
print(f" 项目：{project}")
|
||
print(f" 覆盖率摘要")
|
||
print(f"{'═' * W}")
|
||
print(f" 接口覆盖率 {bar(report.interface_coverage_rate)}")
|
||
print(f" 需求覆盖率 {bar(report.requirement_coverage_rate)}")
|
||
print(f" 入参覆盖率 {bar(report.avg_in_param_coverage_rate)}")
|
||
print(f" 成功返回字段覆盖 {bar(report.avg_success_field_coverage_rate)}")
|
||
print(f" 失败返回字段覆盖 {bar(report.avg_failure_field_coverage_rate)}")
|
||
print(f" 用例通过率 {bar(report.pass_rate)}")
|
||
print(f"{'─' * W}")
|
||
print(f" 测试用例总数 {report.total_test_cases}")
|
||
print(f" 覆盖缺口总数 {len(report.gaps)}")
|
||
print(f" 🔴 严重缺口 {report.critical_gap_count}")
|
||
print(f" 🟠 高优先级缺口 {report.high_gap_count}")
|
||
|
||
# 参数约束达成情况
|
||
if global_constraint.has_param_directive:
|
||
print(f"{'─' * W}")
|
||
print(f" 参数生成约束 {global_constraint}")
|
||
actual = report.total_test_cases
|
||
needed = global_constraint.min_groups
|
||
status = "✅ 已满足" if actual >= needed else f"❌ 未满足（需 {needed} 组，实际 {actual} 组）"
|
||
print(f" 参数组数要求 {status}")
|
||
|
||
if report.gaps:
|
||
print(f"{'─' * W}")
|
||
print(" Top 缺口（最多显示8条）：")
|
||
icons = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🔵"}
|
||
for g in report.gaps[:8]:
|
||
print(f" {icons.get(g.severity, '⚪')} [{g.gap_type}] {g.target}")
|
||
print(f" → {g.suggestion}")
|
||
|
||
print(f"{'─' * W}")
|
||
print(f" 输出目录：{out.resolve()}")
|
||
print(f" • coverage_report.html")
|
||
print(f" • coverage_report.json")
|
||
print(f" • run_results.json")
|
||
print(f" • test_cases_summary.json")
|
||
print(f"{'═' * W}\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |