python+pytest 之接口测试框架搭建

大家好，我是讯享网，很高兴认识大家。

一. 接口框架介绍

接口框架主要由python+request+pytest+yaml+allure搭建，集成了logging模块，框架的目录结构如下：

讯享网

项目目录结构

api – 模仿PO模式, 抽象出页面类, 页面类内包含页面所包含所有接口, 并封装成方法可供其他模块直接调用
config – 配置文件目录，只要有setting.ini，只要用于存储项目需要的配置信息
data – 测试数据目录
logs – 日志
reports – 测试报告
testcases– 测试脚本存放目录
common– 工具类目录
main_run.py – 命令行启动入口
pytest.ini – pytest测试框架配置文件
README.md – 开发说明文档

二. 接口框架代码实现

1.request的封装：

1.1 request的介绍

request的get post put delete 方法示例

url：请的的路径，一般是项目的跟路径和接口路径才是完整的请求路径
data/params：请求的数据，有时候是json，有时候是字典，具体看接口的定义
headers：请求头，有些接口需要加入一些特殊的请求头，比如有些接口依赖与登录的token,会将token 放入headers

import requests #requests get 方法 res = requests.get(url="",params="",headers="") #requests post 方法 res1 = requests.post(url="",data="",headers="") #requests put 方法 res2 = requests.put(url="",data="",headers="") #requests delete 方法 res3 = requests.delete(url="",data="",headers="")

讯享网

所以抽取以上方法的共性，来封装请求基类

common/httpClient.py

讯享网import json import requests from common.parseData import base_data from common.log_util import logger api_root_url = base_data.get_ini_data()["host"]["api_sit_url"] class HttpClient: """ 封装requests的get/post/put/delete请求 """ def __init__(self): self.api_root_url = api_root_url pass # def send_request(self,url,method,kwargs): # """ # 测试用例执行调用的方法 # :param url: # :param method: # :param kwargs: # :return: # """ # return self.request(url,method,kwargs) def get(self,url,kwargs): return self.request(url,"GET",kwargs) def post(self, url, kwargs): return self.request(url, "POST", kwargs) def put(self, url, kwargs): return self.request(url, "PUT", kwargs) def delete(self,url,kwargs): return self.request(url,"DELETE",kwargs) def request(self,url,method,kwargs): """ 根据request的method来调用requests的具体方法 :param url: :param method: :param kwargs: :return: 返回response """ self.request_log(url,method,kwargs) #调用request_log方法来输出日志 if method=="GET": return requests.get(self.api_root_url+url,kwargs) if method == "POST": return requests.post(self.api_root_url + url, kwargs) if method == "PUT": return requests.put(self.api_root_url + url, kwargs) if method == "DELETE": return requests.delete(self.api_root_url + url, kwargs) def request_log(self,url,method,kwargs): """ 解包参数为字典，然后通过字典的get()方法（传入key）获取入参的请求数据 :param url: :param method: :param kwargs: :return: """ # get(key)方法在key（键）不在字典中时，可以返回默认值None或者设置的默认值 # 而dict[key]在key（键）不在字典中时，会触发KeyError异常 data = dict(kwargs).get("data")#kwargs是关键字位置参数，可以转化为字典，通过key来获取 json_data = dict(kwargs).get("json") params = dict(kwargs).get("params") hearders = dict(kwargs).get("headers") logger.info("接口的请求地址>>{}".format(self.api_root_url+url)) logger.info("接口的请求方法>>{}".format(method)) if data is not None: #json.dumps()将json对象转化为字符串，indent表示缩进两个字符显示 logger.info("接口请求的data参数>>>\n{}".format(json.dumps(data,indent=2))) if params is not None: logger.info("接口请求的params参数>>>\n{}".format(json.dumps(params,indent=2))) if json_data is not None: logger.info("接口请求的json_data参数>>>\n{}".format(json.dumps(json_data,indent=2))) if hearders is not None: logger.info("接口请求的hearders>>>\n{}".format(json.dumps(hearders,indent=2)))

以上代码中引用的logger 是用来做log打印，base_data是获取在配置文件中的url

common/log_util.py

