测试策略设计：从项目需求到测试计划的完整流程_千锋教育

import requests
import threading
import time
from concurrent.futures import ThreadPoolExecutor

# 测试配置
BASE_URL = "http://test-api.example.com"
PRODUCT_ID = "seckill_001"
INITIAL_STOCK = 10
USER_COUNT = 100

# 测试结果统计
success_count = 0
fail_count = 0
lock = threading.Lock()

def seckill_request(user_id):
    """模拟用户发起秒杀请求"""
    global success_count, fail_count
    
    payload = {
        "user_id": user_id,
        "product_id": PRODUCT_ID,
        "timestamp": int(time.time() * 1000)
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/seckill",
            json=payload,
            timeout=5
        )
        
        if response.status_code == 200:
            data = response.json()
            if data.get("code") == 0:
                with lock:
                    success_count += 1
            else:
                with lock:
                    fail_count += 1
        else:
            with lock:
                fail_count += 1
    except Exception as e:
        with lock:
            fail_count += 1

def run_concurrent_seckill():
    """执行并发秒杀测试"""
    print(f"开始并发测试：{USER_COUNT}个用户，库存{INITIAL_STOCK}件")
    
    start_time = time.time()
    
    # 使用线程池模拟并发
    with ThreadPoolExecutor(max_workers=50) as executor:
        user_ids = [f"user_{i}" for i in range(USER_COUNT)]
        executor.map(seckill_request, user_ids)
    
    end_time = time.time()
    duration = end_time - start_time
    
    # 输出测试结果
    print(f"测试完成，耗时：{duration:.2f}秒")
    print(f"成功下单：{success_count}人")
    print(f"失败下单：{fail_count}人")
    print(f"库存应有：{INITIAL_STOCK}件")
    print(f"库存实际：{INITIAL_STOCK - success_count}件")
    
    # 断言验证
    assert success_count == INITIAL_STOCK, f"库存超卖！成功{success_count}人，库存{INITIAL_STOCK}件"
    assert fail_count == USER_COUNT - INITIAL_STOCK, f"失败人数异常！失败{fail_count}人，预期{USER_COUNT - INITIAL_STOCK}人"
    
    print("测试通过：库存扣减逻辑正确，无超卖现象")

if __name__ == "__main__":
    run_concurrent_seckill()

# 测试重复请求
def test_duplicate_request():
    """同一用户重复提交，应返回已购买"""
    payload = {
        "user_id": "user_001",
        "product_id": PRODUCT_ID,
        "timestamp": int(time.time() * 1000)
    }
    
    # 第一次请求
    response1 = requests.post(f"{BASE_URL}/api/seckill", json=payload)
    assert response1.json().get("code") == 0, "第一次请求失败"
    
    # 第二次请求
    response2 = requests.post(f"{BASE_URL}/api/seckill", json=payload)
    assert response2.json().get("code") == 1, "重复请求未拦截"
    assert response2.json().get("message") == "已购买过该商品"
    
    print("重复请求测试通过")

# 测试库存耗尽
def test_stock_exhausted():
    """库存为0时，所有请求应返回售罄"""
    payload = {
        "user_id": "user_999",
        "product_id": PRODUCT_ID,
        "timestamp": int(time.time() * 1000)
    }
    
    # 先消耗完库存
    for i in range(INITIAL_STOCK):
        requests.post(f"{BASE_URL}/api/seckill", json={
            "user_id": f"user_{i+1000}",
            "product_id": PRODUCT_ID,
            "timestamp": int(time.time() * 1000)
        })
    
    # 库存耗尽的请求
    response = requests.post(f"{BASE_URL}/api/seckill", json=payload)
    assert response.json().get("code") == 2, "库存耗尽未正确处理"
    assert response.json().get("message") == "商品已售罄"
    
    print("库存耗尽测试通过")