Python+Pytest+Yaml+Request+Allure框架源代码之（一）common公共方法封装

2024-06-20 14:12
文章标签 python 封装 方法 框架 源代码 公共 request common yaml pytest allure

本文主要是介绍Python+Pytest+Yaml+Request+Allure框架源代码之（一）common公共方法封装，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

common模块：

在这里插入图片描述

  • get_path.py：获取路径方法
# -*- coding: UTF-8 -*-
import os# 项目根目录
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))# 配置文件目录
CONFIG_DIR = os.path.join(BASE_DIR,'config')# 测试用例文件目录
TESTCASES_DIR = os.path.join(BASE_DIR,'testcases')#data文件目录
DATA_DIR = os.path.join(BASE_DIR,'data')#日志文件目录
LOGS_DIR=os.path.join(BASE_DIR,'logs')if __name__ == '__main__':print(LOGS_DIR)
  • logger_util.py：日志封装
import logging
import time
from common.get_path import *
from common.yaml_util import read_fileclass LoggerUitl:def create_log(self,logger_name='log'):# 创建一个日志对象self.logger = logging.getLogger(logger_name)# 设置全局的日志级别(DEBUG<INFO<WARNING<ERROR<CRITICAL)self.logger.setLevel(logging.DEBUG)# 防止日志重复if not self.logger.handlers:#------------文件日志--------------# 获取日志文件的名称self.file_log_path = LOGS_DIR+'/'+ read_file('/config/config.yml','log','log_name') + str(int(time.time()))+".log"# 创建文件日志的控制器self.file_handler = logging.FileHandler(self.file_log_path,encoding='utf-8')# 设置文件日志的级别file_log_level= str(read_file('/config/config.yml','log','log_level')).lower()if file_log_level == 'debug':self.file_handler.setLevel(logging.DEBUG)elif file_log_level == 'info':self.file_handler.setLevel(logging.INFO)elif file_log_level == 'waring':self.file_handler.setLevel(logging.WARNING)elif file_log_level == 'error':self.file_handler.setLevel(logging.ERROR)elif file_log_level == 'critical':self.file_handler.setLevel(logging.CRITICAL)# 设置文件日志的格式self.file_handler.setFormatter(logging.Formatter(read_file('/config/config.yml','log','log_format')))# 将控制器加入到日志对象self.logger.addHandler(self.file_handler)#------------控制台日志--------------# 创建控制台日志的控制器self.console_handler = logging.StreamHandler()# 设置控制台日志的级别console_log_level= read_file('/config/config.yml','log','log_level').lower()if console_log_level == 'debug':self.console_handler.setLevel(logging.DEBUG)elif console_log_level == 'info':self.console_handler.setLevel(logging.INFO)elif console_log_level == 'waring':self.console_handler.setLevel(logging.WARNING)elif console_log_level == 'error':self.console_handler.setLevel(logging.ERROR)elif console_log_level == 'critical':self.console_handler.setLevel(logging.CRITICAL)# 设置控制台日志的格式self.console_handler.setFormatter(logging.Formatter(read_file('/config/config.yml','log','log_format')))# 将控制器加入到日志对象self.logger.addHandler(self.console_handler)return self.logger# 函数:输出正常日志
def my_log(log_massage):LoggerUitl().create_log().info(log_massage)# 函数:输出错误日志
def error_log(log_massage):LoggerUitl().create_log().error(log_massage)raise Exception(log_massage)if __name__ == '__main__':my_log('zhangweixu')

- parameters_until.py：传参方式方法封装

