近期準(zhǔn)備優(yōu)先做接口測試的覆蓋，為此需要開發(fā)一個(gè)測試框架警儒，經(jīng)過思考，這次依然想做點(diǎn)兒不一樣的東西边琉。

接口測試是比較講究效率的，測試人員會(huì)希望很快能得到結(jié)果反饋记劝，然而接口的數(shù)量一般都很多，而且會(huì)越來越多厌丑，所以提高執(zhí)行效率很有必要
接口測試的用例其實(shí)也可以用來兼做簡單的壓力測試，而壓力測試需要并發(fā)
接口測試的用例有很多重復(fù)的東西砍鸠，測試人員應(yīng)該只需要關(guān)注接口測試的設(shè)計(jì)耕驰，這些重復(fù)勞動(dòng)最好自動(dòng)化來做
pytest和allure太好用了，新框架要集成它們
接口測試的用例應(yīng)該盡量簡潔，最好用yaml饭弓，這樣數(shù)據(jù)能直接映射為請(qǐng)求數(shù)據(jù)双饥，寫起用例來跟做填空題一樣，便于向沒有自動(dòng)化經(jīng)驗(yàn)的成員推廣 加上我對(duì)Python的協(xié)程很感興趣示启，也學(xué)了一段時(shí)間，一直希望學(xué)以致用夫嗓，所以http請(qǐng)求我決定用aiohttp來實(shí)現(xiàn)。 但是pytest是不支持事件循環(huán)的矩父，如果想把它們結(jié)合還需要一番功夫排霉。于是繼續(xù)思考，思考的結(jié)果是其實(shí)我可以把整個(gè)事情分為兩部分攻柠。 第一部分，讀取yaml測試用例瑰钮，http請(qǐng)求測試接口，收集測試數(shù)據(jù)开睡。 第二部分苟耻，根據(jù)測試數(shù)據(jù)，動(dòng)態(tài)生成pytest認(rèn)可的測試用例凶杖，然后執(zhí)行，生成測試報(bào)告蝗茁。 這樣一來寻咒，兩者就能完美結(jié)合了，也完美符合我所做的設(shè)想毛秘。想法既定阻课，接著 就是實(shí)現(xiàn)了限煞。
第一部分（整個(gè)過程都要求是異步非阻塞的）
讀取yaml測試用例
一份簡單的用例模板我是這樣設(shè)計(jì)的员凝，這樣的好處是，參數(shù)名和aiohttp.ClientSession().request(method,url,**kwargs)是直接對(duì)應(yīng)上的健霹，我可以不費(fèi)力氣的直接傳給請(qǐng)求方法，避免各種轉(zhuǎn)換宣吱，簡潔優(yōu)雅瞳别，表達(dá)力又強(qiáng)。

args:

• post
• /xxx/add
  kwargs:
• caseName: 新增xxx
  data:
  name: ${gen_uid(10)}
  validator:
• json:
  successed: True
  異步讀取文件可以使用aiofiles這個(gè)第三方庫疤坝，yaml_load是一個(gè)協(xié)程馆铁，可以保證主進(jìn)程讀取yaml測試用例時(shí)不被阻塞，通過 await yaml_load() 便能獲取測試用例的數(shù)據(jù)

