自动化测试框架开发pytest数据库备份和恢复

景点排名 2025年11月05日 22:13 0 admin

在做接口自动化测试的时候，我们一般都需要使用对数据库的数据进行备份和恢复，主要是为了避免在数据库遗留大量的脏数据，保持数据库数据的干净！#python##pytest##自动化测试#

使用场景：

假设我们有一个测试项目，需要在测试前将数据库恢复到一个已知的初始状态（例如，包含一些基础数据），测试过程中可能会修改数据库，测试结束后再恢复回测试前的状态，以免影响后续的测试。

步骤：

1，在测试会话开始时，根据配置决定是否执行备份和恢复初始数据：

2，从配置文件中读取数据库配置（根据cmdopt获取的环境，例如'test'）

3，备份当前数据库到db_bak_path（如果配置中backup为True）

4，然后恢复到一个初始状态的数据库（db_re_path，这个文件应该是一个已知的包含测试数据的sql文件）

5，执行所有测试。

6，在测试会话结束后，根据配置决定是否恢复数据库到备份的状态：

7，如果配置中recovery为True，则用之前备份的sql文件恢复数据库。

这样，每次测试会话都会在一个干净的数据库环境中开始，并且测试结束后会恢复之前的状态，避免测试数据残留。

项目结构如下：

project/├── conftest.py          # 包含上述两个函数├── test_example.py      # 测试用例文件└── pytest.ini           # pytest配置文件（可选）

假设我们有一个配置文件，例如config.yaml，其中包含：

database:      test:          data: "test_database_connection_string"          Operations_db:                backup: True                recovery: True                       prod:          data: "prod_database_connection_string"          Operations_db:                backup: True                recovery: True

具体的fixture文件如下：

import pytestdef pytest_addoption(parser):    parser.addoption("--env", action="store", default="test", help=None)@pytest.fixture(scope='session', autouse=True)def cmdopt(pytestconfig):    global cmdopt_env    cmdopt_env = pytestconfig.getoption("--env")    return cmdopt_env@pytest.fixture(scope="session", autouse=True)def bac_rec(cmdopt):    db_data = dict(ReadFile.read_config('$.database.%s'%cmdopt))    BR = BackupRecovery(db_data['data'])    db_bak_path = './config/ry-vue_bak.sql'  # 当前数据库备份文件    db_re_path = './config/ry-vue_re.sql'  # 初始化好测试数据的数据库sql文件    backup = ReadFile.read_config(f'$.database.{cmdopt}.Operations_db.backup')    if backup:        print("开始备份数据库...")        BR.backup(db_bak_path)        print("备份完成，开始恢复初始测试数据库...")        BR.recovery(db_re_path)        print("恢复初始测试数据库完成")    else:        print("跳过备份")    yield    recovery = ReadFile.read_config(f'$.database.{cmdopt}.Operations_db.recovery')    if recovery:        print("测试结束，开始恢复备份数据库...")        BR.recovery(db_bak_path)        print("恢复备份数据库完成")    else:        print("跳过恢复")

当运行pytest -v test_example.py --env test的时候：

首先会通过--env test获取到环境参数为test

然后再从config.yaml里面获取到test环境字段里面的内容，比如Operations_db的backup: True和recovery: True

最后在将Operations_db的backup: True和recovery: True传给bac_rec(cmdopt)函数使用

运行结果如下：

开始备份数据库...备份完成，开始恢复初始测试数据库...恢复初始测试数据库完成...（测试输出）...测试结束，开始恢复备份数据库...恢复备份数据库完成

面向对象封装如下：

# config.yamldatabase:  test:    host: "localhost"    user: "test_user"    password: "test_password"    database: "test_db"    port: 3306Operations_db:  backup: true  recovery: true   class BackupRecovery:    """数据库备份恢复工具类"""        def __init__(self, db_config):        self.db_config = db_config        def backup(self, backup_path):        """备份数据库"""        import subprocess        import os                # 构建备份命令        cmd = [            'mysqldump',            f'-h{self.db_config["host"]}',            f'-u{self.db_config["user"]}',            f'-p{self.db_config["password"]}',            self.db_config["database"]        ]                # 执行备份        with open(backup_path, 'w') as f:            result = subprocess.run(cmd, stdout=f, stderr=subprocess.PIPE, text=True)                if result.returncode != 0:            raise Exception(f"备份失败: {result.stderr}")                print(f"数据库备份成功: {backup_path}")        def recovery(self, recovery_path):        """恢复数据库"""        import subprocess                # 构建恢复命令        cmd = [            'mysql',            f'-h{self.db_config["host"]}',            f'-u{self.db_config["user"]}',            f'-p{self.db_config["password"]}',            self.db_config["database"]        ]                # 执行恢复        with open(recovery_path, 'r') as f:            result = subprocess.run(cmd, stdin=f, stderr=subprocess.PIPE, text=True)                if result.returncode != 0:            raise Exception(f"恢复失败: {result.stderr}")                print(f"数据库恢复成功: {recovery_path}")