您现在的位置是:首页 >其他 >从零搭建完整 Python 接口自动化测试框架—持续更新网站首页其他
从零搭建完整 Python 接口自动化测试框架—持续更新
本接口自动化框架采用 python + unittest + request + openpyxl + myddt + pymysql 来实现接口自动化。
1、总体框架
2、单元测试框架 unittest
unittest 是 Python 自带的一个单元测试框架
2.1 作用
-
管理用例
-
批量执行用例
-
组织运行结果/报告
-
让代码更稳健
-
可拓展
2.2 unittest 框架中,有以下几个组件:
TestCase:即测试用例,Unittest提供testCase类来编写测试用例,一个TestCase的实例就是一个测试用例。一条测试用例就是一个完整的测试流程,包括测试前准备环境的搭建(setUp),执行测试代码(run),以及测试后环境的还原(tearDown),通过运行一条测试用例,可以对某一个问题进行验证。
Fixture:即测试固件,用于测试用例环境的搭建和销毁。在测试步骤执行前需要为该测试用例准备环境(SetUp),如启动app或打开浏览器,测试步骤执行后需要恢复环境(TearDown),如关闭app或浏览器,这时候就需要用到Fixture,使代码更简洁。
TestSuite:即测试套件,把需要执行的测试用例集合在一起就是TestSuite。使用TestLoader来加载TestCase到TestSuite中
TextTestRunner:即测试执行器,用于执行测试用例。该模块中提供run方法执行TestSuite中的测试用例,并返回测试用例的执行结果,如运行的用例总数、用例通过数、用例失败数。
report:即测试报告。unittest框架没有自带的用于生成测试报告的模块或接口,需要使用第三方的扩展模块HTMLTestRunner。
2.3 跳过执行测试用例共有四种写法
- @unittest.skip(reason) :跳过测试用例,reason 为测试被跳过的原因。
- @unittest.skipIf(condition, reason) :当 condition 为真时,跳过测试用例。
- @unittest.skipUnless(condition, reason) :跳过测试用例,除非 condition 为真。
2.4 断言
2.5 报告
from BeautifulReport import BeautifulReport
from common.HTMLTestRunnerNew import HTMLTestRunner
# 4种测试报告
"""
1、生成 HTML 类型
2、生成 Br 类型
3、生成 txt 类型
"""
# ts0 = unittest.TestLoader().discover('test_cases')
# with open('reports/html_do接口自动化.html','wb') as f:
# runner = HTMLTestRunner(f)
# runner.run(ts0)
#
# ts1 = unittest.TestLoader().discover('test_cases')
# br = BeautifulReport(ts1)
# br.report(description='DO',filename='br_do接口自动化',report_dir='reports',theme='theme_memories')
#
#
# ts2 = unittest.TestLoader().discover('test_cases')
# with open('reports/txt_do接口自动化.txt','w+') as f:
# unittest.TextTestRunner(f,2).run(ts2)
if __name__ == "__main__":
unittest.main()
3、基础框架搭建
在项目根目录下新建 common 文件夹,用来存储公用方法。
在项目根目录下新建 reports 文件夹,用来存储项目报告。
在项目根目录下新建 logs 文件夹,用来存储结果日志。
在项目根目录下新建 test_data 文件夹,用来存储用例数据。
在项目根目录下新建 test_cases 文件夹,用例存储测试用例模块。
在项目根目录下新建 main.py 文件,作为入口函数,方便项目调试。
3.1 common公用方法文件
3.1.1 init.py
# /usr/bin/env python
# __*__ coding: utf-8 __*__
# @Time : 2021/9/9 22:22
# @Author: 夜华
import settings
from common.log_handler import get_logger
from common.db_handler import DB
# 日志
logger = get_logger(**settings.LOG_CONFIG)
# 数据库
db = DB(settings.DB_CONFIG)
3.1.2 http_requests.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 3:38 PM
@File : cliend_http_requests.py
@Project : PyCharm
"""
import requests
def cliend_http_requests(url,method,**kwargs):
method = method.lower()
return getattr(requests,method)(url,**kwargs)
if __name__ == "__main__":
case = {
'url' : 'http://10.21.5.74:33140/api/v1/login',
'method' : 'post',
'requests':{
'json' : {"email": "name", "password": "password"},
'headers' : {"Content-Type": "application/json;charset=UTF-8"}
}
}
response = cliend_http_requests(url=case['url'],method=case['method'],**case['requests'])
print(response.json())
3.1.3 data_handler.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 4:10 PM
@File : data_handler.py
@Project : PyCharm
"""
import json
from openpyxl import load_workbook
def get_test_data(filename,sheet_name):
wb = load_workbook(filename=filename)
sh = wb[sheet_name]
row = sh.max_row
column = sh.max_column
data = []
keys = []
for i in range(1,column+1):
keys.append(sh.cell(1,i).value)
for i in range(2,row+1):
temp = {}
for j in range(1,column+1):
temp[keys[j-1]] = sh.cell(i,j).value
try:
temp['request'] = json.loads(temp['request'])
temp['exportx_code'] = json.loads(temp['exportx_code'])
except json.decoder.JSONDecodeError:
raise ValueError('json数据转换错误')
data.append(temp)
return data
if __name__ == "__main__":
res = get_test_data(filename='../test_data/test_cases.xlsx',sheet_name='login')
print(res[0])
3.1.4 db_handler.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 2:58 PM
@File : db_config_handler.py
@Project : PyCharm
"""
import settings
import pymysql
class DB:
def __init__(self,db_config):
self.conn = pymysql.connect(**db_config)
def sql_one(self,sql):
with self.conn.cursor() as cursor:
cursor.execute(sql)
return cursor.fetchone()
def sql_many(self,sql,size=int):
with self.conn.cursor() as cursor:
cursor.execute(sql)
return cursor.fetchmany(size)
def sql_all(self,sql):
with self.conn.cursor() as cursor:
cursor.execute(sql)
return cursor.fetchall()
def exisx(self,sql):
with self.conn.cursor() as cursor:
cursor.execute(sql)
if cursor.fetchone():
return True
else:
return False
def sql_update(self,sql):
with self.conn.cursor() as cursor:
try:
cursor.execute(sql)
self.conn.commit()
except:
self.conn.rollback()
return cursor.fetchone()
def __del__(self):
self.conn.close()
if __name__ == "__main__":
db = DB(db_config=settings.DB_CONFIG)
print(db.sql_one("select * from help_category;"))
print(db.sql_many("select * from help_category;",2))
print(db.sql_all("select name from help_category;"))
print(db.exisx("select * from help_category where name = 'Contents';"))
print(db.sql_update("update help_category set url='' where name = 'Contents';"))
3.1.5 fixtrue
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 8:27 PM
@File : fixtrue.py
@Project : PyCharm
"""
import requests
import settings
from common import logger
def login(email,password):
data = {
'email': email,
'password': password
}
headers = {"Content-Type":"application/json;charset=UTF-8"}
url = settings.PROJECT_URL + settings.INTERFACE['login']
res = requests.post(url=url,json=data,headers=headers)
if res.status_code == 200:
logger.info('用户登录成功')
return res.json()
else:
logger.warning('用户登录失败')
if __name__ == "__main__":
res = login(email='name',password='password')
3.1.6 logs_handler.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 3:28 PM
@File : log_handler.py
@Project : PyCharm
"""
import logging
def get_logger(name,filename,debug=False,fmt=None,mode='w',encoding='utf-8'):
logger = logging.getLogger(name=name)
logger.setLevel(level=logging.DEBUG)
if debug:
file_level = logging.DEBUG
console_level = logging.DEBUG
else:
file_level = logging.WARNING
console_level = logging.INFO
if fmt is None:
#fmt = '%(asctime)s %(levelname)s [%(name)s] [%(filename)s (%(funcName)s:%(lineno)d] - %(message)s'
fmt = '%(asctime)s-[%(filename)s-->line:%(lineno)d]-%(levelname)s:%(message)s'
format = logging.Formatter(fmt)
file_handler = logging.FileHandler(filename=filename,mode=mode,encoding=encoding)
file_handler.setLevel(level=file_level)
console_handler = logging.StreamHandler()
console_handler.setLevel(level=console_level)
file_handler.setFormatter(format)
console_handler.setFormatter(format)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
if __name__ == "__main__":
logger = get_logger(name='do',filename='../logs/do.txt',debug=False,mode='a')
logger.debug(10)
logger.info(20)
logger.warning(30)
logger.error(40)
logger.critical(50)
3.1.7 reports_handler.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 3:42 PM
@File : reports_handler.py
@Project : PyCharm
"""
import os
from BeautifulReport import BeautifulReport
from common.HTMLTestRunnerNew import HTMLTestRunner
from datetime import datetime
def reports(ts,filename,report_dir,theme='theme_default',title=None,description=None,tester=None,_type='br'):
time_prefix=datetime.now().strftime('%Y-%m-%d_%H:%M')
filename = '{}_{}'.format(time_prefix,filename)
if _type == 'br':
br = BeautifulReport(ts)
br.report(description=description,filename=filename,report_dir=report_dir,theme=theme)
else:
with open(os.path.join(report_dir,filename),'wb') as f:
runner = HTMLTestRunner(f,title=title,description=description,tester=tester)
runner.run(ts)
4、config 配置文件夹
4.1 config_dev.ini
[URL]
api_url = http://10.21.5.74:33140
4.2 config_handler.py
# /usr/bin/env python
# __*__ coding: utf-8 __*__
# @Time : 2021/9/10 21:00
# @Author: 夜华
"""
封装配置文件
"""
import yaml
from configparser import ConfigParser
def get_config(filename,encoding='utf-8'):
# 根据 . 获取文件后缀,并获取后面的内容
suffix = filename.split('.')[-1]
if suffix in ['ini','cfg','cng']: # 判断文件后缀是否存在列表内
# 就是ini 配置
config = ConfigParser() # 实例
config.read(filename,encoding=encoding) # 读取文件
data = {} #
for section in config.sections(): #获取 文件里面的所有段名
data[section] = dict(config.items(section))
elif suffix in ['yaml','yml']:
# 就是 yaml 配置
with open(filename,'r',encoding=encoding) as f:
data = yaml.load(f,Loader=yaml.FullLoader)
else:
raise ValueError('不能识别的配置后缀')
return data
if __name__ == '__main__':
get = get_config('../config.ini')
print(get)
res = get_config('../config.yaml')
print(res)
4.3 init.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/6/1 2:18 PM
@File : __init__.py
@Project : PyCharm
"""
import os
import sys
from config.config_handler import get_config
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
if sys.argv[1] == "DEV":
Config = get_config(os.path.join(BASE_DIR, './config_dev.ini'))
else:
Config = get_config(os.path.join(BASE_DIR, './config_test.ini'))
5、 logs 文件夹
保存接口测试过程中输出的日志
6、reports 文件夹
保存接口测试报告
7、test_cases 文件夹
7.1 base_case.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 4:39 PM
@File : base_case.py
@Project : PyCharm
"""
import unittest
from common import logger,db
import requests
import settings
class Basic_test_case(unittest.TestCase):
name = '基类'
logger = logger
requests = requests
session = requests.session()
db = db
settings = settings
@classmethod
def setUpClass(cls) -> None:
cls.logger.info('---------------【{}】开始测试---------------'.format(cls.name))
@classmethod
def tearDownClass(cls) -> None:
cls.logger.info('---------------【{}】结束测试---------------'.format(cls.name))
def check(self,case):
self.logger.info('---------------【{}】开始测试---------------'.format(self.name))
self.case = case # 测试用例
self.step() # 测试步骤
self.assert_status_code() # 断言状态码
self.assert_json() # 断言响应信息
self.assert_db() # 断言数据库是否存在数据
self.logger.info('---------------【{}】结束测试---------------'.format(self.name))
def step(self):
self.case['url'] = self.settings.PROJECT_URL + self.settings.INTERFACE[self.case['url']]
try:
self.resposen = self.http_requests(url=self.case['url'],method=self.case['method'],**self.case['request'])
except Exception as e:
self.logger.warning('用例【{}】发送请求错误'.format(self.case['title']))
self.logger.debug('url:【{}】'.format(self.case['url']))
self.logger.debug('method:【{}】'.format(self.case['method']))
raise e
else:
self.logger.info('用例【{}】发送请求成功'.format(self.case['title']))
def assert_status_code(self):
try:
self.assertEqual(self.resposen.status_code,self.case['status_code'])
except AssertionError as e:
self.logger.warning('用例【{}】状态码断言错误'.format(self.case['title']))
self.logger.debug('预期状态码:【{}】'.format(self.case['status_code']))
self.logger.debug('实际状态码:【{}】'.format(self.resposen.status_code))
raise e
else:
self.logger.info('用例【{}】状态码断言成功'.format(self.case['title']))
def assert_json(self):
res = self.resposen.json()
res_data = {
'phone':res.get('phone',None),
'roleType':res.get('roleType',None)
}
try:
self.assertEqual(res_data,self.case['exportx_code'])
except AssertionError as e:
self.logger.warning('用例【{}】响应信息断言错误'.format(self.case['title']))
self.logger.debug('预期内容:【{}】'.format(self.case['exportx_code']))
self.logger.debug('实际内容:【{}】'.format(res_data))
self.logger.debug('响应内容:【{}】'.format(res))
raise e
else:
self.logger.info('用例【{}】响应信息断言成功'.format(self.case['title']))
def assert_db(self):
if self.case.get('sql'):
try:
db_res = self.db.exisx(self.case['sql'])
self.assertTrue(db_res)
except Exception as e:
self.logger.warning('用例【{}】数据库查询失败'.format(self.case['title']))
self.logger.debug('sql:【{}】'.format(self.case['sql']))
raise e
else:
self.logger.info('用例【{}】数据库查询成功'.format(self.case['title']))
def http_requests(self,url,method,**kwargs)->requests.Response:
method = method.lower()
return getattr(self.session,method)(url=url,**kwargs)
if __name__ == "__main__":
unittest.main()
7.2 test_login.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 3:43 PM
@File : test_login.py
@Project : PyCharm
"""
import unittest
import settings
from test_cases.base_case import Basic_test_case
from common.data_handler import get_test_data
from common.myddt import data,ddt
cases = get_test_data(settings.TEST_DATA,sheet_name='login')
@ddt
class Test(Basic_test_case):
name = '登录'
@data(*cases)
def test_01(self,case):
self.check(case)
if __name__ == "__main__":
unittest.main()
8、test_data 文件夹
8.1 使用Excel表格维护测试用例
用例:id、title、url、method、requests、status_code、exportx_code、sql
9、main.py
main.py 为 测试入口。
from BeautifulReport import BeautifulReport
from common.HTMLTestRunnerNew import HTMLTestRunner
import settings
import unittest
from common.reports_handler import reports
ts = unittest.TestLoader().discover('test_cases')
runner = reports(ts,**settings.REPORTS_CONFIG)
if __name__ == "__main__":
unittest.main()
10、settings.py
"""
-*- coding: utf-8 -*-
@Author : 夜华
@Time : 2023/5/14 2:53 PM
@File : settings.py
@Project : PyCharm
"""
import os
import project_api
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
TEST_DATA = os.path.join(BASE_DIR,'test_data/test_cases.xlsx')
PROJECT_URL = 'http://10.00.5.74:00000'
INTERFACE = {
'login' : '/api/v1/login',
'query' : '/api/v1/approve/query?page=1&size=4294967295'
}
# 数据库配置
DB_CONFIG = {
'user': 'root',
'password': '123456',
'host': '127.0.0.1',
'database': 'mysql',
'port': 3306,
'autocommit': False
}
# 输出日志
LOG_CONFIG = {
'name' : 'DPO',
'filename' : os.path.join(BASE_DIR,'logs/dpo.txt'),
'debug' : True,
'mode' : 'w',
'fmt' : None,
'encoding': 'utf-8'
}
# 报告
REPORTS_CONFIG = {
'filename':'do接口自动化',
'report_dir' : os.path.join(BASE_DIR,'reports'),
'theme' : 'theme_default',
'description' : 'DO',
'title': 'DO1期',
'_type': 'br'
}
USER_LOGIN = {"email":"name","password":"password"}
11、终端内执行
注意:DEV表示开发环境,如果想在非开发环境进行测试,就输入TEST。也可以在4.3init.py 修改