Pytest-BDD实现接口自动化测试，并附全部代码

引言

在之前的文章中简单的介绍了怎么使用Pytest-BDD进行接口测试，可以参考《pytest-bdd 行为驱动自动化测试》。本篇文章主要介绍使用Pytest-BDD实现接口自动化测试。后面的文章会介绍生成测试报告，和流程性接口测试。

feature文件

首先我们先整理好一个接口测试，需要用到的场景、参数等信息，写到feature文件中，以便我们根据各种测试中的行为来进行脚本实现。

demo.feature

Feature: 测试模块Scenario: demo接口测试Given 初始化Given 登录When 调用 "/test" 接口# 这里只配置接口路径，具体的域名和IP，通过config文件配置，以便用了切换环境When 使用 "post" 请求# 目前支持的请求方式 post get put deleteWhen 参数类型 "json"# 目前支持的参数类型 params json form-data form-urlencodedWhen 请求头 "{'Content-type': 'application/json'}"When 参数 "{'id': '1'}"When 校验类型 "包含校验"# 支持的校验类型： 包含校验 相等校验 字段类型校验 字段值校验 状态码校验When 校验文本 "成功"When 校验字段类型为 "int"# 字段类型：int str float list dict boolWhen 校验字段 "loginInfo"When 上传文件 "test.txt"When 退出Then 调用成功

从上面看出，我们要实现的行为有初始化、登录，步骤中需要实现接口地址、请求方式、参数和断言。目前我们需要的接口测试行为已经列出来，下面就是对这些行为的编码实现。

py文件

首先，我们来实现feature文件中的{初始化}。将feature文件中的参数进行初始化。初始化调用api工具的函数。

class ApiTest:def __init__(self):self.api = Noneself.methods = 'get'self.data_type = 'params'self.params = {}self.headers = {}self.test_result_type = Noneself.result_text = Noneself.result_type = Noneself.result_data_type = Noneself.result_key = Noneself.file = Noneself.is_logout = Falsedef call_api(self, test_body):response = api_test(test_body)return [e-mail]
@given('初始化')
def api_tool():return ApiTest()

实现feature文件中的{登录}。登录的目的是拿到接口的授权，所以我们调用登录方法，拿到鉴权后，将鉴权保存到headers中，以便后面进行接口请求时使用。具体实现登录的方法就不展示了，根据自己的项目来实现就可以。

@given('登录')
def login(api_tool):headers = get_token()for key, value in headers.items():api_tool.headers[key] = value

实现feature文件中{调用 “/test” 接口}，拿到feature文件中的接口路径，然后赋值给api，后面接口工具会根据环境和接口路径拼接成完整的接口地址。

@when(parsers.parse('调用 "{url}" 接口'))
def api(api_tool, url):api_tool.api = url

实现feature文件中{使用 “post” 请求}，拿到feat文件中的请求方法，赋值给methods，后面接口工具使用该请求方法。

@when(parsers.parse('使用 "{methods}" 请求'))
def methods(api_tool, methods):api_tool.methods = methods

实现feature文件中{参数类型 “json”}

@when(parsers.parse('参数类型 "{data_type}"'))
def data_type(api_tool, data_type):api_tool.data_type = data_type

实现feature文件中{请求头 “{‘Content-type’: ‘application/json’}”}，因为通过feature文件拿到的参数都是字符串类型，所以要将字符串转成dict，并添加到请求头中。

@when(parsers.parse('请求头 "{headers}"'))
def headers(api_tool, headers):headers_dict = ast.literal_eval(headers)for key, value in headers_dict.items():api_tool.headers[key] = value

实现feature文件中{参数 “{‘id’: ‘1’}”}，将字符串类型的参数转成dict。

@when(parsers.parse('参数 "{params}"'))
def params(api_tool, params):api_tool.params = ast.literal_eval(params)

实现feature文件中{校验类型 “包含校验”}，根据映射关系，将校验类型数据赋值给test_result_type。

test_result_type_list = {'相等校验': 0,'包含校验': 1,'字段类型校验': 2,'字段值校验': 3,'状态码校验': 4
}@when(parsers.parse('校验类型 "{test_result_type}"'))
def result_test(api_tool, test_result_type):api_tool.test_result_type = test_result_type_list[test_result_type]

实现feature文件中的{校验文本 “成功”}、{校验字段类型为 “int”}、{校验字段 “loginInfo”}，这三个加上校验类型，是断言工具需要的，如果是相当校验，会将测试接口返回的response和result_text进行相等校验，如果是包含校验，会校验测试接口的response是否包含result_text，如果是字段类型校验，会校验result_key的类型是否为result_type，如果是字段值校验，会校验测试接口的response返回的result_key的value值，是否等于result_text，如果是状态码校验，会校验response的状态码，是否为result_text。

@when(parsers.parse('校验文本 "{result_text}"'))
def result_text(api_tool, result_text):api_tool.result_text = result_text@when(parsers.parse('校验字段类型为 "{result_type}"'))
def result_type(api_tool, result_type):api_tool.result_type = result_type@when(parsers.parse('校验字段 "{result_key}"'))
def result_key(api_tool, result_key):