import csv
import json
import traceback
import yaml
from common.get_path import *
from common.logger_util import error_log# 读取csv文件
def read_csv_file(csv_file):'''c参数说明'''csv_list = []path = BASE_DIR+"/"+csv_filewith open(path,encoding='utf-8') as f:csv_data = csv.reader(f)for row in csv_data:csv_list.append(row)return csv_list# 读取yaml文件
def read_file(yml_file):try:path = BASE_DIR+yml_filewith open(path,encoding='utf-8') as f:caseinfo = yaml.load(f,Loader=yaml.FullLoader)if len(caseinfo)>=2:return caseinfoelse:caseinfo_keys = dict(*caseinfo).keys()if 'parameters' in caseinfo_keys:new_caseinfo = analysis_parameters(*caseinfo)return new_caseinfoelse:return caseinfoexcept Exception as f:error_log("读取用例文件报错：异常信息：%s"%str(traceback.format_exc()))# 分析参数化
def analysis_parameters(caseinfo):try:caseinfo_keys = dict(caseinfo).keys()if 'parameters' in caseinfo_keys:for key, value in dict(caseinfo['parameters']).items():caseinfo_str = json.dumps(caseinfo)key_list = str(key).split('-')# 规范csv数据的写法length_flag = Truecsv_list = read_csv_file(value)one_row_data = csv_list[0]for csv_data in csv_list:if len(csv_data) != len(one_row_data):length_flag = Falsebreak# 解析new_caseinfo = []if length_flag:for x in range(1, len(csv_list)):  # x代表行temp_caseinfo = caseinfo_strfor y in range(0, len(csv_list[x])):  # y代表列if csv_list[0][y] in key_list:temp_caseinfo = temp_caseinfo.replace("$csv{" + csv_list[0][y] + "}", csv_list[x][y])new_caseinfo.append(json.loads(temp_caseinfo))return new_caseinfoelse:return caseinfoexcept Exception as f:error_log("分析parameters参数异常：异常信息：%s"%str(traceback.format_exc()))if __name__ == '__main__':print(read_file('/testcases/weixin/get_token.yml'))
  • requests_util.py：请求方式方法封装