import logging import os import time project_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # 项目的根路径 log_path = os.path.join(project_path, "logs") # log的存放路径 if not os.path.exists(log_path): os.mkdir(log_path) # 不存在log文件夹，则自动创建 class Logger(object): default_formats = { # 终端输出格式 'console_fmt': '%(log_color)s%(asctime)s-%(name)s-%(filename)s-[line:%(lineno)d]-%(levelname)s-[日志信息]: %(message)s', # 日志输出格式 'file_fmt': '%(asctime)s-%(filename)s-[line:%(lineno)d]-%(levelname)s-[日志信息]: %(message)s' } def __init__(self, name=None, log_level=logging.DEBUG): self.name = name # ①创建一个记录器 self.logger = logging.getLogger(self.name) self.logger.setLevel("INFO") # 设置日志级别为 'level'，即只有日志级别大于等于'level'的日志才会输出 self.log_formatter = logging.Formatter(self.default_formats["file_fmt"]) # 创建formatter self.console_formatter = logging.Formatter(self.default_formats["file_fmt"]) # 创建formatter # ②创建屏幕-输出到控制台，设置输出等级 self.streamHandler = logging.StreamHandler() self.streamHandler.setLevel("DEBUG") # ③创建log文件，设置输出等级 time_now = time.strftime('%Y_%m%d_%H', time.localtime()) + '.log' # log文件命名：2022_0402_21.log self.fileHandler = logging.FileHandler(os.path.join(log_path, time_now), 'a', encoding='utf-8') self.fileHandler.setLevel("DEBUG") # ④用formatter渲染这两个Handler self.streamHandler.setFormatter(self.console_formatter) self.fileHandler.setFormatter(self.log_formatter) # ⑤将这两个Handler加入logger内 if not self.logger.handlers: # 在新增handler时判断是否为空,解决log重复打印的问题 self.logger.addHandler(self.streamHandler) self.logger.addHandler(self.fileHandler) # def getLogger(self): # return self.logger logger = Logger().logger if __name__ == '__main__': logger.warning("warning") logger.error("error") logger.info("info") logger.debug("debug") logger.critical("critical")

base_data在common/parseData.py中,用来从ini和yaml中处理数据

common/parseData.py如下

讯享网import configparser import os import yaml # 项目的路径 project_path = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))) # 项目配置文件ini的路径 ini_path = os.path.join(project_path, "conf", "setting.ini") # 项目数据文件的路径 data_path = os.path.join(project_path, "data", "data.yaml") class ParseData: """ 用来封装读取各种类型文件的类 """ def __init__(self): self.ini_path = ini_path self.data_path = data_path def get_yaml_data(self): """ 读取yaml文件内容 :return: 返回yaml读取的数据 """ with open(self.data_path, mode='r',encoding="utf8") as file: data = yaml.safe_load(file) return data def get_ini_data(self): """ 获取数据文件的内容 :return: """ config = configparser.ConfigParser() config.read(self.ini_path, encoding="utf8") return config base_data = ParseData() # if __name__ == '__main__': # # base_data = ParseData() # print(base_data.get_ini_data()["host"]["api_sit_url"]) # print(base_data.get_yaml_data())

conf/setting.ini

[host] shouji_api_sit_url:https://api.binstd.com api_sit_url:http://admin.5istudy.online [mysql] MYSQL_HOST=47.110.151.xxx MYSQL_PORT=3306 MYSQL_USER=xxxx MYSQL_PASSWD=xxxx MYSQL_DB=xxxx

2.request基类的调用：

2.1 ApiUtil 用来调用实现各个api的url的写入
api/api_util.py

讯享网from common.httpClient import HttpClient class ApiUtil(HttpClient): """ Api继承HttpClient API主要是各个api的调用，主要是添加url """ def __init__(self): #super()继承父类的构造方法 super().__init__() #below methods work for the meikefresh apis def get_code(self,kwargs): #因为继承了HttpClient，所以可以调用HttpClient的post方法 return self.post("/code/",kwargs) def register_mobile(self,kwargs): return self.post("/users/", kwargs) def user_login(self,kwargs): return self.post("/login/", kwargs) def banner(self,kwargs): return self.get("/banners/",kwargs) def shopping_cart(self,kwargs): return self.post("/shopcarts/",kwargs) api_util = ApiUtil()

2.2 调用接口封装的方法，处理response（process_response（）就是对response的处理）

api/meikefresh_api.py

from api.api_util import api_util from common.ResponseUtil import process_response """ 此文件为美客生鲜项目的api的方法 """ def send_code(json_data): """ 获取短信验证码 :param json_data: :return: """ response = api_util.get_code(json=json_data) return process_response(response) def register(mobile,code): """ 用户注册方法 :param mobile: :param code: :return: """ json_data={ "code": str(code), "password": "", "username": str(mobile) } response = api_util.register_mobile(json=json_data) return process_response(response) def login(username,password): """ 用户登录方法 :param username: :param password: :return: """ json_data = { 'username':username, 'password':password } response = api_util.user_login(json=json_data) return process_response(response) """googs center""" def load_banners(): """ 加载banner页面 :return: """ response = api_util.banner() return process_response(response) def add_shopping_cart(json_data,token): """ 加入购物车需要先登录 :param json_data: :param token: :return: """ headers = { "Authorization":"JWT "+token } response = api_util.shopping_cart(json=json_data,headers=headers) return process_response(response)

