#!/usr/bin/env python3 """ 综合系统测试 全面测试订单处理系统的所有关键功能 """ import asyncio import sys import os import json from pathlib import Path from datetime import datetime, timedelta import random sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'backend')) os.environ['DATABASE_URL'] = 'sqlite+aiosqlite:///./backend/ordersys.db' from app.core.database import AsyncSessionLocal from app.services.import_service import ImportService from app.services.products_master_service import ProductsMasterService from app.services.procurement_order_service_v2 import ProcurementOrderServiceV2 from app.services.procurement_order_sync_service import ProcurementOrderSyncService from app.models.raw_orders import RawOrder, ProcessedFile from app.models.products_master import ProductMaster from app.models.procurement_orders import ProcurementOrder from sqlalchemy import select, func, and_, or_ class SystemTestSuite: """系统综合测试套件""" def __init__(self): self.test_results = [] self.passed = 0 self.failed = 0 def add_result(self, test_name: str, passed: bool, message: str = None): """添加测试结果""" self.test_results.append({ 'test': test_name, 'passed': passed, 'message': message }) if passed: self.passed += 1 else: self.failed += 1 async def test_database_connection(self, db): """测试1: 数据库连接""" print("\n📋 测试1: 数据库连接") try: # 测试基本查询 result = await db.execute(select(func.count(RawOrder.id))) count = result.scalar() print(f" ✅ 数据库连接正常，原始订单数: {count}") self.add_result("数据库连接", True, f"订单数: {count}") return True except Exception as e: print(f" ❌ 数据库连接失败: {e}") self.add_result("数据库连接", False, str(e)) return False async def test_import_service(self, db): """测试2: 导入服务""" print("\n📋 测试2: 导入服务") import_service = ImportService() # 测试文件扫描 try: files = await import_service.scan_files() if files: print(f" ✅ 扫描到 {len(files)} 个文件") self.add_result("文件扫描", True, f"找到 {len(files)} 个文件") else: print(f" ⚠️ 没有扫描到文件") self.add_result("文件扫描", True, "无文件") # 测试处理状态检查 if files: file_path = files[0] processed = await import_service.check_file_processed(db, file_path) if processed: print(f" ✅ 文件已处理检查正常") self.add_result("文件处理检查", True) else: print(f" ℹ️ 文件未处理") self.add_result("文件处理检查", True, "文件未处理") return True except Exception as e: print(f" ❌ 导入服务测试失败: {e}") self.add_result("导入服务", False, str(e)) return False async def test_update_mode(self, db): """测试3: Update模式功能""" print("\n📋 测试3: Update模式更新功能") import_service = ImportService() try: # 获取一个测试订单 result = await db.execute(select(RawOrder).limit(1)) test_order = result.scalar_one_or_none() if not test_order: print(" ⚠️ 没有订单可供测试") self.add_result("Update模式", False, "无测试数据") return False # 创建更新数据 original_state = test_order.交易状态 original_refund = test_order.退款状态 update_data = { '原始订单编号': test_order.原始订单编号, '线上宝贝名称': test_order.线上宝贝名称, '线上销售属性': test_order.线上销售属性, '数量': test_order.数量, '付款时间': test_order.付款时间, '交易状态': '测试状态' if original_state != '测试状态' else '已付款', '退款状态': '测试退款' if original_refund != '测试退款' else '正常', '卖家备注': f'测试备注 {datetime.now():%H:%M:%S}' } # 执行更新 updated = await import_service._update_existing_order(db, update_data) if updated: # 验证更新 await db.refresh(test_order) if test_order.卖家备注 and '测试备注' in test_order.卖家备注: print(f" ✅ Update模式正常工作") self.add_result("Update模式", True) # 恢复原始状态 restore_data = update_data.copy() restore_data['交易状态'] = original_state restore_data['退款状态'] = original_refund restore_data['卖家备注'] = None await import_service._update_existing_order(db, restore_data) return True else: print(f" ❌ 更新未生效") self.add_result("Update模式", False, "更新未生效") return False else: print(f" ❌ 更新失败") self.add_result("Update模式", False, "更新失败") return False except Exception as e: print(f" ❌ Update模式测试失败: {e}") self.add_result("Update模式", False, str(e)) return False async def test_products_master(self, db): """测试4: 产品主表生成""" print("\n📋 测试4: 产品主表生成") try: products_service = ProductsMasterService() # 获取当前产品数 result = await db.execute(select(func.count(ProductMaster.id))) initial_count = result.scalar() # 测试生成（增量模式） result = await products_service.generate_products_master( db, force_rebuild=False, batch_size=100 ) print(f" ✅ 产品主表生成正常") print(f" - 初始产品数: {initial_count}") print(f" - 创建: {result.get('created', 0)}") print(f" - 更新: {result.get('updated', 0)}") self.add_result("产品主表生成", True, f"创建{result.get('created', 0)}/更新{result.get('updated', 0)}") # 测试SKU生成 if initial_count > 0: result = await db.execute( select(ProductMaster).where(ProductMaster.sku_key.is_not(None)).limit(1) ) product = result.scalar_one_or_none() if product and product.sku_key: print(f" ✅ SKU生成正常: {product.sku_key[:16]}...") self.add_result("SKU生成", True) else: print(f" ⚠️ 未找到带SKU的产品") self.add_result("SKU生成", False, "无SKU") return True except Exception as e: print(f" ❌ 产品主表测试失败: {e}") self.add_result("产品主表", False, str(e)) return False async def test_procurement_orders(self, db): """测试5: 采购订单生成""" print("\n📋 测试5: 采购订单生成") try: procurement_service = ProcurementOrderServiceV2() # 获取当前采购订单数 result = await db.execute(select(func.count(ProcurementOrder.id))) initial_count = result.scalar() # 测试生成 result = await procurement_service.generate_procurement_orders( db, force_rebuild=False, batch_size=100 ) print(f" ✅ 采购订单生成正常") print(f" - 初始订单数: {initial_count}") print(f" - 创建: {result.get('created_orders', 0)}") print(f" - 更新: {result.get('updated_orders', 0)}") self.add_result("采购订单生成", True, f"创建{result.get('created_orders', 0)}") # 测试采购方式分类 result = await db.execute( select(ProcurementOrder.procurement_method, func.count(ProcurementOrder.id)) .group_by(ProcurementOrder.procurement_method) ) methods = result.all() if methods: print(" ✅ 采购方式分类:") for method, count in methods: print(f" - {method}: {count}") self.add_result("采购方式分类", True, f"{len(methods)}种方式") return True except Exception as e: print(f" ❌ 采购订单测试失败: {e}") self.add_result("采购订单", False, str(e)) return False async def test_status_sync(self, db): """测试6: 状态同步功能""" print("\n📋 测试6: 状态同步功能") try: sync_service = ProcurementOrderSyncService() # 获取同步统计 stats = await sync_service.get_sync_statistics(db) print(f" ✅ 状态同步统计:") print(f" - 总采购订单: {stats.get('total_orders', 0)}") print(f" - 交易状态不一致: {stats.get('trade_status_mismatch', 0)}") print(f" - 退款状态不一致: {stats.get('refund_status_mismatch', 0)}") print(f" - 卖家备注不一致: {stats.get('seller_note_mismatch', 0)}") # 执行同步 if stats.get('total_orders', 0) > 0: sync_result = await sync_service.sync_order_status(db, batch_size=100) print(f" ✅ 同步执行成功:") print(f" - 同步成功: {sync_result.get('synced_orders', 0)}") print(f" - 同步失败: {sync_result.get('failed_orders', 0)}") self.add_result("状态同步", True, f"同步{sync_result.get('synced_orders', 0)}个订单") else: print(" ⚠️ 没有采购订单可同步") self.add_result("状态同步", True, "无订单需同步") return True except Exception as e: print(f" ❌ 状态同步测试失败: {e}") self.add_result("状态同步", False, str(e)) return False async def test_data_integrity(self, db): """测试7: 数据完整性检查""" print("\n📋 测试7: 数据完整性检查") try: # 检查原始订单去重 result = await db.execute( select( RawOrder.原始订单编号, RawOrder.线上宝贝名称, RawOrder.线上销售属性, RawOrder.数量, RawOrder.付款时间, func.count(RawOrder.id) ).group_by( RawOrder.原始订单编号, RawOrder.线上宝贝名称, RawOrder.线上销售属性, RawOrder.数量, RawOrder.付款时间 ).having(func.count(RawOrder.id) > 1) ) duplicates = result.all() if duplicates: print(f" ⚠️ 发现 {len(duplicates)} 组重复订单") self.add_result("订单去重", False, f"{len(duplicates)}组重复") else: print(f" ✅ 订单去重正常") self.add_result("订单去重", True) # 检查采购订单关联 result = await db.execute( select(func.count(ProcurementOrder.id)) .where(ProcurementOrder.original_order_id.not_in( select(RawOrder.id) )) ) orphaned = result.scalar() if orphaned > 0: print(f" ⚠️ 发现 {orphaned} 个孤立采购订单") self.add_result("数据关联", False, f"{orphaned}个孤立订单") else: print(f" ✅ 数据关联正常") self.add_result("数据关联", True) # 检查产品主表数据质量 result = await db.execute( select(func.count(ProductMaster.id)) .where(and_( ProductMaster.品牌.is_not(None), ProductMaster.货号.is_not(None), ProductMaster.sku_key.is_not(None) )) ) complete_products = result.scalar() result = await db.execute(select(func.count(ProductMaster.id))) total_products = result.scalar() if total_products > 0: completeness = (complete_products / total_products) * 100 print(f" ✅ 产品数据完整性: {completeness:.1f}%") self.add_result("产品完整性", True, f"{completeness:.1f}%") return True except Exception as e: print(f" ❌ 数据完整性检查失败: {e}") self.add_result("数据完整性", False, str(e)) return False async def test_performance(self, db): """测试8: 性能测试""" print("\n📋 测试8: 性能测试") try: import time # 测试查询性能 start = time.time() result = await db.execute( select(RawOrder) .where(RawOrder.交易状态 == '已付款') .limit(100) ) orders = result.scalars().all() query_time = (time.time() - start) * 1000 print(f" ✅ 查询性能: {query_time:.2f}ms (100条记录)") if query_time < 500: self.add_result("查询性能", True, f"{query_time:.2f}ms") else: self.add_result("查询性能", False, f"{query_time:.2f}ms 过慢") # 测试批量更新性能 if orders: start = time.time() import_service = ImportService() for order in orders[:10]: update_data = { '原始订单编号': order.原始订单编号, '线上宝贝名称': order.线上宝贝名称, '线上销售属性': order.线上销售属性, '数量': order.数量, '付款时间': order.付款时间, '卖家备注': f'性能测试 {time.time()}' } await import_service._update_existing_order(db, update_data) update_time = (time.time() - start) * 1000 avg_time = update_time / 10 print(f" ✅ 更新性能: {avg_time:.2f}ms/条") if avg_time < 100: self.add_result("更新性能", True, f"{avg_time:.2f}ms/条") else: self.add_result("更新性能", False, f"{avg_time:.2f}ms/条 过慢") return True except Exception as e: print(f" ❌ 性能测试失败: {e}") self.add_result("性能测试", False, str(e)) return False async def test_edge_cases(self, db): """测试9: 边界情况测试""" print("\n📋 测试9: 边界情况测试") try: import_service = ImportService() # 测试空值处理 test_data = { '原始订单编号': 'TEST_EDGE_001', '线上宝贝名称': None, '线上销售属性': '', '数量': 0, '付款时间': datetime.now(), '交易状态': None, '退款状态': '', '卖家备注': None } # 尝试更新不存在的订单 result = await import_service._update_existing_order(db, test_data) if not result: print(f" ✅ 正确处理不存在的订单") self.add_result("边界情况-不存在订单", True) else: print(f" ❌ 未正确处理不存在的订单") self.add_result("边界情况-不存在订单", False) # 测试特殊字符处理 result = await db.execute( select(RawOrder).where( or_( RawOrder.线上宝贝名称.contains("'"), RawOrder.线上宝贝名称.contains('"'), RawOrder.线上宝贝名称.contains("\\") ) ).limit(1) ) special_order = result.scalar_one_or_none() if special_order: print(f" ✅ 系统支持特殊字符") self.add_result("特殊字符处理", True) else: print(f" ℹ️ 未找到包含特殊字符的订单") self.add_result("特殊字符处理", True, "无测试数据") return True except Exception as e: print(f" ❌ 边界情况测试失败: {e}") self.add_result("边界情况", False, str(e)) return False async def test_concurrent_operations(self, db): """测试10: 并发操作测试""" print("\n📋 测试10: 并发操作测试") try: import_service = ImportService() # 获取测试订单 result = await db.execute(select(RawOrder).limit(5)) test_orders = result.scalars().all() if len(test_orders) < 5: print(f" ⚠️ 测试数据不足") self.add_result("并发操作", False, "测试数据不足") return False # 创建并发更新任务 async def update_order(order, index): update_data = { '原始订单编号': order.原始订单编号, '线上宝贝名称': order.线上宝贝名称, '线上销售属性': order.线上销售属性, '数量': order.数量, '付款时间': order.付款时间, '卖家备注': f'并发测试 {index} - {datetime.now()}' } return await import_service._update_existing_order(db, update_data) # 并发执行 tasks = [update_order(order, i) for i, order in enumerate(test_orders)] results = await asyncio.gather(*tasks, return_exceptions=True) # 检查结果 success_count = sum(1 for r in results if r is True) error_count = sum(1 for r in results if isinstance(r, Exception)) print(f" ✅ 并发更新完成:") print(f" - 成功: {success_count}") print(f" - 失败: {error_count}") if error_count == 0: self.add_result("并发操作", True, f"成功{success_count}个") else: self.add_result("并发操作", False, f"失败{error_count}个") return error_count == 0 except Exception as e: print(f" ❌ 并发操作测试失败: {e}") self.add_result("并发操作", False, str(e)) return False def print_summary(self): """打印测试总结""" print("\n" + "=" * 60) print("📊 测试总结") print("=" * 60) for result in self.test_results: status = "✅" if result['passed'] else "❌" message = f" - {result['message']}" if result['message'] else "" print(f"{status} {result['test']}{message}") print("\n" + "-" * 60) total = self.passed + self.failed if total > 0: pass_rate = (self.passed / total) * 100 print(f"总计: {total} 个测试") print(f"通过: {self.passed} 个") print(f"失败: {self.failed} 个") print(f"通过率: {pass_rate:.1f}%") if pass_rate == 100: print("\n🎉 所有测试通过！系统运行正常！") elif pass_rate >= 80: print("\n✅ 大部分测试通过，系统基本正常") elif pass_rate >= 60: print("\n⚠️ 部分测试失败，需要关注") else: print("\n❌ 多数测试失败，系统存在问题") print("=" * 60) async def main(): """主测试函数""" print("=" * 60) print("🚀 开始综合系统测试") print(f"时间: {datetime.now():%Y-%m-%d %H:%M:%S}") print("=" * 60) test_suite = SystemTestSuite() async with AsyncSessionLocal() as db: # 运行所有测试 await test_suite.test_database_connection(db) await test_suite.test_import_service(db) await test_suite.test_update_mode(db) await test_suite.test_products_master(db) await test_suite.test_procurement_orders(db) await test_suite.test_status_sync(db) await test_suite.test_data_integrity(db) await test_suite.test_performance(db) await test_suite.test_edge_cases(db) await test_suite.test_concurrent_operations(db) # 打印总结 test_suite.print_summary() if __name__ == "__main__": asyncio.run(main())