# -*- coding: UTF-8 -*-
import json
import re
import traceback
import jsonpath
import requests
# from common.parameters_until import read_file
from common.logger_util import my_log, error_log
from common.yaml_util import *
from debugtalk import DebugTalkclass Requestutil:session = requests.session()def __init__(self):self.base_url =""self.last_headers={}# 规范功能测试YAML测试用例文件的写法def analysis_yaml(self,caseinfo):try:# 1.必须有的四个一级关键字：name，base_url，requests，validatecaseinfo_keys = dict(caseinfo).keys()if 'name' in caseinfo_keys and 'base_url' in caseinfo and 'request' in caseinfo and 'validate' in caseinfo:# 2.request关键字必须包含两个二级关键字：method，urlrequest_keys = dict(caseinfo['request']).keys()if 'method' in request_keys and 'url' in request_keys:# 参数(params,data,json),请求头,文件上传这些都不能约束.name = caseinfo['name']self.base_url = caseinfo['base_url']method = caseinfo['request']['method']del caseinfo['request']['method']url = caseinfo['request']['url']del caseinfo['request']['url']headers = Noneif jsonpath.jsonpath(caseinfo,'$..headers'):headers = caseinfo['request']['headers']del caseinfo['request']['headers']files = Noneif jsonpath.jsonpath(caseinfo, '$..files'):files = caseinfo['request']['files']for key,value in dict(files).items():files[key] = open(value,'rb')del caseinfo['request']['files']# 把method,url,headers,files这四个数据从caseinfo['request']去掉之后再把剩下的传给kwargsres = self.send_request(name=name,method=method,url=url,headers=headers,files=files,**caseinfo['request'])return_text = res.textstatus_code = res.status_codemy_log("响应文本信息：%s"%return_text)my_log("响应json信息：%s"%res.json())# 提取接口关联的变量,既要支持正则表达式,又要支持json提取if 'extract' in caseinfo_keys:for key,value in dict(caseinfo['extract']).items():# 正则表达式提取if '(.*?)' in value or '(.+?)' in value:ze_value = re.search(value,return_text)if ze_value:extract_data = {key:ze_value.group(1)}write_file('/config/extract.yml',extract_data)print(extract_data)else:   # json提取return_json = res.json()  # 前提是要返回json格式extract_data = {key:return_json[value]}write_file('/config/extract.yml', extract_data)print(extract_data)# 断言的封装yq_result = caseinfo['validate']sj_result = res.json()self.validate_result(yq_result, sj_result, status_code)else:error_log('request关键字必须包含两个二级关键字：method，url')else:error_log('必须有的四个一级关键字：name，base_url，request，validate')except Exception as f:error_log("分析YAML文件异常：异常信息：%s" % str(traceback.format_exc()))#  统一替换方法,data可以是url(string),也可以是参数(字典,字典中包含有列表),也可以是请求头(字典).def replace_value(self,data):# 字典类型转换成字符串if data and isinstance(data,dict):  # 如果data不为空并且数据类型为字典str_data = json.dumps(data)else:str_data = data# 替换值for i in range(1, str_data.count('{{') + 1):if "{{" in str_data and "}}" in str_data:start_index = str_data.index("{{")end_index = str_data.index("}}",start_index)old_value = str_data[start_index:end_index + 2]new_value = read_file("/config/extract.yml", old_value[2:-2])str_data = str_data.replace(old_value, new_value)# 还原数据类型if data and isinstance(data,dict):  # 如果data不为空并且数据类型为字典data = json.loads(str_data)else:data = str_datareturn data#  统一替换方法,data可以是url(string),也可以是参数(字典,字典中包含有列表),也可以是请求头(字典).def replace_load(self, data):# 字典类型转换成字符串if data and isinstance(data, dict): # 如果data不为空并且数据类型为字典str_data = json.dumps(data)else:str_data = data# 替换值for i in range(1, str_data.count('${') + 1):if "${" in str_data and "}" in str_data:start_index = str_data.index("${")end_index = str_data.index("}", start_index)old_value = str_data[start_index:end_index + 1]function_name = old_value[2:old_value.index('(')]args_value = old_value[old_value.index('(')+1:old_value.index(')')]# 反射(通过一个函数的字符串直接去调用这个方法)new_value = getattr(DebugTalk(),function_name)(*args_value.split(','))str_data = str_data.replace(old_value, str(new_value))# 还原数据类型if data and isinstance(data, dict):   # 如果data不为空并且数据类型为字典data = json.loads(str_data)else:data = str_datareturn data# 统一发送请求def send_request(self,name,method,url,headers=None,files=None,**kwargs):try:# 处理methodself.last_method = str(method).lower()# 处理基础路径self.url=self.replace_load(self.base_url) + self.replace_value(url)# 处理请求头if headers and isinstance(headers,dict):self.last_headers=self.replace_value(headers)# 最核心的地方:请求数据如何去替换:可能是params,data,jsonfor key,value in kwargs.items():if key in ['params','data','json']:# 替换{{}}格式value = self.replace_value(value)# 替换${}格式value = self.replace_load(value)kwargs[key] = value# 收集日志my_log('-----------------接口请求开始-----------------')my_log("接口名称：%s"%name)my_log("请求方式：%s"%self.last_method)my_log("请求路径：%s"%self.url)my_log("请求头：%s"%self.last_headers)if 'params' in kwargs.keys():my_log("请求参数：%s"%kwargs['params'])elif 'data' in kwargs.keys():my_log("请求参数：%s"%kwargs['data'])elif 'json' in kwargs.keys():my_log("请求参数：%s"%kwargs['json'])my_log("文件上传：%s"%files)# 发送请求res = Requestutil.session.request(method=self.last_method,url=self.url,headers=self.last_headers,**kwargs)# print(res.request.headers)# print(res.text)# print(res.json())return resexcept Exception as f:error_log("发送请求异常：异常信息：%s"%str(traceback.format_exc()))# 断言封装def validate_result(self,yq_result,sj_result,status_code):try:''':param yq_result:预期结果:param sj_result:实际结果:param status_code:实际状态码:return:'''# 收集日志my_log("预期结果：%s"%yq_result)my_log("实际结果：%s"%sj_result)#判断是否断言成功,0成功,1失败flag = 0# 解析ƒif yq_result and isinstance(yq_result,list):for yq in yq_result:for key,value in dict(yq).items():# 判断断言方式if key=='equals':for assert_key,assert_value in dict(value).items():if assert_key=='status_code':if status_code!=assert_value:flag=flag+1error_log("断言失败:"+assert_key+"不等于"+str(assert_value)+"")else:key_list = jsonpath.jsonpath(sj_result,'$..%s'%assert_key)if key_list:if assert_value not in key_list:flag = flag + 1error_log("断言失败:"+assert_key+"不等于"+str(assert_value)+"")else:flag = flag + 1error_log("断言失败:返回结果中不存在"+assert_key+"")elif key=='contains':if value not in json.dumps(sj_result):flag = flag + 1error_log("断言失败:返回结果中不包含字符串"+value+"")else:error_log('框架不支持此断言方式')assert flag==0my_log('接口请求成功')my_log('-----------------接口请求结束-----------------\n')except Exception as f:my_log('接口请求失败')my_log('-----------------接口请求结束-----------------\n')error_log("断言异常：异常信息：%s" % str(traceback.format_exc()))if __name__ == '__main__':# url = "/cgi-bin/tags/update?access_token={{access_token}}&a=c{{csrf_token}}"# for i in range(1,url.count("{{")+1):#     if "{{" in url and "}}" in url:#         start_index = url.index("{{")#         end_index = url.index("}}")#         old_value = url[start_index:end_index+2]#         new_value = read_file('/config/extract.yml',old_value[2:-2])#         url = url.replace(old_value,new_value)##         print(old_value,new_value)#         print(url)# dict_data = {'name': '获取access_token统一鉴权码', 'base_url': 'https://api.weixin.qq.com', 'request': {'method': 'GET', 'url': '/cgi-bin/token', 'params': {'grant_type': 'client_credential', 'appid': 'wx9b755d429f6fb216', 'secret': 'b963db0b97c8487b0cb920a240bd78e3'}}, 'validate': [{'eq': ['status_code', 200]}]}# # print(dict_data.pop('name'))# del dict_data['name']# print(dict_data)json_data = {"tag": {"id": 100, "name": "CesareCheung${get_random_number(100000,999999)}" }}result = Requestutil('base', 'base_weixin_url').replace_load(json_data)print(result)
  • yaml_util.py：文件读取写入方法
