release/test_admin_user_query_optimization.py

#!/usr/bin/env python3
"""
管理员查询逻辑优化测试脚本
测试优化前后的性能差异和功能正确性
"""

import requests
import time
import json
from typing import Dict, List

class AdminUserQueryOptimizationTester:
    def __init__(self, base_url: str = "http://localhost:8081"):
        self.base_url = base_url
        self.headers = {
            'Content-Type': 'application/json'
        }
    
    def test_optimized_query(self, org_id: str, customer_id: int) -> Dict:
        """测试优化后的查询接口"""
        print(f"🔍 测试优化后的查询逻辑: orgId={org_id}, customerId={customer_id}")
        
        start_time = time.time()
        
        try:
            # 调用优化后的接口
            url = f"{self.base_url}/sys_user/get_users_by_org_id"
            params = {
                'orgId': org_id,
                'customerId': customer_id
            }
            
            response = requests.get(url, params=params, headers=self.headers, timeout=30)
            end_time = time.time()
            
            response_time = (end_time - start_time) * 1000  # 转换为毫秒
            
            if response.status_code == 200:
                data = response.json()
                user_count = len(data.get('data', {}).get('userMap', {})) if data.get('data') else 0
                
                result = {
                    'success': True,
                    'response_time_ms': round(response_time, 2),
                    'user_count': user_count,
                    'status_code': response.status_code,
                    'data': data
                }
                
                print(f"✅ 查询成功: {user_count} 个用户，耗时: {response_time:.2f}ms")
                return result
            else:
                print(f"❌ 查询失败: HTTP {response.status_code}")
                return {
                    'success': False,
                    'response_time_ms': round(response_time, 2),
                    'status_code': response.status_code,
                    'error': response.text
                }
                
        except requests.exceptions.RequestException as e:
            end_time = time.time()
            response_time = (end_time - start_time) * 1000
            print(f"❌ 网络错误: {e}")
            return {
                'success': False,
                'response_time_ms': round(response_time, 2),
                'error': str(e)
            }
    
    def analyze_user_types(self, users_data: Dict) -> Dict:
        """分析用户类型分布"""
        if not users_data or not users_data.get('data'):
            return {'普通用户': 0, '管理员': 0, '总计': 0}
        
        # 这里只是演示分析逻辑，实际需要根据API返回格式调整
        user_map = users_data['data'].get('userMap', {})
        total_users = len(user_map)
        
        return {
            '普通用户': total_users,  # 优化后应该只返回普通用户
            '管理员': 0,  # 优化后应该为0
            '总计': total_users
        }
    
    def performance_benchmark(self, org_id: str, customer_id: int, iterations: int = 5):
        """性能基准测试"""
        print(f"🚀 执行性能基准测试: {iterations} 次查询")
        
        response_times = []
        success_count = 0
        
        for i in range(iterations):
            print(f"  第 {i+1}/{iterations} 次查询...")
            result = self.test_optimized_query(org_id, customer_id)
            
            if result['success']:
                response_times.append(result['response_time_ms'])
                success_count += 1
            
            time.sleep(0.5)  # 避免请求过于频繁
        
        if response_times:
            avg_time = sum(response_times) / len(response_times)
            min_time = min(response_times)
            max_time = max(response_times)
            
            print(f"📊 性能统计结果:")
            print(f"  ✅ 成功查询: {success_count}/{iterations}")
            print(f"  ⏱️  平均响应时间: {avg_time:.2f}ms")
            print(f"  ⚡ 最快响应时间: {min_time:.2f}ms")
            print(f"  🐌 最慢响应时间: {max_time:.2f}ms")
            
            return {
                'success_rate': success_count / iterations,
                'avg_response_time_ms': avg_time,
                'min_response_time_ms': min_time,
                'max_response_time_ms': max_time,
                'response_times': response_times
            }
        else:
            print("❌ 所有查询都失败了")
            return None
    
    def validate_optimization(self, org_id: str, customer_id: int):
        """验证优化效果"""
        print(f"🎯 开始验证管理员查询优化效果")
        print(f"📋 测试参数: orgId={org_id}, customerId={customer_id}")
        print("=" * 60)
        
        # 1. 功能测试
        print("1️⃣ 功能正确性测试")
        result = self.test_optimized_query(org_id, customer_id)
        
        if result['success']:
            user_analysis = self.analyze_user_types(result['data'])
            print(f"   👥 用户分析: {user_analysis}")
            
            if user_analysis['管理员'] == 0:
                print("   ✅ 管理员过滤功能正常")
            else:
                print("   ⚠️ 管理员过滤可能有问题")
        
        # 2. 性能测试
        print("\n2️⃣ 性能基准测试")
        performance_result = self.performance_benchmark(org_id, customer_id, 3)
        
        if performance_result and performance_result['avg_response_time_ms'] < 1000:
            print("   ✅ 性能表现良好 (< 1秒)")
        elif performance_result:
            print(f"   ⚠️ 性能需要关注 ({performance_result['avg_response_time_ms']:.2f}ms)")
        
        # 3. 优化效果评估
        print("\n3️⃣ 优化效果评估")
        if result['success'] and performance_result:
            print(f"   🎯 查询成功率: {performance_result['success_rate']*100:.1f}%")
            
            # 评估性能等级
            avg_time = performance_result['avg_response_time_ms']
            if avg_time < 100:
                grade = "🌟 优秀"
            elif avg_time < 300:
                grade = "✅ 良好"
            elif avg_time < 1000:
                grade = "⚠️ 一般"
            else:
                grade = "❌ 需要优化"
            
            print(f"   🏆 性能等级: {grade} ({avg_time:.2f}ms)")
            
        print("\n" + "=" * 60)
        print("🎉 管理员查询优化验证完成！")

def main():
    """主测试函数"""
    print("🔧 管理员查询逻辑优化测试")
    print("=" * 60)
    
    tester = AdminUserQueryOptimizationTester()
    
    # 测试参数 - 根据实际情况调整
    test_org_id = "1939964806110937090"
    test_customer_id = 0
    
    try:
        tester.validate_optimization(test_org_id, test_customer_id)
    except Exception as e:
        print(f"❌ 测试执行出错: {e}")
        print("💡 请确保:")
        print("   1. ljwx-boot 服务正在运行")
        print("   2. 端口 8081 可访问")
        print("   3. 测试的 orgId 和 customerId 存在于数据库中")

if __name__ == "__main__":
    main()