前言
前些天写了pytest+yaml+allure接口自动化测试框架
这篇文章。
今天采用Excel
继续写一个接口自动化测试框架。
设计流程图
这张图是我的excel接口测试框架的一些设计思路。
首先读取excel文件,得到测试信息,然后通过封装的requests方法,用unittest进行测试。
其中,接口关联的参数通过正则进行查找和替换,为此我专门开辟了一个全局变量池,用于管理各种各样的变量。
最后通过HTMLrunner生成测试报告。如果执行失败,发送测试报告结果邮件。
Excel和结果预览
这个时excel的测试测试用例组织结构图。
这个是运行之后生成的HTML测试报告。
这个时运行之后生成的excel报告。可以看到我故意在预期正则
中设置了错误的值,然后用例失败的同时也把失败的预期值标记出来了。
测试失败之后收到的邮件
好了上面就是一些简单的介绍,我们开始进入正题把。
框架结构
首先,要开发这样一个excel接口自动化测试项目必须有一个设计清晰的思路,这样我们在开发框架的过程中才会明白自己要干什么。
目录/文件 | 说明 | 是否为python包 |
---|---|---|
common | 公共类 | 是 |
core | 核心类,封装requests 等 | 是 |
data | 测试使用的excel文件存放目录 | |
logs | 日志目录 | |
tests | 测试用例目录 | 是 |
utils | 工具类,如:日志 | 是 |
config.py | 配置文件 | |
run.py | 执行文件 |
Excel相关
用例设计
本次依然采用的是智学网
登录接口。使用了智学网中的登录接口和登录验证接口,这两个接口之间有依赖的参数。
配置文件
在项目的根目录创建config.py
,把你能想到的配置信息,全部丢在这个文件中进行统一的管理。
#!/usr/bin/env python3
# coding=utf-8
import osclass CF:"""配置文件"""# 项目目录BASE_DIR = os.path.abspath(os.path.dirname(__file__))# Excel首行配置NUMBER = 0NAME = 1METHOD = 2URL = 3ROUTE = 4HEADERS = 5PARAMETER = 6 # 参数EXPECTED_CODE = 7 # 预期响应码EXPECTED_REGULAR = 8 # 预期正则EXPECTED_VALUE = 9 # 预期结果值SPEND_TIME = 10 # 响应时间TEST_RESULTS = 11 # 测试结果EXTRACT_VARIABLE = 12 # 提取变量RESPONSE_TEXT = 13 # 响应文本# 字体大小FONT_SET = "微软雅黑"FONT_SIZE = 16# 颜色配置COLOR_PASSED = "90EE90"COLOR_FAILED = "FA8072"# 邮箱配置EMAIL_INFO = {'username': '1084502012@qq.com','password': 2,'smtp_host': 'smtp.qq.com','smtp_port': 465}# 收件人ADDRESSEE = ['1084502012@qq.com']if __name__ == '__main__':print(CF.EXPECTED_CODE)
读取/写入excel#
在common目录中新建excelset.py
文件,在这个文件中我们要实现,读取excel中的用例,写入测试结果并绘制相应的颜色,写入测试耗费时长。
#!/usr/bin/env python
# coding=utf-8
import shutil
import openpyxl
from config import CF
from openpyxl.styles import Font
from openpyxl.styles import PatternFill
from common.variables import VariablePoolclass ExcelSet:"""Excel配置"""def __init__(self):shutil.copyfile(VariablePool.get('excel_input'), VariablePool.get('excel_output'))self.path = VariablePool.get('excel_output')self.wb = openpyxl.load_workbook(self.path)self.table = self.wb.activedef get_cases(self, min_row=2):"""获取用例"""all_cases = []for row in self.table.iter_rows(min_row=min_row):all_cases.append((self.table.cell(min_row, CF.NAME + 1).value,min_row, [cell.value for cell in row]))min_row += 1return all_casesdef write_color(self, row_n, col_n, color=CF.COLOR_FAILED):"""写入颜色"""cell = self.table.cell(row_n, col_n + 1)fill = PatternFill("solid", fgColor=color)cell.fill = filldef write_results(self, row_n, col_n, value, color=True):"""写入结果"""cell = self.table.cell(row_n, col_n + 1)cell.value = valuefont = Font(name=CF.FONT_SET, size=CF.FONT_SIZE)cell.font = fontif color:if value.lower() in ("fail", 'failed'):fill = PatternFill("solid", fgColor=CF.COLOR_FAILED)cell.fill = fillelif value.lower() in ("pass", "ok"):fill = PatternFill("solid", fgColor=CF.COLOR_PASSED)cell.fill = fillself.wb.save(self.path)excel_set = ExcelSet()
if __name__ == '__main__':print(excel_set.get_cases())
日志封装
logger.py
在一个项目中日志是必不可少的东西,可以第一时间反馈问题。
#!/usr/bin/env python3
# coding=utf-8
import os
import logging
from config import CF
from datetime import datetimeclass Logger:def __init__(self):self.logger = logging.getLogger()if not self.logger.handlers:self.logger.setLevel(logging.DEBUG)# 创建一个handler,用于写入日志文件fh = logging.FileHandler(self.log_path, encoding='utf-8')fh.setLevel(logging.DEBUG)# 创建一个handler,用于输出到控制台ch = logging.StreamHandler()ch.setLevel(logging.INFO)# 定义handler的输出格式formatter = logging.Formatter(self.fmt)fh.setFormatter(formatter)ch.setFormatter(formatter)# 给logger添加handlerself.logger.addHandler(fh)self.logger.addHandler(ch)@propertydef log_path(self):logs_path = os.path.join(CF.BASE_DIR, 'logs')if not os.path.exists(logs_path):os.makedirs(logs_path)now_month = datetime.now().strftime("%Y%m")return os.path.join(logs_path, '{}.log'.format(now_month))@propertydef fmt(self):return '%(levelname)s %(asctime)s %(filename)s:%(lineno)d %(message)s'log = Logger().logger
if __name__ == '__main__':log.info("你好")
正则操作
regular.py
在接口关联参数的提取和传参中的起到了决定性的作用。
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import re
from utils.logger import log
from common.variables import VariablePool
from core.serialize import is_json_strclass Regular:"""正则类"""def __init__(self):self.reg = re.compiledef finds(self, string):return self.reg(r'\{{(.*?)}\}').findall(string)def subs(self, keys, string):result = Nonelog.info("提取变量:{}".format(keys))for i in keys:if VariablePool.has(i):log.info("替换变量:{}".format(i))comment = self.reg(r"\{{%s}}" % i)result = comment.sub(VariablePool.get(i), string)log.info("替换结果:{}".format(result))return resultdef find_res(self, exp, string):"""在结果中查找"""if is_json_str(string):return self.reg(r'\"%s":"(.*?)"' % exp).findall(string)[0]else:return self.reg(r'%s' % exp).findall(string)[0]if __name__ == '__main__':a = "{'data': {'loginName': 18291900215, 'password': '{{dd636482aca022}}', 'code': None, 'description': 'encrypt'}}"print(Regular().finds(a))
核心操作
定义变量池
variables.py
全局变量池来了,是不是很简单,但是作用确实很巨大的。
#!/usr/bin/env python3
# -*- coding:utf-8 -*-class VariablePool:"""全局变量池"""@staticmethoddef get(name):"""获取变量"""return getattr(VariablePool, name)@staticmethoddef set(name, value):"""设置变量"""setattr(VariablePool, name, value)@staticmethoddef has(name):return hasattr(VariablePool, name)if __name__ == '__main__':VariablePool.set('name', 'wxhou')print(VariablePool.get('name'))
封装requests
request.py
最最核心的部分,对于python requests库的二次封装。用以实现接口的请求和返回结果的获取。
#!/usr/bin/env python
# coding=utf-8
import urllib3
import requests
from config import CF
from utils.logger import log
from common.regular import Regular
from common.setResult import replace_param
from core.serialize import deserialization
from requests.exceptions import RequestException
from common.variables import VariablePoolurllib3.disable_warnings()class HttpRequest:"""二次封装requests方法"""http_method_names = 'get', 'post', 'put', 'delete', 'patch', 'head', 'options'def __init__(self):self.r = requests.session()self.reg = Regular()def send_request(self, case, **kwargs):"""发送请求:param case: 测试用例:param kwargs: 其他参数:return: request响应"""if case[CF.URL]:VariablePool.set('url', case[CF.URL])if case[CF.HEADERS]:VariablePool.set('headers', deserialization(case[CF.HEADERS]))method = case[CF.METHOD].upper()url = VariablePool.get('url') + case[CF.ROUTE]self.r.headers = VariablePool.get('headers')params = replace_param(case)if params: kwargs = paramstry:log.info("Request Url: {}".format(url))log.info("Request Method: {}".format(method))log.info("Request Data: {}".format(kwargs))def dispatch(method, *args, **kwargs):if method in self.http_method_names:handler = getattr(self.r, method)return handler(*args, **kwargs)else:raise AttributeError('request method is ERROR!')response = dispatch(method.lower(), url, **kwargs)log.info(response)log.info("Response Data: {}".format(response.text))return responseexcept RequestException as e:log.exception(format(e))except Exception as e:raise e
序列化与反序列化
serialize.py
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import json
from json.decoder import JSONDecodeErrordef deserialization(content: json):"""反序列化json对象 -> python数据类型"""return json.loads(content)def serialization(content, ensure_ascii=True):"""序列化python数据类型 -> json对象"""return json.dumps(content, ensure_ascii=ensure_ascii)def is_json_str(string):"""判断是否是json格式字符串"""if isinstance(string, str):try:json.loads(string)return Trueexcept JSONDecodeError:return Falsereturn Falseif __name__ == '__main__':a = "{'data': {'loginName': 18291900215, 'password': 'dd636482aca022', 'code': None, 'description': 'encrypt'}}"print(is_json_str(a))
检查结果
checkResult.py
在这个文件中,我们将对测试返回的结果进行预期的验证。
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import re
from config import CF
from utils.logger import log
from requests import Response
from common.excelset import excel_setdef check_result(r: Response, number, case):"""获取结果"""results = []excel_set.write_results(number, CF.SPEND_TIME, r.elapsed.total_seconds(), color=False)if case[CF.EXPECTED_CODE]:res = int(case[CF.EXPECTED_CODE]) == r.status_coderesults.append(res)if not res: excel_set.write_color(number, CF.EXPECTED_CODE)log.info(f"预期响应码:{case[CF.EXPECTED_CODE]},实际响应码:{r.status_code}")if case[CF.EXPECTED_VALUE]:res = case[CF.EXPECTED_VALUE] in r.textresults.append(res)if not res: excel_set.write_color(number, CF.EXPECTED_VALUE)log.info(f"预期响应值:{case[CF.EXPECTED_VALUE]},实际响应值:{r.text}")if case[CF.EXPECTED_REGULAR]:res = r'%s' % case[CF.EXPECTED_REGULAR]ref = re.findall(res, r.text)results.append(ref)if not ref: excel_set.write_color(number, CF.EXPECTED_REGULAR)log.info(f"预期正则:{res},响应{ref}")if all(results):excel_set.write_results(number, CF.TEST_RESULTS, 'Pass')log.info(f"用例【{case[CF.NAME]}】测试成功!")else:excel_set.write_results(number, CF.TEST_RESULTS, 'Failed')assert all(results), f"用例【{case[CF.NUMBER]}{case[CF.NAME]}】测试失败:{results}"
设置参数
setResult.py
在这个文件中我们实现了接口返回值的提取,实现了接口传递参数的函数。
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
from requests import Response
from utils.logger import log
from common.regular import Regular
from common.excelset import excel_set
from common.variables import VariablePool
from core.serialize import is_json_str, deserialization
from config import CFreg = Regular()def get_var_result(r: Response, number, case):"""替换变量"""if case[CF.EXTRACT_VARIABLE]:for i in case[CF.EXTRACT_VARIABLE].split(','):result = reg.find_res(i, r.text)VariablePool.set(i, result)log.info(f"提取变量{i}={result}")if not VariablePool.get(i):excel_set.write_results(number, CF.EXTRACT_VARIABLE, f"提变量{i}失败")excel_set.write_results(number, CF.RESPONSE_TEXT,f"ResponseCode:{r.status_code}\nResponseText:{r.text}")def replace_param(case):"""传入参数"""if case[CF.PARAMETER]:if is_json_str(case[CF.PARAMETER]):is_extract = reg.finds(case[CF.PARAMETER])if is_extract:return deserialization(reg.subs(is_extract, case[CF.PARAMETER]))return deserialization(case[CF.PARAMETER])
测试操作
test_api.py
我们采用unittest进行测试,在前置条件和后置条件中我们对封装的HttpRequest方法进行了初始化和关闭会话操作。
使用parameterized库中的expend方法对excel中的用例进行参数化读取执行。
#!/usr/bin/env python
# coding=utf-8
import unittest
from parameterized import parameterized
from common.excelset import excel_set
from core.request import HttpRequest
from common.checkResult import check_result
from common.setResult import get_var_resultclass TestApi(unittest.TestCase):"""测试接口"""@classmethoddef setUpClass(cls) -> None:cls.req = HttpRequest()@classmethoddef tearDownClass(cls) -> None:cls.req.r.close()@parameterized.expand(excel_set.get_cases())def test_api(self, name, number, case):"""测试excel接口用例"""r = self.req.send_request(case)get_var_result(r, number, case)check_result(r, number, case)if __name__ == '__main__':unittest.main(verbosity=2)
测试报告发送邮件类
run.py
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import os
import platform
import argparse
import unittest
from common.variables import VariablePool
from utils.send_mail import send_report_mail
from utils.HTMLTestRunner import HTMLTestRunnerdef running(path):"""运行"""test_case = unittest.defaultTestLoader.discover('tests', 'test*.py')with open(path, 'wb') as fp:runner = HTMLTestRunner(stream=fp,title='Excel接口测试',description="用例执行情况",verbosity=2)result = runner.run(test_case)if result.failure_count:send_report_mail(path)def file_path(arg):"""获取输入的文件路径"""if 'Windows' in platform.platform():_dir = os.popen('chdir').read().strip()else:_dir = os.popen('pwd').read().strip()if _dir in arg:return argreturn os.path.join(_dir, arg)def main():"""主函数"""parser = argparse.ArgumentParser(description="运行Excel接口测试")parser.add_argument('-i', type=str, help='原始文件')parser.add_argument('-o', type=str, default='report.xlsx', help="输出文件")parser.add_argument('-html', type=str, default='report.html', help="报告文件")args = parser.parse_args()VariablePool.set('excel_input', file_path(args.i))VariablePool.set('excel_output', file_path(args.o))VariablePool.set('report_path', file_path(args.html))running(VariablePool.get('report_path'))if __name__ == '__main__':main()
运行
值得注意的是,运行测试时要关闭office打开该excel文件。
最后的文件中我是使用了argparse进行了命令行管理,意味着我们可以通过命令行进行测试而无需关心excel在那个目录下存放着。
python run.py -i data\usercase.xlsx
输入下面的命令执行一下。
INFO 2020-07-30 22:07:52,713 request.py:40 Request Url: https://www.zhixue.com/weakPwdLogin/?from=web_login
INFO 2020-07-30 22:07:52,714 request.py:41 Request Method: POST
INFO 2020-07-30 22:07:52,715 request.py:42 Request Data: {'data': {'loginName': 18291900215, 'password': 'dd636482aca022', 'code': None, 'descriptio
n': 'encrypt'}}
INFO 2020-07-30 22:08:17,204 request.py:55 <Response [200]>
INFO 2020-07-30 22:08:17,204 request.py:56 Response Data: {"data":"1500000100070008427","result":"success"}
INFO 2020-07-30 22:08:17,207 setResult.py:20 提取变量data=1500000100070008427
INFO 2020-07-30 22:08:17,307 checkResult.py:18 预期响应码:200,实际响应码:200
INFO 2020-07-30 22:08:17,308 checkResult.py:23 预期响应值:"result":"success",实际响应值:{"data":"1500000100070008427","result":"success"}
INFO 2020-07-30 22:08:17,310 checkResult.py:29 预期正则:[\d]{16},响应['1500000100070008']
INFO 2020-07-30 22:08:17,356 checkResult.py:32 用例【登录】测试成功!
ok test_api_0__ (test_api.TestApi)
INFO 2020-07-30 22:08:17,358 regular.py:20 提取变量:['data']
INFO 2020-07-30 22:08:17,359 regular.py:23 替换变量:data
INFO 2020-07-30 22:08:17,361 regular.py:26 替换结果:{"data": {"userId": "1500000100070008427"}}
INFO 2020-07-30 22:08:17,363 request.py:40 Request Url: https://www.zhixue.com/loginSuccess/
INFO 2020-07-30 22:08:17,366 request.py:41 Request Method: POST
INFO 2020-07-30 22:08:17,367 request.py:42 Request Data: {'data': {'userId': '1500000100070008427'}}
INFO 2020-07-30 22:08:20,850 request.py:55 <Response [200]>
INFO 2020-07-30 22:08:20,851 request.py:56 Response Data: {"result":"success"}
INFO 2020-07-30 22:08:20,932 checkResult.py:18 预期响应码:200,实际响应码:200
INFO 2020-07-30 22:08:20,933 checkResult.py:23 预期响应值:"result":"success",实际响应值:{"result":"success"}
INFO 2020-07-30 22:08:20,935 checkResult.py:29 预期正则:11,响应[]
F test_api_1__ (test_api.TestApi)Time Elapsed: 0:00:28.281434
测试结果邮件发送成功!
执行规则
(venv) C:\Users\hoou\PycharmProjects\httptest-excel>python run.py -h
usage: run.py [-h] [-i I] [-o O] [-html HTML]运行Excel接口测试optional arguments:-h, --help show this help message and exit-i I 原始文件-o O 输出文件-html HTML 报告文件
SHELL 复制 全屏
在命令行输入python run.py excel路径 新excel路径 报告路径
如果不输入新excel路径
和报告路径
,会在run.py
所在目录生成两个report.xlsx,report.html。
本篇的excel测试框架就完成了。
Python接口自动化测试零基础入门到精通(2023最新版)