# -*- coding: UTF-8 -*-
import os
import yaml
from common.get_path import *# 读取yml文件
def read_file(yml_file,one_node=None,two_node=None):path = BASE_DIR+yml_filewith open(path,encoding='utf-8') as f:value = yaml.load(f,Loader=yaml.FullLoader)if one_node and two_node:return value[one_node][two_node]elif one_node:return value[one_node]else:return value# 写入yml文件
def write_file(yml_file,data):path = BASE_DIR+yml_filewith open(path,encoding='utf-8',mode='a') as f:yaml.dump(data, stream=f,allow_unicode=True)# 清空yml文件
def clean_file(yml_file):path = BASE_DIR+yml_filewith open(path,encoding='utf-8',mode='w') as f:f.truncate()if __name__ == '__main__':# print(read_file('/config.yml',"base",'base_info_url'))print(read_file('/config/config.yml','log','log_name'))

这篇关于Python+Pytest+Yaml+Request+Allure框架源代码之（一）common公共方法封装的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1078357。

相关文章

使用Python和OpenCV库实现实时颜色识别系统

使用Python和OpenCV库实现实时颜色识别系统

《使用Python和OpenCV库实现实时颜色识别系统》：本文主要介绍使用Python和OpenCV库实现的实时颜色识别系统,这个系统能够通过摄像头捕捉视频流,并在视频中指定区域内识别主要颜色（红... 目录一、引言二、系统概述三、代码解析1. 导入库2. 颜色识别函数3. 主程序循环四、HSV色彩空间详解
阅读更多...
一文深入详解Python的secrets模块