async def yaml_load(dir='', file=''):
"""
異步讀取yaml文件叼架，并轉(zhuǎn)義其中的特殊值
: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)

# 匹配函數(shù)調(diào)用形式的語法
pattern_function = re.compile(r'^\${([A-Za-z_]+\w*\(.*\))}$')
pattern_function2 = re.compile(r'^\${(.*)}$')
# 匹配取默認(rèn)值的語法
pattern_function3 = re.compile(r'^\$\((.*)\)$')

def my_iter(data):
    """
    遞歸測試用例乖订，根據(jù)不同數(shù)據(jù)類型做相應(yīng)處理具练，將模板語法轉(zhuǎn)化為正常值
    :param data:
    :return:
    """
    if isinstance(data, (list, tuple)):
        for index, _data in enumerate(data):
            data[index] = my_iter(_data) or _data
    elif isinstance(data, dict):
        for k, v in data.items():
            data[k] = my_iter(v) or v
    elif 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 data

my_iter(data)

return BXMDict(data)

可以看到，測試用例還支持一定的模板語法扛点，如 (a:b) 等眠饮，這能在很大程度上拓展測試人員用例編寫的能力

http請(qǐng)求測試接口
http請(qǐng)求可以直接用 aiohttp.ClientSession().request(method,url,**kwargs) 铜邮，http也是一個(gè)協(xié)程寨蹋，可以保證網(wǎng)絡(luò)請(qǐng)求時(shí)不被阻塞扔茅，通過 await http() 便可以拿到接口測試數(shù)據(jù)

async def http(domain, *args, **kwargs):
"""
http請(qǐng)求處理器
:param domain: 服務(wù)地址
:param args:
:param kwargs:
:return:
"""
method, api = args
arguments = kwargs.get('data') or kwargs.get('params') or kwargs.get('json') or {}

# kwargs中加入token
kwargs.setdefault('headers', {}).update({'token': bxmat.token})
# 拼接服務(wù)地址和api
url = ''.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
        }

收集測試數(shù)據(jù)
協(xié)程的并發(fā)真的很快召娜，這里為了避免服務(wù)響應(yīng)不過來導(dǎo)致熔斷，可以引入 asyncio.Semaphore(num) 來控制并發(fā)

async def entrace(test_cases, loop, semaphore=None):
"""
http執(zhí)行入口
:param test_cases:
:param semaphore:
:return:
"""
res = BXMDict()
# 在CookieJar的update_cookies方法中玖瘸，如果unsafe=False并且訪問的是IP地址，客戶端是不會(huì)更新cookie信息
# 這就導(dǎo)致session不能正確處理登錄態(tài)的問題
# 所以這里使用的cookie_jar參數(shù)使用手動(dòng)生成的CookieJar對(duì)象嗦枢，并將其unsafe設(shè)置為True
async 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 res

async def one(session, case_dir='', case_name=''):
"""
一份測試用例執(zhí)行的全過程屯断，包括讀取.yml測試用例，執(zhí)行http請(qǐng)求氧秘，返回請(qǐng)求結(jié)果
所有操作都是異步非阻塞的
:param session: session會(huì)話
: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

事件循環(huán)負(fù)責(zé)執(zhí)行協(xié)程并返回結(jié)果趴久，在最后的結(jié)果收集中，我用測試用例目錄來對(duì)結(jié)果進(jìn)行了分類灭忠，這為接下來的自動(dòng)生成pytest認(rèn)可的測試用例打下了良好的基礎(chǔ)

def main(test_cases):
"""
事件循環(huán)主函數(shù)座硕，負(fù)責(zé)所有接口請(qǐng)求的執(zhí)行
:param test_cases:
:return:
"""
loop = asyncio.get_event_loop()
semaphore = asyncio.Semaphore(bxmat.semaphore)
# 需要處理的任務(wù)
# 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))
# 將協(xié)程注冊(cè)到事件循環(huán)，并啟動(dòng)事件循環(huán)
try:
# loop.run_until_complete(asyncio.gather(*tasks))
loop.run_until_complete(task)
finally:
loop.close()

return task.result()

