# 使用 Locust 进行性能测试
from locust import HttpUser, task, between

class EcommerceUser(HttpUser):
    wait_time = between(1, 3)
    
    @task(3)  # 权重 3
    def browse_products(self):
        """浏览商品"""
        self.client.get("/products")
    
    @task(2)  # 权重 2
    def search_products(self):
        """搜索商品"""
        self.client.get("/products/search?q=iPhone")
    
    @task(1)  # 权重 1
    def add_to_cart(self):
        """加入购物车"""
        self.client.post("/cart/add", json={
            "product_id": 123,
            "quantity": 1
        })

# 运行：locust -f test_performance.py --users 1000 --spawn-rate 10
# 结果：
# - 浏览商品：平均响应时间 200ms ✅
# - 搜索商品：平均响应时间 500ms ✅
# - 加入购物车：平均响应时间 2000ms ❌ 瓶颈！

# 优化前：每次都查询数据库
def add_to_cart(user_id, product_id, quantity):
    product = db.query(f"SELECT * FROM products WHERE id = {product_id}")
    cart = db.query(f"SELECT * FROM carts WHERE user_id = {user_id}")
    # ...

# 优化后：使用 Redis 缓存
def add_to_cart(user_id, product_id, quantity):
    # 从缓存获取商品信息
    product = redis.get(f"product:{product_id}")
    if not product:
        product = db.query(f"SELECT * FROM products WHERE id = {product_id}")
        redis.set(f"product:{product_id}", product, ex=3600)
    
    # 购物车直接写入 Redis
    redis.sadd(f"cart:{user_id}", product_id)
    
    # 异步同步到数据库
    celery.send_task("sync_cart_to_db", args=[user_id, product_id])

# 优化后响应时间：200ms ✅

# 使用 JMeter 进行容量测试
# 测试计划：
# 1. 线程组：1000 个用户
# 2. 递增时间：10 秒（每秒增加 100 个用户）
# 3. 持续时间：60 秒

# 结果分析：
# - 0-500 用户：响应时间 < 1秒 ✅
# - 500-800 用户：响应时间 1-3秒 ⚠️
# - 800-1000 用户：响应时间 > 5秒 ❌
# - 结论：系统容量上限为 800 并发用户

# 前端性能测试
from playwright.sync_api import sync_playwright

def test_page_load_time():
    with sync_playwright() as p:
        browser = p.chromium.launch()
        page = browser.new_page()
        
        # 测试页面加载时间
        start_time = time.time()
        page.goto("https://example.com")
        page.wait_for_load_state("networkidle")
        end_time = time.time()
        
        load_time = end_time - start_time
        
        # 验证加载时间 < 3秒
        assert load_time < 3, f"页面加载时间过长：{load_time}秒"
        
        # 测试首屏渲染时间
        fcp = page.evaluate("""
            () => {
                const paint = performance.getEntriesByType('paint');
                const fcp = paint.find(entry => entry.name === 'first-contentful-paint');
                return fcp ? fcp.startTime : 0;
            }
        """)
        
        # 验证首屏渲染时间 < 1秒
        assert fcp < 1000, f"首屏渲染时间过长：{fcp}ms"
        
        browser.close()

# 测试系统在 1000 个并发用户下的性能
locust -f test_load.py --users 1000 --spawn-rate 10 --run-time 10m

from locust import HttpUser, task, between
import random

class EcommerceUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        """用户登录"""
        response = self.client.post("/login", json={
            "email": "test@example.com",
            "password": "password123"
        })
        self.token = response.json()["token"]

    @task(5)  # 权重 5
    def browse_products(self):
        """浏览商品"""
        self.client.get("/products", headers={
            "Authorization": f"Bearer {self.token}"
        })

    @task(3)  # 权重 3
    def search_products(self):
        """搜索商品"""
        keywords = ["iPhone", "iPad", "MacBook", "AirPods"]
        keyword = random.choice(keywords)
        self.client.get(f"/products/search?q={keyword}", headers={
            "Authorization": f"Bearer {self.token}"
        })

    @task(2)  # 权重 2
    def view_product(self):
        """查看商品详情"""
        product_id = random.randint(1, 1000)
        self.client.get(f"/products/{product_id}", headers={
            "Authorization": f"Bearer {self.token}"
        })

    @task(1)  # 权重 1
    def add_to_cart(self):
        """加入购物车"""
        product_id = random.randint(1, 1000)
        self.client.post("/cart/add", json={
            "product_id": product_id,
            "quantity": 1
        }, headers={
            "Authorization": f"Bearer {self.token}"
        })

# 运行：
# locust -f locustfile.py --users 1000 --spawn-rate 10 --run-time 10m --host https://api.example.com

from locust import HttpUser, task, between, SequentialTaskSet

