区块链链上测试完整指南

📚 目录

概述
测试环境准备
入金测试
出金测试
多站点测试
对账测试
钱包币种测试
熔断测试
测试工具和脚本
常见问题排查
测试检查清单

概述

什么是链上测试？

链上测试（On-chain Testing） 是指对区块链交易系统进行全面的功能、性能和安全性测试，确保系统能够正确处理用户的充币（入金）、提币（出金）等操作，并保证资金安全和数据一致性。

测试工程师的核心职责

作为区块链测试工程师，你需要验证：

✅ 资金流转正确性：入金、出金流程是否正常
✅ 数据一致性：链上数据与系统数据是否一致
✅ 多站点同步：不同站点之间的数据是否同步
✅ 对账准确性：系统对账是否准确无误
✅ 币种支持：各种币种是否正常工作
✅ 异常处理：熔断机制是否有效

测试流程概览

测试准备 → 环境搭建 → 执行测试 → 验证结果 → 问题排查 → 报告输出

测试环境准备

1. 测试环境分类

1.1 开发环境（Dev）

用途：日常开发和调试
特点：数据可随时重置，测试币充足
使用场景：功能开发、单元测试

1.2 测试环境（Test/QA）

用途：功能测试和集成测试
特点：模拟生产环境，数据相对稳定
使用场景：功能测试、回归测试

1.3 预生产环境（Staging）

用途：生产前最后验证
特点：与生产环境配置一致
使用场景：上线前验证、压力测试

1.4 生产环境（Production）

用途：真实用户使用
特点：真实资金，不可随意操作
使用场景：监控、紧急问题排查

2. 测试账号准备

2.1 创建测试钱包地址

# 示例：生成测试钱包地址（以太坊）
from eth_account import Account
import secrets

def create_test_wallet():
    """创建测试钱包"""
    # 生成随机私钥
    private_key = secrets.token_hex(32)
    
    # 从私钥创建账户
    account = Account.from_key(private_key)
    
    return {
        'address': account.address,
        'private_key': private_key,
        'public_key': account.key.hex()
    }

# 使用示例
wallet = create_test_wallet()
print(f"地址: {wallet['address']}")
print(f"私钥: {wallet['private_key']}")

2.2 获取测试币

以太坊测试网（Goerli/Sepolia）：

访问水龙头：https://goerli-faucet.pk910.de/
或使用：https://sepoliafaucet.com/

币安智能链测试网：

访问：https://testnet.bnbchain.org/faucet-smart

波场测试网：

访问：https://www.trongrid.io/

3. 测试工具准备

3.1 必需工具清单

工具	用途	下载地址
Postman/Insomnia	API测试	https://www.postman.com/
MySQL Workbench	数据库查询	https://www.mysql.com/products/workbench/
Redis Desktop Manager	Redis查看	https://redisdesktop.com/
区块链浏览器	查看链上数据	Etherscan、BscScan等
命令行工具	脚本执行	Git Bash、PowerShell

3.2 Python测试环境

# 安装Python依赖
pip install requests
pip install web3
pip install mysql-connector-python
pip install redis
pip install pytest
pip install python-dotenv

3.3 配置文件示例

创建 .env 文件：

# 数据库配置
DB_HOST=test-db.example.com
DB_PORT=3306
DB_USER=test_user
DB_PASSWORD=test_password
DB_NAME=test_db

# Redis配置
REDIS_HOST=test-redis.example.com
REDIS_PORT=6379
REDIS_PASSWORD=test_redis_password

# API配置
API_BASE_URL=https://test-api.example.com
API_KEY=test_api_key_12345

# 区块链RPC
ETH_RPC_URL=https://goerli.infura.io/v3/YOUR_PROJECT_ID
BSC_RPC_URL=https://data-seed-prebsc-1-s1.binance.org:8545

# 测试钱包
TEST_WALLET_ADDRESS=0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb
TEST_WALLET_PRIVATE_KEY=your_private_key_here

4. 测试数据准备

4.1 测试地址库

创建 test_addresses.json：

{
  "ethereum": {
    "mainnet": [
      {
        "address": "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb",
        "label": "测试地址1",
        "has_balance": true
      }
    ],
    "goerli": [
      {
        "address": "0x...",
        "label": "Goerli测试地址",
        "has_balance": true
      }
    ]
  },
  "bsc": {
    "mainnet": [],
    "testnet": []
  }
}

入金测试

1. 什么是入金？

入金（Deposit） 是指用户将数字资产从外部钱包转入到平台钱包地址的过程。

2. 入金流程

用户发起转账 → 链上确认 → 系统扫链检测 → 入金到账 → 用户余额更新

3. 测试场景

3.1 正常入金测试

测试步骤：

准备测试地址

# 获取平台入金地址
def get_deposit_address(user_id, currency):
    """
    获取用户的入金地址
    :param user_id: 用户ID
    :param currency: 币种（如：ETH、USDT）
    :return: 入金地址
    """
    url = f"{API_BASE_URL}/api/v1/deposit/address"
    headers = {
        'Authorization': f'Bearer {API_KEY}',
        'Content-Type': 'application/json'
    }
    params = {
        'user_id': user_id,
        'currency': currency
    }
    
    response = requests.get(url, headers=headers, params=params)
    data = response.json()
    
    if data['code'] == 0:
        return data['data']['address']
    else:
        raise Exception(f"获取入金地址失败: {data['message']}")

发送测试转账

from web3 import Web3

def send_test_deposit(to_address, amount_eth, private_key):
    """
    发送测试入金
    :param to_address: 平台入金地址
    :param amount_eth: 转账金额（ETH）
    :param private_key: 发送方私钥
    :return: 交易哈希
    """
    w3 = Web3(Web3.HTTPProvider(ETH_RPC_URL))
    
    # 从私钥获取账户
    account = w3.eth.account.from_key(private_key)
    
    # 构建交易
    transaction = {
        'to': to_address,
        'value': w3.to_wei(amount_eth, 'ether'),
        'gas': 21000,
        'gasPrice': w3.eth.gas_price,
        'nonce': w3.eth.get_transaction_count(account.address),
        'chainId': 5  # Goerli测试网
    }
    
    # 签名交易
    signed_txn = w3.eth.account.sign_transaction(transaction, private_key)
    
    # 发送交易
    tx_hash = w3.eth.send_raw_transaction(signed_txn.rawTransaction)
    
    print(f"交易已发送，哈希: {tx_hash.hex()}")
    return tx_hash.hex()

验证入金到账

import time

def verify_deposit(user_id, tx_hash, expected_amount, timeout=300):
    """
    验证入金是否到账
    :param user_id: 用户ID
    :param tx_hash: 交易哈希
    :param expected_amount: 预期金额
    :param timeout: 超时时间（秒）
    :return: 是否到账
    """
    start_time = time.time()
    
    while time.time() - start_time < timeout:
        # 查询用户余额
        url = f"{API_BASE_URL}/api/v1/user/balance"
        headers = {'Authorization': f'Bearer {API_KEY}'}
        params = {'user_id': user_id, 'currency': 'ETH'}
        
        response = requests.get(url, headers=headers, params=params)
        data = response.json()
        
        if data['code'] == 0:
            balance = float(data['data']['balance'])
            
            # 查询入金记录
            deposit_url = f"{API_BASE_URL}/api/v1/deposit/records"
            deposit_params = {
                'user_id': user_id,
                'tx_hash': tx_hash
            }
            deposit_response = requests.get(deposit_url, headers=headers, params=deposit_params)
            deposit_data = deposit_response.json()
            
            if deposit_data['code'] == 0 and len(deposit_data['data']) > 0:
                record = deposit_data['data'][0]
                if record['status'] == 'confirmed' and float(record['amount']) == expected_amount:
                    print(f"✅ 入金成功！金额: {expected_amount} ETH")
                    return True
        
        print(f"等待入金确认... ({int(time.time() - start_time)}秒)")
        time.sleep(10)
    
    print(f"❌ 入金验证超时")
    return False

检查数据库记录

import mysql.connector

def check_deposit_in_db(tx_hash):
    """检查数据库中的入金记录"""
    conn = mysql.connector.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        database=DB_NAME
    )
    
    cursor = conn.cursor(dictionary=True)
    
    # 查询入金记录
    query = """
        SELECT * FROM deposit_records 
        WHERE tx_hash = %s
    """
    cursor.execute(query, (tx_hash,))
    record = cursor.fetchone()
    
    cursor.close()
    conn.close()
    
    if record:
        print(f"数据库记录:")
        print(f"  用户ID: {record['user_id']}")
        print(f"  金额: {record['amount']}")
        print(f"  状态: {record['status']}")
        print(f"  确认数: {record['confirmations']}")
        return record
    else:
        print("❌ 数据库中未找到记录")
        return None

测试检查清单：

3.2 小额入金测试

测试目的： 验证系统能否正确处理小额入金

测试步骤：

def test_small_deposit():
    """测试小额入金（0.0001 ETH）"""
    user_id = "test_user_001"
    amount = 0.0001  # 非常小的金额
    
    # 获取入金地址
    deposit_address = get_deposit_address(user_id, "ETH")
    
    # 发送小额转账
    tx_hash = send_test_deposit(deposit_address, amount, TEST_PRIVATE_KEY)
    
    # 验证入金
    assert verify_deposit(user_id, tx_hash, amount), "小额入金失败"
    
    print("✅ 小额入金测试通过")

检查项：

系统能识别极小金额
余额精确到小数点后足够位数
手续费计算正确

3.3 大额入金测试

测试目的： 验证大额入金的安全性和准确性

测试步骤：

def test_large_deposit():
    """测试大额入金（100 ETH）"""
    user_id = "test_user_002"
    amount = 100.0
    
    deposit_address = get_deposit_address(user_id, "ETH")
    tx_hash = send_test_deposit(deposit_address, amount, TEST_PRIVATE_KEY)
    
    # 验证大额入金
    assert verify_deposit(user_id, tx_hash, amount), "大额入金失败"
    
    # 检查是否需要人工审核
    record = check_deposit_in_db(tx_hash)
    if record['amount'] >= 100:
        assert record['need_review'] == 1, "大额入金应触发审核"
    
    print("✅ 大额入金测试通过")

检查项：

大额入金是否触发风控审核
余额计算准确（无精度丢失）
系统性能正常（无卡顿）

3.4 重复入金测试

测试目的： 验证相同交易哈希不会被重复入账

测试步骤：

