近期準備優先作接口測試的覆蓋,爲此須要開發一個測試框架,通過思考,此次依然想作點兒不同的東西。ajax
讀取yaml測試用例express
一份簡單的用例模板我是這樣設計的,這樣的好處是,參數名和aiohttp.ClientSession().request(method,url,**kwargs)是直接對應上的,我能夠不費力氣的直接傳給請求方法,避免各類轉換,簡潔優雅,表達力又強。json
args: - post - /xxx/add kwargs: - caseName: 新增xxx data: name: ${gen_uid(10)} validator: - json: successed: True
異步讀取文件可使用aiofiles這個第三方庫,yaml_load是一個協程,能夠保證主進程讀取yaml測試用例時不被阻塞,經過await yaml_load()
便能獲取測試用例的數據bootstrap
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 _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)
能夠看到,測試用例還支持必定的模板語法,如${function}
、$(a:b)
等,這能在很大程度上拓展測試人員用例編寫的能力api
http請求測試接口瀏覽器
http請求能夠直接用aiohttp.ClientSession().request(method,url,**kwargs)
,http也是一個協程,能夠保證網絡請求時不被阻塞,經過await http()
即可以拿到接口測試數據cookie
async def http(domain, *args, **kwargs): """ http請求處理器 :param domain: 服務地址 :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}) # 拼接服務地址和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 }
收集測試數據網絡
協程的併發真的很快,這裏爲了不服務響應不過來致使熔斷,能夠引入asyncio.Semaphore(num)
來控制併發session
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設置爲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=''): """ 一份測試用例執行的全過程,包括讀取.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.py import pytest class TestStarter(object): def test_start(self): pytest.skip('此爲測試啓動方法, 不執行')
我想到的是經過fixture,由於fixture有setup的能力,這樣我經過定義一個scope爲session的fixture,而後在TestStarter上面標記use,就能夠在導入TestStarter以前預先處理一些事情,那麼我把生成用例的操做放在這個fixture裏就能完成目標了。
# test_bootstrap.py import 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 allure from 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 = 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
能夠看到,測試文件中有一個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' + 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
是一個元類,它讀取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參數傳進去。
至此,框架的核心功能已經完成了,通過幾個項目的實踐,效果徹底超過預期,寫起用例來不要太爽,運行起來不要太快,測試報告也整的明明白白漂漂亮亮的,但我發現仍是有些累,爲何呢?
我目前作接口測試的流程是,若是項目集成了swagger,經過swagger去獲取接口信息,根據這些接口信息來手工起項目建立用例。這個過程很重複很繁瑣,由於咱們的用例模板已經大體固定了,其實用例之間就是一些參數好比目錄、用例名稱、method等等的區別,那麼這個過程我以爲徹底能夠自動化。
由於swagger有個網頁啊,我能夠去提取關鍵信息來自動建立.yml測試文件,就像搭起架子同樣,待項目架子生成後,我再去設計用例填傳參就能夠了。
因而我試着去解析請求swagger首頁獲得的HTML,而後失望的是並無實際數據,後來猜測應該是用了ajax,打開瀏覽器控制檯的時,我發現了api-docs
的請求,一看果真是json數據,那麼問題就簡單了,網頁分析都不用了。
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): """ 根據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 = 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_)
如今要開始一個項目的接口測試覆蓋,只要該項目集成了swagger,就能秒生成項目架子,測試人員只須要專心設計接口測試用例便可,我以爲對於測試團隊的推廣使用是頗有意義的,也更方便了我這樣的懶人。