#!/usr/bin/env python3 """ API端点测试 测试系统的所有REST API接口 """ import asyncio import aiohttp import json from datetime import datetime from typing import Dict, Any, List class APITestSuite: """API测试套件""" def __init__(self, base_url: str = "http://localhost:8000"): self.base_url = base_url self.api_base = f"{base_url}/api/v1" self.session = None self.test_results = [] self.passed = 0 self.failed = 0 async def setup(self): """初始化测试会话""" self.session = aiohttp.ClientSession() async def cleanup(self): """清理测试会话""" if self.session: await self.session.close() def add_result(self, endpoint: str, method: str, passed: bool, message: str = None): """添加测试结果""" self.test_results.append({ 'endpoint': endpoint, 'method': method, 'passed': passed, 'message': message }) if passed: self.passed += 1 else: self.failed += 1 async def test_health_check(self): """测试健康检查端点""" print("\n📋 测试健康检查API") try: async with self.session.get(f"{self.base_url}/health") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ 健康检查: {data.get('status', 'unknown')}") self.add_result("/health", "GET", True) else: print(f" ❌ 健康检查失败: HTTP {resp.status}") self.add_result("/health", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ 健康检查异常: {e}") self.add_result("/health", "GET", False, str(e)) async def test_import_endpoints(self): """测试导入相关端点""" print("\n📋 测试导入API") # 1. 获取导入状态 try: async with self.session.get(f"{self.api_base}/import/status") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /import/status - 文件数: {data.get('files_in_inbox', 0)}") self.add_result("/import/status", "GET", True) else: print(f" ❌ GET /import/status - HTTP {resp.status}") self.add_result("/import/status", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /import/status - {e}") self.add_result("/import/status", "GET", False, str(e)) # 2. 获取已处理文件 try: async with self.session.get(f"{self.api_base}/import/files?limit=10") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /import/files - 总数: {data.get('total', 0)}") self.add_result("/import/files", "GET", True) else: print(f" ❌ GET /import/files - HTTP {resp.status}") self.add_result("/import/files", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /import/files - {e}") self.add_result("/import/files", "GET", False, str(e)) # 3. 获取原始订单 try: async with self.session.get(f"{self.api_base}/import/orders?page=1&page_size=10") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /import/orders - 总数: {data.get('total', 0)}") self.add_result("/import/orders", "GET", True) else: print(f" ❌ GET /import/orders - HTTP {resp.status}") self.add_result("/import/orders", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /import/orders - {e}") self.add_result("/import/orders", "GET", False, str(e)) async def test_products_endpoints(self): """测试产品主表相关端点""" print("\n📋 测试产品主表API") # 1. 获取产品统计 try: async with self.session.get(f"{self.api_base}/products-master/stats") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /products-master/stats - 总产品数: {data.get('total_products', 0)}") self.add_result("/products-master/stats", "GET", True) else: print(f" ❌ GET /products-master/stats - HTTP {resp.status}") self.add_result("/products-master/stats", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /products-master/stats - {e}") self.add_result("/products-master/stats", "GET", False, str(e)) # 2. 获取产品列表 try: async with self.session.get(f"{self.api_base}/products-master?page=1&page_size=10") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /products-master - 返回 {len(data.get('items', []))} 个产品") self.add_result("/products-master", "GET", True) else: print(f" ❌ GET /products-master - HTTP {resp.status}") self.add_result("/products-master", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /products-master - {e}") self.add_result("/products-master", "GET", False, str(e)) # 3. 品牌统计 try: async with self.session.get(f"{self.api_base}/products-master/brands") as resp: if resp.status == 200: data = await resp.json() brands = data.get('brands', []) print(f" ✅ GET /products-master/brands - {len(brands)} 个品牌") self.add_result("/products-master/brands", "GET", True) else: print(f" ❌ GET /products-master/brands - HTTP {resp.status}") self.add_result("/products-master/brands", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /products-master/brands - {e}") self.add_result("/products-master/brands", "GET", False, str(e)) async def test_procurement_endpoints(self): """测试采购订单相关端点""" print("\n📋 测试采购订单API") # 1. 获取采购订单列表 try: params = { "page": 1, "page_size": 10, "procurement_status": "PENDING" } async with self.session.get(f"{self.api_base}/procurement/orders", params=params) as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /procurement/orders - 总数: {data.get('total', 0)}") self.add_result("/procurement/orders", "GET", True) else: print(f" ❌ GET /procurement/orders - HTTP {resp.status}") self.add_result("/procurement/orders", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /procurement/orders - {e}") self.add_result("/procurement/orders", "GET", False, str(e)) # 2. 获取采购统计 try: async with self.session.get(f"{self.api_base}/procurement/statistics") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /procurement/statistics - 待采购: {data.get('status_counts', {}).get('PENDING', 0)}") self.add_result("/procurement/statistics", "GET", True) else: print(f" ❌ GET /procurement/statistics - HTTP {resp.status}") self.add_result("/procurement/statistics", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /procurement/statistics - {e}") self.add_result("/procurement/statistics", "GET", False, str(e)) # 3. 获取采购方式统计 try: async with self.session.get(f"{self.api_base}/procurement/methods/summary") as resp: if resp.status == 200: data = await resp.json() methods = data.get('method_counts', {}) print(f" ✅ GET /procurement/methods/summary - {len(methods)} 种采购方式") self.add_result("/procurement/methods/summary", "GET", True) else: print(f" ❌ GET /procurement/methods/summary - HTTP {resp.status}") self.add_result("/procurement/methods/summary", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /procurement/methods/summary - {e}") self.add_result("/procurement/methods/summary", "GET", False, str(e)) async def test_monitoring_endpoints(self): """测试监控相关端点""" print("\n📋 测试监控API") # 1. 性能指标 try: async with self.session.get(f"{self.api_base}/monitoring/performance") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /monitoring/performance - 记录数: {data.get('total', 0)}") self.add_result("/monitoring/performance", "GET", True) else: print(f" ❌ GET /monitoring/performance - HTTP {resp.status}") self.add_result("/monitoring/performance", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /monitoring/performance - {e}") self.add_result("/monitoring/performance", "GET", False, str(e)) # 2. 审计日志 try: async with self.session.get(f"{self.api_base}/monitoring/audit-logs?limit=10") as resp: if resp.status == 200: data = await resp.json() print(f" ✅ GET /monitoring/audit-logs - 日志数: {data.get('total', 0)}") self.add_result("/monitoring/audit-logs", "GET", True) else: print(f" ❌ GET /monitoring/audit-logs - HTTP {resp.status}") self.add_result("/monitoring/audit-logs", "GET", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ GET /monitoring/audit-logs - {e}") self.add_result("/monitoring/audit-logs", "GET", False, str(e)) async def test_post_endpoints(self): """测试POST端点（只测试安全的操作）""" print("\n📋 测试POST API") # 1. 触发导入（带dry_run参数，不实际执行） try: payload = {"force": False, "dry_run": True} async with self.session.post( f"{self.api_base}/import/trigger", json=payload ) as resp: if resp.status in [200, 422]: # 422可能是验证错误，也算测试通过 print(f" ✅ POST /import/trigger (dry_run) - HTTP {resp.status}") self.add_result("/import/trigger", "POST", True) else: print(f" ❌ POST /import/trigger - HTTP {resp.status}") self.add_result("/import/trigger", "POST", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ POST /import/trigger - {e}") self.add_result("/import/trigger", "POST", False, str(e)) # 2. 生成产品主表（dry_run模式） try: payload = {"force_rebuild": False, "batch_size": 10} async with self.session.post( f"{self.api_base}/products-master/generate", json=payload ) as resp: if resp.status in [200, 422]: print(f" ✅ POST /products-master/generate - HTTP {resp.status}") self.add_result("/products-master/generate", "POST", True) else: print(f" ❌ POST /products-master/generate - HTTP {resp.status}") self.add_result("/products-master/generate", "POST", False, f"HTTP {resp.status}") except Exception as e: print(f" ❌ POST /products-master/generate - {e}") self.add_result("/products-master/generate", "POST", False, str(e)) def print_summary(self): """打印测试总结""" print("\n" + "=" * 60) print("📊 API测试总结") print("=" * 60) # 按端点分组显示结果 endpoints_by_group = {} for result in self.test_results: group = result['endpoint'].split('/')[2] if len(result['endpoint'].split('/')) > 2 else 'other' if group not in endpoints_by_group: endpoints_by_group[group] = [] endpoints_by_group[group].append(result) for group, results in endpoints_by_group.items(): print(f"\n{group.upper()} 端点:") for result in results: status = "✅" if result['passed'] else "❌" message = f" - {result['message']}" if result['message'] else "" print(f" {status} {result['method']} {result['endpoint']}{message}") print("\n" + "-" * 60) total = self.passed + self.failed if total > 0: pass_rate = (self.passed / total) * 100 print(f"总计: {total} 个API测试") print(f"通过: {self.passed} 个") print(f"失败: {self.failed} 个") print(f"通过率: {pass_rate:.1f}%") if pass_rate == 100: print("\n🎉 所有API测试通过！") elif pass_rate >= 80: print("\n✅ 大部分API测试通过") else: print("\n⚠️ 多个API测试失败，需要检查") print("=" * 60) async def main(): """主测试函数""" print("=" * 60) print("🚀 开始API端点测试") print(f"时间: {datetime.now():%Y-%m-%d %H:%M:%S}") print("=" * 60) # 检查服务是否运行 print("\n⏳ 检查服务状态...") test_suite = APITestSuite() await test_suite.setup() try: # 先测试健康检查 await test_suite.test_health_check() # 如果健康检查通过，继续其他测试 if test_suite.passed > 0: await test_suite.test_import_endpoints() await test_suite.test_products_endpoints() await test_suite.test_procurement_endpoints() await test_suite.test_monitoring_endpoints() await test_suite.test_post_endpoints() else: print("\n⚠️ 服务未运行，请先启动后端服务:") print(" cd ordersys && make dev-backend") finally: # 打印总结 test_suite.print_summary() await test_suite.cleanup() if __name__ == "__main__": asyncio.run(main())