近期準(zhǔn)備優(yōu)先做接口測(cè)試的覆蓋柑土，為此需要開(kāi)發(fā)一個(gè)測(cè)試框架慢哈，經(jīng)過(guò)思考，這次依然想做點(diǎn)兒不一樣的東西观游。

接口測(cè)試是比較講究效率的搂捧，測(cè)試人員會(huì)希望很快能得到結(jié)果反饋，然而接口的數(shù)量一般都很多懂缕，而且會(huì)越來(lái)越多允跑，所以提高執(zhí)行效率很有必要
接口測(cè)試的用例其實(shí)也可以用來(lái)兼做簡(jiǎn)單的壓力測(cè)試，而壓力測(cè)試需要并發(fā)
接口測(cè)試的用例有很多重復(fù)的東西搪柑，測(cè)試人員應(yīng)該只需要關(guān)注接口測(cè)試的設(shè)計(jì)聋丝，這些重復(fù)勞動(dòng)最好自動(dòng)化來(lái)做
pytest和allure太好用了，新框架要集成它們
接口測(cè)試的用例應(yīng)該盡量簡(jiǎn)潔工碾，最好用yaml弱睦，這樣數(shù)據(jù)能直接映射為請(qǐng)求數(shù)據(jù)，寫(xiě)起用例來(lái)跟做填空題一樣渊额，便于向沒(méi)有自動(dòng)化經(jīng)驗(yàn)的成員推廣 加上我對(duì)Python的協(xié)程很感興趣况木，也學(xué)了一段時(shí)間，一直希望學(xué)以致用旬迹，所以http請(qǐng)求我決定用aiohttp來(lái)實(shí)現(xiàn)火惊。 但是pytest是不支持事件循環(huán)的，如果想把它們結(jié)合還需要一番功夫奔垦。于是繼續(xù)思考屹耐，思考的結(jié)果是其實(shí)我可以把整個(gè)事情分為兩部分。 第一部分宴倍，讀取yaml測(cè)試用例张症，http請(qǐng)求測(cè)試接口仓技，收集測(cè)試數(shù)據(jù)。 第二部分俗他，根據(jù)測(cè)試數(shù)據(jù)脖捻，動(dòng)態(tài)生成pytest認(rèn)可的測(cè)試用例，然后執(zhí)行兆衅，生成測(cè)試報(bào)告地沮。 這樣一來(lái)，兩者就能完美結(jié)合了羡亩，也完美符合我所做的設(shè)想摩疑。想法既定，接著 就是實(shí)現(xiàn)了畏铆。
第一部分（整個(gè)過(guò)程都要求是異步非阻塞的）
讀取yaml測(cè)試用例
一份簡(jiǎn)單的用例模板我是這樣設(shè)計(jì)的雷袋，這樣的好處是，參數(shù)名和aiohttp.ClientSession().request(method,url,**kwargs)是直接對(duì)應(yīng)上的辞居，我可以不費(fèi)力氣的直接傳給請(qǐng)求方法楷怒，避免各種轉(zhuǎn)換，簡(jiǎn)潔優(yōu)雅瓦灶，表達(dá)力又強(qiáng)鸠删。

args:

• post
• /xxx/add
  kwargs:
• caseName: 新增xxx
  data:
  name: ${gen_uid(10)}
  validator:
• json:
  successed: True
  異步讀取文件可以使用aiofiles這個(gè)第三方庫(kù)，yaml_load是一個(gè)協(xié)程贼陶，可以保證主進(jìn)程讀取yaml測(cè)試用例時(shí)不被阻塞刃泡，通過(guò)await yaml_load()便能獲取測(cè)試用例的數(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)用形式的語(yǔ)法
pattern_function = re.compile(r'^\${([A-Za-z_]+\w*\(.*\))}$')
pattern_function2 = re.compile(r'^\${(.*)}$')
# 匹配取默認(rèn)值的語(yǔ)法
pattern_function3 = re.compile(r'^\$\((.*)\)$')