def test_duplicate_deposit():
    """测试重复入金（相同tx_hash）"""
    user_id = "test_user_003"
    amount = 1.0
    
    deposit_address = get_deposit_address(user_id, "ETH")
    tx_hash = send_test_deposit(deposit_address, amount, TEST_PRIVATE_KEY)
    
    # 第一次入金
    assert verify_deposit(user_id, tx_hash, amount), "第一次入金失败"
    
    # 获取第一次入金后的余额
    balance1 = get_user_balance(user_id, "ETH")
    
    # 模拟系统再次扫描到相同交易（不应该重复入账）
    # 这里需要触发扫链服务重新处理该交易
    trigger_rescan(tx_hash)
    time.sleep(30)
    
    # 检查余额是否未变化
    balance2 = get_user_balance(user_id, "ETH")
    assert balance1 == balance2, f"余额不应变化: {balance1} != {balance2}"
    
    # 检查数据库记录数量
    records = get_deposit_records_by_tx(tx_hash)
    assert len(records) == 1, f"应只有1条记录，实际: {len(records)}"
    
    print("✅ 重复入金测试通过（未重复入账）")

检查项：

相同tx_hash不会重复入账
数据库有唯一索引防止重复
系统日志记录重复检测

3.5 多币种入金测试

测试目的： 验证不同币种的入金功能

测试币种清单：

ETH（以太坊主币）
USDT（ERC-20代币）
USDC（ERC-20代币）
BTC（比特币，如果支持）

测试步骤：

def test_multi_currency_deposit():
    """测试多币种入金"""
    user_id = "test_user_004"
    currencies = ['ETH', 'USDT', 'USDC']
    
    for currency in currencies:
        print(f"\n测试 {currency} 入金...")
        
        # 获取该币种的入金地址
        deposit_address = get_deposit_address(user_id, currency)
        
        # 根据币种类型发送转账
        if currency == 'ETH':
            tx_hash = send_eth_deposit(deposit_address, 1.0, TEST_PRIVATE_KEY)
        elif currency in ['USDT', 'USDC']:
            tx_hash = send_token_deposit(
                deposit_address, 
                100.0,  # 100 USDT/USDC
                currency,
                TEST_PRIVATE_KEY
            )
        
        # 验证入金
        assert verify_deposit(user_id, tx_hash, 100.0 if currency != 'ETH' else 1.0), \
            f"{currency} 入金失败"
        
        print(f"✅ {currency} 入金测试通过")

3.6 入金确认数测试

测试目的： 验证不同确认数下的入金状态

测试步骤：

def test_deposit_confirmations():
    """测试入金确认数"""
    user_id = "test_user_005"
    amount = 1.0
    
    deposit_address = get_deposit_address(user_id, "ETH")
    tx_hash = send_test_deposit(deposit_address, amount, TEST_PRIVATE_KEY)
    
    # 监控确认数变化
    confirmations_required = 12  # 通常需要12个确认
    
    for i in range(20):  # 最多等待20次检查
        time.sleep(12)  # 每12秒检查一次（假设12秒一个区块）
        
        record = get_deposit_record(tx_hash)
        confirmations = record['confirmations']
        status = record['status']
        
        print(f"确认数: {confirmations}, 状态: {status}")
        
        if confirmations < confirmations_required:
            assert status == 'pending', "未达到确认数时状态应为pending"
        else:
            assert status == 'confirmed', "达到确认数后状态应为confirmed"
            break
    
    print("✅ 入金确认数测试通过")

检查项：

0-11个确认：状态为"pending"，余额未到账
12个确认：状态为"confirmed"，余额到账
确认数实时更新

3.7 异常入金测试

测试场景：

无效地址入金

def test_invalid_address_deposit():
    """测试向无效地址入金"""
    invalid_address = "0x0000000000000000000000000000000000000000"
    
    # 尝试获取无效地址的入金记录
    # 系统应该拒绝或忽略
    pass

合约地址入金

def test_contract_address_deposit():
    """测试向合约地址入金"""
    # 某些系统可能不支持合约地址入金
    # 需要验证系统行为
    pass

Gas不足的入金

def test_insufficient_gas_deposit():
    """测试Gas不足的入金交易"""
    # 发送Gas不足的交易
    # 验证系统如何处理失败的交易
    pass

4. 入金测试自动化脚本

import pytest
import time
from datetime import datetime

class DepositTestSuite:
    def __init__(self):
        self.api_base_url = os.getenv('API_BASE_URL')
        self.api_key = os.getenv('API_KEY')
        self.test_user_id = f"test_{int(time.time())}"
    
    def test_normal_deposit(self):
        """正常入金测试"""
        # 实现代码...
        pass
    
    def test_small_deposit(self):
        """小额入金测试"""
        # 实现代码...
        pass
    
    def test_large_deposit(self):
        """大额入金测试"""
        # 实现代码...
        pass
    
    def test_duplicate_deposit(self):
        """重复入金测试"""
        # 实现代码...
        pass
    
    def test_multi_currency_deposit(self):
        """多币种入金测试"""
        # 实现代码...
        pass

# 运行测试
if __name__ == "__main__":
    pytest.main([__file__, "-v"])

出金测试

1. 什么是出金？

出金（Withdrawal） 是指用户将数字资产从平台钱包提取到外部钱包地址的过程。

2. 出金流程

用户发起出金 → 风控检查 → 系统审核 → 构建交易 → 签名交易 → 广播交易 → 链上确认 → 出金完成

3. 测试场景

3.1 正常出金测试

测试步骤：

准备测试环境

def setup_withdrawal_test():
    """准备出金测试环境"""
    # 确保测试用户有足够余额
    user_id = "test_user_001"
    currency = "ETH"
    
    # 先入金确保有余额
    deposit_address = get_deposit_address(user_id, currency)
    send_test_deposit(deposit_address, 10.0, TEST_PRIVATE_KEY)
    
    # 等待入金确认
    time.sleep(60)
    
    # 验证余额
    balance = get_user_balance(user_id, currency)
    assert balance >= 10.0, "余额不足，无法进行出金测试"
    
    return user_id

发起出金请求

def create_withdrawal(user_id, to_address, amount, currency="ETH"):
    """
    创建出金请求
    :param user_id: 用户ID
    :param to_address: 接收地址
    :param amount: 出金金额
    :param currency: 币种
    :return: 出金订单ID
    """
    url = f"{API_BASE_URL}/api/v1/withdrawal/create"
    headers = {
        'Authorization': f'Bearer {API_KEY}',
        'Content-Type': 'application/json'
    }
    
    data = {
        'user_id': user_id,
        'to_address': to_address,
        'amount': amount,
        'currency': currency
    }
    
    response = requests.post(url, headers=headers, json=data)
    result = response.json()
    
    if result['code'] == 0:
        order_id = result['data']['order_id']
        print(f"出金订单创建成功，订单ID: {order_id}")
        return order_id
    else:
        raise Exception(f"创建出金订单失败: {result['message']}")

验证出金状态

def check_withdrawal_status(order_id):
    """检查出金状态"""
    url = f"{API_BASE_URL}/api/v1/withdrawal/status"
    headers = {
        'Authorization': f'Bearer {API_KEY}',
        'Content-Type': 'application/json'
    }
    params = {'order_id': order_id}
    
    response = requests.get(url, headers=headers, params=params)
    result = response.json()
    
    if result['code'] == 0:
        return result['data']
    else:
        raise Exception(f"查询出金状态失败: {result['message']}")

验证链上交易

from web3 import Web3

def verify_withdrawal_on_chain(tx_hash, expected_to, expected_amount):
    """
    验证链上出金交易
    :param tx_hash: 交易哈希
    :param expected_to: 预期接收地址
    :param expected_amount: 预期金额
    :return: 是否验证通过
    """
    w3 = Web3(Web3.HTTPProvider(ETH_RPC_URL))
    
    # 等待交易确认
    receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=300)
    
    if receipt['status'] == 1:  # 交易成功
        tx = w3.eth.get_transaction(tx_hash)
        
        # 验证接收地址
        assert tx['to'].lower() == expected_to.lower(), \
            f"接收地址不匹配: {tx['to']} != {expected_to}"
        
        # 验证金额（考虑Gas费用）
        actual_amount = w3.from_wei(tx['value'], 'ether')
        assert abs(actual_amount - expected_amount) < 0.0001, \
            f"金额不匹配: {actual_amount} != {expected_amount}"
        
        print(f"✅ 链上验证通过")
        print(f"  交易哈希: {tx_hash}")
        print(f"  接收地址: {tx['to']}")
        print(f"  金额: {actual_amount} ETH")
        print(f"  确认数: {w3.eth.block_number - receipt['blockNumber']}")
        return True
    else:
        print(f"❌ 交易失败")
        return False

完整出金测试流程

def test_normal_withdrawal():
    """正常出金完整测试"""
    # 1. 准备测试环境
    user_id = setup_withdrawal_test()
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"  # 外部钱包地址
    amount = 1.0  # 出金1 ETH
    
    # 2. 获取出金前余额
    balance_before = get_user_balance(user_id, "ETH")
    print(f"出金前余额: {balance_before} ETH")
    
    # 3. 创建出金订单
    order_id = create_withdrawal(user_id, to_address, amount, "ETH")
    
    # 4. 检查订单状态（应该是pending或processing）
    status = check_withdrawal_status(order_id)
    assert status['status'] in ['pending', 'processing'], \
        f"订单状态异常: {status['status']}"
    
    # 5. 等待出金处理（通常需要几分钟）
    print("等待出金处理...")
    for i in range(30):  # 最多等待5分钟
        time.sleep(10)
        status = check_withdrawal_status(order_id)
        
        if status['status'] == 'completed':
            tx_hash = status['tx_hash']
            print(f"出金完成，交易哈希: {tx_hash}")
            break
        elif status['status'] == 'failed':
            raise Exception(f"出金失败: {status['message']}")
    
    # 6. 验证链上交易
    assert verify_withdrawal_on_chain(tx_hash, to_address, amount), \
        "链上验证失败"
    
    # 7. 验证用户余额减少
    time.sleep(10)  # 等待余额更新
    balance_after = get_user_balance(user_id, "ETH")
    expected_balance = balance_before - amount
    
    assert abs(balance_after - expected_balance) < 0.0001, \
        f"余额不正确: {balance_after} != {expected_balance}"
    
    print(f"✅ 出金测试通过")
    print(f"  出金前: {balance_before} ETH")
    print(f"  出金后: {balance_after} ETH")
    print(f"  出金金额: {amount} ETH")

测试检查清单：

出金订单创建成功
订单状态正确流转（pending → processing → completed）
链上交易成功广播
交易哈希正确记录
用户余额正确减少
接收地址收到正确金额
数据库记录完整
出金手续费计算正确

3.2 小额出金测试

测试目的： 验证系统能否正确处理小额出金

