文章目錄

• 前言
• 一、框架目錄介紹
• • 1、common模塊
  • • 讀取Excel代碼
    • 讀取yaml代碼（支持場景關聯）
    • jsonpath斷言封裝代碼
    • requests二次封裝（get、post）
    • configparser讀取配置文件
    • 遞歸遍歷字典常用方法
    • log日志封裝
  • 2、conf模塊
  • 3、data模塊
  • 4、case模塊
  • 5、run_main.py執行文件
  • 6、log模塊
  • 7、report模塊
• 二、接口關聯（場景測試）
• 三、接口自動化平臺
• 總結

---

前言

之前因項目需求，自己學習并寫了一個python接口自動化的小框架，基于python+requests+pytest+allure，支持Excel、yaml用例數據存儲并參數化，生成領導都喜歡的allure報告。博主的代碼功底不咋樣，望大佬們多指點指點。后面會分不同的章節一一講解，希望能對你有所幫助。

---

一、框架目錄介紹

case：用例存放目錄
common：存放公共方法目錄
conf：存放配置文件目錄
data：存放測試數據目錄
logs ：存放日志目錄
report：存放報告目錄
run_main.py：用例總執行入口

如上所示，框架的整體目錄結構還是分工明確的，封裝的方法并不適用每個項目，需要根據自己所在的項目改動，那么下面將每個模塊的代碼及功能展示。

1、common模塊

該模塊主要存放公共方法，如讀取Excel、yaml用例數據，jsonpath斷言、日志、configparser讀取配置文件、requests二次封裝等。根據自己項目所需，封裝適合自己的方法，方便后續操作。

讀取Excel代碼

# coding=utf-8 # @Time : 2022/3/11 9:27 # @Author : 梗小旭 # @File : get_excel_data.pyimport os from openpyxl import load_workbook from common.public_path import DIR from common.config_operate_api import Configclass GetExcelData():"""封裝讀取Excel數據"""def __init__(self,sheet):self.path=eval(Config().getconf("excel_path").path) #excel文件路徑self.excel_path=os.path.join(DIR,self.path)self.wb = load_workbook(self.excel_path)self.ws = self.wb[sheet]self.max_columns = self.ws.max_column #最大列數self.max_rows=self.ws.max_row #最大行數def get_row_case_list(self,row=None):"""按行獲取Excel中的用例數據，返回list,如果row=None時，返回整個sheet頁所有數據（除表頭），如果row為具體數字時，讀取sheet對應的行數數據:param row: 行數，第一行數據為title，默認已把值加1:return:"""case_list = [] #返回的所有case數據#當row為None返回當前sheet頁中所有用例數據if row==None:for i in range(self.max_rows):temp_case_list=[]for each in self.ws.iter_cols(min_col=0):temp_case_list.append(each[i].value)#openpy的iter_cols用法會讀取所有行包含空行（做了格式其他的改變，也會讀取），加判斷去除空行if temp_case_list[0]!=None and temp_case_list[:-1]!=None:case_list.append(temp_case_list)#去除表頭數據del case_list[0]return case_listelse:for i in range(1,self.max_columns+1):value=self.ws.cell(row=row+1,column=i).valuecase_list.append(value)return case_listdef get_row_case_dict(self,row=None):"""按行獲取Excel中的用例數據，如果row=None時，返回的數據是全部用例數據，格式為list中存在多個dict如果row等于具體數字時，讀取對應行的數據:param row: 行數:return:"""case_title_list=self.get_row_case_list(row=0) #獲取sheet頁第一行，即titleif row==None:all_case_dict_list=[] #存每個用例的dict格式的listall_case_list = self.get_row_case_list()for case in all_case_list:temp_case_dict=dict(zip(case_title_list,case))all_case_dict_list.append(temp_case_dict)return all_case_dict_listelse:case_list=self.get_row_case_list(row=row)#通過title和一行的數據使用zip合并成字典case_dict=dict(zip(case_title_list,case_list))return case_dictdef get_case_data(self,row=None):"""按行獲取Excel中用例數據，并把數據中提取url、data、expected_result值，返回tuple,其中從Excel中讀取的鍵值對數據需要用eval格式轉成字典格式row==None時返回全部用例數據:param row: 行數:return:"""if row==None:all_case_list = [] #list存多個tuple，每個tuple中有url，data，expected_resultall_case_dict_list = self.get_row_case_dict()for temp_case_dict in all_case_dict_list:temp_list=[]temp_list.append(temp_case_dict["url"])data = temp_case_dict["data"]temp_list.append(eval(data))temp_list.append(temp_case_dict["expected_result"])all_case_list.append(tuple(temp_list))return all_case_listelse:case_dict=self.get_row_case_dict(row=row)new_case_list=[]new_case_list.append(case_dict["url"])data=case_dict["data"]new_case_list.append(eval(data))new_case_list.append(case_dict["expected_result"])return tuple(new_case_list)

