近期準備優先做介面測驗的覆寫，為此需要開發一個測驗框架，經過思考，這次依然想做點兒不一樣的東西，
- 介面測驗是比較講究效率的，測驗人員會希望很快能得到結果反饋，然而介面的數量一般都很多，而且會越來越多，所以提高執行效率很有必要
- 介面測驗的用例其實也可以用來兼做簡單的壓力測驗，而壓力測驗需要并發
- 介面測驗的用例有很多重復的東西，測驗人員應該只需要關注介面測驗的設計，這些重復勞動最好自動化來做
pytest和allure太好用了，新框架要集成它們
- 介面測驗的用例應該盡量簡潔，最好用yaml，這樣資料能直接映射為請求資料，寫起用例來跟做填空題一樣，便于向沒有自動化經驗的成員推廣
加上我對Python的協程很感興趣，也學了一段時間，一直希望學以致用，所以http請求我決定用aiohttp來實作，
但是pytest是不支持事件回圈的，如果想把它們結合還需要一番功夫，于是繼續思考，思考的結果是其實我可以把整個事情分為兩部分，
第一部分，讀取yaml測驗用例，http請求測驗介面，收集測驗資料，
第二部分，根據測驗資料，動態生成pytest認可的測驗用例，然后執行，生成測驗報告，
這樣一來，兩者就能完美結合了，也完美符合我所做的設想，想法既定，接著 就是實作了，
第一部分（整個程序都要求是異步非阻塞的）
讀取yaml測驗用例
一份簡單的用例模板我是這樣設計的，這樣的好處是，引數名和aiohttp.ClientSession().request(method,url,**kwargs)是直接對應上的，我可以不費力氣的直接傳給請求方法，避免各種轉換，簡潔優雅，表達力又強，
args: - post - /xxx/add kwargs: - caseName: 新增xxx data: 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 _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) 等，這能在很大程度上拓展測驗人員用例撰寫的能力
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 = 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) 來控制并發
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 = https://www.cnblogs.com/cshuangxi/p/{}""" 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引數傳進去，
后續（yml測驗檔案自動生成）
至此，框架的核心功能已經完成了，經過幾個專案的實踐，效果完全超過預期，寫起用例來不要太爽，運行起來不要太快，測驗報告也整的明明白白漂漂亮亮的，但我發現還是有些累，為什么呢？
我目前做介面測驗的流程是，如果專案集成了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，就能秒生成專案架子，測驗人員只需要專心設計介面測驗用例即可，我覺得對于測驗團隊的推廣使用是很有意義的，也更方便了我這樣的懶人，
如果你
①從事功能測驗，想進階自動化測驗
②在測驗界混了1、2年，依然不會敲代碼
③面試大廠卻屢屢碰壁
我推薦一個群吧！來吧~~測驗員，313782132（Q群里有技術大牛一起交流分享，學習資源的價值取決于你的行動，莫做“收藏家”）獲取更多大廠技術、面試資料
最后：
凡事要趁早，特別是技術行業，一定要提升技術功底，如果文章對你有幫助，麻煩伸出發財小手點個贊，感謝您的支持，你的點贊是我持續更新的動力，
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/242236.html
標籤：其他
下一篇：介面測驗基礎知識