common/ResponseUtil.py

讯享网import json from api.ResultBase import ResultResponse from common.log_util import logger def process_response(response): """ 将response处理下 :param response: :return: """ if response.status_code==200 or response.status_code==201: #构造字典{success:True} ResultResponse.success = True ResultResponse.body = response.json() # logger.info("组装的ResultResponse是",ResultResponse) else: ResultResponse.success = False logger.info("接口状态码不是2开头的") #response.status_code不是200和201的时候，打印log信息 logger.info("接口的返回内容是>>\n{}:".format(json.dumps(response.json(),ensure_ascii=False,indent=2))) return ResultResponse

ResultResponse 类 用来中转response的，因为返回的数据是response.json() 没有办法来断言response.status_code,所以封装这个类来保存ResultResponse.success=True ResultResponse.body=response.json() 来给用例断言的时候调用

common/ResponseUtil.py

讯享网class ResultResponse: """ 用来中转response的，因为返回的数据是response.json() 没有办法来断言response.status_code,所以封装这个类来保存ResultResponse.success=True ResultResponse.body=response.json() """ def __init__(self): pass

2.3. 用例的调用

1.调用接口实现登录

调试登录用例

testcases/usercenter/test_user.py

import allure import pytest from api.meikefresh_api import send_code, register, login, add_shopping_cart from testcases.conftest import get_base_data from testcases.usercenter.conftest import get_code, delete_user, delete_code, get_goods_num @allure.feature("用户中心模块") class TestUser: # 参数化实现登录 @pytest.mark.parametrize("username,password", get_base_data()["user_login"]) @allure.story("用户使用手机进行注册后登录") @allure.title("用户登录") def test_user_login(self, username, password): result = login(username, password) assert result.success is True # token每次的值都不一样，没有办法断言具体的值，所以断言不为空有内容即可 assert len(result.body['token']) != 0

2.4.接口依赖的处理

将依赖放在conftest中，将token传入环境变量中，实现一次登录，多次调用别的接口的目的

testcases/conftest.py

讯享网import os from api.meikefresh_api import login from common.log_util import logger import pytest from common.parseData import base_data #testcases文件夹下下面的每一个 @pytest.fixture(scope="function", autouse=True) def run_case_mark(): """ 执行用例开始前和结束后打印log信息 :return: """ logger.info("开始执行用例") yield logger.info("用例执行完成") def get_base_data(): """ 获取测试数据，作为fixture传入测试用例 :return: """ res = base_data.get_yaml_data() return res @pytest.fixture() def login_fixture(): """ 登录需要的fixture,未将token写入全局中，如果接口需要多次登录得多次调用 :return: """ data = get_base_data()["login_fixture"] username = data["username"] password = data["password"] res = login(username,password) #因为断言加入购物车的时候，需要传入username去获取user_id，然后查询加入购物车商品的数量，所以需要 #返回了元祖，所以取值的时候是login_fixture[0],login_fixture[1] return res.body['token'],username @pytest.fixture() def login_fixture_full(): """ 登录需要的fixture,将token写入环境变量中，如果环境变量中有token不用多次调用login_fixture :return: """ if "token" not in os.environ: data = get_base_data()["login_fixture"] username = data["username"] password = data["password"] res = login(username,password) os.environ["token"] = res.body['token'] os.environ["mobile"] = str(username) return os.environ["token"], os.environ["mobile"] else: return os.environ["token"], os.environ["mobile"]

用例引用登录接口

import allure import pytest from api.meikefresh_api import send_code, register, login, add_shopping_cart from testcases.conftest import get_base_data from testcases.usercenter.conftest import get_code, delete_user, delete_code, get_goods_num @allure.feature("用户中心模块") class TestUser: @allure.story("用户登录后加商品到购物车") @allure.title("用户加商品到购物车用例-使用login_fixture处理") def test_shopping_cart_fixture_full(self, login_fixture_full): """ 使用login_fixture_full来处理登录 :param login_fixture: :return: """ json_data = get_base_data()["shopping_cart"] token = login_fixture_full[0] username = login_fixture_full[1] result = add_shopping_cart(json_data, token) # 断言 username就是登录的mobile,goods_id从要加入的商品信息中拿到json_data["goods"] good_num = get_goods_num(username, json_data["goods"]) assert result.success is True assert result.body["nums"] == good_num

data.yaml -data/data.yaml