讀取yaml代碼（支持場景關聯）

# coding=utf-8 # @Time : 2022/3/16 14:58 # @Author : 梗小旭 # @File : read_yaml_data.py import osimport yaml from common.public_path import DIR from common.read_file_func import execute_func from common.get_dict_api import update_dict_val,add_params class ReadYamlData():def __init__(self,filename):self.path=os.path.join(DIR,f"data/{filename}.yaml")def read_yaml_case(self):"""讀取yaml文件中數據并返回:return:"""with open(self.path,"r",encoding="utf-8") as f:data=f.read()result=yaml.load(data,Loader=yaml.FullLoader)return resultdef yaml_to_list(self,n=None):"""把讀取yaml的數據轉成list中多個tuple，每個tuple放url，data,expected_result,參數化使用當yaml文件中存在rules規則時，表明該條用例存在接受其他接口傳參，讀取rules下的規則數據，如下：position：想要修改數據字典中的key的路徑，例如["department","id"],配置文件中寫department.id，通過split分解method：需要調用的函數名稱module：需要調用的函數所在模塊及文件路徑，例如：interface_data.jiekou，interface_data模塊名，jiekou文件名稱params：調用函數所需要的傳參，不需要傳參時，默認寫[]:param n 對應第幾條用例，n為None時，返回全部用例:return:"""result=self.read_yaml_case()all_case_list=[]for temp_case in result:case_list=[]#判斷讀取的數據中是否存在rules規則if "rules" in temp_case:data = temp_case['data']for rules in temp_case["rules"]:position=rules["position"].split('.')func_name=rules["method"]module_name=rules["module"]params=rules["params"]#讀取配置文件中的函數，并執行函數返回值result=execute_func(func_name=func_name, module_name=module_name,params=params)#更新data值update_dict_val(data,position,val=result)#把數據加到all_case_list中case_list.append(temp_case["url"])case_list.append(data)case_list.append(temp_case["expected_result"])all_case_list.append(tuple(case_list))else:case_list.append(temp_case["url"])case_list.append(temp_case["data"])case_list.append(temp_case["expected_result"])all_case_list.append(tuple(case_list))#判斷n的值，為None時，返回所有的值，n為具體數字時，返回某個案例if n==None:return all_case_listelse:return all_case_list[n-1]

jsonpath斷言封裝代碼

# coding=utf-8 # @Time : 2022/3/17 21:23 # @Author : 梗小旭 # @File : public_assert.py import jsonpathdef assert_res(res,expected_result):"""傳入響應體的json格式數據和Excel或yaml中讀取的預期結果值，預期結果逐一判斷，有一個不符合則返回False:param res: 請求返回的響應體數據:param expected_result: 預期結果值，例如：'$.code=201;$.success=False;$.message=用戶名或密碼錯誤'注意：字符串里面不能寫引號，比如不能$.message=“用戶名或密碼錯誤”，正確寫法是：$.message=用戶名或密碼錯誤:return:"""for exp in expected_result.split(";"):rule=exp.split("=")[0] #jsonpath提取規則exp_value=exp.split("=")[1] #預期結果值reality_value=jsonpath.jsonpath(res,rule)[0] #真實返回值#預期結果中存在特殊False和True，讀取時要用eval把str類型轉成bool，才能和返回值對比判斷if exp_value=='False' or exp_value=='True':exp_value=eval(exp_value)if str(exp_value)==str(reality_value):continueelse:return Falsereturn True