实现feature文件中{When 上传文件 “test.txt”}，这个步骤只有接口需要上传文件时，才需要写，上传文件放到项目的statics目录下。

@when(parsers.parse('上传文件 "{files}"'))
def file(api_tool, files):api_tool.file = files

实现feature文件中的{退出}，is_logout为True时，调用完测试接口会执行退出登录操作。

@when(parsers.parse("退出"))
def logout(api_tool):api_tool.is_logout = True

实现feature文件中{调用成功}，将上面步骤保存的参数，拼装成两个dict，test_data为断言工具需要的参数，test_body为接口请求需要的参数，然后调用封装好的接口测试工具和断言工具，进行接口测试。

@then('调用成功')
def asserts(api_tool):test_data = {'test_result': {'text': api_tool.result_text, 'type': api_tool.result_type, 'result_type': api_tool.result_type, 'key': api_tool.result_key}}test_body = {'URL': api_tool.api, 'method': api_tool.methods, 'data_type': api_tool.data_type, 'headers': api_tool.headers, 'params': api_tool.params, 'file': api_tool.file}response = api_tool.call_api(test_body)if api_tool.is_logout:do_logout()if api_tool.test_result_type is not None:assert_tool(response, test_data)

登录和退出的函数就不介绍了，根据自己的项目来实现就可以，主要介绍一下接口测试工具和断言工具。

接口测试工具

接口测试工具中，读取配置文件的参数，放到文章最后，会把测试文件的内容，还有yaml工具的代码附上。

import requests
import os
import json
from common.yaml_tool import read_yamldef api_test(test_body):config = read_yaml('config.yaml')config_env = config['env']data_type = test_body['data_type'].lower()url = config_env + test_body['URL']method = test_body['method'].lower()data = test_body['params']headers = test_body['headers']if headers is None:headers = {}file = test_body['file']if data_type.lower() == 'params':response = params(url, method, data, headers)elif data_type == 'json':response = json(url, method, data, headers)elif data_type == 'form-data':response = formdata(url, method, data, headers, file)elif data_type == 'form-urlencoded':response = urlencoded(url, method, data, headers)else:return {'code': '5000', 'msg': '测试脚本暂时不支持的参数类型'}# print('接口返回内容：' + response.text)return responsedef params(url, method, data, headers):try:if method == 'post':response = requests.post(url, params=data, headers=headers)elif method == 'put':response = requests.put(url, params=data, headers=headers)elif method == 'get':response = requests.get(url, params=data, headers=headers)elif method == 'delete':response = requests.delete(url, params=data, headers=headers)else:response = {'code': '5000', 'msg': '测试脚本暂时不支持的请求方法'}response.encoding = 'utf-8'return responseexcept Exception as e:return {"code": "5000", "msg": str(e)}def json(url, method, data, headers):try:if method == 'post':response = requests.post(url, data=data, headers=headers)elif method == 'put':response = requests.put(url, data=data, headers=headers)elif method == 'get':response = requests.get(url, data=data, headers=headers)elif method == 'delete':response = requests.delete(url, data=data, headers=headers)else:response = {'code': '5000', 'msg': '测试脚本暂时不支持的请求方法'.encode('utf-8')}return responseexcept Exception as e:return json.dumps({'code': '5000', 'msg': '测试脚本报错：' + str(e)})def urlencoded(url, method, data, headers):try:if method == 'post':response = requests.post(url, data=data, headers=headers)elif method == 'put':response = requests.put(url, data=data, headers=headers)elif method == 'get':response = requests.get(url, data=data, headers=headers)elif method == 'delete':response = requests.delete(url, data=data, headers=headers)else:response = {'code': '5000', 'msg': '测试脚本暂时不支持的请求方法'.encode('utf-8')}return responseexcept Exception as e:return {'code': '5000', 'msg': '测试脚本报错：' + str(e)}def formdata(url, method, data, headers, file):if file:file_name = filename = os.path.splitext(file_name)[-1]if name == '.doc':file_type = 'application/msword'elif name == '.docx':file_type = 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'elif name == '.pdf':file_type = 'application/pdf'elif name == '.jpg' or name == 'jpeg':file_type = 'image/jpeg'elif name == '.png':file_type = 'image/png'elif name == '.ppt':file_type = 'application/vnd.ms-powerpoint'elif name == '.xls':file_type = 'application/vnd.ms-excel'elif name == '.xlsx':file_type = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'elif name == '.txt':file_type = 'text/plain'path = os.path.dirname(os.path.dirname(__file__))file = [('file', ('Test'+name, open(path + '/statics/' + file_name, 'rb'), file_type))]try:if method == 'post':response = requests.post(url, data=data, files=file, headers=headers)elif method == 'put':response = requests.put(url, data=data, files=file, headers=headers)elif method == 'get':response = requests.get(url, data=data, files=file, headers=headers)elif method == 'delete':response = requests.delete(url, data=data, files=file, headers=headers)else:response = {'code': '5000', 'msg': '测试脚本暂时不支持的请求方法'.encode('utf-8')}return responseexcept Exception as e:return {'code': '5000', 'msg': '测试脚本报错：' + str(e)}

