01 恢复测试:系统崩溃后,你能多快“起死回生”?
恢复测试,不是测系统正常时有多稳,而是测系统崩溃后,你能否把数据、服务、业务一条条拉回来。
核心逻辑:模拟灾难,验证恢复能力。
为什么需要恢复测试?
想象一个场景:电商平台支付模块突然宕机。用户订单提交一半,数据库写入中断。如果恢复测试没做,后果是:
- ►数据不一致:部分订单已扣款,但未生成记录
- ►服务中断:用户无法访问,流失率飙升
- ►修复时间长:开发人员手忙脚乱,恢复时间超过SLA(服务等级协议)
恢复测试的目标:在可控时间内,把系统从“死亡”状态拉回“正常”状态。
恢复测试怎么做?
三步走:
1. 模拟故障:人为制造崩溃、断电、网络中断、数据库损坏
2. 执行恢复:按照预先制定的恢复流程操作
3. 验证结果:检查数据完整性、系统功能、性能是否恢复
代码示例:模拟数据库崩溃并恢复
使用Python和MySQL,模拟数据库损坏并恢复:
python
import mysql.connector
from mysql.connector import Error
import time
import subprocess
import os
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">1. </span><span style="font-size:17px;font-weight:700;color:#1565C0;">创建数据库连接</span></div><div style="padding:16px 20px;"><p>def create_connection():</p>
<p>try:</p>
<p>conn = mysql.connector.connect(</p>
<p>host='localhost',</p>
<p>database='test_db',</p>
<p>user='root',</p>
<p>password='password'</p>
<p>)</p>
<p>return conn</p>
<p>except Error as e:</p>
<p>print(f"连接失败: {e}")</p>
<p>return None</p></div></div>
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">2. </span><span style="font-size:17px;font-weight:700;color:#1565C0;">模拟数据库崩溃(删除关键表)</span></div><div style="padding:16px 20px;"><p>def simulate_crash():</p>
<p>conn = create_connection()</p>
<p>if conn:</p>
<p>cursor = conn.cursor()</p>
<p>try:</p></div></div>
# 模拟误操作:删除orders表
cursor.execute("DROP TABLE IF EXISTS orders")
conn.commit()
print("模拟崩溃:orders表已删除")
except Error as e:
print(f"模拟崩溃失败: {e}")
finally:
cursor.close()
conn.close()
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">3. </span><span style="font-size:17px;font-weight:700;color:#1565C0;">执行恢复流程(从备份恢复)</span></div><div style="padding:16px 20px;"><p>def execute_recovery():</p>
<p>print("开始恢复流程...")</p></div></div>
# 假设有备份文件 orders_backup.sql
backup_file = "orders_backup.sql"
if os.path.exists(backup_file):
# 使用mysql命令行恢复
cmd = f"mysql -u root -ppassword test_db < {backup_file}"
subprocess.run(cmd, shell=True)
print("恢复完成:orders表已从备份恢复")
return True
else:
print("备份文件不存在,恢复失败")
return False
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">4. </span><span style="font-size:17px;font-weight:700;color:#1565C0;">验证恢复结果</span></div><div style="padding:16px 20px;"><p>def verify_recovery():</p>
<p>conn = create_connection()</p>
<p>if conn:</p>
<p>cursor = conn.cursor()</p>
<p>cursor.execute("SELECT COUNT(*) FROM orders")</p>
<p>count = cursor.fetchone()[0]</p>
<p>print(f"验证结果:orders表有 {count} 条记录")</p>
<p>cursor.close()</p>
<p>conn.close()</p></div></div>
# 执行恢复测试
if __name__ == "__main__":
# 模拟崩溃
simulate_crash()
time.sleep(2)
# 执行恢复
if execute_recovery():
time.sleep(2)
# 验证恢复
verify_recovery()
else:
print("恢复测试失败")
金句:恢复测试不是测系统有多强,而是测你有多快承认失败并重新站起来。
恢复测试的关键指标
- ►恢复时间目标(RTO):从崩溃到恢复服务,最多能容忍多久?
- ►恢复点目标(RPO):能容忍丢失多少数据?(比如最多丢10分钟数据)
如果RTO是30分钟,RPO是5分钟,恢复测试必须证明:崩溃后30分钟内恢复,且数据丢失不超过5分钟。
---
02 容错测试:系统不完美,但必须“带伤作战”
容错测试和恢复测试是孪生兄弟,但分工不同:
- ►恢复测试:系统死了,怎么救活?
- ►容错测试:系统受伤了,怎么继续跑?
核心逻辑:模拟部分组件故障,验证系统能否继续提供服务。
容错测试的三种模式
1. 故障屏蔽:某个组件出问题,系统自动切换到备用组件(如主数据库宕机,切换到从数据库)
2. 降级服务:部分功能不可用,但核心功能继续运行(如推荐系统挂了,搜索功能照常)
3. 优雅降级:系统性能下降,但不会完全崩溃(如响应时间从100ms变成500ms,但还能用)
代码示例:模拟Redis缓存故障并降级
假设一个电商系统,用Redis缓存商品信息。当Redis宕机时,系统应该降级到直接查数据库:
python
import redis
from flask import Flask, jsonify
import mysql.connector
import time
app = Flask(__name__)
# 初始化Redis连接
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 数据库连接
def get_db_connection():
return mysql.connector.connect(
host='localhost',
database='ecommerce',
user='root',
password='password'
)
# 获取商品信息(带容错)
def get_product_info(product_id):
try:
# 尝试从Redis获取
product_json = redis_client.get(f"product:{product_id}")
if product_json:
print("从Redis获取数据")
return jsonify({"source": "cache", "data": product_json})
except redis.exceptions.ConnectionError:
print("Redis连接失败,降级到数据库")
# 容错:Redis挂了,直接查数据库
pass
# 降级方案:查数据库
try:
conn = get_db_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT * FROM products WHERE id = %s", (product_id,))
product = cursor.fetchone()
cursor.close()
conn.close()
if product:
print("从数据库获取数据(降级模式)")
return jsonify({"source": "database", "data": product})
else:
return jsonify({"error": "商品不存在"}), 404
except mysql.connector.Error as e:
print(f"数据库查询失败: {e}")
return jsonify({"error": "服务暂时不可用"}), 503
@app.route('/product/<int:product_id>')
def get_product(product_id):
return get_product_info(product_id)
# 模拟Redis故障测试
def test_redis_fault_tolerance():
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">1. </span><span style="font-size:17px;font-weight:700;color:#1565C0;">正常情况:Redis可用</span></div><div style="padding:16px 20px;"><p>print("=== 测试1:Redis正常 ===")</p></div></div>
# 先往Redis写入数据
redis_client.set("product:1", '{"id": 1, "name": "iPhone 15", "price": 7999}')
# 模拟请求
response = get_product_info(1)
assert "cache" in response.get_json()["source"], "应使用缓存"
print("Redis正常:从缓存获取数据")
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">2. </span><span style="font-size:17px;font-weight:700;color:#1565C0;">模拟Redis宕机</span></div><div style="padding:16px 20px;"><p>print("=== 测试2:Redis宕机 ===")</p></div></div>
# 关闭Redis服务(这里假设用外部命令模拟)
import subprocess
subprocess.run(["redis-cli", "SHUTDOWN"])
time.sleep(1)
# 再次请求
response = get_product_info(1)
assert "database" in response.get_json()["source"], "应降级到数据库"
print("Redis宕机:成功降级到数据库")
<div style="margin:24px 0;border:1px solid #E3F2FD;border-radius:8px;background:#FAFAFA;overflow:hidden;"><div style="display:flex;align-items:center;gap:12px;padding:12px 20px;background:linear-gradient(135deg,#E3F2FD,#BBDEFB);"><span style="width:36px;height:36px;background:#2196F3;color:#fff;border-radius:50%;display:inline-flex;align-items:center;justify-content:center;font-weight:bold;font-size:16px;flex-shrink:0;">3. </span><span style="font-size:17px;font-weight:700;color:#1565C0;">恢复Redis</span></div><div style="padding:16px 20px;"><p>print("=== 测试3:Redis恢复 ===")</p>
<p>subprocess.run(["redis-server", "--daemonize", "yes"])</p>
<p>time.sleep(2)</p></div></div>
# 再次请求
response = get_product_info(1)
assert "cache" in response.get_json()["source"], "恢复后应使用缓存"
print("Redis恢复:重新使用缓存")
if __name__ == "__main__":
test_redis_fault_tolerance()
金句:容错测试的终极目标:系统可以慢,但不能停;可以降级,但不能崩。
容错测试的常见故障模拟
- ►网络故障:断网、丢包、延迟增加
- ►服务故障:某个微服务宕机、数据库连接池满
- ►资源耗尽:CPU 100%、内存溢出、磁盘满
- ►数据异常:数据格式错误、重复数据、空值
---
03 恢复测试 vs 容错测试:双剑合璧才是真测试
很多人混淆恢复测试和容错测试。简单区分:
- ►恢复测试:系统死了,怎么救?重点在恢复流程和数据完整性
- ►容错测试:系统受伤了,怎么继续跑?重点在降级策略和故障隔离
对比表
| 维度 | 恢复测试 | 容错测试 |
| 目标 | 验证恢复能力 | 验证容错能力 |
| 触发条件 | 系统完全崩溃 | 部分组件故障 |
| 测试方式 | 模拟灾难→执行恢复→验证 | 模拟故障→验证降级→恢复 |
| 关键指标 | RTO、RPO | 降级成功率、故障隔离性 |
| 典型场景 | 数据库损坏、服务器宕机 | 缓存故障、网络丢包 |
实战案例:电商系统双十一测试
场景:一个电商平台,日订单量100万。双十一前必须通过恢复测试和容错测试。
恢复测试步骤:
1. 模拟支付服务数据库崩溃
2. 执行恢复流程:从备份恢复数据库
3. 验证:恢复后数据是否完整?订单状态是否一致?
4. 目标:RTO=15分钟,RPO=1分钟
容错测试步骤:
1. 模拟推荐服务宕机
2. 验证:商品详情页是否正常显示?搜索功能是否可用?
3. 验证:系统是否自动降级到静态推荐(如热门商品列表)
4. 目标:推荐功能降级,但核心购买流程不受影响
代码示例:综合恢复与容错测试
python
import time
import random
class SystemTester:
def __init__(self):
self.recovery_time = 0
self.fault_tolerance_count = 0
def test_recovery(self):
"""
恢复测试:模拟数据库崩溃并恢复
"""
print("开始恢复测试...")
# 模拟崩溃
print("1. 模拟数据库崩溃")
time.sleep(1)
# 执行恢复
start_time = time.time()
print("2. 执行恢复流程(从备份恢复)...")
time.sleep(2) # 模拟恢复时间
self.recovery_time = time.time() - start_time
# 验证恢复
print(f"3. 恢复完成,耗时 {self.recovery_time:.2f} 秒")
assert self.recovery_time < 30, "恢复时间超过RTO 30秒"
print("恢复测试通过")
def test_fault_tolerance(self):
"""
容错测试:模拟Redis故障并验证降级
"""
print("\n开始容错测试...")
# 模拟故障
print("1. 模拟Redis缓存故障")
time.sleep(1)
# 验证降级
for i in range(5):
# 模拟降级成功
if random.random() > 0.1: # 90%成功率
self.fault_tolerance_count += 1
print(f" 请求{i+1}: 降级成功(从数据库获取)")
else:
print(f" 请求{i+1}: 降级失败")
# 验证容错率
success_rate = self.fault_tolerance_count / 5
print(f"容错成功率: {success_rate * 100:.0f}%")
assert success_rate >= 0.8, "容错成功率低于80%"
print("容错测试通过")
def run_all_tests(self):
print("=== 系统可靠性测试套件 ===")
self.test_recovery()
self.test_fault_tolerance()
print("\n=== 所有测试通过 ===")
print(f"恢复时间: {self.recovery_time:.2f}秒(目标<30秒)")
print(f"容错成功率: {self.fault_tolerance_count/5*100:.0f}%(目标>80%)")
# 执行测试
tester = SystemTester()
tester.run_all_tests()
金句:恢复测试保底线,容错测试保上限。两者缺一,系统就是纸糊的。
常见的误区
- ►误区1:只做功能测试,不做恢复容错测试。→ 结果:系统上线后崩溃,无人能救
- ►误区2:恢复测试和容错测试是一回事。→ 结果:只测了恢复没测容错,系统降级时直接崩
- ►误区3:测试环境做一次就够了。→ 结果:生产环境配置不同,恢复流程执行失败
---
总结:测试不只是找bug,更是给系统买保险
恢复测试和容错测试,是软件测试中最容易被忽略,但最重要的部分。它们不测功能是否正常,而是测系统面对灾难时的生存能力。
行动清单:
1. 下周内,为你的核心系统制定恢复测试计划
2. 模拟一次数据库崩溃,验证恢复流程
3. 模拟一次缓存故障,验证降级策略
4. 记录RTO和RPO,确保在可接受范围内
5. 定期重复测试(每月至少一次)
记住:系统可以不完美,但必须能“死而复生”,能“带伤作战”。这是测试人员对用户最后的承诺。
终极金句:恢复测试和容错测试,不是测试的锦上添花,而是系统生存的底线。