测试步骤：

def test_small_withdrawal():
    """测试小额出金（0.001 ETH）"""
    user_id = setup_withdrawal_test()
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    amount = 0.001  # 非常小的金额
    
    # 检查最小出金限制
    min_withdrawal = get_min_withdrawal("ETH")
    if amount < min_withdrawal:
        print(f"⚠️ 金额小于最小出金限制: {min_withdrawal}")
        return
    
    order_id = create_withdrawal(user_id, to_address, amount, "ETH")
    
    # 验证小额出金
    assert verify_withdrawal_complete(order_id), "小额出金失败"
    
    print("✅ 小额出金测试通过")

检查项：

系统有最小出金限制
小额出金能正常处理
手续费计算正确（可能手续费大于出金金额）

3.3 大额出金测试

测试目的： 验证大额出金的安全性和审核流程

测试步骤：

def test_large_withdrawal():
    """测试大额出金（100 ETH）"""
    user_id = setup_withdrawal_test()
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    amount = 100.0
    
    # 创建大额出金订单
    order_id = create_withdrawal(user_id, to_address, amount, "ETH")
    
    # 检查是否需要人工审核
    status = check_withdrawal_status(order_id)
    assert status['need_review'] == True, "大额出金应触发审核"
    assert status['status'] == 'pending_review', "状态应为待审核"
    
    # 模拟人工审核通过（需要管理员权限）
    approve_withdrawal(order_id, admin_token)
    
    # 等待出金处理
    assert verify_withdrawal_complete(order_id), "大额出金失败"
    
    print("✅ 大额出金测试通过")

检查项：

大额出金触发风控审核
审核流程正常
审核通过后出金正常
审核拒绝后订单取消，余额不减少

3.4 余额不足测试

测试目的： 验证余额不足时的处理

测试步骤：

def test_insufficient_balance():
    """测试余额不足的出金"""
    user_id = "test_user_insufficient"
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    
    # 获取当前余额
    balance = get_user_balance(user_id, "ETH")
    print(f"当前余额: {balance} ETH")
    
    # 尝试出金超过余额的金额
    amount = balance + 1.0
    
    try:
        order_id = create_withdrawal(user_id, to_address, amount, "ETH")
        assert False, "应该拒绝余额不足的出金"
    except Exception as e:
        assert "余额不足" in str(e) or "insufficient" in str(e).lower(), \
            f"错误信息不正确: {e}"
        print("✅ 余额不足测试通过（正确拒绝）")

检查项：

系统拒绝余额不足的出金
错误提示清晰
用户余额未变化

3.5 地址白名单测试

测试目的： 验证地址白名单功能

测试步骤：

def test_address_whitelist():
    """测试地址白名单"""
    user_id = "test_user_whitelist"
    whitelisted_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    non_whitelisted_address = "0x1111111111111111111111111111111111111111"
    
    # 1. 添加地址到白名单
    add_to_whitelist(user_id, whitelisted_address, "ETH")
    
    # 2. 测试白名单地址出金（应该成功）
    order_id1 = create_withdrawal(user_id, whitelisted_address, 1.0, "ETH")
    assert verify_withdrawal_complete(order_id1), "白名单地址出金失败"
    
    # 3. 测试非白名单地址出金（应该失败或需要额外验证）
    try:
        order_id2 = create_withdrawal(user_id, non_whitelisted_address, 1.0, "ETH")
        # 如果系统允许，检查是否需要额外验证
        status = check_withdrawal_status(order_id2)
        if status['status'] == 'pending_verification':
            print("✅ 非白名单地址需要额外验证")
    except Exception as e:
        assert "地址未在白名单" in str(e) or "not whitelisted" in str(e).lower(), \
            f"错误信息不正确: {e}"
        print("✅ 非白名单地址被正确拒绝")

检查项：

白名单地址可以正常出金
非白名单地址被拒绝或需要验证
白名单管理功能正常

3.6 出金手续费测试

测试目的： 验证出金手续费计算

测试步骤：

def test_withdrawal_fee():
    """测试出金手续费"""
    user_id = setup_withdrawal_test()
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    amount = 1.0
    
    balance_before = get_user_balance(user_id, "ETH")
    
    # 获取手续费信息
    fee_info = get_withdrawal_fee("ETH")
    print(f"出金手续费: {fee_info['fee']} {fee_info['currency']}")
    
    # 创建出金订单
    order_id = create_withdrawal(user_id, to_address, amount, "ETH")
    
    # 验证手续费扣除
    order_detail = get_withdrawal_detail(order_id)
    fee = order_detail['fee']
    
    # 等待出金完成
    verify_withdrawal_complete(order_id)
    
    # 检查余额扣除（金额 + 手续费）
    balance_after = get_user_balance(user_id, "ETH")
    expected_balance = balance_before - amount - fee
    
    assert abs(balance_after - expected_balance) < 0.0001, \
        f"余额扣除不正确: {balance_after} != {expected_balance}"
    
    print(f"✅ 手续费测试通过")
    print(f"  出金金额: {amount} ETH")
    print(f"  手续费: {fee} ETH")
    print(f"  总扣除: {amount + fee} ETH")

检查项：

手续费计算正确
手续费从用户余额扣除
不同币种手续费不同
手续费配置可调整

3.7 出金失败处理测试

测试场景：

Gas不足导致失败

def test_withdrawal_gas_failure():
    """测试Gas不足导致的出金失败"""
    # 模拟Gas不足的情况
    # 系统应该能够检测到失败并回滚
    pass

地址无效

def test_invalid_address_withdrawal():
    """测试无效地址出金"""
    user_id = setup_withdrawal_test()
    invalid_address = "0x0000000000000000000000000000000000000000"
    
    try:
        order_id = create_withdrawal(user_id, invalid_address, 1.0, "ETH")
        assert False, "应该拒绝无效地址"
    except Exception as e:
        assert "无效地址" in str(e) or "invalid" in str(e).lower()
        print("✅ 无效地址被正确拒绝")

网络拥堵

def test_withdrawal_network_congestion():
    """测试网络拥堵时的出金处理"""
    # 模拟网络拥堵
    # 系统应该有重试机制
    pass

4. 出金测试自动化脚本

import pytest

class WithdrawalTestSuite:
    def __init__(self):
        self.api_base_url = os.getenv('API_BASE_URL')
        self.api_key = os.getenv('API_KEY')
    
    def test_normal_withdrawal(self):
        """正常出金测试"""
        # 实现代码...
        pass
    
    def test_small_withdrawal(self):
        """小额出金测试"""
        # 实现代码...
        pass
    
    def test_large_withdrawal(self):
        """大额出金测试"""
        # 实现代码...
        pass
    
    def test_insufficient_balance(self):
        """余额不足测试"""
        # 实现代码...
        pass

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

多站点测试

1. 什么是多站点？

多站点（Multi-site） 是指同一个平台在多个地区或服务器部署，需要保证数据同步和一致性。

2. 多站点架构

站点A（主站） ←→ 数据同步 ←→ 站点B（分站）
     ↓                           ↓
   数据库A                   数据库B
     ↓                           ↓
   缓存A                      缓存B

3. 测试场景

3.1 数据同步测试

测试目的： 验证不同站点之间的数据同步

测试步骤：

def test_multi_site_sync():
    """测试多站点数据同步"""
    # 站点配置
    sites = {
        'site_a': {
            'api_url': 'https://site-a.example.com/api',
            'api_key': 'key_a'
        },
        'site_b': {
            'api_url': 'https://site-b.example.com/api',
            'api_key': 'key_b'
        }
    }
    
    user_id = "test_user_multi_site"
    
    # 1. 在站点A创建入金
    print("在站点A创建入金...")
    deposit_address_a = get_deposit_address(
        user_id, "ETH", 
        site_config=sites['site_a']
    )
    tx_hash = send_test_deposit(deposit_address_a, 1.0, TEST_PRIVATE_KEY)
    
    # 2. 等待入金确认
    verify_deposit(user_id, tx_hash, 1.0, site_config=sites['site_a'])
    
    # 3. 检查站点B是否同步
    print("检查站点B数据同步...")
    time.sleep(5)  # 等待同步
    
    balance_b = get_user_balance(
        user_id, "ETH", 
        site_config=sites['site_b']
    )
    balance_a = get_user_balance(
        user_id, "ETH", 
        site_config=sites['site_a']
    )
    
    assert abs(balance_a - balance_b) < 0.0001, \
        f"余额不同步: 站点A={balance_a}, 站点B={balance_b}"
    
    # 4. 检查入金记录是否同步
    records_a = get_deposit_records(user_id, site_config=sites['site_a'])
    records_b = get_deposit_records(user_id, site_config=sites['site_b'])
    
    assert len(records_a) == len(records_b), \
        f"记录数量不一致: A={len(records_a)}, B={len(records_b)}"
    
    print("✅ 多站点数据同步测试通过")

检查项：

入金数据同步到所有站点
出金数据同步到所有站点
余额数据一致
交易记录一致
同步延迟在可接受范围内（通常<5秒）

3.2 跨站点出金测试

测试目的： 验证在站点A出金，站点B能正确显示

测试步骤：

def test_cross_site_withdrawal():
    """测试跨站点出金"""
    sites = get_site_configs()
    user_id = "test_user_cross_site"
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    
    # 1. 在站点A准备余额
    setup_balance(user_id, "ETH", 10.0, site_config=sites['site_a'])
    
    # 2. 在站点A创建出金
    order_id = create_withdrawal(
        user_id, to_address, 1.0, "ETH",
        site_config=sites['site_a']
    )
    
    # 3. 等待出金完成
    verify_withdrawal_complete(order_id, site_config=sites['site_a'])
    
    # 4. 检查站点B的余额和记录
    time.sleep(5)  # 等待同步
    
    balance_b = get_user_balance(
        user_id, "ETH", 
        site_config=sites['site_b']
    )
    
    # 站点B的余额应该也减少了
    assert balance_b == 9.0, f"站点B余额不正确: {balance_b}"
    
    # 检查站点B是否有出金记录
    withdrawals_b = get_withdrawal_records(
        user_id, 
        site_config=sites['site_b']
    )
    
    assert len(withdrawals_b) > 0, "站点B没有出金记录"
    assert withdrawals_b[0]['order_id'] == order_id, "订单ID不一致"
    
    print("✅ 跨站点出金测试通过")

3.3 并发操作测试

测试目的： 验证多站点并发操作的数据一致性

测试步骤：

import threading