requests二次封裝（get、post）

# coding=utf-8 # @Time : 2022/3/10 15:52 # @Author : 梗小旭 # @File : base_method_api.pyfrom common.log import log import requests import traceback from common.config_operate_api import Configclass BaseMethodApi():def __init__(self):self.conf = Config().getconf("enviro")self.host=self.conf.hostself.url=self.conf.urlself.data=self.conf.datadef get_token_data(self):"""獲取當前環境下的token值:return: 返回登錄成功的token值"""complete_ulr = "http://" + self.host + self.url # 完整urlres=requests.post(url=complete_ulr,json=eval(self.data),headers=self.choice_headers())token=res.json()['data']['token']['access_token']return tokendef choice_headers(self,type=None):"""封裝選擇請求頭信息，type等于None時，請求頭不傳token，等于其他值時傳token:param type::return:"""headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36","Content-Type": "application/json; charset=utf-8"}if type:headers["Authorization"]=self.get_token_data()return headerselse:return headersdef get(self,url,params=None,headers=None,files=None):"""get請求:param url: 請求路徑:param params: 請求參數:param headers: 請求頭:param files: 請求文件:return:"""try:log.info("============請求信息============")complete_ulr="http://"+self.host+url#完整urlif not headers:headers=self.choice_headers(type=1)else:headers=self.choice_headers()res=requests.get(url=complete_ulr,params=params,headers=headers,files=files)log.info(f"請求url：{complete_ulr}")log.info(f"請求參數:{params}")log.info(f"請求頭:{headers}")log.info("============響應信息============")log.info(f"響應狀態碼：{res.status_code}")log.info(f"響應結果：{res.text}")return resexcept:log.error("============請求失敗信息============")log.error(f"請求異常：{traceback.print_exc()}")def post(self,url,data=None,json_data=None,headers=None,files=None):"""post請求:param url: 請求路徑:param data: 原始請求參數:param json_data: json格式請求參數:param headers: 請求頭:param files: 請求文件:return:"""try:log.info("============請求信息============")complete_ulr="http://"+self.host+url#完整urlif not headers:headers=self.choice_headers(type=1)else:headers=self.choice_headers()res=requests.post(url=complete_ulr,data=data,json=json_data,headers=headers,files=files)log.info(f"請求url：{complete_ulr}")if json_data==None:log.info(f"請求參數:{data}")else:log.info(f"請求參數:{json_data}")log.info(f"請求頭:{headers}")log.info("============響應信息============")log.info(f"響應狀態碼：{res.status_code}")log.info(f"響應結果：{res.text}")return resexcept:log.error("============請求失敗信息============")log.error(f"請求異常：{traceback.print_exc()}")

configparser讀取配置文件

#coding=utf-8 import os from configparser import ConfigParserclass Dictionary(dict):'''把config.ini中的參數添加值dict'''def __getattr__(self, keyname):#如果key值不存在則返回默認值"not find config keyname"return self.get(keyname, "config.ini中沒有找到對應的keyname")class Config(object):'''ConfigParser二次封裝，在字典中獲取value'''def __init__(self):# 設置配置文件路徑current_dir = os.path.dirname(__file__)top_one_dir = os.path.dirname(current_dir)file_name = top_one_dir + "/conf/config.ini"# 實例化ConfigParser對象self.config = ConfigParser()self.config.read(file_name,encoding="utf-8")#根據section把key、value寫入字典for section in self.config.sections():setattr(self, section, Dictionary())for keyname, value in self.config.items(section):setattr(getattr(self, section), keyname, value)def getconf(self, section):'''用法：conf = Config()info = conf.getconf("main").url'''if section in self.config.sections():passelse:print(" 找不到該 section")return getattr(self, section)

遞歸遍歷字典常用方法

