python接口自动化测试框架链接数据库_python+requests接口自动化测试框架
框架思路：common文件夹存放公共方法，data文件夹存放测试用例参数表格，report文件夹存放执行测试用例导出的测试报告，testcases文件夹存放测试用例执行文件，testsuite.py用于将需要执行的测试用例放到一个测试套件中批量执行
封装公共方法：common.common_method.py
1、初始化公共方法，传入环境参数，测试环境传参test，生产环境传参production
2、获取各个环境对应的url，方便后续拼接完整请求地址
3、封装request.post、get、patch、delete、put请求方法，通过传入请求地址和请求参数，返回请求结果，请求头部写死，每次执行用例，请求头部需要手动修改
4、封装总的请求API方法，通过传参方法名称(post、get、patch、delete、put)、请求地址和请求参数，调用具体的请求方法
import requests
import json
class CommonMethod():
# 初始化公共方法，传入环境参数
def __init__(self, env):
self.env = env
# 根据不同的环境取不同的base_url
def get_base_url(self):
if self.env == "test":
self.base_url = "http://*.*.*.*"
elif self.env == "production":
self.base_url = "http://*.*.*.*"
else:
print("传参错误，测试环境：test，生产环境：production")
return self.base_url
# 封装请求头部
def get_headers_with_content_type_1(self):
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Cookie": "XSRF-TOKEN=Z373OTKQhAoErRzmi6uc7qOdbh1TGeEzB7nMWwtT; laravel_session=W97NLF7x1q8R8oAFdlmi3CpdOuA0dFsdvVMOX2mr",
"X-CSRF-TOKEN": "Z373OTKQhAoErRzmi6uc7qOdbh1TGeEzB7nMWwtT",
"X-Requested-With": "XMLHttpRequest",
"X-XSRF-TOKEN": "Z373OTKQhAoErRzmi6uc7qOdbh1TGeEzB7nMWwtT"
}
return headers
def get_headers_without_content_type(self):
headers = {
"Cookie": "XSRF-TOKEN=Z373OTKQhAoErRzmi6uc7qOdbh1TGeEzB7nMWwtT; laravel_session=W97NLF7x1q8R8oAFdlmi3CpdOuA0dFsdvVMOX2mr",
"X-CSRF-TOKEN": "Z373OTKQhAoErRzmi6uc7qOdbh1TGeEzB7nMWwtT",
"X-Requested-With": "XMLHttpRequest",
"X-XSRF-TOKEN": "Z373OTKQhAoErRzmi6uc7qOdbh1TGeEzB7nMWwtT"
}
return headers
# 封装post请求方法，传入需要的参数url和data
def send_post(self, url, data):
headers = self.get_headers_with_content_type_1()
result = requests.post(url=url, data=data, headers=headers).json()
# 将dict类型结果result转换成json格式
# res = json.dumps(result, ensure_ascii=False, sort_keys=True, indent=2)
return result
# 封装get请求方法，传入需要的参数url和data
def send_get(self, url, data):
headers = self.get_headers_without_content_type()
result = requests.get(url=url, params=data, headers=headers).json()
return result
# 封装put请求方法，传入需要的参数url和data
def send_put(self, url, data):
headers = self.get_headers_with_content_type_1()
result = requests.put(url=url, params=data, headers=headers).json()
return result
# 封装patch请求方法，传入需要的参数url和data
def send_patch(self, url, data):
headers = self.get_headers_with_content_type_1()
result = requests.patch(url=url, params=data, headers=headers).json()
return result
# 封装delete请求方法，传入需要的参数url和data
def send_delete(self, url, data):
headers = self.get_headers_with_content_type_1()
result = requests.delete(url=url, params=data, headers=headers).json()
return result
# 定义一个method_main函数，通过传过来的method来进行不同的get或post请求
def method_main(self, method, url=None, data=None):
result = None
if method == 'post':
result = self.send_post(url, data)
elif method == 'get':
result = self.send_get(url, data)
elif method == 'put':
result = self.send_put(url, data)
elif method == 'patch':
result = self.send_post(url, data)
elif method == 'delete':
result = self.send_delete(url, data)
else:
print("method值错误！！！")
return result
获取测试用例表格数据：common.get_excel_data.py
1、测试用例表格设计，一个模块所有API放到一个表格，该模块不同API放到不同Sheet中，表头字段：功能模块、用例描述、case_name、method、path、param
2、通过传参表格名称和sheet名称，将每行用例存到列表中，方便后续执行测试用例提取测试参数
from xlrd import open_workbook
class GetExcelData():
def get_xls(self, xls_name, sheet_name): # xls_name填写用例的Excel名称 sheet_name该Excel的sheet名称
list = []
# 获取用例文件路径
xlsPath = "D:/pycharmproject/data/" + xls_name
file = open_workbook(xlsPath) # 打开用例Excel
sheet = file.sheet_by_name(sheet_name) # 获得打开Excel的sheet
# 获取这个sheet内容行数
nrows = sheet.nrows
for i in range(nrows): # 根据行数做循环
if sheet.row_values(i)[2] != u'case_name': # 如果这个Excel的这个sheet的第i行的第三列不等于case_name那么我们把这行的数据添加到cls[]
list.append(sheet.row_values(i))
return list
if __name__ == '__main__': # 我们执行该文件测试一下是否可以正确获取Excel中的值
print(GetExcelData().get_xls('管理中心-资产管理-资源管理.xls', '添加资产API'))
print(GetExcelData().get_xls('管理中心-资产管理-资源管理.xls', '添加资产API')[0][2])
查询mysql数据库：common.select_mysql.py
1、通过传sql查询语句，查询数据库中对应数据，若未找到则捕获异常返回“数据库未找到该数据”，用于后续进行数据库一致性断言
import pymysql
class SelectMysql():
def select_mysql(self, sql):
self.host = "*.*.*.*"
self.user = "root"
self.password = "OYXb0IJOgBs2pOzX3Nw2"
self.database = "qyjc"
self.conn = pymysql.connect(self.host, self.user, self.password, self.database, charset="utf8")
self.cur = self.conn.cursor()
self.sql1 = sql
try:
self.cur.execute(self.sql1)
sth = self.cur.fetchall()[0][0]
except:
sth = '数据库未找到该数据！'
return sth
测试用例执行文件-添加资源测试用例：testcases.zjtest_add_source.py
继承unittest.TestCase类，使用unittest单元测试框架，setUpClass实例化需要调用的类对象，获取表格API请求方法，获取表格API请求完整路径
定义添加资源方法，获取表格中添加资源请求参数，由于requests.post传参需要字典类型，取出表格的数据为str类型，需要将str类型转换为字典类型
调用请求API主方法，传入(请求方法，请求URL，请求参数)，请求API成功后通过调用查询数据库公共方法查询库中对应的资产url，若添加成功则数据库能找到该资产url，若添加失败则返回“数据库未找到该数据”
from common.common_method import CommonMethod
import unittest
from common.get_excel_data import GetExcelData
from common.select_mysql import SelectMysql
class TestAddSource(unittest.TestCase):
@classmethod
def setUpClass(cls):
env = "test"
# 实例化，声明CommonMethod，GetExcelData，SelectMysql类对象
cls.commonMethod = CommonMethod(env)
cls.getExcelData = GetExcelData()
cls.selectMysql = SelectMysql()
# 获取表格中API对应的请求方法
cls.method = cls.getExcelData.get_xls('管理中心-资产管理-资源管理.xls', '添加资产API')[0][3]
# 获取表格中API对应的完整请求路径
cls.url = cls.commonMethod.get_base_url() + GetExcelData().get_xls('管理中心-资产管理-资源管理.xls', '添加资产API')[0][4]
def setUp(self):
pass
def tearDown(self):
pass
@classmethod
def tearDownClass(cls):
pass
def test_add_source_01(self):
# 取出表格中param列内容，为str类型
data = GetExcelData().get_xls('管理中心-资产管理-资源管理.xls', '添加资产API')[0][5]
# 将取出的str类型数据转换成字典类型
data_dict = {}
for i in range(0,100):
# 捕获异常，当index i超出范围，则跳出循环，不再执行
try:
data1 = data.split('\n')[i]
key = data1.split('|')[0]
value = data1.split('|')[1]
data_dict[key] = value
except:
break
select_site = "select site from q_host_contacts where site = " + '"' + data_dict.get("url") + '"' + " and user_id = 1"
self.result = self.commonMethod.method_main(self.method, self.url, data_dict)
self.assertEqual(self.selectMysql.select_mysql(select_site), data_dict.get("url"), self.selectMysql.select_mysql(select_site))
def test_add_source_02(self):
# 取出表格中param列内容，为str类型
data = GetExcelData().get_xls('管理中心-资产管理-资源管理.xls', '添加资产API')[1][5]
# 将取出的str类型数据转换成字典类型
data_dict = {}
for i in range(0,100):
# 捕获异常，当index i超出范围，则跳出循环，不再执行
try:
data1 = data.split('\n')[i]
key = data1.split('|')[0]
value = data1.split('|')[1]
data_dict[key] = value
except:
break
self.result = self.commonMethod.method_main(self.method, self.url, data_dict)
# 断言，当API返回结果中的message字段为“当前资产已存在”，则用例执行成功，否则执行失败
self.assertEqual(self.result['error']['message'],"当前资产已存在","用例执行失败！")
if __name__ == "__main__":
unittest.main()
测试套件，执行代码入口：testsuite.py
使用discover方法将指定路径下"zjtest"开头的用例全部放到测试套件中，并导入HTMLTestRunner模块，执行完生成对应的测试报告
import unittest
import HTMLTestRunner
if __name__ == "__main__":
filepath = r".\report\htmlreport.html"
test_dir = r'.\testcases'
ftp = open(filepath,'wb')
suite = unittest.defaultTestLoader.discover(test_dir, pattern='zjtest*.py')
runner = HTMLTestRunner.HTMLTestRunner(stream = ftp, title = "测试报告")
runner.run(suite)
ftp.close()