一、前言

本文章主要会讲解接口自动化测试中Python如何操作数据库、为何要操作数据库，有哪些利弊，以及数据库断言、相关的接口关联的测试，废话我就不都说了，咱们直接进入主题哟。

二、自动化数据库理论与操作

2.1 接口自动化为何要操作数据库

接口自动化中操作数据库主要是根据业务层面决定的，部分情况例如查询手机号、或个人信息时需要操作数据库，有时候也有可能需要删除某个内容，通常而言不会这么做罢了。

2.2 接口自动化操作数据库的利弊

"""利：1、能够根据我们需要的业务情况来解决一定的业务问题弊：1、数据库的操作本身就会让自动化变得慢，需要建立连接 -- 查询 等等2、数据库本身是一种依赖， 我们能不操作数据库尽可能不操作"""

2.3 Python操作数据库

要操作数据库，需要先进行pymysql库的安装，按照对应语法填写好对应数据即可：

import pymysql"""1、连接数据库：输入用户名，密码，地址，端口2、游标：使用游标读取数据、修改数据（执行sql语句）3、获取操作结果4、关闭游标5、关闭数据库连接"""conn = pymysql.connect(user="future",                password="XXXXXX",                host="XX.XX.XX.XX",                port=3306)cursor = conn.cursor()# 执行sqlsql = "SELECT mobile_phone,id,reg_name FROM XXXXXXX.member WHERE mobile_phone = 137XXXXXXXX"cursor.execute(sql)# 获取一个结果每条记录用元组表示res = cursor.fetchone()print(res)# 关闭cursor.close()conn.close()

输出的数据是元组，元组数据不可修改，我们需要进行类型转换，如要输出字典，需要加上DictCursor：

import pymysqlfrom pymysql.cursors import DictCursor"""1、连接数据库：输入用户名，密码，地址，端口2、游标：使用游标读取数据、修改数据（执行sql语句）3、获取操作结果4、关闭游标5、关闭数据库连接"""conn = pymysql.connect(user="future",                password="XXXXXX",                host="XX.XX.XX.XX",                port=3306)cursor = conn.cursor(DictCursor)# 执行sqlsql = "SELECT mobile_phone,id,reg_name FROM XXXXXXX.member WHERE mobile_phone = 137XXXXXXXX"cursor.execute(sql)# 获取一个结果每条记录用元组表示res = cursor.fetchone()print(res)# 关闭cursor.close()conn.close()

2.4 操作数据库封装

我们需要进行函数封装，数据库内容也属于公用内容，也可以放入至common包下，函数封装后：

class DBHandler:    def __init__(self, host=setting.db_host, port=setting.db_port,                 user=setting.db_user, password=setting.db_pwd):        self.conn = pymysql.connect(user=user,                                    password=password,                                    host=host,                                    port=port,                                    autocommit=True                                    )    def query_one(self, sql, cursor_type=DictCursor):        cursor = self.conn.cursor(cursor_type)        cursor.execute(sql)        data = cursor.fetchone()        cursor.close()        return data    def query_all(self, sql, cursor_type=DictCursor):        cursor = self.conn.cursor(cursor_type)        cursor.execute(sql)        data = cursor.fetchall()        cursor.close()        return data    def query(self, sql, one=True, cursor_type=DictCursor):        if one:            return self.query_one(sql, cursor_type)        return self.query_all(sql, cursor_type)    def close(self):        self.conn.close()    # 自己实现上下文管理器    # def __enter__(self):    #     return self    #    # def __exit__(self, exc_type, exc_val, exc_tb):    #     return self.close()db_module = DBHandler()if __name__ == '__main__':    db = DBHandler()    sql = 'select mobile_phone from futureloan.member limit 5'    res = db.query(sql, one=False)    db.close()    print(res)    #    # with DBHandler_2() as db:    #     sql = 'select mobile_phone from futureloan.member limit 5'    #     res = db.query(sql, one=False)    #     print(res)

# 数据库配置项db_host = "XX.XX.XX.XX"db_port = XXXXdb_user = "future"db_pwd = "XXXXXX"

自动化测试用例执行后，判断成功的测试用例才进行数据库的校验：