#coding=utf-8 from typing import Dict,Listdef get_dict(dict_value ,obj_key ,default=None):"""遍歷字典，得到想要的value:param dict_value: 所需要遍歷的字典:param obj_key: 所需要value的鍵:param default:進行取值中報錯時所返回的默認值 (default: None):return:"""for k ,v in dict_value.items():if k == obj_key:return velse:if type(v) is dict : # 如果鍵對應的值還是字典re = get_dict(v ,obj_key ,default) # 遞歸if re is not default:return redef get_list_dict(list_value ,obj_key,obj_value):"""遍歷列表中的每個字典，判斷obj_key,obj_value值是否存在，存在則任何True，否則False:param list_value: 所需要遍歷的列表:param obj_key: 想要判斷的key:param obj_value: 想要判斷的value:return:"""for dict_value in list_value:for k ,v in dict_value.items():if k == obj_key and v == obj_value:return Trueelse:continue#列表中所有數據都不存在時，返回Falsereturn Falsedef updata_dict_value(dict_data ,obj_key,update_value=None):"""遍歷字典，得到想要的key對象，給讀取文件時修改值,如果obj_key存在一樣的情況下，就會改錯:param dict_value: 所需要遍歷的字典:param obj_key: 所需要value的鍵:return:"""for k ,v in dict_data.items():if k == obj_key:dict_data[k]=update_valueelse:if type(v) is dict : # 如果鍵對應的值還是字典updata_dict_value(v ,obj_key,update_value) # 遞歸def update_dict_val(data:Dict, key_list:List, val:int,i=0):"""傳入data字典格式數據，根據對應的key_list,把對應的key的val值修改:param data: 傳入的字典數據:param key_list: 傳入修改的key list，例如["department","id"],配置文件中寫department.id，通過split分解:param val: 想要修改的值:param i: i值默認為0，遞歸時默認+1:return:"""if i==len(key_list)-1:data[key_list[i]] = valreturnreturn update_dict_val(data[key_list[i]], key_list, val,i=i+1)def add_params(func_str,params):"""根據傳入的函數名稱，和函數所需要的數據來拼接成函數傳參的字符串格式，通過eval轉成可以執行的函數:param func_str: 函數的名稱，必須傳字符串:param params: 函數所需要的參數，params是一個list，例如函數為：add(a，b,c=4),huanc:return: """value = ",".join([str(i) for i in params])val = f'{func_str}({value})'return eval(val)

log日志封裝

#coding=utf-8 import logging from common.public_path import DIR import time import osdef get_log(logger_name):""":param logger_name: 填項目名稱表示哪個項目:return:"""#創建一個loggerlogger = logging.getLogger(logger_name)logger.setLevel(logging.INFO)#獲取本地時間，轉換為設置的格式#rq = time.strftime('%Y%m%d%H%M',time.localtime(time.time()))rq = time.strftime("%Y_%m_%d_")#設置日志文件存放路徑，日志文件名#設置所有日志和錯誤日志的存放路徑# 通過getcwd.py文件的絕對路徑來拼接日志存放路徑all_log_path = os.path.join(DIR,'logs/info_logs/')error_log_path = os.path.join(DIR,'logs/error_logs/')#設置日志文件名all_log_name = all_log_path + rq + '.log'error_log_name = error_log_path + rq + '.log'#創建handler#創建一個handler，寫入所有日志fh = logging.FileHandler(all_log_name,encoding="utf-8")fh.setLevel(logging.INFO)#創建一個handler，寫入錯誤日志eh = logging.FileHandler(error_log_name,encoding="utf-8")eh.setLevel(logging.ERROR)#創建一個handler，輸出到控制臺ch = logging.StreamHandler()ch.setLevel(logging.INFO)#定義日志輸出格式#以時間-日志器名稱-日志級別-日志內容的形式展示all_log_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')# 以時間-日志器名稱-日志級別-文件名-函數行號-錯誤內容error_log_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(module)s - %(lineno)s - %(message)s')# 將定義好的輸出形式添加到handlerfh.setFormatter(all_log_formatter)ch.setFormatter(all_log_formatter)eh.setFormatter(error_log_formatter)# 給logger添加handlerlogger.addHandler(fh)logger.addHandler(eh)logger.addHandler(ch)return logger#實例化log，調用時，直接調用log log = get_log("CL接口自動化")

2、conf模塊

該模塊主要放配置數據，ini文件中數據格式如下，可存放環境數據、文件路徑、郵箱等，根據自己需要配置

[enviro] host=127.0.0.1:8000 url=/interface/login data={'username':'zwx','password':'123123'}[excel_path] path='data/case_data.xlsx'