第二部分
動(dòng)態(tài)生成pytest認(rèn)可的測試用例
首先說明下pytest的運(yùn)行機(jī)制映琳，pytest首先會(huì)在當(dāng)前目錄下找conftest.py文件蜘拉，如果找到了，則先運(yùn)行它旭旭，然后根據(jù)命令行參數(shù)去指定的目錄下找test開頭或結(jié)尾的.py文件，如果找到了穿肄，如果找到了，再分析fixture咸产，如果有session或module類型的，并且參數(shù)autotest=True或標(biāo)記了pytest.mark.usefixtures(a…)僵朗，則先運(yùn)行它們屑彻；再去依次找類、方法等社牲，規(guī)則類似。大概就是這樣一個(gè)過程违寿。
可以看出熟空，pytest測試運(yùn)行起來的關(guān)鍵是，必須有至少一個(gè)被pytest發(fā)現(xiàn)機(jī)制認(rèn)可的 testxx.py 文件息罗，文件中有 TestxxClass 類迈喉，類中至少有一個(gè) def testxx(self) 方法。
現(xiàn)在并沒有任何pytest認(rèn)可的測試文件挨摸，所以我的想法是先創(chuàng)建一個(gè)引導(dǎo)型的測試文件，它負(fù)責(zé)讓pytest動(dòng)起來∨亳桑可以用 pytest.skip() 讓其中的測試方法跳過。然后我們的目標(biāo)是在pytest動(dòng)起來之后瞬女，怎么動(dòng)態(tài)生成用例努潘，然后發(fā)現(xiàn)這些用例坤学，執(zhí)行這些用例报慕，生成測試報(bào)告，一氣呵成眠冈。

test_bootstrap.py

import pytest

class TestStarter(object):

def test_start(self):
    pytest.skip('此為測試啟動(dòng)方法, 不執(zhí)行')

我想到的是通過fixture，因?yàn)閒ixture有setup的能力布卡，這樣我通過定義一個(gè)scope為session的fixture雇盖，然后在TestStarter上面標(biāo)記use，就可以在導(dǎo)入TestStarter之前預(yù)先處理一些事情崔挖，那么我把生成用例的操作放在這個(gè)fixture里就能完成目標(biāo)了。

test_bootstrap.py

import pytest

@pytest.mark.usefixtures('te', 'test_cases')
class TestStarter(object):

def test_start(self):
    pytest.skip('此為測試啟動(dòng)方法, 不執(zhí)行')

pytest有個(gè) --rootdir 參數(shù)匾浪，該fixture的核心目的就是卷哩，通過 --rootdir 獲取到目標(biāo)目錄，找出里面的 .yml 測試文件冷溶，運(yùn)行后獲得測試數(shù)據(jù)尊浓，然后為每個(gè)目錄創(chuàng)建一份 testxx.py 的測試文件，文件內(nèi)容就是 content 變量的內(nèi)容栋齿，然后把這些參數(shù)再傳給 pytest.main() 方法執(zhí)行測試用例的測試，也就是在pytest內(nèi)部再運(yùn)行了一個(gè)pytest基协！最后把生成的測試文件刪除菇用。注意該fixture要定義在 conftest.py 里面，因?yàn)閜ytest對(duì)于 conftest 中定義的內(nèi)容有自發(fā)現(xiàn)能力惋鸥，不需要額外導(dǎo)入悍缠。

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 allure

from conftest import CaseMetaClass

@allure.feature('{}接口測試({}項(xiàng)目)')
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 = var

pytest.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

可以看到飞蚓，測試文件中有一個(gè) TestxxAPI 的類拴鸵，它只有一個(gè) test_cases_data 屬性，并沒有 testxx 方法八堡，所以還不是被pytest認(rèn)可的測試用例聘芜，根本運(yùn)行不起來。那么它是怎么解決這個(gè)問題的呢汰现？答案就是 CaseMetaClass 。

function_express = """
def {}(self, response, validata):
with allure.step(response.pop('case_name')):
validator(response,validata)"""

