本文目录导读：
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))
使用注意事项：
- 测试环境隔离 - 使用专门的测试数据库和账号
- 金额控制 - 避免真实支付，使用测试支付环境
- 数据清理 - 测试后清理测试数据
- 日志记录 - 记录所有测试操作
- 安全控制 - 避免影响生产环境
需要根据你的具体系统架构和需求选择合适的方案。