def test_concurrent_operations():
    """测试并发操作"""
    sites = get_site_configs()
    user_id = "test_user_concurrent"
    
    # 准备初始余额
    setup_balance(user_id, "ETH", 100.0, site_config=sites['site_a'])
    
    results = []
    
    def concurrent_deposit(site_name, amount):
        """并发入金"""
        site_config = sites[site_name]
        deposit_address = get_deposit_address(
            user_id, "ETH", 
            site_config=site_config
        )
        tx_hash = send_test_deposit(deposit_address, amount, TEST_PRIVATE_KEY)
        results.append(('deposit', site_name, tx_hash, amount))
    
    def concurrent_withdrawal(site_name, amount):
        """并发出金"""
        site_config = sites[site_name]
        to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
        order_id = create_withdrawal(
            user_id, to_address, amount, "ETH",
            site_config=site_config
        )
        results.append(('withdrawal', site_name, order_id, amount))
    
    # 创建多个线程同时操作
    threads = []
    
    # 站点A并发入金
    for i in range(5):
        t = threading.Thread(
            target=concurrent_deposit, 
            args=('site_a', 1.0)
        )
        threads.append(t)
    
    # 站点B并发出金
    for i in range(3):
        t = threading.Thread(
            target=concurrent_withdrawal, 
            args=('site_b', 1.0)
        )
        threads.append(t)
    
    # 启动所有线程
    for t in threads:
        t.start()
    
    # 等待所有线程完成
    for t in threads:
        t.join()
    
    # 等待所有操作完成
    time.sleep(30)
    
    # 验证最终余额一致性
    balance_a = get_user_balance(user_id, "ETH", site_config=sites['site_a'])
    balance_b = get_user_balance(user_id, "ETH", site_config=sites['site_b'])
    
    assert abs(balance_a - balance_b) < 0.0001, \
        f"并发操作后余额不一致: A={balance_a}, B={balance_b}"
    
    print("✅ 并发操作测试通过")

3.4 站点故障恢复测试

测试目的： 验证站点故障后的数据恢复

测试步骤：

def test_site_failure_recovery():
    """测试站点故障恢复"""
    sites = get_site_configs()
    user_id = "test_user_failure"
    
    # 1. 正常操作
    setup_balance(user_id, "ETH", 10.0, site_config=sites['site_a'])
    
    # 2. 模拟站点A故障（停止服务）
    stop_site('site_a')
    print("站点A已停止")
    
    # 3. 在站点B进行操作
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    order_id = create_withdrawal(
        user_id, to_address, 1.0, "ETH",
        site_config=sites['site_b']
    )
    
    # 4. 恢复站点A
    start_site('site_a')
    print("站点A已恢复")
    
    # 5. 等待数据同步
    time.sleep(10)
    
    # 6. 验证站点A数据已恢复
    balance_a = get_user_balance(user_id, "ETH", site_config=sites['site_a'])
    assert balance_a == 9.0, f"站点A数据未恢复: {balance_a}"
    
    # 检查出金记录
    withdrawals_a = get_withdrawal_records(
        user_id, 
        site_config=sites['site_a']
    )
    assert len(withdrawals_a) > 0, "站点A没有出金记录"
    
    print("✅ 站点故障恢复测试通过")

对账测试

1. 什么是对账？

对账（Reconciliation） 是指将系统内部记录与链上实际数据进行比较，确保数据一致性。

2. 对账类型

2.1 余额对账

目的： 验证系统记录的余额与链上实际余额一致

测试步骤：

def test_balance_reconciliation():
    """测试余额对账"""
    # 获取所有需要对账的地址
    addresses = get_all_deposit_addresses()
    
    discrepancies = []
    
    for addr_info in addresses:
        address = addr_info['address']
        currency = addr_info['currency']
        user_id = addr_info['user_id']
        
        # 1. 获取系统记录的余额
        system_balance = get_user_balance(user_id, currency)
        
        # 2. 获取链上实际余额
        if currency == 'ETH':
            on_chain_balance = get_on_chain_balance_eth(address)
        else:
            on_chain_balance = get_on_chain_balance_token(
                address, 
                currency
            )
        
        # 3. 计算系统应记录的余额（链上余额 - 未确认出金）
        pending_withdrawals = get_pending_withdrawals(address, currency)
        pending_amount = sum([w['amount'] for w in pending_withdrawals])
        
        expected_balance = on_chain_balance - pending_amount
        
        # 4. 比较
        diff = abs(system_balance - expected_balance)
        
        if diff > 0.0001:  # 允许小的精度误差
            discrepancies.append({
                'user_id': user_id,
                'address': address,
                'currency': currency,
                'system_balance': system_balance,
                'on_chain_balance': on_chain_balance,
                'expected_balance': expected_balance,
                'difference': diff
            })
    
    # 5. 报告差异
    if discrepancies:
        print(f"❌ 发现 {len(discrepancies)} 个余额差异:")
        for disc in discrepancies:
            print(f"  用户: {disc['user_id']}")
            print(f"  地址: {disc['address']}")
            print(f"  币种: {disc['currency']}")
            print(f"  系统余额: {disc['system_balance']}")
            print(f"  链上余额: {disc['on_chain_balance']}")
            print(f"  差异: {disc['difference']}")
        return False
    else:
        print("✅ 余额对账通过，无差异")
        return True

2.2 交易对账

目的： 验证系统记录的交易与链上交易一致

测试步骤：

def test_transaction_reconciliation():
    """测试交易对账"""
    # 获取时间范围（例如最近24小时）
    end_time = int(time.time())
    start_time = end_time - 24 * 3600
    
    # 1. 获取系统记录的所有交易
    system_deposits = get_deposit_records_by_time(start_time, end_time)
    system_withdrawals = get_withdrawal_records_by_time(start_time, end_time)
    
    # 2. 获取链上所有交易
    on_chain_txs = scan_chain_transactions(start_time, end_time)
    
    # 3. 比对入金交易
    deposit_discrepancies = []
    
    for sys_deposit in system_deposits:
        tx_hash = sys_deposit['tx_hash']
        
        # 查找链上对应交易
        on_chain_tx = find_transaction_by_hash(on_chain_txs, tx_hash)
        
        if not on_chain_tx:
            deposit_discrepancies.append({
                'type': 'missing_on_chain',
                'tx_hash': tx_hash,
                'system_record': sys_deposit
            })
        else:
            # 比较金额
            if abs(float(sys_deposit['amount']) - float(on_chain_tx['amount'])) > 0.0001:
                deposit_discrepancies.append({
                    'type': 'amount_mismatch',
                    'tx_hash': tx_hash,
                    'system_amount': sys_deposit['amount'],
                    'on_chain_amount': on_chain_tx['amount']
                })
    
    # 4. 检查链上有但系统没有的交易（漏扫）
    for on_chain_tx in on_chain_txs:
        if on_chain_tx['type'] == 'deposit':
            sys_record = find_deposit_by_hash(system_deposits, on_chain_tx['hash'])
            if not sys_record:
                deposit_discrepancies.append({
                    'type': 'missing_in_system',
                    'tx_hash': on_chain_tx['hash'],
                    'on_chain_record': on_chain_tx
                })
    
    # 5. 报告差异
    if deposit_discrepancies:
        print(f"❌ 发现 {len(deposit_discrepancies)} 个交易差异")
        for disc in deposit_discrepancies:
            print(f"  类型: {disc['type']}")
            print(f"  交易哈希: {disc['tx_hash']}")
        return False
    else:
        print("✅ 交易对账通过，无差异")
        return True

2.3 自动对账脚本

def auto_reconciliation_daily():
    """每日自动对账"""
    print("开始每日自动对账...")
    print(f"时间: {datetime.now()}")
    
    # 1. 余额对账
    balance_ok = test_balance_reconciliation()
    
    # 2. 交易对账
    tx_ok = test_transaction_reconciliation()
    
    # 3. 生成对账报告
    report = {
        'date': datetime.now().strftime('%Y-%m-%d'),
        'balance_reconciliation': 'PASS' if balance_ok else 'FAIL',
        'transaction_reconciliation': 'PASS' if tx_ok else 'FAIL',
        'timestamp': int(time.time())
    }
    
    # 保存报告
    save_reconciliation_report(report)
    
    # 如果有差异，发送告警
    if not balance_ok or not tx_ok:
        send_alert("对账发现差异，请及时处理！", report)
    
    print("每日自动对账完成")
    return report

# 使用示例
if __name__ == "__main__":
    auto_reconciliation_daily()

测试检查清单：

余额对账无差异
交易对账无差异
对账报告生成正确
差异告警及时发送
对账性能满足要求（通常需要在1小时内完成）

钱包币种测试

1. 什么是钱包币种测试？

钱包币种测试 是指验证平台支持的各种数字货币（币种）在钱包系统中的功能是否正常，包括：

主链币（如ETH、BTC、BNB）
ERC-20代币（如USDT、USDC）
BEP-20代币（如BUSD）
其他链的代币

2. 测试场景

2.1 币种列表测试

测试目的： 验证系统支持的币种列表是否正确

测试步骤：

def test_currency_list():
    """测试币种列表"""
    url = f"{API_BASE_URL}/api/v1/currencies"
    response = requests.get(url)
    result = response.json()
    
    if result['code'] == 0:
        currencies = result['data']
        
        # 检查必需币种是否存在
        required_currencies = ['ETH', 'BTC', 'USDT', 'USDC']
        for req_curr in required_currencies:
            found = any(c['symbol'] == req_curr for c in currencies)
            assert found, f"缺少必需币种: {req_curr}"
        
        # 检查币种信息完整性
        for currency in currencies:
            assert 'symbol' in currency, "币种缺少symbol字段"
            assert 'name' in currency, "币种缺少name字段"
            assert 'decimals' in currency, "币种缺少decimals字段"
            assert 'chain' in currency, "币种缺少chain字段"
            assert 'contract_address' in currency or currency['symbol'] in ['ETH', 'BTC'], \
                f"代币缺少合约地址: {currency['symbol']}"
        
        print(f"✅ 币种列表测试通过，共 {len(currencies)} 个币种")
        return currencies
    else:
        raise Exception(f"获取币种列表失败: {result['message']}")

检查项：

所有必需币种都存在
币种信息完整（名称、精度、链、合约地址等）
币种状态正确（启用/禁用）
币种排序合理

2.2 币种精度测试

测试目的： 验证不同精度的币种处理是否正确

测试步骤：