3、data模塊

該模塊下存放參數化用例數據，支持yaml或者Excel，使用common中封裝讀取方法，讀取測試用例并參數化使用。
yaml文件格式如下（如果不需要接口關聯，可以不用寫rules，后續文章會講解）：

Excel文件格式如下（可以用sheet頁區分不同模塊或接口的用例）：

4、case模塊

case模塊主要用于存放測試用例，簡單寫了兩條查詢項目的接口用例，用例存放在yaml中，這里采用jsonpath斷言，通過封裝好的讀取yaml數據的方法，讀取數據后通過parametrize參數化，如下：

yaml文件：

- case_id : case_01case_name: 驗證查詢項目列表數據成功url: /interface/select/itemmethod: POSTdata: {"pagenum":1,"pagesize":10}expected_result: $.code=200;$.success=True - case_id : case_02case_name: 驗證查詢頁數pagenum為-1時，查詢失敗url: /interface/select/itemmethod: POSTdata: {"pagenum":-1,"pagesize":10}expected_result: $.code=2003;$.success=False;$.msg=查詢項目數據失敗

用例文件：

# coding=utf-8 # @Time : 2022/5/24 11:17 # @Author : 梗小旭 # @File : test_search_item.py import pytestfrom common.base_method_api import BaseMethodApi from common.read_yaml_data import ReadYamlData from common.public_assert import assert_res from common.get_excel_data import GetExcelData case_list=ReadYamlData("search_item").yaml_to_list() #讀取該文件下所有測試用例@pytest.mark.parametrize("url,data,expected_result",case_list) def test_login(url,data,expected_result):bma=BaseMethodApi()res=bma.post(url=url,json_data=data)result=res.json()assert assert_res(result,expected_result)if __name__ == '__main__':pytest.main(["-s","test_search_item.py"])

5、run_main.py執行文件

該文件執行所有用例，代碼如下：

# coding=utf-8 # @Time : 2022/3/10 15:33 # @Author : 梗小旭 # @File : run_main.py import os import shutil from common.public_path import DIR path=DIR+'/report' if os.path.exists(path):shutil.rmtree(path) os.system("pytest -s -q --alluredir report")#生成allure報告 os.system("allure generate report/ -o report/html --clean")#清除報告數據

6、log模塊

該模塊主要放info和error日志數據，如下：

7、report模塊

該模塊存放執行生成的allure報告數據，可以通過allure添加步驟，描述、優先級等詳細信息，本文未添加，用例標題可通過在parametrize中ids參數中添加，報告如下：

---

二、接口關聯（場景測試）

在我們做接口測試時，接口關聯的測試必不可少且非常重要的，那么我們在做接口自動化時，接口關聯的場景我們如何做呢？
1、假如B接口的入參需要A接口的返回值，那么執行A接口的用例后把返回值存在文件中，B接口用例執行時讀取文件中數據，形成接口關聯。這樣做雖然可以，但是存在一個問題，會導致每條用例不獨立，如果接口A失敗了，接口B的用例全部失敗。
2、假如B接口的入參需要A接口的返回值，接口A單獨寫用例，不存返回值數據。單獨封裝一個接口A的方法返回值，接口B使用數據時調用封裝的A方法，實現接口A和接口B的用例解耦。

本文中使用的是第2種方式，在yaml文件中增加rules規則，存在rules規則時，會調用對應模塊的方法獲取返回值，并修改這條用例的position字段值。大佬們有其他方法可以在評論區下留言。

rules:- position: idmodule: common.comm_ret_valuemethod: get_item_idparams: []

---

三、接口自動化平臺

最近自己也寫了一個接口自動化小平臺，僅供自己學習使用，功能還未完善，完善后續更新出來，采用vue+fastapi前后端分離實現，話不多說，上圖：

總結

目前框架只實現了基本的功能，未涉及到當接口用例過多時，接口變動，如何快速修改用例，pytest的失敗重試、數據庫校驗、Jenkins集成等問題，后續在根據項目需求加上對應功能。希望大佬們給點好的建議改進改進。

總結

以上是生活随笔為你收集整理的如何在项目中搭建python接口自动化框架？的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。