buff3.0/backend-client/test_selection_api.py

# -*- coding: utf-8 -*-
"""
测试选品接口 - 验证 selection_condition_json 字段是否正确保存
"""
import json
import hashlib
import base64
import time
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
import requests

BASE_URL = "http://localhost:9000"
TIMEOUT = 30

# AES加密用的key（需与后端协商一致，这里用测试key）
AES_KEY = b"1234567890123456"  # 16字节

def md5(s):
    return hashlib.md5(s.encode('utf-8')).hexdigest()

def aes_encrypt(plaintext: str, key: bytes) -> str:
    """AES/CBC/PKCS7 加密，返回 base64 编码"""
    cipher = AES.new(key, AES.MODE_CBC, iv=b'\x00' * 16)
    padded = pad(plaintext.encode('utf-8'), AES.block_size)
    return base64.b64encode(cipher.encrypt(padded)).decode('ascii')

def encrypt_params(params: dict) -> dict:
    """加密参数：params -> {encrypted_params, aes_key, key_version}"""
    # AES加密后得到encrypted_params
    # key_version默认1
    json_str = json.dumps(params, ensure_ascii=False, separators=(',', ':'))
    encrypted = aes_encrypt(json_str, AES_KEY)
    return {
        "encrypted_params": encrypted,
        "aes_key": base64.b64encode(AES_KEY).decode('ascii'),
        "key_version": 1
    }

def raw_machine_info_to_machine_code(raw: str) -> str:
    return md5(raw)

def register_and_login():
    """注册并登录，返回 session_id 和 user_id"""
    raw_machine = "TEST_MACHINE_CODE_12345"
    machine_code = raw_machine_info_to_machine_code(raw_machine)

    # 注册
    reg_params = {
        "username": f"test_user_{int(time.time())}",
        "password": "Test123456",
        "card_code": "TEST_CARD_001",  # 测试卡密
        "software_code": "buff-xp-steam-sh",
        "raw_machine_info": raw_machine
    }
    enc_reg = encrypt_params(reg_params)
    try:
        r = requests.post(f"{BASE_URL}/api/v1/auth/register", json=enc_reg, timeout=TIMEOUT)
        print(f"注册响应: {r.status_code} {r.text[:200]}")
    except Exception as e:
        print(f"注册异常: {e}")

    # 登录
    login_params = {
        "username": reg_params["username"],
        "password": reg_params["password"],
        "raw_machine_info": raw_machine
    }
    enc_login = encrypt_params(login_params)
    try:
        r = requests.post(f"{BASE_URL}/api/v1/auth/login", json=enc_login, timeout=TIMEOUT)
        print(f"登录响应: {r.status_code} {r.text[:500]}")
        resp = r.json()
        if resp.get("code") == 200 and resp.get("data"):
            data = resp["data"]
            # 解密data
            encrypted_data = data.get("encrypted_data") or ""
            aes_key_b64 = data.get("aes_key") or ""
            if encrypted_data and aes_key_b64:
                aes_key = base64.b64decode(aes_key_b64)
                encrypted_bytes = base64.b64decode(encrypted_data)
                cipher = AES.new(aes_key, AES.MODE_CBC, iv=b'\x00' * 16)
                decrypted = cipher.decrypt(encrypted_bytes)
                decrypted = decrypted.rstrip(b'\x00').rstrip(b'\x10')
                login_data = json.loads(decrypted.decode('utf-8'))
                return login_data.get("session_id"), login_data.get("user_id"), machine_code
    except Exception as e:
        print(f"登录异常: {e}")
        return None, None, None

def test_selection_result(session_id, machine_code, user_id):
    """查询选品结果"""
    body = {
        "session_id": session_id,
        "machine_code": machine_code,
    }
    try:
        r = requests.post(f"{BASE_URL}/api/v1/buff/selection-result", json=body, timeout=TIMEOUT)
        print(f"\n选品结果查询响应: {r.status_code} {r.text[:800]}")
        resp = r.json()
        if resp.get("code") == 200 and resp.get("data"):
            data = resp["data"]
            req_id = data.get("request_id")
            status = data.get("status")
            items_count = len(data.get("items") or [])
            print(f"\n选品状态: status={status}, items_count={items_count}, request_id={req_id}")
    except Exception as e:
        print(f"选品结果查询异常: {e}")

def test_selection_config(session_id, machine_code):
    """测试获取选品配置接口"""
    body = {
        "session_id": session_id,
        "machine_code": machine_code,
    }
    try:
        r = requests.post(f"{BASE_URL}/api/v1/buff/secure/selection-config", json=body, timeout=TIMEOUT)
        print(f"\n选品配置查询响应: {r.status_code}")
        resp = r.json()
        if resp.get("code") == 200 and resp.get("data"):
            configs = resp["data"].get("configs") or []
            print(f"配置数量: {len(configs)}")
            if configs:
                print(f"默认配置: {json.dumps(configs[0], ensure_ascii=False, indent=2)[:1000]}")
        else:
            print(f"响应内容: {r.text[:500]}")
    except Exception as e:
        print(f"选品配置查询异常: {e}")

if __name__ == "__main__":
    print("=" * 60)
    print("开始测试选品接口")
    print("=" * 60)

    # 测试获取选品配置（不需要登录）
    print("\n[测试1] 公开选品配置接口")
    try:
        r = requests.get(f"{BASE_URL}/api/v1/buff/selection-config", timeout=TIMEOUT)
        print(f"响应: {r.status_code} {r.text[:300]}")
    except Exception as e:
        print(f"异常: {e}")

    # 测试公开的筛选选项接口
    print("\n[测试2] 筛选选项接口")
    try:
        r = requests.get(f"{BASE_URL}/api/v1/buff/filter-options", timeout=TIMEOUT)
        print(f"响应: {r.status_code} {r.text[:300]}")
    except Exception as e:
        print(f"异常: {e}")

    # 注册并登录测试
    print("\n[测试3] 注册并登录")
    session_id, user_id, machine_code = register_and_login()
    if session_id:
        print(f"登录成功: user_id={user_id}, session_id={session_id[:20]}...")
    else:
        print("登录失败，跳过后续测试")
        exit(1)

    # 测试选品配置查询
    test_selection_config(session_id, machine_code)

    # 查询选品结果
    test_selection_result(session_id, machine_code, user_id)

    print("\n" + "=" * 60)
    print("测试完成")
    print("=" * 60)