def test_currency_precision():
    """测试币种精度"""
    test_cases = [
        {'symbol': 'BTC', 'decimals': 8, 'test_amount': 0.00000001},  # 最小单位
        {'symbol': 'ETH', 'decimals': 18, 'test_amount': 0.000000000000000001},
        {'symbol': 'USDT', 'decimals': 6, 'test_amount': 0.000001},
    ]
    
    for case in test_cases:
        symbol = case['symbol']
        decimals = case['decimals']
        test_amount = case['test_amount']
        
        print(f"\n测试币种: {symbol}, 精度: {decimals}")
        
        # 1. 获取币种信息
        currency_info = get_currency_info(symbol)
        assert currency_info['decimals'] == decimals, \
            f"{symbol} 精度不匹配: {currency_info['decimals']} != {decimals}"
        
        # 2. 测试最小金额处理
        if symbol == 'ETH':
            # 测试入金最小金额
            deposit_address = get_deposit_address("test_user", symbol)
            # 发送最小金额的测试转账
            tx_hash = send_test_deposit(deposit_address, test_amount, TEST_PRIVATE_KEY)
            
            # 验证系统能正确处理最小金额
            time.sleep(60)
            balance = get_user_balance("test_user", symbol)
            assert balance >= test_amount, f"最小金额处理失败: {balance} < {test_amount}"
        
        print(f"✅ {symbol} 精度测试通过")

检查项：

币种精度配置正确
系统能正确处理最小单位
余额显示精度正确
计算无精度丢失

2.3 主链币测试

测试目的： 验证主链币（ETH、BTC等）功能

测试步骤：

def test_main_chain_currency():
    """测试主链币（ETH）"""
    currency = "ETH"
    user_id = "test_user_main"
    
    # 1. 测试入金
    print("测试ETH入金...")
    deposit_address = get_deposit_address(user_id, currency)
    tx_hash = send_test_deposit(deposit_address, 1.0, TEST_PRIVATE_KEY)
    assert verify_deposit(user_id, tx_hash, 1.0), "ETH入金失败"
    
    # 2. 测试余额查询
    balance = get_user_balance(user_id, currency)
    assert balance >= 1.0, "ETH余额不正确"
    
    # 3. 测试出金
    print("测试ETH出金...")
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    order_id = create_withdrawal(user_id, to_address, 0.5, currency)
    assert verify_withdrawal_complete(order_id), "ETH出金失败"
    
    # 4. 验证Gas费用处理
    order_detail = get_withdrawal_detail(order_id)
    assert 'gas_fee' in order_detail, "缺少Gas费用信息"
    
    print("✅ 主链币测试通过")

检查项：

主链币入金正常
主链币出金正常
Gas费用计算正确
余额查询准确

2.4 ERC-20代币测试

测试目的： 验证ERC-20代币功能

测试步骤：

from web3 import Web3

def test_erc20_token():
    """测试ERC-20代币（USDT）"""
    currency = "USDT"
    user_id = "test_user_erc20"
    contract_address = "0xdAC17F958D2ee523a2206206994597C13D831ec7"  # USDT合约地址
    
    w3 = Web3(Web3.HTTPProvider(ETH_RPC_URL))
    
    # 1. 获取代币信息
    token_info = get_currency_info(currency)
    assert token_info['contract_address'].lower() == contract_address.lower(), \
        "合约地址不匹配"
    assert token_info['decimals'] == 6, "USDT精度应为6"
    
    # 2. 测试代币入金
    print("测试USDT入金...")
    deposit_address = get_deposit_address(user_id, currency)
    
    # 发送ERC-20代币转账
    tx_hash = send_erc20_token(
        deposit_address,
        100.0,  # 100 USDT
        contract_address,
        TEST_PRIVATE_KEY
    )
    
    assert verify_deposit(user_id, tx_hash, 100.0), "USDT入金失败"
    
    # 3. 测试代币余额
    balance = get_user_balance(user_id, currency)
    assert balance >= 100.0, "USDT余额不正确"
    
    # 4. 测试代币出金
    print("测试USDT出金...")
    to_address = "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
    order_id = create_withdrawal(user_id, to_address, 50.0, currency)
    assert verify_withdrawal_complete(order_id), "USDT出金失败"
    
    print("✅ ERC-20代币测试通过")

def send_erc20_token(to_address, amount, contract_address, private_key):
    """发送ERC-20代币转账"""
    w3 = Web3(Web3.HTTPProvider(ETH_RPC_URL))
    account = w3.eth.account.from_key(private_key)
    
    # ERC-20 transfer函数ABI
    transfer_abi = [
        {
            "constant": False,
            "inputs": [
                {"name": "_to", "type": "address"},
                {"name": "_value", "type": "uint256"}
            ],
            "name": "transfer",
            "outputs": [{"name": "", "type": "bool"}],
            "type": "function"
        }
    ]
    
    # 创建合约实例
    contract = w3.eth.contract(address=contract_address, abi=transfer_abi)
    
    # 构建交易（USDT精度为6）
    amount_wei = int(amount * 10**6)
    
    # 构建transfer函数调用
    tx = contract.functions.transfer(
        to_address,
        amount_wei
    ).build_transaction({
        'from': account.address,
        'gas': 100000,
        'gasPrice': w3.eth.gas_price,
        'nonce': w3.eth.get_transaction_count(account.address),
        'chainId': 1  # 主网
    })
    
    # 签名并发送
    signed_txn = w3.eth.account.sign_transaction(tx, private_key)
    tx_hash = w3.eth.send_raw_transaction(signed_txn.rawTransaction)
    
    print(f"ERC-20代币转账已发送，哈希: {tx_hash.hex()}")
    return tx_hash.hex()

检查项：

2.5 多链币种测试

测试目的： 验证同一币种在不同链上的支持（如USDT在ETH、BSC、TRX等链上）

测试步骤：

def test_multi_chain_currency():
    """测试多链币种（USDT）"""
    currency = "USDT"
    
    chains = [
        {'chain': 'ETH', 'contract': '0xdAC17F958D2ee523a2206206994597C13D831ec7'},
        {'chain': 'BSC', 'contract': '0x55d398326f99059fF775485246999027B3197955'},
        {'chain': 'TRX', 'contract': 'TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t'},
    ]
    
    for chain_info in chains:
        chain = chain_info['chain']
        contract = chain_info['contract']
        
        print(f"\n测试 {currency} 在 {chain} 链上...")
        
        # 1. 获取该链的币种信息
        currency_info = get_currency_info(currency, chain=chain)
        assert currency_info['contract_address'] == contract, \
            f"{chain}链上合约地址不匹配"
        
        # 2. 测试该链的入金
        user_id = f"test_user_{chain.lower()}"
        deposit_address = get_deposit_address(user_id, currency, chain=chain)
        assert deposit_address, f"无法获取{chain}链的入金地址"
        
        # 3. 测试该链的出金
        to_address = get_test_address(chain)
        # 先确保有余额
        # ... 入金操作 ...
        
        # 测试出金
        order_id = create_withdrawal(user_id, to_address, 10.0, currency, chain=chain)
        assert verify_withdrawal_complete(order_id), f"{chain}链出金失败"
        
        print(f"✅ {currency} 在 {chain} 链测试通过")

检查项：

各链的合约地址配置正确
各链的入金功能正常
各链的出金功能正常
链标识正确（避免混淆）

2.6 币种启用/禁用测试

测试目的： 验证币种的启用和禁用功能

测试步骤：

def test_currency_enable_disable():
    """测试币种启用/禁用"""
    currency = "TEST_COIN"
    
    # 1. 禁用币种
    print(f"禁用币种: {currency}")
    disable_currency(currency, admin_token)
    
    # 验证禁用后无法入金
    try:
        deposit_address = get_deposit_address("test_user", currency)
        assert False, "禁用币种不应能获取入金地址"
    except Exception as e:
        assert "币种已禁用" in str(e) or "disabled" in str(e).lower(), \
            f"错误信息不正确: {e}"
    
    # 验证禁用后无法出金
    try:
        order_id = create_withdrawal("test_user", "0x...", 1.0, currency)
        assert False, "禁用币种不应能出金"
    except Exception as e:
        assert "币种已禁用" in str(e), f"错误信息不正确: {e}"
    
    # 2. 启用币种
    print(f"启用币种: {currency}")
    enable_currency(currency, admin_token)
    
    # 验证启用后功能正常
    deposit_address = get_deposit_address("test_user", currency)
    assert deposit_address, "启用后应能获取入金地址"
    
    print("✅ 币种启用/禁用测试通过")

检查项：

禁用后无法入金
禁用后无法出金
禁用后余额查询正常（历史数据）
启用后功能恢复正常

熔断测试

1. 什么是熔断机制？

熔断机制（Circuit Breaker） 是一种保护机制，当系统检测到异常情况时，自动暂停某些操作，防止问题扩大。

2. 熔断场景

2.1 余额不足熔断

测试目的： 验证当热钱包余额不足时，系统自动暂停出金

测试步骤：

def test_insufficient_hot_wallet_balance():
    """测试热钱包余额不足熔断"""
    currency = "ETH"
    
    # 1. 获取热钱包地址
    hot_wallet_address = get_hot_wallet_address(currency)
    
    # 2. 检查当前热钱包余额
    hot_wallet_balance = get_chain_balance(hot_wallet_address, currency)
    print(f"热钱包当前余额: {hot_wallet_balance} {currency}")
    
    # 3. 设置熔断阈值（例如：余额低于10 ETH时熔断）
    set_circuit_breaker_threshold(currency, 10.0, admin_token)
    
    # 4. 模拟余额不足（通过大量出金或手动调整）
    # 注意：测试环境操作，生产环境谨慎！
    if hot_wallet_balance > 10.0:
        # 触发大量出金，使余额低于阈值
        trigger_multiple_withdrawals(currency, hot_wallet_balance - 5.0)
        time.sleep(30)  # 等待出金处理
    
    # 5. 验证熔断是否触发
    circuit_breaker_status = get_circuit_breaker_status(currency)
    assert circuit_breaker_status['status'] == 'OPEN', \
        "余额不足时应触发熔断"
    assert circuit_breaker_status['reason'] == 'insufficient_balance', \
        "熔断原因应为余额不足"
    
    # 6. 验证出金被拒绝
    try:
        order_id = create_withdrawal("test_user", "0x...", 1.0, currency)
        assert False, "熔断状态下不应能创建出金订单"
    except Exception as e:
        assert "熔断" in str(e) or "circuit breaker" in str(e).lower(), \
            f"错误信息不正确: {e}"
    
    # 7. 补充热钱包余额
    top_up_hot_wallet(currency, 20.0, admin_token)
    time.sleep(30)
    
    # 8. 验证熔断恢复
    circuit_breaker_status = get_circuit_breaker_status(currency)
    assert circuit_breaker_status['status'] == 'CLOSED', \
        "余额充足后熔断应恢复"
    
    # 9. 验证出金恢复正常
    order_id = create_withdrawal("test_user", "0x...", 1.0, currency)
    assert order_id, "熔断恢复后应能正常出金"
    
    print("✅ 余额不足熔断测试通过")