讯享网json_data: { title: foo,body: bar,userId: 1 } #test data for test_mobile.py mobile_belong: { shouji: , appkey: 0cd38759e1 } #test data for meikeshengxian register test_register: { mobile:  } #test data for login user_login: # 手机号，密码 - [ , ] #test case for login fixture login_fixture: {username: ,password: } #test data for 加入购物车 shopping_cart: {goods: "1", nums: 1}

2.5 request基类及调用方式的更新

有部分测试数据和url都写在yaml，比如以下格式：

user_login_new: - url: /login/ method: POST data: { username: ,password:  } validate: - eq: [ $.success, true ] - nq: [ $.body.token, null ]

2.5.1 request类的重新封装

因为接口的路径在数据中存放着，所以在拿到数据之后，添加一个参数method。

更加requests的方法，可以使用最好一个request（）方法

调用过程

1. 测试用例文件：定义测试用例，调用接口的组装方法，同时给这个方法传入读取的数据

testcases/usercenter/test_user_login_new.py

讯享网import allure import pytest from api.meikefresh_api import send_code, register, login, add_shopping_cart from api.meikefresh_api_new import login_new from testcases.conftest import get_base_data @allure.feature("用户中心模块") class TestUser: # 参数化实现登录 @pytest.mark.parametrize("data", get_base_data()["user_login_new"]) @allure.story("用户使用手机进行注册后登录") @allure.title("用户登录") def test_user_login(self, data): #调用接口方法，并传入用例 result = login_new(data) assert result.success is True # token每次的值都不一样，没有办法断言具体的值，所以断言不为空有内容即可 assert len(result.body['token']) != 0

2. 接口组装请求需要的方法

解析到的数据，因为使用了，pytest的参数化，这里是列表嵌套字典，所以可以直接去列表中拿值

api/meikefresh_api_new.py

from api.api_util import api_util from common.ResponseUtil import process_response from common.httpClient_new import api_util_new """ 测试数据定义在yaml，格式如下 user_login_new: - url: /login/ method: POST data: { username: ,password:  } validate: - eq: [ $.success, true ] - nq: [ $.body.token, null ] """ def login_new(data): """ 从json :param data: :return: """ response = api_util_new.send_request(url=data["url"],json=data["data"],method=data["method"]) return process_response(response)

3. api调用的是封装好的request方法

讯享网import json import requests from common.parseData import base_data from common.log_util import logger api_root_url = base_data.get_ini_data()["host"]["api_sit_url"] class HttpClient: """ 封装requests的get/post/put/delete请求 """ def __init__(self): self.api_root_url = api_root_url pass def send_request(self,url,method,kwargs): """ 封装requests.request()方法，需要传入参数method :param url: :param method: :param kwargs: :return: """ return self.request(url,method,kwargs) def request(self,url,method,kwargs): """ 根据request的method来调用requests的具体方法 :param url: :param method: :param kwargs: :return: 返回response """ self.request_log(url,method,kwargs) #调用request_log方法来输出日志 if method=="GET": return requests.get(self.api_root_url+url,kwargs) if method == "POST": return requests.post(self.api_root_url + url, kwargs) if method == "PUT": return requests.put(self.api_root_url + url, kwargs) if method == "DELETE": return requests.delete(self.api_root_url + url, kwargs) def request_log(self,url,method,kwargs): """ 解包参数为字典，然后通过字典的get()方法（传入key）获取入参的请求数据 :param url: :param method: :param kwargs: :return: """ # get(key)方法在key（键）不在字典中时，可以返回默认值None或者设置的默认值 # 而dict[key]在key（键）不在字典中时，会触发KeyError异常 data = dict(kwargs).get("data")#kwargs是关键字位置参数，可以转化为字典，通过key来获取 json_data = dict(kwargs).get("json") params = dict(kwargs).get("params") hearders = dict(kwargs).get("headers") logger.info("接口的请求地址>>{}".format(self.api_root_url+url)) logger.info("接口的请求方法>>{}".format(method)) if data is not None: #json.dumps()将json对象转化为字符串，indent表示缩进两个字符显示 logger.info("接口请求的data参数>>>\n{}".format(json.dumps(data,indent=2))) if params is not None: logger.info("接口请求的params参数>>>\n{}".format(json.dumps(params,indent=2))) if json_data is not None: logger.info("接口请求的json_data参数>>>\n{}".format(json.dumps(json_data,indent=2))) if hearders is not None: logger.info("接口请求的hearders>>>\n{}".format(json.dumps(hearders,indent=2))) api_util_new = HttpClient()

此方法跳过了url的拼接层，更为简单，主要看项目具体的测试数据怎么安排，跟httprunner比较起来，需要编写额外的代码实现