软件测试开发实战｜接口自动化测试框架开发 _框架

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

接口测试是比较讲究效率的，测试人员会希望很快能得到结果反馈，然而接口的数量一般都很多，而且会越来越多，所以提高执行效率很有必要
接口测试的用例其实也可以用来兼做简单的压力测试，而压力测试需要并发
接口测试的用例有很多重复的东西，测试人员应该只需要关注接口测试的设计，这些重复劳动最好自动化来做
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 = https://www.isolves.com/it/cxkf/kj/2021-04-01/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