WebDemo/modules/performance/performance_tester.py

import time
import requests
from locust import HttpUser, task, between
from core.config.config_manager import ConfigManager
from core.utils.logger import Logger


class PerformanceTester:
    def __init__(self):
        self.config = ConfigManager()
        self.logger = Logger.get_logger()
        self.base_url = self.config.get('api.base_url', 'https://example.com')

    def run_tests(self):
        """运行性能测试"""
        self.logger.info("开始执行性能测试")

        test_results = []

        # 响应时间测试
        test_results.extend(self.run_response_time_tests())

        # 负载测试 (使用Locust)
        test_results.extend(self.run_load_tests())

        # 压力测试
        test_results.extend(self.run_stress_tests())

        self.logger.info(f"性能测试完成，共执行 {len(test_results)} 个测试用例")
        return test_results

    def run_response_time_tests(self):
        """运行响应时间测试"""
        self.logger.info("执行响应时间测试")
        results = []

        # 测试端点
        test_endpoints = [
            "/",
            "/api/users",
            "/api/products",
            "/api/orders"
        ]

        for endpoint in test_endpoints:
            url = urljoin(self.base_url, endpoint)
            test_name = f"响应时间测试 - {endpoint}"

            try:
                # 多次请求取平均值
                total_time = 0
                num_requests = 10

                for i in range(num_requests):
                    start_time = time.time()
                    response = requests.get(url, timeout=30)
                    end_time = time.time()

                    if response.status_code == 200:
                        total_time += (end_time - start_time)
                    else:
                        result = {
                            'name': test_name,
                            'status': 'FAIL',
                            'message': f"请求失败，状态码: {response.status_code}",
                            'response_time': 0,
                            'response_code': response.status_code
                        }
                        results.append(result)
                        break

                else:  # 所有请求都成功
                    avg_time = total_time / num_requests

                    # 根据响应时间判断性能
                    if avg_time < 1.0:
                        status = 'PASS'
                        message = f"响应时间良好: {avg_time:.3f}秒"
                    elif avg_time < 3.0:
                        status = 'WARN'
                        message = f"响应时间一般: {avg_time:.3f}秒"
                    else:
                        status = 'FAIL'
                        message = f"响应时间较差: {avg_time:.3f}秒"

                    result = {
                        'name': test_name,
                        'status': status,
                        'message': message,
                        'response_time': avg_time,
                        'response_code': 200
                    }
                    results.append(result)

            except Exception as e:
                result = {
                    'name': test_name,
                    'status': 'ERROR',
                    'message': f"测试执行出错: {str(e)}",
                    'response_time': 0
                }
                results.append(result)

        return results

    def run_load_tests(self):
        """运行负载测试"""
        self.logger.info("执行负载测试")
        results = []

        # 这里使用Locust进行负载测试
        # 在实际应用中，可以启动Locust并运行测试，然后解析结果

        # 模拟测试结果
        test_cases = [
            {
                'name': '负载测试 - 50用户',
                'status': 'PASS',
                'message': '平均响应时间: 1.2秒, 失败率: 0%',
                'avg_response_time': 1.2,
                'fail_rate': 0.0
            },
            {
                'name': '负载测试 - 100用户',
                'status': 'PASS',
                'message': '平均响应时间: 1.8秒, 失败率: 0%',
                'avg_response_time': 1.8,
                'fail_rate': 0.0
            },
            {
                'name': '负载测试 - 200用户',
                'status': 'WARN',
                'message': '平均响应时间: 3.5秒, 失败率: 2%',
                'avg_response_time': 3.5,
                'fail_rate': 2.0
            }
        ]

        results.extend(test_cases)
        return results

    def run_stress_tests(self):
        """运行压力测试"""
        self.logger.info("执行压力测试")
        results = []

        # 模拟测试结果
        test_cases = [
            {
                'name': '压力测试 - 持续高负载',
                'status': 'PASS',
                'message': '系统在持续高负载下稳定运行',
                'max_response_time': 4.2,
                'error_rate': 0.5
            },
            {
                'name': '压力测试 - 峰值负载',
                'status': 'WARN',
                'message': '系统在峰值负载下出现部分错误',
                'max_response_time': 8.7,
                'error_rate': 5.2
            }
        ]

        results.extend(test_cases)
        return results


# Locust测试类
class WebsiteUser(HttpUser):
    wait_time = between(1, 5)

    def on_start(self):
        """在用户启动时执行"""
        pass

    @task(2)
    def view_products(self):
        """查看产品列表"""
        self.client.get("/api/products")

    @task(1)
    def view_product_detail(self):
        """查看产品详情"""
        product_id = 1  # 实际应用中可以从列表获取
        self.client.get(f"/api/products/{product_id}")

    @task(1)
    def create_order(self):
        """创建订单"""
        order_data = {
            "product_id": 1,
            "quantity": 2,
            "customer_name": "Test Customer",
            "customer_email": "test@example.com"
        }
        self.client.post("/api/orders", json=order_data)