一文深入详解Python的secrets模块

《一文深入详解Python的secrets模块》在构建涉及用户身份认证、权限管理、加密通信等系统时,开发者最不能忽视的一个问题就是“安全性”,Python在3.6版本中引入了专门面向安全用途的secr... 目录引言一、背景与动机：为什么需要 secrets 模块？二、secrets 模块的核心功能1. 基
阅读更多...
python常见环境管理工具超全解析

python常见环境管理工具超全解析

《python常见环境管理工具超全解析》在Python开发中,管理多个项目及其依赖项通常是一个挑战,下面：本文主要介绍python常见环境管理工具的相关资料,文中通过代码介绍的非常详细,需要的朋友... 目录1. conda2. pip3. uvuv 工具自动创建和管理环境的特点4. setup.py5.
阅读更多...
Python常用命令提示符使用方法详解

Python常用命令提示符使用方法详解

《Python常用命令提示符使用方法详解》在学习python的过程中,我们需要用到命令提示符(CMD)进行环境的配置,：本文主要介绍Python常用命令提示符使用方法的相关资料,文中通过代码介绍的... 目录一、python环境基础命令【Windows】1、检查Python是否安装2、 查看Python的安
阅读更多...
Python UV安装、升级、卸载详细步骤记录

Python UV安装、升级、卸载详细步骤记录

《PythonUV安装、升级、卸载详细步骤记录》：本文主要介绍PythonUV安装、升级、卸载的详细步骤,uv是Astral推出的下一代Python包与项目管理器,主打单一可执行文件、极致性能... 目录安装检查升级设置自动补全卸载UV 命令总结 官方文档详见：https://docs.astral.sh/
阅读更多...
Python并行处理实战之如何使用ProcessPoolExecutor加速计算

Python并行处理实战之如何使用ProcessPoolExecutor加速计算

《Python并行处理实战之如何使用ProcessPoolExecutor加速计算》Python提供了多种并行处理的方式,其中concurrent.futures模块的ProcessPoolExecu... 目录简介完整代码示例代码解释1. 导入必要的模块2. 定义处理函数3. 主函数4. 生成数字列表5.
阅读更多...
Python中help()和dir()函数的使用

Python中help()和dir()函数的使用

《Python中help()和dir()函数的使用》我们经常需要查看某个对象（如模块、类、函数等）的属性和方法,Python提供了两个内置函数help()和dir(),它们可以帮助我们快速了解代... 目录1. 引言2. help() 函数2.1 作用2.2 使用方法2.3 示例(1) 查看内置函数的帮助(
阅读更多...
Python虚拟环境与Conda使用指南分享

Python虚拟环境与Conda使用指南分享

《Python虚拟环境与Conda使用指南分享》：本文主要介绍Python虚拟环境与Conda使用指南,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、python 虚拟环境概述1.1 什么是虚拟环境1.2 为什么需要虚拟环境二、Python 内置的虚拟环境工具
阅读更多...
Python实例题之pygame开发打飞机游戏实例代码

Python实例题之pygame开发打飞机游戏实例代码

《Python实例题之pygame开发打飞机游戏实例代码》对于python的学习者,能够写出一个飞机大战的程序代码,是不是感觉到非常的开心,：本文主要介绍Python实例题之pygame开发打飞机... 目录题目pygame-aircraft-game使用 Pygame 开发的打飞机游戏脚本代码解释初始化部
阅读更多...
Python pip下载包及所有依赖到指定文件夹的步骤说明

Python pip下载包及所有依赖到指定文件夹的步骤说明

《Pythonpip下载包及所有依赖到指定文件夹的步骤说明》为了方便开发和部署,我们常常需要将Python项目所依赖的第三方包导出到本地文件夹中,：本文主要介绍Pythonpip下载包及所有依... 目录步骤说明命令格式示例参数说明离线安装方法注意事项总结要使用pip下载包及其所有依赖到指定文件夹，请按照以
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2