import unittestimport requestsimport jsonfrom common.db import DBHandler, db_modulefrom common.logger import logfrom common.excel import read_excelfrom common import helperfrom config import path_configfrom unittestreport import ddt, list_datafrom config import setting# 获取数据data = read_excel(path_config.case_path , 'XXXXXX')@ddtclass TestRegister(unittest.TestCase):    @classmethod    def setUpClass(cls) -> None:        """测试类的前置"""        cls.db = DBHandler()    @classmethod    def tearDownClass(cls) -> None:        """测试类的前置"""        cls.db.close()    @list_data(data)    def test_register(self, case_data):        print(case_data)        json_data = case_data['json']        if '#new_phone#' in json_data:            new_phone = helper.generate_new_phone()            json_data = json_data.replace('#new_phone#', new_phone)        # 把json格式的字符串转化成字典        json_data = json.loads(json_data)        headers = json.loads(case_data['headers'])        print("替换之后", json_data)        resp = requests.request(            method=case_data['method'],            url= setting.host +  case_data['url'],            json=json_data,            headers=headers        )        actual = resp.json()        print(actual)        try:            self.assertEqual(case_data['expected'], actual['code'])        except AssertionError as e:            raise e"""1、判断是否是注册成功的测试用例2、查询数据库是否包含了手机号的记录3、判断数据库记录条数是否为1"""        if actual['msg'] == "OK":            # 通过 setUp 创建数据库连接            sql = f"select id from XXXX.XXXX where XXXXXX = {json_data['XXXXXX']}"            result = self.db.query_all(sql)            self.assertEqual(1, len(result))

往往项目中的接口都会存在接口关联，而我们就需要对接口进行单独处理：

import unittestimport requestsimport jsonfrom jsonpath import jsonpathfrom decimal import Decimalfrom common.db import DBHandlerfrom common.logger import logfrom common.excel import read_excelfrom common import helperfrom config import path_configfrom unittestreport import ddt, list_datafrom config import setting# 获取数据from middle import apidata = read_excel(path_config.case_path , 'XXXXXX')@ddtclass TestRecharge(unittest.TestCase):    @classmethod    def setUpClass(cls) -> None:        """        1,访问登录接口，得到返回值        2,从返回值当中提取数据：resp["data"]["id"]        3, 从返回值当中提取数据：resp["data"]["token_info"]["token"]        4， 设置成类属性，方便测试用例函数当中调用        :return:        """        resp_data = api.login()        # 提取方式1：        cls.member_id = resp_data["data"]["id"]        cls.token = resp_data["data"]["token_info"]["token"]        # 提取方式2：jsonpath        cls.member_id = jsonpath(resp_data, '$..id')[0]        cls.token = jsonpath(resp_data, '$..token')[0]        cls.before = jsonpath(resp_data, '$..leave_amount')[0]        # 初始化数据库        cls.db = DBHandler()    @classmethod    def tearDownClass(cls) -> None:        pass    @list_data(data)    def test_recharge(self, case_data):        """        1, 获取 case_data 当中的数据，headers, json 是要重点关注的。        2， 数据的预处理：数据替换，数据转化成 字典格式：headers, json, expected        3, 发送请求        4， 断言        :param case_data:        :return:        """        # 获取 case_data当中的数据，headers, json是要重点关注的。        headers_string = case_data['headers']        json_string = case_data['json']        # 数据的预处理：数据替换，数据转化成字典格式：headers, json, expected        if "#token#" in headers_string:            headers_string = headers_string.replace("#token#", self.token)        if "#member_id#" in json_string:            json_string = json_string.replace("#member_id#", str(self.member_id))        headers = json.loads(headers_string)        json_data = json.loads(json_string)        expected = json.loads(case_data['expected'])        # 获取充值之前数据库当中用户的余额        sql = f"select XXXXXXX from XXXX.XXXXX where id={self.XXXXX_id}"        before = self.db.query_one(sql)        # {"leave_amount": 200}        # 3, 发送请求        resp = requests.request(            method=case_data['method'],            url=setting.host + case_data['url'],            json=json_data,            headers=headers        )        # 4, 断言        json_response = resp.json()        try:            for key, value in expected.items():                self.assertEqual(value, json_response[key])        except AssertionError as e:            log.error(f"测试用例失败：{e}")            raise e        if json_response['msg'] == 'OK':            print(json_response)            # 通过数据库断言            sql = f"select leave_amount from futureloan.member where id={self.member_id}"            after = self.db.query_one(sql)            money = str(json_data['amount'])            after = str(after['leave_amount'])            before = str(before['leave_amount'])            self.assertEqual(Decimal(money), Decimal(after) - Decimal(before))

关于接口自动化测试中Python如何操作数据库、为何要操作数据库，有哪些利弊，以及数据库断言、相关的接口关联的测试，笔者就说到这里了，喜欢的小伙伴可以收藏点赞评论，加关注哟，关注我每天给你不同的惊喜。