检查项：

余额低于阈值时自动触发熔断
熔断状态下出金被拒绝
熔断状态正确记录和通知
余额恢复后自动解除熔断
熔断解除后功能恢复正常

2.2 异常交易熔断

测试目的： 验证检测到异常交易时触发熔断

测试步骤：

def test_abnormal_transaction_circuit_breaker():
    """测试异常交易熔断"""
    currency = "ETH"
    
    # 1. 设置异常交易检测规则
    # 例如：1小时内超过10笔大额出金（>10 ETH）
    set_abnormal_transaction_rule(
        currency,
        {
            'max_large_withdrawals': 10,
            'large_amount_threshold': 10.0,
            'time_window': 3600  # 1小时
        },
        admin_token
    )
    
    # 2. 模拟异常交易（快速大量出金）
    print("模拟异常交易...")
    user_ids = [f"test_user_{i}" for i in range(15)]
    to_addresses = [f"0x{'1'*40}", f"0x{'2'*40}", ...]  # 不同地址
    
    for i, user_id in enumerate(user_ids):
        try:
            order_id = create_withdrawal(
                user_id,
                to_addresses[i % len(to_addresses)],
                15.0,  # 大额
                currency
            )
            print(f"创建出金订单 {i+1}: {order_id}")
        except Exception as e:
            print(f"出金被拒绝: {e}")
    
    time.sleep(60)  # 等待系统检测
    
    # 3. 验证熔断是否触发
    circuit_breaker_status = get_circuit_breaker_status(currency)
    assert circuit_breaker_status['status'] == 'OPEN', \
        "异常交易应触发熔断"
    assert 'abnormal_transaction' in circuit_breaker_status['reason'], \
        "熔断原因应为异常交易"
    
    # 4. 验证告警发送
    alerts = get_recent_alerts()
    assert any('异常交易' in a['message'] or 'abnormal' in a['message'].lower() 
               for a in alerts), "应发送异常交易告警"
    
    print("✅ 异常交易熔断测试通过")

检查项：

异常交易模式被正确识别
熔断及时触发
告警及时发送
相关出金被暂停

2.3 链异常熔断

测试目的： 验证链异常（如节点不可用、网络拥堵）时触发熔断

测试步骤：

def test_chain_abnormal_circuit_breaker():
    """测试链异常熔断"""
    currency = "ETH"
    
    # 1. 设置链健康检查规则
    set_chain_health_check(
        currency,
        {
            'rpc_timeout': 5,  # RPC超时5秒
            'max_failed_requests': 3,  # 连续3次失败触发熔断
            'check_interval': 10  # 每10秒检查一次
        },
        admin_token
    )
    
    # 2. 模拟RPC节点故障（测试环境）
    # 注意：这需要能控制RPC节点，或使用Mock
    print("模拟RPC节点故障...")
    simulate_rpc_failure(currency, duration=60)  # 故障60秒
    
    time.sleep(40)  # 等待检测
    
    # 3. 验证熔断触发
    circuit_breaker_status = get_circuit_breaker_status(currency)
    assert circuit_breaker_status['status'] == 'OPEN', \
        "链异常应触发熔断"
    assert 'chain_unavailable' in circuit_breaker_status['reason'] or \
           'rpc_error' in circuit_breaker_status['reason'], \
        "熔断原因应为链异常"
    
    # 4. 验证出金被暂停
    try:
        order_id = create_withdrawal("test_user", "0x...", 1.0, currency)
        assert False, "链异常时不应能创建出金订单"
    except Exception as e:
        assert "链异常" in str(e) or "chain" in str(e).lower(), \
            f"错误信息不正确: {e}"
    
    # 5. 恢复RPC节点
    restore_rpc_node(currency)
    time.sleep(30)
    
    # 6. 验证熔断恢复
    circuit_breaker_status = get_circuit_breaker_status(currency)
    assert circuit_breaker_status['status'] == 'CLOSED', \
        "链恢复后熔断应解除"
    
    print("✅ 链异常熔断测试通过")

检查项：

RPC节点故障被及时检测
故障时触发熔断
熔断期间出金被暂停
节点恢复后自动解除熔断

2.4 手动熔断测试

测试目的： 验证管理员手动触发熔断功能

测试步骤：

def test_manual_circuit_breaker():
    """测试手动熔断"""
    currency = "ETH"
    
    # 1. 手动触发熔断
    print("手动触发熔断...")
    trigger_manual_circuit_breaker(
        currency,
        reason="系统维护",
        admin_token
    )
    
    # 2. 验证熔断状态
    circuit_breaker_status = get_circuit_breaker_status(currency)
    assert circuit_breaker_status['status'] == 'OPEN', \
        "手动熔断应生效"
    assert circuit_breaker_status['reason'] == 'manual', \
        "熔断原因应为手动"
    assert circuit_breaker_status['manual_reason'] == '系统维护', \
        "应记录手动原因"
    
    # 3. 验证出金被暂停
    try:
        order_id = create_withdrawal("test_user", "0x...", 1.0, currency)
        assert False, "手动熔断时不应能出金"
    except Exception as e:
        assert "系统维护" in str(e) or "maintenance" in str(e).lower(), \
            f"错误信息应包含维护原因: {e}"
    
    # 4. 手动解除熔断
    print("手动解除熔断...")
    close_circuit_breaker(currency, admin_token)
    
    # 5. 验证熔断解除
    circuit_breaker_status = get_circuit_breaker_status(currency)
    assert circuit_breaker_status['status'] == 'CLOSED', \
        "手动解除后熔断应关闭"
    
    # 6. 验证功能恢复
    order_id = create_withdrawal("test_user", "0x...", 1.0, currency)
    assert order_id, "熔断解除后应能正常出金"
    
    print("✅ 手动熔断测试通过")

检查项：

手动熔断能正确触发
熔断原因正确记录
用户收到明确的提示信息
手动解除后功能恢复

2.5 熔断状态查询和监控

测试步骤：

def test_circuit_breaker_monitoring():
    """测试熔断监控"""
    currency = "ETH"
    
    # 1. 查询当前熔断状态
    status = get_circuit_breaker_status(currency)
    print(f"熔断状态: {status['status']}")
    print(f"熔断原因: {status.get('reason', 'N/A')}")
    print(f"触发时间: {status.get('triggered_at', 'N/A')}")
    print(f"持续时间: {status.get('duration', 'N/A')}秒")
    
    # 2. 查询熔断历史
    history = get_circuit_breaker_history(currency, days=7)
    print(f"\n最近7天熔断历史: {len(history)} 次")
    for record in history:
        print(f"  - {record['triggered_at']}: {record['reason']} "
              f"({record['duration']}秒)")
    
    # 3. 验证监控告警
    alerts = get_circuit_breaker_alerts(currency)
    assert len(alerts) > 0, "应有熔断告警记录"
    
    print("✅ 熔断监控测试通过")

测试工具和脚本

1. 测试工具类

import os
import time
import requests
import mysql.connector
from web3 import Web3
from dotenv import load_dotenv

load_dotenv()

