软件测试开发实战|接口自动化测试框架开发(pytest+allure+aiohttp+用例自动生成)

近期准备优先做接口测试的覆盖，为此需要开发一个测试框架，经过思考，这次依然想做点儿不一样的东西。

• 接口测试是比较讲究效率的，测试人员会希望很快能得到结果反馈，然而接口的数量一般都很多，而且会越来越多，所以提高执行效率很有必要
• 接口测试的用例其实也可以用来兼做简单的压力测试，而压力测试需要并发
• 接口测试的用例有很多重复的东西，测试人员应该只需要关注接口测试的设计，这些重复劳动最好自动化来做
  pytest和allure太好用了，新框架要集成它们
• 接口测试的用例应该尽量简洁，最好用yaml，这样数据能直接映射为请求数据，写起用例来跟做填空题一样，便于向没有自动化经验的成员推广 加上我对Python的协程很感兴趣，也学了一段时间，一直希望学以致用，所以http请求我决定用aiohttp来实现。 但是pytest是不支持事件循环的，如果想把它们结合还需要一番功夫。于是继续思考，思考的结果是其实我可以把整个事情分为两部分。 第一部分，读取yaml测试用例，http请求测试接口，收集测试数据。 第二部分，根据测试数据，动态生成pytest认可的测试用例，然后执行，生成测试报告。 这样一来，两者就能完美结合了，也完美符合我所做的设想。想法既定，接着 就是实现了。
  

  第一部分（整个过程都要求是异步非阻塞的）

  读取yaml测试用例

  一份简单的用例模板我是这样设计的，这样的好处是，参数名和aiohttp.ClientSession().request(method,url,**kwargs)是直接对应上的，我可以不费力气的直接传给请求方法，避免各种转换，简洁优雅，表达力又强。

args:- post- /xxx/addkwargs:-caseName: 新增xxxdata:name: ${gen_uid(10)}validator:-json:successed: True

异步读取文件可以使用aiofiles这个第三方库，yaml_load是一个协程，可以保证主进程读取yaml测试用例时不被阻塞，通过 await yaml_load() 便能获取测试用例的数据

