怎么用脚本自动充值测试

wen 实用脚本 2026-08-01 202

本文目录导读：

怎么用脚本自动充值测试

API接口测试脚本（最常用）
数据库直接操作脚本
自动化测试框架
浏览器自动化测试
建议的测试流程
使用注意事项：

API接口测试脚本（最常用）

Python + Requests

import requests
import json
import time
import random
class AutoRechargeTest:
    def __init__(self):
        self.base_url = "http://your-api-server.com"
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": "Bearer your_test_token"
        }
    def recharge(self, user_id, amount, payment_method="test_card"):
        """模拟充值请求"""
        payload = {
            "user_id": user_id,
            "amount": amount,
            "payment_method": payment_method,
            "order_id": f"test_{int(time.time())}_{random.randint(1000, 9999)}"
        }
        try:
            response = requests.post(
                f"{self.base_url}/api/v1/recharge",
                headers=self.headers,
                json=payload,
                timeout=10
            )
            return {
                "status_code": response.status_code,
                "response": response.json() if response.status_code == 200 else response.text
            }
        except Exception as e:
            return {"error": str(e)}
    def run_tests(self, cases):
        """批量测试"""
        results = []
        for case in cases:
            result = self.recharge(**case)
            results.append(result)
            print(f"订单 {case.get('order_id', 'N/A')}: {result}")
            # 添加延时，避免请求过于频繁
            time.sleep(1)
        return results
# 测试用例
test_cases = [
    {"user_id": "test_user_01", "amount": 100},
    {"user_id": "test_user_02", "amount": 500},
    {"user_id": "test_user_03", "amount": 1000},
    {"user_id": "test_user_04", "amount": 0},  # 边界测试
    {"user_id": "test_user_05", "amount": -100},  # 异常测试
]
# 执行测试
tester = AutoRechargeTest()
results = tester.run_tests(test_cases)

数据库直接操作脚本

MySQL/SQLite脚本

import sqlite3
import hashlib
import uuid
from datetime import datetime
class DatabaseRecharge:
    def __init__(self, db_path="test.db"):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
        self.setup_tables()
    def setup_tables(self):
        # 创建测试表
        self.cursor.execute("""
            CREATE TABLE IF NOT EXISTS wallet_balance (
                user_id TEXT PRIMARY KEY,
                balance DECIMAL(10,2) DEFAULT 0,
                updated_at DATETIME
            )
        """)
        self.cursor.execute("""
            CREATE TABLE IF NOT EXISTS transaction_records (
                transaction_id TEXT PRIMARY KEY,
                user_id TEXT,
                amount DECIMAL(10,2),
                type TEXT,
                status TEXT,
                created_at DATETIME
            )
        """)
        self.conn.commit()
    def direct_recharge(self, user_id, amount):
        """直接修改数据库余额"""
        transaction_id = str(uuid.uuid4())
        now = datetime.now().isoformat()
        try:
            # 更新余额
            self.cursor.execute("""
                INSERT INTO wallet_balance (user_id, balance, updated_at)
                VALUES (?, ?, ?)
                ON CONFLICT(user_id) 
                DO UPDATE SET balance = balance + ?, updated_at = ?
            """, (user_id, amount, now, amount, now))
            # 记录交易
            self.cursor.execute("""
                INSERT INTO transaction_records 
                (transaction_id, user_id, amount, type, status, created_at)
                VALUES (?, ?, ?, 'test_recharge', 'success', ?)
            """, (transaction_id, user_id, amount, now))
            self.conn.commit()
            return {"success": True, "transaction_id": transaction_id, "amount": amount}
        except Exception as e:
            self.conn.rollback()
            return {"success": False, "error": str(e)}
    def batch_recharge(self, user_amounts):
        """批量充值"""
        results = []
        for user_id, amount in user_amounts.items():
            result = self.direct_recharge(user_id, amount)
            results.append({"user": user_id, "result": result})
        return results
    def get_balance(self, user_id):
        self.cursor.execute(
            "SELECT balance FROM wallet_balance WHERE user_id = ?", 
            (user_id,)
        )
        result = self.cursor.fetchone()
        return result[0] if result else 0
# 使用示例
db_recharge = DatabaseRecharge("test.db")
# 批量充值
test_users = {
    "test_user_001": 100.00,
    "test_user_002": 250.50,
    "test_user_003": 999.99,
}
results = db_recharge.batch_recharge(test_users)
print(f"批量充值结果: {results}")
# 验证余额
for user_id in test_users:
    balance = db_recharge.get_balance(user_id)
    print(f"{user_id} 余额: {balance}")

自动化测试框架

Pytest测试脚本