class BlockchainTestHelper:
    """区块链测试辅助类"""
    
    def __init__(self):
        self.api_base_url = os.getenv('API_BASE_URL')
        self.api_key = os.getenv('API_KEY')
        self.db_config = {
            'host': os.getenv('DB_HOST'),
            'port': int(os.getenv('DB_PORT')),
            'user': os.getenv('DB_USER'),
            'password': os.getenv('DB_PASSWORD'),
            'database': os.getenv('DB_NAME')
        }
        self.eth_rpc = os.getenv('ETH_RPC_URL')
        self.w3 = Web3(Web3.HTTPProvider(self.eth_rpc))
    
    def get_headers(self):
        """获取API请求头"""
        return {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
    
    def query_db(self, sql, params=None):
        """查询数据库"""
        conn = mysql.connector.connect(**self.db_config)
        cursor = conn.cursor(dictionary=True)
        cursor.execute(sql, params or ())
        results = cursor.fetchall()
        cursor.close()
        conn.close()
        return results
    
    def wait_for_confirmation(self, tx_hash, timeout=300):
        """等待交易确认"""
        start_time = time.time()
        while time.time() - start_time < timeout:
            try:
                receipt = self.w3.eth.get_transaction_receipt(tx_hash)
                if receipt['status'] == 1:
                    return receipt
            except:
                pass
            time.sleep(5)
        raise TimeoutError(f"交易确认超时: {tx_hash}")
    
    def get_chain_balance(self, address, currency):
        """获取链上余额"""
        if currency == 'ETH':
            balance_wei = self.w3.eth.get_balance(address)
            return self.w3.from_wei(balance_wei, 'ether')
        else:
            # ERC-20代币余额查询
            # 实现代币余额查询逻辑
            pass

# 使用示例
helper = BlockchainTestHelper()

2. 测试数据生成器

class TestDataGenerator:
    """测试数据生成器"""
    
    @staticmethod
    def generate_test_address():
        """生成测试地址"""
        from eth_account import Account
        import secrets
        private_key = secrets.token_hex(32)
        account = Account.from_key(private_key)
        return {
            'address': account.address,
            'private_key': private_key
        }
    
    @staticmethod
    def generate_test_user_id(prefix="test"):
        """生成测试用户ID"""
        return f"{prefix}_{int(time.time())}_{secrets.token_hex(4)}"
    
    @staticmethod
    def generate_test_amount(min_amount=0.001, max_amount=100):
        """生成测试金额"""
        import random
        return round(random.uniform(min_amount, max_amount), 6)

3. 测试报告生成器

class TestReportGenerator:
    """测试报告生成器"""
    
    def __init__(self):
        self.results = []
    
    def add_result(self, test_name, status, message="", details=None):
        """添加测试结果"""
        self.results.append({
            'test_name': test_name,
            'status': status,  # PASS, FAIL, SKIP
            'message': message,
            'details': details or {},
            'timestamp': time.time()
        })
    
    def generate_report(self, output_file="test_report.html"):
        """生成HTML测试报告"""
        html = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <title>链上测试报告</title>
            <style>
                body {{ font-family: Arial, sans-serif; margin: 20px; }}
                .pass {{ color: green; }}
                .fail {{ color: red; }}
                .skip {{ color: orange; }}
                table {{ border-collapse: collapse; width: 100%; }}
                th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
                th {{ background-color: #f2f2f2; }}
            </style>
        </head>
        <body>
            <h1>链上测试报告</h1>
            <p>生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}</p>
            <table>
                <tr>
                    <th>测试名称</th>
                    <th>状态</th>
                    <th>消息</th>
                    <th>时间</th>
                </tr>
        """
        
        for result in self.results:
            status_class = result['status'].lower()
            html += f"""
                <tr>
                    <td>{result['test_name']}</td>
                    <td class="{status_class}">{result['status']}</td>
                    <td>{result['message']}</td>
                    <td>{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(result['timestamp']))}</td>
                </tr>
            """
        
        html += """
            </table>
        </body>
        </html>
        """
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(html)
        
        print(f"测试报告已生成: {output_file}")

# 使用示例
report_gen = TestReportGenerator()
report_gen.add_result("入金测试", "PASS", "所有入金测试通过")
report_gen.add_result("出金测试", "FAIL", "大额出金审核失败", {"order_id": "12345"})
report_gen.generate_report("test_report.html")

4. 测试执行脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
链上测试执行脚本
用法: python run_tests.py --test-type deposit --env test
"""

import argparse
import sys
import time
from datetime import datetime

class TestRunner:
    """测试执行器"""
    
    def __init__(self, test_type, environment):
        self.test_type = test_type
        self.environment = environment
        self.helper = BlockchainTestHelper()
        self.report_gen = TestReportGenerator()
    
    def run_deposit_tests(self):
        """执行入金测试"""
        print("=" * 50)
        print("开始执行入金测试...")
        print("=" * 50)
        
        tests = [
            ("正常入金测试", self.test_normal_deposit),
            ("小额入金测试", self.test_small_deposit),
            ("大额入金测试", self.test_large_deposit),
            ("重复入金测试", self.test_duplicate_deposit),
            ("多币种入金测试", self.test_multi_currency_deposit),
        ]
        
        for test_name, test_func in tests:
            try:
                print(f"\n执行: {test_name}")
                test_func()
                self.report_gen.add_result(test_name, "PASS")
                print(f"✅ {test_name} 通过")
            except Exception as e:
                self.report_gen.add_result(test_name, "FAIL", str(e))
                print(f"❌ {test_name} 失败: {e}")
    
    def run_withdrawal_tests(self):
        """执行出金测试"""
        print("=" * 50)
        print("开始执行出金测试...")
        print("=" * 50)
        
        tests = [
            ("正常出金测试", self.test_normal_withdrawal),
            ("小额出金测试", self.test_small_withdrawal),
            ("大额出金测试", self.test_large_withdrawal),
            ("余额不足测试", self.test_insufficient_balance),
            ("地址白名单测试", self.test_address_whitelist),
        ]
        
        for test_name, test_func in tests:
            try:
                print(f"\n执行: {test_name}")
                test_func()
                self.report_gen.add_result(test_name, "PASS")
                print(f"✅ {test_name} 通过")
            except Exception as e:
                self.report_gen.add_result(test_name, "FAIL", str(e))
                print(f"❌ {test_name} 失败: {e}")
    
    def run_all_tests(self):
        """执行所有测试"""
        self.run_deposit_tests()
        self.run_withdrawal_tests()
        # 添加其他测试...
    
    def generate_final_report(self):
        """生成最终报告"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        report_file = f"test_report_{timestamp}.html"
        self.report_gen.generate_report(report_file)
        
        # 统计结果
        total = len(self.report_gen.results)
        passed = sum(1 for r in self.report_gen.results if r['status'] == 'PASS')
        failed = sum(1 for r in self.report_gen.results if r['status'] == 'FAIL')
        
        print("\n" + "=" * 50)
        print("测试总结")
        print("=" * 50)
        print(f"总测试数: {total}")
        print(f"通过: {passed}")
        print(f"失败: {failed}")
        print(f"通过率: {passed/total*100:.2f}%")
        print(f"报告文件: {report_file}")

def main():
    parser = argparse.ArgumentParser(description='链上测试执行脚本')
    parser.add_argument('--test-type', choices=['deposit', 'withdrawal', 'all'], 
                       default='all', help='测试类型')
    parser.add_argument('--env', choices=['dev', 'test', 'staging'], 
                       default='test', help='测试环境')
    
    args = parser.parse_args()
    
    runner = TestRunner(args.test_type, args.env)
    
    if args.test_type == 'deposit':
        runner.run_deposit_tests()
    elif args.test_type == 'withdrawal':
        runner.run_withdrawal_tests()
    else:
        runner.run_all_tests()
    
    runner.generate_final_report()

if __name__ == '__main__':
    main()

常见问题排查

1. 入金相关问题

问题1：入金未到账

症状：

链上交易已确认，但用户余额未增加
系统未检测到入金交易

排查步骤：

def troubleshoot_deposit_not_arrived(tx_hash, user_id):
    """排查入金未到账问题"""
    print(f"排查入金问题: {tx_hash}")
    
    # 1. 检查链上交易状态
    w3 = Web3(Web3.HTTPProvider(ETH_RPC_URL))
    try:
        tx = w3.eth.get_transaction(tx_hash)
        receipt = w3.eth.get_transaction_receipt(tx_hash)
        
        print(f"链上交易状态: {'成功' if receipt['status'] == 1 else '失败'}")
        print(f"确认数: {w3.eth.block_number - receipt['blockNumber']}")
        print(f"接收地址: {tx['to']}")
        print(f"金额: {w3.from_wei(tx['value'], 'ether')} ETH")
        
        # 2. 检查系统是否记录该交易
        db_record = check_deposit_in_db(tx_hash)
        if not db_record:
            print("❌ 问题: 数据库中未找到该交易记录")
            print("   可能原因: 扫链服务未检测到该交易")
            print("   解决方案: 手动触发扫链或补扫")
        else:
            print(f"✅ 数据库记录存在")
            print(f"   状态: {db_record['status']}")
            print(f"   确认数: {db_record['confirmations']}")
            
            if db_record['status'] != 'confirmed':
                print(f"❌ 问题: 交易状态未确认")
                print(f"   当前确认数: {db_record['confirmations']}")
                print(f"   需要确认数: 12")
        
        # 3. 检查接收地址是否正确
        deposit_address = get_deposit_address(user_id, "ETH")
        if tx['to'].lower() != deposit_address.lower():
            print(f"❌ 问题: 接收地址不匹配")
            print(f"   交易接收地址: {tx['to']}")
            print(f"   用户入金地址: {deposit_address}")
        
        # 4. 检查用户余额
        balance = get_user_balance(user_id, "ETH")
        print(f"用户当前余额: {balance} ETH")
        
    except Exception as e:
        print(f"❌ 检查链上交易失败: {e}")

常见原因和解决方案：

原因	解决方案
扫链服务未运行	重启扫链服务
确认数不足	等待更多确认
地址不匹配	检查用户入金地址配置
交易失败	检查Gas费用和交易状态
数据库记录异常	手动修复数据库记录

问题2：重复入账

症状：

同一笔交易被多次入账
用户余额异常增加

排查步骤：

def troubleshoot_duplicate_deposit(tx_hash):
    """排查重复入账问题"""
    print(f"排查重复入账: {tx_hash}")
    
    # 1. 查询数据库中该交易的所有记录
    records = get_all_deposit_records_by_tx(tx_hash)
    
    if len(records) > 1:
        print(f"❌ 发现重复记录: {len(records)} 条")
        for i, record in enumerate(records, 1):
            print(f"   记录{i}: 用户ID={record['user_id']}, 金额={record['amount']}, 状态={record['status']}")
        
        # 2. 检查是否有唯一索引
        print("\n检查数据库索引...")
        # 执行SQL检查索引
        
        # 3. 建议修复方案
        print("\n修复建议:")
        print("1. 保留最早的一条记录")
        print("2. 删除其他重复记录")
        print("3. 修正用户余额")
        print("4. 添加唯一索引防止再次发生")
    else:
        print("✅ 未发现重复记录")

2. 出金相关问题

问题1：出金订单卡在pending状态

症状：

出金订单创建后一直处于pending状态
长时间未处理

排查步骤：

def troubleshoot_withdrawal_pending(order_id):
    """排查出金pending问题"""
    print(f"排查出金pending: {order_id}")
    
    # 1. 检查订单状态
    order = get_withdrawal_order(order_id)
    print(f"订单状态: {order['status']}")
    print(f"创建时间: {order['create_time']}")
    print(f"等待时间: {int(time.time()) - order['create_time']} 秒")
    
    # 2. 检查是否需要审核
    if order['need_review']:
        print("❌ 问题: 订单需要人工审核")
        print("   解决方案: 联系管理员审核")
        return
    
    # 3. 检查风控状态
    risk_status = check_risk_status(order_id)
    if risk_status['blocked']:
        print(f"❌ 问题: 订单被风控拦截")
        print(f"   原因: {risk_status['reason']}")
        return
    
    # 4. 检查钱包余额
    wallet_balance = get_wallet_balance(order['currency'])
    if wallet_balance < order['amount']:
        print(f"❌ 问题: 钱包余额不足")
        print(f"   需要: {order['amount']} {order['currency']}")
        print(f"   实际: {wallet_balance} {order['currency']}")
        return
    
    # 5. 检查出金服务状态
    service_status = check_withdrawal_service_status()
    if not service_status['running']:
        print(f"❌ 问题: 出金服务未运行")
        print("   解决方案: 重启出金服务")
        return
    
    print("✅ 订单状态正常，等待处理中...")

问题2：出金交易失败

症状：

出金订单状态为failed
链上交易失败或未广播

排查步骤：

def troubleshoot_withdrawal_failed(order_id):
    """排查出金失败问题"""
    print(f"排查出金失败: {order_id}")
    
    order = get_withdrawal_order(order_id)
    
    # 1. 检查失败原因
    print(f"失败原因: {order.get('fail_reason', '未知')}")
    
    # 2. 如果有交易哈希，检查链上状态
    if order.get('tx_hash'):
        tx_hash = order['tx_hash']
        w3 = Web3(Web3.HTTPProvider(ETH_RPC_URL))
        
        try:
            receipt = w3.eth.get_transaction_receipt(tx_hash)
            if receipt['status'] == 0:
                print(f"❌ 链上交易失败")
                print(f"   可能原因: Gas不足、合约执行失败等")
                
                # 分析失败原因
                tx = w3.eth.get_transaction(tx_hash)
                gas_used = receipt['gasUsed']
                gas_limit = tx['gas']
                
                if gas_used == gas_limit:
                    print("   原因: Gas不足（Out of Gas）")
        except:
            print("   交易可能未上链或已被丢弃")
    
    # 3. 检查是否需要重试
    if order.get('retryable', False):
        print("\n建议: 可以重试该出金订单")
    else:
        print("\n建议: 需要人工处理")

3. 对账相关问题

问题1：余额对账不一致

症状：

系统余额与链上余额不一致
对账报告显示差异

排查步骤：

def troubleshoot_balance_mismatch(user_id, currency, address):
    """排查余额不一致问题"""
    print(f"排查余额不一致: 用户={user_id}, 币种={currency}, 地址={address}")
    
    # 1. 获取系统余额
    system_balance = get_user_balance(user_id, currency)
    print(f"系统余额: {system_balance} {currency}")
    
    # 2. 获取链上余额
    on_chain_balance = get_chain_balance(address, currency)
    print(f"链上余额: {on_chain_balance} {currency}")
    
    # 3. 计算差异
    difference = system_balance - on_chain_balance
    print(f"差异: {difference} {currency}")
    
    if abs(difference) > 0.0001:  # 允许小误差
        print("❌ 余额不一致")
        
        # 4. 分析可能原因
        print("\n可能原因分析:")
        
        # 检查是否有未确认的入金
        pending_deposits = get_pending_deposits(user_id, currency)
        if pending_deposits:
            total_pending = sum(d['amount'] for d in pending_deposits)
            print(f"   未确认入金: {total_pending} {currency}")
        
        # 检查是否有处理中的出金
        processing_withdrawals = get_processing_withdrawals(user_id, currency)
        if processing_withdrawals:
            total_processing = sum(w['amount'] for w in processing_withdrawals)
            print(f"   处理中出金: {total_processing} {currency}")
        
        # 检查是否有失败的交易未回滚
        failed_txs = get_failed_transactions(user_id, currency)
        if failed_txs:
            print(f"   失败的交易: {len(failed_txs)} 笔")
        
        # 5. 建议修复方案
        print("\n修复建议:")
        print("1. 检查最近的入金/出金记录")
        print("2. 检查是否有重复入账或漏账")
        print("3. 手动调整余额（需要审批）")
        print("4. 记录差异原因")
    else:
        print("✅ 余额一致")

4. 多站点相关问题

问题1：站点间数据不同步

症状：

不同站点显示的数据不一致
用户在一个站点操作后，另一个站点未更新

排查步骤：

def troubleshoot_site_sync_issue(user_id, site1, site2):
    """排查站点同步问题"""
    print(f"排查站点同步: 用户={user_id}, 站点1={site1}, 站点2={site2}")
    
    # 1. 检查两个站点的用户余额
    balance1 = get_user_balance_from_site(user_id, site1)
    balance2 = get_user_balance_from_site(user_id, site2)
    
    print(f"{site1} 余额: {balance1}")
    print(f"{site2} 余额: {balance2}")
    
    if balance1 != balance2:
        print("❌ 余额不一致")
        
        # 2. 检查最近交易记录
        txs1 = get_recent_transactions_from_site(user_id, site1, limit=10)
        txs2 = get_recent_transactions_from_site(user_id, site2, limit=10)
        
        print(f"\n{site1} 最近交易数: {len(txs1)}")
        print(f"{site2} 最近交易数: {len(txs2)}")
        
        # 3. 检查同步服务状态
        sync_status = check_sync_service_status()
        if not sync_status['running']:
            print("❌ 同步服务未运行")
            print("   解决方案: 重启同步服务")
        
        # 4. 检查消息队列
        queue_status = check_message_queue_status()
        if queue_status['backlog'] > 1000:
            print(f"⚠️ 消息队列积压: {queue_status['backlog']} 条")
            print("   建议: 检查消息队列处理速度")
    else:
        print("✅ 余额一致")

5. 熔断相关问题

问题1：熔断未触发

症状：

异常情况下熔断机制未生效
系统继续处理异常请求

排查步骤：

def troubleshoot_circuit_breaker_not_triggered():
    """排查熔断未触发问题"""
    print("排查熔断机制...")
    
    # 1. 检查熔断配置
    config = get_circuit_breaker_config()
    print(f"熔断配置:")
    print(f"  失败阈值: {config['failure_threshold']}")
    print(f"  时间窗口: {config['time_window']} 秒")
    print(f"  恢复时间: {config['recovery_time']} 秒")
    
    # 2. 检查当前错误率
    error_rate = get_current_error_rate()
    print(f"当前错误率: {error_rate:.2f}%")
    
    if error_rate > config['failure_threshold']:
        print("❌ 错误率超过阈值，但熔断未触发")
        print("   可能原因: 熔断服务异常")
        print("   解决方案: 检查熔断服务状态")
    else:
        print("✅ 错误率正常，熔断无需触发")
    
    # 3. 检查熔断服务状态
    cb_status = get_circuit_breaker_status()
    print(f"熔断状态: {cb_status['state']}")  # CLOSED, OPEN, HALF_OPEN
    print(f"失败计数: {cb_status['failure_count']}")

测试检查清单

1. 入金测试检查清单

功能测试

正常入金流程完整
小额入金（0.0001 ETH）正常
大额入金（100 ETH）正常
重复入金不会重复到账
多币种入金（ETH、USDT、USDC）正常
入金确认数正确（0-11个确认为pending，12个确认为confirmed）
入金地址生成正确
入金记录数据库保存完整

异常测试

无效地址入金被拒绝
Gas不足的交易正确处理
交易失败时系统不增加余额
网络异常时能正确恢复

性能测试

大量并发入金处理正常
扫链服务能及时检测到入金
入金到账时间在可接受范围内（<5分钟）

安全测试

入金地址唯一性
交易签名验证
防止重放攻击
防止SQL注入等安全问题

2. 出金测试检查清单

功能测试

正常出金流程完整
小额出金正常
大额出金触发审核流程
余额不足时拒绝出金
地址白名单功能正常
出金手续费计算正确
出金订单状态流转正确（pending → processing → completed）
链上交易成功广播

异常测试

无效地址出金被拒绝
出金交易失败时正确处理
审核拒绝时余额不减少
网络异常时能正确重试

性能测试

出金处理时间在可接受范围内（<10分钟）
大量并发出金处理正常
出金队列处理正常

安全测试

出金需要身份验证
大额出金需要额外审核
防止重复出金
防止未授权出金

3. 多站点测试检查清单

站点A的入金在站点B可见
站点A的出金在站点B可见
站点A的余额变化在站点B同步
站点间数据同步延迟<30秒
站点A故障不影响站点B
数据同步服务高可用

4. 对账测试检查清单

余额对账准确（系统余额 = 链上余额）
交易对账准确（系统记录 = 链上交易）
对账报告生成正确
差异能及时发现和告警
对账脚本能自动运行
对账数据可追溯

5. 钱包币种测试检查清单

所有支持的币种都能正常入金
所有支持的币种都能正常出金
币种余额查询准确
币种间转换（如适用）正常
新币种上线流程完整
币种配置管理正常

6. 熔断测试检查清单

错误率超过阈值时熔断触发
熔断后系统停止处理新请求
熔断恢复机制正常
熔断状态正确记录和告警
熔断配置可动态调整
熔断不影响已处理的请求

7. 综合测试检查清单

数据一致性

数据库数据与链上数据一致
缓存数据与数据库数据一致
多站点数据一致

系统稳定性

7x24小时运行稳定
异常情况下能自动恢复
日志记录完整
监控告警正常

性能指标

入金到账时间 < 5分钟
出金处理时间 < 10分钟
API响应时间 < 1秒
系统可用性 > 99.9%

总结

测试工程师日常工作流程

每日检查
- 检查对账报告
- 检查系统告警
- 检查异常交易
功能测试
- 新功能上线前测试
- 回归测试
- 性能测试
问题排查
- 用户反馈问题
- 系统异常问题
- 数据不一致问题
测试报告
- 测试结果记录
- 问题跟踪
- 改进建议

关键技能要求

技术技能
- 区块链基础知识
- 编程能力（Python/JavaScript）
- 数据库操作
- API测试
业务理解
- 理解入金/出金流程
- 理解风控规则
- 理解对账逻辑
问题解决能力
- 快速定位问题
- 分析问题原因
- 提出解决方案

学习资源

区块链基础: 《精通比特币》、《精通以太坊》
测试方法: 《软件测试的艺术》
工具文档: Web3.py、Etherscan API文档
实践: 在测试环境多练习

注意事项

⚠️ 重要提醒：

测试环境与生产环境隔离
- 永远不要在生产环境随意测试
- 测试数据不要影响生产数据
资金安全
- 测试钱包私钥妥善保管
- 不要将私钥提交到代码仓库
数据备份
- 重要测试前备份数据
- 测试后及时清理测试数据
文档记录
- 详细记录测试步骤
- 记录发现的问题
- 记录解决方案

祝您测试工作顺利！ 🚀

链上测试专题（扫盲专用）

区块链链上测试完整指南

📚 目录

概述

什么是链上测试？

测试工程师的核心职责

测试流程概览

测试环境准备

1. 测试环境分类

1.1 开发环境（Dev）

1.2 测试环境（Test/QA）

1.3 预生产环境（Staging）

1.4 生产环境（Production）

2. 测试账号准备

2.1 创建测试钱包地址

2.2 获取测试币

3. 测试工具准备

3.1 必需工具清单

3.2 Python测试环境

3.3 配置文件示例

4. 测试数据准备

4.1 测试地址库

入金测试

1. 什么是入金？

2. 入金流程

3. 测试场景

3.1 正常入金测试

3.2 小额入金测试

3.3 大额入金测试

3.4 重复入金测试

3.5 多币种入金测试

3.6 入金确认数测试

3.7 异常入金测试

4. 入金测试自动化脚本

出金测试

1. 什么是出金？

2. 出金流程

3. 测试场景

3.1 正常出金测试

3.2 小额出金测试

3.3 大额出金测试

3.4 余额不足测试

3.5 地址白名单测试

3.6 出金手续费测试

3.7 出金失败处理测试

4. 出金测试自动化脚本

多站点测试

1. 什么是多站点？

2. 多站点架构

3. 测试场景

3.1 数据同步测试

3.2 跨站点出金测试

3.3 并发操作测试

3.4 站点故障恢复测试

对账测试

1. 什么是对账？

2. 对账类型

2.1 余额对账

2.2 交易对账

2.3 自动对账脚本

钱包币种测试

1. 什么是钱包币种测试？

2. 测试场景

2.1 币种列表测试

2.2 币种精度测试

2.3 主链币测试

2.4 ERC-20代币测试

2.5 多链币种测试

2.6 币种启用/禁用测试

熔断测试

1. 什么是熔断机制？

2. 熔断场景

2.1 余额不足熔断

2.2 异常交易熔断

2.3 链异常熔断

2.4 手动熔断测试

2.5 熔断状态查询和监控

测试工具和脚本

1. 测试工具类

2. 测试数据生成器