class CaseMetaClass(type):
"""
根據(jù)接口調(diào)用的結(jié)果自動(dòng)生成測試用例
"""

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' + api
        test_data = [tuple(x.values()) for x in each.get('responses')]
        function = gen_function(function_express.format(function_name),
                                namespace={'validator': validator, 'allure': allure})
        # 集成allure
        story_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 是一個(gè)元類口叙，它讀取test_cases_data屬性的內(nèi)容嗅战，然后動(dòng)態(tài)生成方法對(duì)象，每一個(gè)接口都是單獨(dú)一個(gè)方法疟呐，在相繼被allure的細(xì)粒度測試報(bào)告功能和pytest提供的參數(shù)化測試功能裝飾后东且，把該方法對(duì)象賦值給 test+api 的類屬性，也就是說珊泳， TestxxAPI 在生成之后便有了若干 testxx 的方法，此時(shí)內(nèi)部再運(yùn)行起pytest晓褪，pytest也就能發(fā)現(xiàn)這些用例并執(zhí)行了综慎。

def gen_function(function_express, namespace={}):
"""
動(dòng)態(tài)生成函數(shù)對(duì)象, 函數(shù)作用域默認(rèn)設(shè)置為builtins.dict勤庐，并合并namespace的變量
:param function_express: 函數(shù)表達(dá)式好港，示例 '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)
在生成方法對(duì)象時(shí)要注意namespace的問題米罚，最好默認(rèn)傳 builtins.dict ，然后自定義的方法通過namespace參數(shù)傳進(jìn)去拔莱。

后續(xù)（yml測試文件自動(dòng)生成）
至此隘竭，框架的核心功能已經(jīng)完成了，經(jīng)過幾個(gè)項(xiàng)目的實(shí)踐动看，效果完全超過預(yù)期，寫起用例來不要太爽须误，運(yùn)行起來不要太快仇轻，測試報(bào)告也整的明明白白漂漂亮亮的，但我發(fā)現(xiàn)還是有些累篷店，為什么呢？
我目前做接口測試的流程是吭产，如果項(xiàng)目集成了swagger鸭轮，通過swagger去獲取接口信息，根據(jù)這些接口信息來手工起項(xiàng)目創(chuàng)建用例窃爷。這個(gè)過程很重復(fù)很繁瑣，因?yàn)槲覀兊挠美０逡呀?jīng)大致固定了医吊，其實(shí)用例之間就是一些參數(shù)比如目錄逮京、用例名稱、method等等的區(qū)別，那么這個(gè)過程我覺得完全可以自動(dòng)化览绿。
因?yàn)閟wagger有個(gè)網(wǎng)頁啊穗慕，我可以去提取關(guān)鍵信息來自動(dòng)創(chuàng)建.yml測試文件，就像搭起架子一樣逛绵，待項(xiàng)目架子生成后，我再去設(shè)計(jì)用例填傳參就可以了瓢对。
于是我試著去解析請(qǐng)求swagger首頁得到的HTML添吗，然后失望的是并沒有實(shí)際數(shù)據(jù)，后來猜想應(yīng)該是用了ajax碟联，打開瀏覽器控制臺(tái)的時(shí)鲤孵，我發(fā)現(xiàn)了 api-docs 的請(qǐng)求，一看果然是json數(shù)據(jù)普监，那么問題就簡單了，網(wǎng)頁分析都不用了毙玻。

import re
import os
import sys

from requests import Session

template ="""
args:

• {method}
• {api}
  kwargs:
• caseName: {caseName}
  {data_or_params}:
  {data}
  validator:
• json:
  successed: True
  """

def auto_gen_cases(swagger_url, project_name):
"""
根據(jù)swagger返回的json數(shù)據(jù)自動(dòng)生成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 = dir

    file += '.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 = _k
        api = k
        caseName = _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 += ' ' * 8
        except 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_)

現(xiàn)在要開始一個(gè)項(xiàng)目的接口測試覆蓋桑滩，只要該項(xiàng)目集成了swagger允睹，就能秒生成項(xiàng)目架子，測試人員只需要專心設(shè)計(jì)接口測試用例即可缭受，我覺得對(duì)于測試團(tuán)隊(duì)的推廣使用是很有意義的，也更方便了我這樣的懶人韭畸。

（文章來源于霍格沃茲測試學(xué)院）