def my_iter(data):
    """
    遞歸測(cè)試用例碉怔，根據(jù)不同數(shù)據(jù)類(lèi)型做相應(yīng)處理烘贴，將模板語(yǔ)法轉(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)

可以看到，測(cè)試用例還支持一定的模板語(yǔ)法眨层，如(a:b)等，這能在很大程度上拓展測(cè)試人員用例編寫(xiě)的能力

http請(qǐng)求測(cè)試接口
http請(qǐng)求可以直接用aiohttp.ClientSession().request(method,url,**kwargs)趴樱，http也是一個(gè)協(xié)程，可以保證網(wǎng)絡(luò)請(qǐng)求時(shí)不被阻塞酪捡，通過(guò)await http()便可以拿到接口測(cè)試數(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
        }

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

async def entrace(test_cases, loop, semaphore=None):
"""
http執(zhí)行入口
:param test_cases:
:param semaphore:
:return:
"""
res = BXMDict()
# 在CookieJar的update_cookies方法中逛薇，如果unsafe=False并且訪問(wèn)的是IP地址捺疼，客戶端是不會(huì)更新cookie信息
# 這就導(dǎo)致session不能正確處理登錄態(tài)的問(wèn)題
# 所以這里使用的cookie_jar參數(shù)使用手動(dòng)生成的CookieJar對(duì)象，并將其unsafe設(shè)置為T(mén)rue
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=''):
"""
一份測(cè)試用例執(zhí)行的全過(guò)程永罚，包括讀取.yml測(cè)試用例啤呼，執(zhí)行http請(qǐng)求卧秘，返回請(qǐng)求結(jié)果
所有操作都是異步非阻塞的
:param session: session會(huì)話
:param case_dir: 用例目錄
:param case_name: 用例名稱(chēng)
: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é)果收集中官扣，我用測(cè)試用例目錄來(lái)對(duì)結(jié)果進(jìn)行了分類(lèi)翅敌，這為接下來(lái)的自動(dòng)生成pytest認(rèn)可的測(cè)試用例打下了良好的基礎(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)可的測(cè)試用例
首先說(shuō)明下pytest的運(yùn)行機(jī)制蚯涮，pytest首先會(huì)在當(dāng)前目錄下找conftest.py文件，如果找到了卖陵，則先運(yùn)行它遭顶，然后根據(jù)命令行參數(shù)去指定的目錄下找test開(kāi)頭或結(jié)尾的.py文件，如果找到了泪蔫，如果找到了棒旗，再分析fixture，如果有session或module類(lèi)型的撩荣，并且參數(shù)autotest=True或標(biāo)記了pytest.mark.usefixtures(a...)铣揉，則先運(yùn)行它們；再去依次找類(lèi)婿滓、方法等老速，規(guī)則類(lèi)似。大概就是這樣一個(gè)過(guò)程凸主。
可以看出橘券，pytest測(cè)試運(yùn)行起來(lái)的關(guān)鍵是，必須有至少一個(gè)被pytest發(fā)現(xiàn)機(jī)制認(rèn)可的testxx.py文件卿吐，文件中有TestxxClass類(lèi)旁舰，類(lèi)中至少有一個(gè)def testxx(self)方法。
現(xiàn)在并沒(méi)有任何pytest認(rèn)可的測(cè)試文件嗡官，所以我的想法是先創(chuàng)建一個(gè)引導(dǎo)型的測(cè)試文件箭窜，它負(fù)責(zé)讓pytest動(dòng)起來(lái)⊙苄龋可以用pytest.skip()讓其中的測(cè)試方法跳過(guò)磺樱。然后我們的目標(biāo)是在pytest動(dòng)起來(lái)之后，怎么動(dòng)態(tài)生成用例婆咸，然后發(fā)現(xiàn)這些用例竹捉，執(zhí)行這些用例，生成測(cè)試報(bào)告尚骄，一氣呵成块差。

test_bootstrap.py

import pytest

class TestStarter(object):

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