async def yaml_load(dir=\'\', file=\'\'):\"\"\"异步读取yaml文件，并转义其中的特殊值:param file::return:\"\"\"if dir:file = os.path.join(dir, file)async with aiofiles.open(file, \'r\', encoding=\'utf-8\', errors=\'ignore\') as f:data = await f.read()data = yaml.load(data)# 匹配函数调用形式的语法pattern_function = re.compile(r\'^\\${([A-Za-z_]+\\w*\\(.*\\))}$\')pattern_function2 = re.compile(r\'^\\${(.*)}$\')# 匹配取默认值的语法pattern_function3 = re.compile(r\'^\\$\\((.*)\\)$\')def my_iter(data):\"\"\"递归测试用例，根据不同数据类型做相应处理，将模板语法转化为正常值:param data::return:\"\"\"if isinstance(data, (list, tuple)):for index, _data in enumerate(data):data[index] = my_iter(_data) or _dataelif isinstance(data, dict):for k, v in data.items():data[k] = my_iter(v) or velif isinstance(data, (str, bytes)):m = pattern_function.match(data)if not m:m = pattern_function2.match(data)if m:return eval(m.group(1))if not m:m = pattern_function3.match(data)if m:K, k = m.group(1).split(\':\')return bxmat.default_values.get(K).get(k)return datamy_iter(data)return BXMDict(data)

可以看到，测试用例还支持一定的模板语法，如 ${function} 、 $(a:b) 等，这能在很大程度上拓展测试人员用例编写的能力

http请求测试接口

http请求可以直接用 aiohttp.ClientSession().request(method,url,**kwargs) ，http也是一个协程，可以保证网络请求时不被阻塞，通过 await http() 便可以拿到接口测试数据

async def http(domain, *args, **kwargs):\"\"\"http请求处理器:param domain: 服务地址:param args::param kwargs::return:\"\"\"method, api = argsarguments = kwargs.get(\'data\') or kwargs.get(\'params\') or kwargs.get(\'json\') or {}# kwargs中加入tokenkwargs.setdefault(\'headers\', {}).update({\'token\': bxmat.token})# 拼接服务地址和apiurl = \'\'.join([domain, api])async with ClientSession() as session:async with session.request(method, url, **kwargs) as response:res = await response_handler(response)return {\'response\': res,\'url\': url,\'arguments\': arguments}

收集测试数据

协程的并发真的很快，这里为了避免服务响应不过来导致熔断，可以引入 asyncio.Semaphore(num) 来控制并发

async def entrace(test_cases, loop, semaphore=None):\"\"\"http执行入口:param test_cases::param semaphore::return:\"\"\"res = BXMDict()# 在CookieJar的update_cookies方法中，如果unsafe=False并且访问的是IP地址，客户端是不会更新cookie信息# 这就导致session不能正确处理登录态的问题# 所以这里使用的cookie_jar参数使用手动生成的CookieJar对象，并将其unsafe设置为Trueasync with ClientSession(loop=loop, cookie_jar=CookieJar(unsafe=True), headers={\'token\': bxmat.token}) as session:await advertise_cms_login(session)if semaphore:async with semaphore:for test_case in test_cases:data = await one(session, case_name=test_case)res.setdefault(data.pop(\'case_dir\'), BXMList()).append(data)else:for test_case in test_cases:data = await one(session, case_name=test_case)res.setdefault(data.pop(\'case_dir\'), BXMList()).append(data)return resasync def one(session, case_dir=\'\', case_name=\'\'):\"\"\"一份测试用例执行的全过程，包括读取.yml测试用例，执行http请求，返回请求结果所有操作都是异步非阻塞的:param session: session会话:param case_dir: 用例目录:param case_name: 用例名称:return:\"\"\"project_name = case_name.split(os.sep)[1]domain = bxmat.url.get(project_name)test_data = await yaml_load(dir=case_dir, file=case_name)result = BXMDict({\'case_dir\': os.path.dirname(case_name),\'api\': test_data.args[1].replace(\'/\', \'_\'),})if isinstance(test_data.kwargs, list):for index, each_data in enumerate(test_data.kwargs):step_name = each_data.pop(\'caseName\')r = await http(session, domain, *test_data.args, **each_data)r.update({\'case_name\': step_name})result.setdefault(\'responses\', BXMList()).append({\'response\': r,\'validator\': test_data.validator[index]})else:step_name = test_data.kwargs.pop(\'caseName\')r = await http(session, domain, *test_data.args, **test_data.kwargs)r.update({\'case_name\': step_name})result.setdefault(\'responses\', BXMList()).append({\'response\': r,\'validator\': test_data.validator})return result

事件循环负责执行协程并返回结果，在最后的结果收集中，我用测试用例目录来对结果进行了分类，这为接下来的自动生成pytest认可的测试用例打下了良好的基础

def main(test_cases):\"\"\"事件循环主函数，负责所有接口请求的执行:param test_cases::return:\"\"\"loop = asyncio.get_event_loop()semaphore = asyncio.Semaphore(bxmat.semaphore)# 需要处理的任务# tasks = [asyncio.ensure_future(one(case_name=test_case, semaphore=semaphore)) for test_case in test_cases]task = loop.create_task(entrace(test_cases, loop, semaphore))# 将协程注册到事件循环，并启动事件循环try:# loop.run_until_complete(asyncio.gather(*tasks))loop.run_until_complete(task)finally:loop.close()return task.result()

第二部分

动态生成pytest认可的测试用例

首先说明下pytest的运行机制，pytest首先会在当前目录下找conftest.py文件，如果找到了，则先运行它，然后根据命令行参数去指定的目录下找test开头或结尾的.py文件，如果找到了，如果找到了，再分析fixture，如果有session或module类型的，并且参数autotest=True或标记了pytest.mark.usefixtures(a…)，则先运行它们；再去依次找类、方法等，规则类似。大概就是这样一个过程。
可以看出，pytest测试运行起来的关键是，必须有至少一个被pytest发现机制认可的 testxx.py 文件，文件中有 TestxxClass 类，类中至少有一个 def testxx(self) 方法。
现在并没有任何pytest认可的测试文件，所以我的想法是先创建一个引导型的测试文件，它负责让pytest动起来。可以用 pytest.skip() 让其中的测试方法跳过。然后我们的目标是在pytest动起来之后，怎么动态生成用例，然后发现这些用例，执行这些用例，生成测试报告，一气呵成。

# test_bootstrap.pyimport pytestclass TestStarter(object):def test_start(self):pytest.skip(\'此为测试启动方法, 不执行\')

我想到的是通过fixture，因为fixture有setup的能力，这样我通过定义一个scope为session的fixture，然后在TestStarter上面标记use，就可以在导入TestStarter之前预先处理一些事情，那么我把生成用例的操作放在这个fixture里就能完成目标了。

# test_bootstrap.pyimport pytest@pytest.mark.usefixtures(\'te\', \'test_cases\')class TestStarter(object):def test_start(self):pytest.skip(\'此为测试启动方法, 不执行\')

pytest有个 –rootdir 参数，该fixture的核心目的就是，通过 –rootdir 获取到目标目录，找出里面的 .yml 测试文件，运行后获得测试数据，然后为每个目录创建一份 testxx.py 的测试文件，文件内容就是 content 变量的内容，然后把这些参数再传给 pytest.main() 方法执行测试用例的测试，也就是在pytest内部再运行了一个pytest！最后把生成的测试文件删除。注意该fixture要定义在 conftest.py 里面，因为pytest对于 conftest 中定义的内容有自发现能力，不需要额外导入。

# conftest.py@pytest.fixture(scope=\'session\')def test_cases(request):\"\"\"测试用例生成处理:param request::return:\"\"\"var = request.config.getoption(\"--rootdir\")test_file = request.config.getoption(\"--tf\")env = request.config.getoption(\"--te\")cases = []if test_file:cases = [test_file]else:if os.path.isdir(var):for root, dirs, files in os.walk(var):if re.match(r\'\\w+\', root):if files:cases.extend([os.path.join(root, file) for file in files if file.endswith(\'yml\')])data = main(cases)content = \"\"\"import allurefrom conftest import CaseMetaClass@allure.feature(\'{}接口测试({}项目)\')class Test{}API(object, metaclass=CaseMetaClass):test_cases_data = {}\"\"\"test_cases_files = []if os.path.isdir(var):for root, dirs, files in os.walk(var):if not (\'.\' in root or \'__\' in root):if files:case_name = os.path.basename(root)project_name = os.path.basename(os.path.dirname(root))test_case_file = os.path.join(root, \'test_{}.py\'.format(case_name))with open(test_case_file, \'w\', encoding=\'utf-8\') as fw:fw.write(content.format(case_name, project_name, case_name.title(), data.get(root)))test_cases_files.append(test_case_file)if test_file:temp = os.path.dirname(test_file)py_file = os.path.join(temp, \'test_{}.py\'.format(os.path.basename(temp)))else:py_file = varpytest.main([\'-v\',py_file,\'--alluredir\',\'report\',\'--te\',env,\'--capture\',\'no\',\'--disable-warnings\',])for file in test_cases_files:os.remove(file)return test_cases_files

可以看到，测试文件中有一个 TestxxAPI 的类，它只有一个 test_cases_data 属性，并没有 testxx 方法，所以还不是被pytest认可的测试用例，根本运行不起来。那么它是怎么解决这个问题的呢？答案就是 CaseMetaClass 。

function_express = \"\"\"def {}(self, response, validata):with allure.step(response.pop(\'case_name\')):validator(response,validata)\"\"\"class CaseMetaClass(type):\"\"\"根据接口调用的结果自动生成测试用例\"\"\"def __new__(cls, name, bases, attrs):test_cases_data = attrs.pop(\'test_cases_data\')for each in test_cases_data:api = each.pop(\'api\')function_name = \'test\' + apitest_data = [tuple(x.values()) for x in each.get(\'responses\')]function = gen_function(function_express.format(function_name),namespace={\'validator\': validator, \'allure\': allure})# 集成allurestory_function = allure.story(\'{}\'.format(api.replace(\'_\', \'/\')))(function)attrs[function_name] = pytest.mark.parametrize(\'response,validata\', test_data)(story_function)return super().__new__(cls, name, bases, attrs)

CaseMetaClass 是一个元类，它读取test_cases_data属性的内容，然后动态生成方法对象，每一个接口都是单独一个方法，在相继被allure的细粒度测试报告功能和pytest提供的参数化测试功能装饰后，把该方法对象赋值给 test+api 的类属性，也就是说， TestxxAPI 在生成之后便有了若干 testxx 的方法，此时内部再运行起pytest，pytest也就能发现这些用例并执行了。

def gen_function(function_express, namespace={}):\"\"\"动态生成函数对象, 函数作用域默认设置为builtins.__dict__，并合并namespace的变量:param function_express: 函数表达式，示例 \'def foobar(): return \"foobar\"\':return:\"\"\"builtins.__dict__.update(namespace)module_code = compile(function_express, \'\', \'exec\')function_code = [c for c in module_code.co_consts if isinstance(c, types.CodeType)][0]return types.FunctionType(function_code, builtins.__dict__)

在生成方法对象时要注意namespace的问题，最好默认传 builtins.dict ，然后自定义的方法通过namespace参数传进去。

后续（yml测试文件自动生成）

至此，框架的核心功能已经完成了，经过几个项目的实践，效果完全超过预期，写起用例来不要太爽，运行起来不要太快，测试报告也整的明明白白漂漂亮亮的，但我发现还是有些累，为什么呢？
我目前做接口测试的流程是，如果项目集成了swagger，通过swagger去获取接口信息，根据这些接口信息来手工起项目创建用例。这个过程很重复很繁琐，因为我们的用例模板已经大致固定了，其实用例之间就是一些参数比如目录、用例名称、method等等的区别，那么这个过程我觉得完全可以自动化。
因为swagger有个网页啊，我可以去提取关键信息来自动创建.yml测试文件，就像搭起架子一样，待项目架子生成后，我再去设计用例填传参就可以了。
于是我试着去解析请求swagger首页得到的HTML，然后失望的是并没有实际数据，后来猜想应该是用了ajax，打开浏览器控制台的时，我发现了 api-docs 的请求，一看果然是json数据，那么问题就简单了，网页分析都不用了。

import reimport osimport sysfrom requests import Sessiontemplate =\"\"\"args:- {method}- {api}kwargs:-caseName: {caseName}{data_or_params}:{data}validator:-json:successed: True\"\"\"def auto_gen_cases(swagger_url, project_name):\"\"\"根据swagger返回的json数据自动生成yml测试用例模板:param swagger_url::param project_name::return:\"\"\"res = Session().request(\'get\', swagger_url).json()data = res.get(\'paths\')workspace = os.getcwd()project_ = os.path.join(workspace, project_name)if not os.path.exists(project_):os.mkdir(project_)for k, v in data.items():pa_res = re.split(r\'[/]+\', k)dir, *file = pa_res[1:]if file:file = \'\'.join([x.title() for x in file])else:file = dirfile += \'.yml\'dirs = os.path.join(project_, dir)if not os.path.exists(dirs):os.mkdir(dirs)os.chdir(dirs)if len(v) > 1:v = {\'post\': v.get(\'post\')}for _k, _v in v.items():method = _kapi = kcaseName = _v.get(\'description\')data_or_params = \'params\' if method == \'get\' else \'data\'parameters = _v.get(\'parameters\')data_s = \'\'try:for each in parameters:data_s += each.get(\'name\')data_s += \': \\n\'data_s += \' \' * 8except TypeError:data_s += \'{}\'file_ = os.path.join(dirs, file)with open(file_, \'w\', encoding=\'utf-8\') as fw:fw.write(template.format(method=method,api=api,caseName=caseName,data_or_params=data_or_params,data=data_s))os.chdir(project_)

现在要开始一个项目的接口测试覆盖，只要该项目集成了swagger，就能秒生成项目架子，测试人员只需要专心设计接口测试用例即可，我觉得对于测试团队的推广使用是很有意义的，也更方便了我这样的懒人。