本文章主要会讲解接口自动化测试中Python的数据库断言以及相关的接口关联的测试，除此之外下方有系列文章的传送门，还在持续更新中，感兴趣的小伙伴也可以前往查看，话不多说，让我们一起看看吧~
系列文章： 系列文章【Python自动化测试1】遇见Python之美 系列文章【Python自动化测试2】Python安装配置及PyCharm基本使用 系列文章【Python自动化测试3】初识数据类型与基础语法 系列文章【Python自动化测试4】字符串知识总结 系列文章【Python自动化测试5】列表与元组知识总结 系列文章【Python自动化测试6】字典与集合知识总结 系列文章【Python自动化测试7】数据运算符知识合集 系列文章【Python自动化测试8】流程控制语句讲解 系列文章【Python自动化测试9】函数知识合集 系列文章10：【Python自动化测试10】文件基础操作 系列文章1【Python自动化测试11】模块、包与路径知识合集 系列文章1【Python自动化测试12】异常处理机制知识合集 系列文章1【Python自动化测试13】类、对象、属性与方法知识合集 系列文章1【Python自动化测试14】Python自动化测试基础与进阶练习题 系列文章1【Python自动化测试15】unittest测试框架的核心概念与作用 系列文章1【Python自动化测试16】测试用例数据分离 系列文章1【Python自动化测试17】openpyxl二次封装与数据驱动 系列文章1【Python自动化测试18】配置文件解析与实际应用 系列文章1【Python自动化测试19】日志系统logging讲解 系列文章20：【Python自动化测试20】接口自动化测试框架模型搭建 系列文章2【Python自动化测试21】接口自动化测试实战一_接口概念、项目简介及测试流程问答 系列文章2【Python自动化测试22】接口自动化测试实战二_接口框架修改及用例优化 系列文章2【Python自动化测试23】接口自动化测试实战三_动态参数化与数据伪造 系列文章2【Python自动化测试24】接口自动化测试实战四_Python操作数据库
校验数据库、接口关联及项目优化
自动化测试用例执行后，判断成功的测试用例才进行数据库的校验：
import unittest
import requests
import json
from common.db import DBHandler, db_module
from common.logger import log
from common.excel import read_excel
from common import helper
from config import path_config
from unittestreport import ddt, list_data
from config import setting
# 获取数据
data = read_excel(path_config.case_path , 'XXXXXX')
@ddt
class 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 unittest
import requests
import json
from jsonpath import jsonpath
from decimal import Decimal
from common.db import DBHandler
from common.logger import log
from common.excel import read_excel
from common import helper
from config import path_config
from unittestreport import ddt, list_data
from config import setting
# 获取数据
from middle import api
data = read_excel(path_config.case_path , 'XXXXXX')
@ddt
class 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))
好啦~以上就是本次文章分享的全部内容啦，你学会了吗？希望能给大家带来帮助哦！
文章为作者独立观点，不代表 股票程序化软件自动交易接口观点
相关文章
股民评论