断言工具

import json
from requests import Responsedef assert_tool(response, test_result):response_body = Noneif type(response) is dict:response_body = responseelse:response_body = response.texttext = test_result['test_result']['text']types = test_result['test_result']['type']type_map = {'int': int,'float': float,'str': str,'list': list,'dict': dict,'bool': bool,}if types == 0:assert text == str(response_body)elif types == 1:assert text in str(response_body)elif types == 2:result_type = test_result['test_result']['result_type']key_str = test_result['test_result']['key']assert isinstance(analysis_dict(json.loads(response_body), key_str), type_map.get(result_type))elif types == 3:key_str = test_result['test_result']['key']assert text == str(analysis_dict(json.loads(response_body), key_str))elif types == 4:if type(response) is Response:status = response.status_codeassert text == statuselse:assert text == str(response_body)def analysis_dict(response: dict, key_str):if key_str in response:return response[key_str]for key, value in response.items():if isinstance(value, dict):result = analysis_dict(value, key_str)if result is not None:return resultelif isinstance(value, list):for item in value:if isinstance(item, dict):result = analysis_dict(item, key_str)if result is not None:return resultreturn None

本接口自动化测试工具，基本实现了接口测试和断言。后面会加上测试报告，还有流程性接口测试（如接口2的参数，需要从接口1中获取）

配置文件

env: http://yourdomin.com
login:URL: /loginheaders:content-type: application/jsonparams:username: usernamepassword: password
logout:URL: /logoutheaders:content-type: application/json

yam文件工具

import yamldef read_yaml(yaml_file):with open(yaml_file, 'r', encoding='utf-8') as file:test_data = yaml.safe_load(file)return test_datadef wirte_config(config, config_file):with open(config_file, 'w', encoding='utf-8') as file:yaml.dump(config, file, default_flow_style=False)

python文件

实现feature文件的整体python文件

import pytest
from pytest_bdd import scenarios, given, when, then, parsersfrom common.api_tool import api_test
from common.token_api import get_token, get_cms_token, do_logout
from common.assert_tool import assert_tool
import ast
import ostest_result_type_list = {'相等校验': 0,'包含校验': 1,'字段类型校验': 2,'字段值校验': 3,'状态码校验': 4
}class ApiTest:def __init__(self):self.api = Noneself.methods = 'get'self.data_type = 'params'self.params = {}self.headers = {}self.test_result_type = Noneself.result_text = Noneself.result_type = Noneself.result_data_type = Noneself.result_key = Noneself.file = Noneself.is_logout = Falsedef call_api(self, test_body):response = api_test(test_body)return responsefor root, dirs, files in os.walk('case/'):for case in files:scenarios(root + case)@pytest.fixture
@given('初始化')
def api_tool():return ApiTest()@given('登录')
def login(api_tool):headers = get_token()for key, value in headers.items():api_tool.headers[key] = value@when(parsers.parse('调用 "{url}" 接口'))
def api(api_tool, url):api_tool.api = url@when(parsers.parse('使用 "{methods}" 请求'))
def methods(api_tool, methods):api_tool.methods = methods@when(parsers.parse('参数类型 "{data_type}"'))
def data_type(api_tool, data_type):api_tool.data_type = data_type@when(parsers.parse('请求头 "{headers}"'))
def headers(api_tool, headers):headers_dict = ast.literal_eval(headers)for key, value in headers_dict.items():api_tool.headers[key] = value@when(parsers.parse('参数 "{params}"'))
def params(api_tool, params):api_tool.params = ast.literal_eval(params)@when(parsers.parse('校验类型 "{test_result_type}"'))
def result_test(api_tool, test_result_type):api_tool.test_result_type = test_result_type_list[test_result_type]@when(parsers.parse('校验文本 "{result_text}"'))
def result_text(api_tool, result_text):api_tool.result_text = result_text@when(parsers.parse('校验字段类型为 "{result_type}"'))
def result_type(api_tool, result_type):api_tool.result_type = result_type@when(parsers.parse('校验字段 "{result_key}"'))
def result_key(api_tool, result_key):api_tool.result_key = result_key@when(parsers.parse('上传文件 "{files}"'))
def file(api_tool, files):api_tool.file = files@when(parsers.parse("退出"))
def logout(api_tool):api_tool.is_logout = True@then('调用成功')
def asserts(api_tool):test_data = {'test_result': {'text': api_tool.result_text, 'type': api_tool.result_type, 'result_type': api_tool.result_type, 'key': api_tool.result_key}}test_body = {'URL': api_tool.api, 'method': api_tool.methods, 'data_type': api_tool.data_type, 'headers': api_tool.headers, 'params': api_tool.params, 'file': api_tool.file}response = api_tool.call_api(test_body)if api_tool.is_logout:do_logout()if api_tool.test_result_type is not None:assert_tool(response, test_data)