import pytest
import requests
import random
import time
class TestAutoRecharge:
    BASE_URL = "http://your-api-server.com"
    @pytest.fixture
    def api_client(self):
        session = requests.Session()
        session.headers.update({
            "Content-Type": "application/json",
            "Authorization": "Bearer test_token_123"
        })
        return session
    @pytest.mark.parametrize("amount,tax_rate", [
        (100, 0.1),
        (500, 0.15),
        (1000, 0.2),
        (10000, 0.25)
    ])
    def test_recharge_amounts(self, api_client, amount, tax_rate):
        """测试不同金额的充值"""
        order_id = f"test_{int(time.time())}_{random.randint(1000,9999)}"
        response = api_client.post(
            f"{self.BASE_URL}/api/v1/recharge",
            json={
                "order_id": order_id,
                "amount": amount,
                "tax_rate": tax_rate
            }
        )
        assert response.status_code == 200
        data = response.json()
        assert data["status"] == "success"
        assert data["amount"] == amount
        assert data["actual_amount"] == amount * (1 - tax_rate)
    def test_concurrent_recharge(self, api_client):
        """并发充值测试"""
        import threading
        results = []
        def recharge_user(user_id):
            for i in range(10):
                response = api_client.post(
                    f"{self.BASE_URL}/api/v1/recharge",
                    json={"user_id": user_id, "amount": 10}
                )
                results.append(response.status_code)
        threads = []
        for user_id in range(50):
            t = threading.Thread(target=recharge_user, args=(user_id,))
            threads.append(t)
            t.start()
        for t in threads:
            t.join()
        # 验证所有请求都成功
        assert all(code == 200 for code in results)

浏览器自动化测试

Selenium模拟操作

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time
import random
class BrowserAutoRecharge:
    def __init__(self):
        self.driver = webdriver.Chrome()  # 或使用其他浏览器驱动
        self.wait = WebDriverWait(self.driver, 10)
    def setup_test_environment(self):
        """设置测试环境"""
        self.driver.get("http://your-testing-site.com")
        # 登录测试账号
        self.login_test_account()
    def login_test_account(self):
        """登录测试账号"""
        username_input = self.wait.until(
            EC.presence_of_element_located((By.NAME, "username"))
        )
        username_input.send_keys("test_user")
        password_input = self.driver.find_element(By.NAME, "password")
        password_input.send_keys("test_password")
        login_button = self.driver.find_element(By.ID, "login-btn")
        login_button.click()
        time.sleep(2)  # 等待登录完成
    def auto_recharge(self, times=10):
        """自动充值测试"""
        for i in range(times):
            print(f"开始第 {i+1} 次充值测试")
            # 进入充值页面
            self.driver.find_element(By.ID, "recharge-menu").click()
            # 填写充值金额
            amount_input = self.driver.find_element(By.ID, "amount-input")
            amount = random.randint(10, 1000)
            amount_input.clear()
            amount_input.send_keys(str(amount))
            # 选择支付方式
            payment_select = self.driver.find_element(By.ID, "payment-method")
            payment_select.click()
            payment_option = self.driver.find_element(
                By.XPATH, "//option[contains(text(), '测试支付')]"
            )
            payment_option.click()
            # 提交充值请求
            submit_btn = self.driver.find_element(By.ID, "submit-recharge")
            submit_btn.click()
            # 等待充值完成
            time.sleep(2)
            print(f"充值 {amount} 元完成")

建议的测试流程

测试计划示例

class CompleteRechargeTest:
    def __init__(self):
        self.test_log = {
            "api_tests": [],
            "db_tests": [],
            "balance_tests": []
        }
    def run_complete_test(self):
        """完整测试流程"""
        # 1. 测试正常充值
        self.test_normal_recharge()
        # 2. 测试边界情况
        self.test_boundary_recharge()
        # 3. 测试异常情况
        self.test_error_recharge()
        # 4. 测试并发情况
        self.test_concurrent_recharge()
        # 5. 验证数据库更新
        self.verify_database()
        # 6. 生成测试报告
        self.generate_report()
    def generate_report(self):
        """生成测试报告"""
        report = {
            "total_tests": len(self.test_log["api_tests"]),
            "successful_tests": sum(1 for t in self.test_log["api_tests"] if t["status"] == "success"),
            "failed_tests": sum(1 for t in self.test_log["api_tests"] if t["status"] != "success"),
            "test_log": self.test_log
        }
        print(json.dumps(report, ensure_ascii=False, indent=2))

使用注意事项：

测试环境隔离 - 使用专门的测试数据库和账号
金额控制 - 避免真实支付，使用测试支付环境
数据清理 - 测试后清理测试数据
日志记录 - 记录所有测试操作
安全控制 - 避免影响生产环境

需要根据你的具体系统架构和需求选择合适的方案。

本文地址： https://www.bmjup.com/post/35225.html