class UserBehavior(SequentialTaskSet):
    """模拟真实用户行为"""

    @task
    def login(self):
        """1. 登录"""
        self.client.post("/login", json={
            "email": "test@example.com",
            "password": "password123"
        })

    @task
    def browse_homepage(self):
        """2. 浏览首页"""
        self.client.get("/")

    @task
    def search_products(self):
        """3. 搜索商品"""
        self.client.get("/products/search?q=iPhone")

    @task
    def view_product(self):
        """4. 查看商品详情"""
        self.client.get("/products/123")

    @task
    def add_to_cart(self):
        """5. 加入购物车"""
        self.client.post("/cart/add", json={
            "product_id": 123,
            "quantity": 1
        })

    @task
    def checkout(self):
        """6. 结算"""
        self.client.post("/orders", json={
            "address": "北京市朝阳区xxx",
            "phone": "13800138000"
        })

    @task
    def logout(self):
        """7. 登出"""
        self.client.post("/logout")

class EcommerceUser(HttpUser):
    tasks = [UserBehavior]
    wait_time = between(1, 3)

# 阶段 1：冒烟测试（5分钟）
# 目标：验证系统基本功能
locust -f locustfile.py --users 10 --spawn-rate 1 --run-time 5m

# 阶段 2：负载测试（30分钟）
# 目标：验证系统在预期负载下的性能
locust -f locustfile.py --users 1000 --spawn-rate 10 --run-time 30m

# 阶段 3：压力测试（30分钟）
# 目标：找到系统的性能上限
locust -f locustfile.py --users 5000 --spawn-rate 50 --run-time 30m

# 阶段 4：耐力测试（24小时）
# 目标：验证系统长时间运行的稳定性
locust -f locustfile.py --users 1000 --spawn-rate 10 --run-time 24h

# 使用 Prometheus + Grafana 监控
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from flask import Flask, Response

app = Flask(__name__)

# 定义指标
request_count = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
response_time = Histogram('http_response_time_seconds', 'HTTP response time', ['method', 'endpoint'])
active_users = Gauge('active_users', 'Number of active users')
db_connections = Gauge('db_connections', 'Number of database connections')
cache_hit_rate = Gauge('cache_hit_rate', 'Cache hit rate')

@app.route('/metrics')
def metrics():
    """Prometheus 指标端点"""
    return Response(generate_latest(), mimetype='text/plain')

@app.route('/products')
def get_products():
    # 记录请求
    request_count.labels(method='GET', endpoint='/products', status=200).inc()

    # 记录响应时间
    with response_time.labels(method='GET', endpoint='/products').time():
        products = db.query("SELECT * FROM products")

    # 更新活跃用户数
    active_users.set(get_active_user_count())

    # 更新数据库连接数
    db_connections.set(get_db_connection_count())

    # 更新缓存命中率
    cache_hit_rate.set(get_cache_hit_rate())

    return jsonify(products)

# 测试系统在 10000 个并发用户下的性能（超出预期）
locust -f test_stress.py --users 10000 --spawn-rate 100 --run-time 10m

# 测试系统在 1000 个并发用户下运行 24 小时
locust -f test_endurance.py --users 1000 --spawn-rate 10 --run-time 24h

# 测试系统在突然增加 5000 个用户时的性能
locust -f test_spike.py --users 5000 --spawn-rate 500 --run-time 5m

# 测试系统在 1000万 条数据下的性能
def test_large_dataset():
    # 插入 1000万 条数据
    for i in range(10000000):
        db.insert("INSERT INTO products ...")
    
    # 测试查询性能
    start_time = time.time()
    result = db.query("SELECT * FROM products WHERE price > 1000")
    end_time = time.time()
    
    # 验证查询时间 < 1秒
    assert end_time - start_time < 1

# 使用 Faker 生成测试数据
from faker import Faker
import random

fake = Faker('zh_CN')

def generate_test_data(count=1000000):
    """生成 100万 条测试数据"""
    users = []
    for i in range(count):
        users.append({
            "id": i,
            "name": fake.name(),
            "email": fake.email(),
            "phone": fake.phone_number(),
            "address": fake.address(),
            "created_at": fake.date_time_this_year()
        })
    
    # 批量插入数据库
    db.bulk_insert("users", users, batch_size=10000)

# 生成测试数据
generate_test_data(1000000)

# 使用 Grafana + Prometheus 监控
# 1. 安装 Prometheus
docker run -d -p 9090:9090 prom/prometheus

# 2. 安装 Grafana
docker run -d -p 3000:3000 grafana/grafana

# 3. 配置监控指标
from prometheus_client import Counter, Histogram, Gauge
import time

# 请求计数器
request_count = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint'])

# 响应时间直方图
response_time = Histogram('http_response_time_seconds', 'HTTP response time', ['method', 'endpoint'])

# CPU 使用率
cpu_usage = Gauge('cpu_usage_percent', 'CPU usage percentage')

# 使用
@app.route('/products')
def get_products():
    request_count.labels(method='GET', endpoint='/products').inc()
    
    with response_time.labels(method='GET', endpoint='/products').time():
        products = db.query("SELECT * FROM products")
        return jsonify(products)

# 4. 在 Grafana 中创建仪表板
# - 请求数趋势图
# - 响应时间分布图
# - CPU/内存使用率图
# - 错误率图