我想到的是通過(guò)fixture，因?yàn)閒ixture有setup的能力，這樣我通過(guò)定義一個(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('此為測(cè)試啟動(dòng)方法, 不執(zhí)行')

pytest有個(gè)--rootdir參數(shù)轧坎，該fixture的核心目的就是，通過(guò)--rootdir獲取到目標(biāo)目錄训裆，找出里面的.yml測(cè)試文件眶根，運(yùn)行后獲得測(cè)試數(shù)據(jù)，然后為每個(gè)目錄創(chuàng)建一份testxx.py的測(cè)試文件边琉，文件內(nèi)容就是content變量的內(nèi)容属百，然后把這些參數(shù)再傳給pytest.main()方法執(zhí)行測(cè)試用例的測(cè)試，也就是在pytest內(nèi)部再運(yùn)行了一個(gè)pytest变姨！最后把生成的測(cè)試文件刪除族扰。注意該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):
"""
測(cè)試用例生成處理
: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('{}接口測(cè)試({}項(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

可以看到，測(cè)試文件中有一個(gè)TestxxAPI的類(lèi)砍鸠，它只有一個(gè)test_cases_data屬性扩氢，并沒(méi)有testxx方法，所以還不是被pytest認(rèn)可的測(cè)試用例爷辱，根本運(yùn)行不起來(lái)录豺。那么它是怎么解決這個(gè)問(wèn)題的呢？答案就是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)生成測(cè)試用例
"""

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è)元類(lèi)双饥，它讀取test_cases_data屬性的內(nèi)容，然后動(dòng)態(tài)生成方法對(duì)象弟断，每一個(gè)接口都是單獨(dú)一個(gè)方法咏花，在相繼被allure的細(xì)粒度測(cè)試報(bào)告功能和pytest提供的參數(shù)化測(cè)試功能裝飾后，把該方法對(duì)象賦值給test+api的類(lèi)屬性阀趴，也就是說(shuō)昏翰，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的問(wèn)題，最好默認(rèn)傳builtins.dict，然后自定義的方法通過(guò)namespace參數(shù)傳進(jìn)去攻柠。

后續(xù)（yml測(cè)試文件自動(dòng)生成）
至此球订，框架的核心功能已經(jīng)完成了，經(jīng)過(guò)幾個(gè)項(xiàng)目的實(shí)踐瑰钮，效果完全超過(guò)預(yù)期冒滩，寫(xiě)起用例來(lái)不要太爽，運(yùn)行起來(lái)不要太快浪谴，測(cè)試報(bào)告也整的明明白白漂漂亮亮的开睡，但我發(fā)現(xiàn)還是有些累，為什么呢苟耻？
我目前做接口測(cè)試的流程是篇恒，如果項(xiàng)目集成了swagger，通過(guò)swagger去獲取接口信息凶杖，根據(jù)這些接口信息來(lái)手工起項(xiàng)目創(chuàng)建用例胁艰。這個(gè)過(guò)程很重復(fù)很繁瑣，因?yàn)槲覀兊挠美０逡呀?jīng)大致固定了智蝠，其實(shí)用例之間就是一些參數(shù)比如目錄腾么、用例名稱(chēng)、method等等的區(qū)別杈湾，那么這個(gè)過(guò)程我覺(jué)得完全可以自動(dòng)化解虱。
因?yàn)閟wagger有個(gè)網(wǎng)頁(yè)啊，我可以去提取關(guān)鍵信息來(lái)自動(dòng)創(chuàng)建.yml測(cè)試文件漆撞，就像搭起架子一樣殴泰，待項(xiàng)目架子生成后，我再去設(shè)計(jì)用例填傳參就可以了叫挟。
于是我試著去解析請(qǐng)求swagger首頁(yè)得到的HTML艰匙，然后失望的是并沒(méi)有實(shí)際數(shù)據(jù)，后來(lái)猜想應(yīng)該是用了ajax抹恳，打開(kāi)瀏覽器控制臺(tái)的時(shí)员凝，我發(fā)現(xiàn)了api-docs的請(qǐng)求，一看果然是json數(shù)據(jù)奋献，那么問(wèn)題就簡(jiǎn)單了健霹，網(wǎng)頁(yè)分析都不用了。

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測(cè)試用例模板
: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)在要開(kāi)始一個(gè)項(xiàng)目的接口測(cè)試覆蓋瓶蚂，只要該項(xiàng)目集成了swagger糖埋，就能秒生成項(xiàng)目架子，測(cè)試人員只需要專(zhuān)心設(shè)計(jì)接口測(cè)試用例即可窃这，我覺(jué)得對(duì)于測(cè)試團(tuán)隊(duì)的推廣使用是很有意義的瞳别，也更方便了我這樣的懶人。

https://testerhome.com/topics/16206