就是在代码执行过程当中动态的调用Python中的方法达到或得动态参数的目的
在根目录下创建debug_talk.py，这里我们写入了获得随机数的方法，后面使用值直接调用即可
import random
import time
from common.yaml_util import YamlUtil
class DebugTalk:
#获得随机数
def get_randon_number(self,min,max):
return random.randint(int(min),int(max))
#读取extract.yaml文件中的值
def read_extract_data(self,key):
return YamlUtil().read_yaml(key)另外根目录创建test.py,这个是获取随机时间和本地时间的一个方法，因为我后面的接口输入数据中包含一个参数是时间
import random
import time
from common.yaml_util import YamlUtil
class Test:
#获取随机时间
def get_random_time(self):
# return str(int(time.time()))[1:6] #获取随机时间，得到的是时间戳
return time.strftime('%H:%M:%S', time.localtime(time.time())) #获取本地时间，并将格式转换成时分秒
#读取extract.yaml文件中的值
def read_extract_data(self,key):
return YamlUtil().read_yaml(key)接下来就是request.py文件的修改：
import jsonpath
import requests
import json
from common.yaml_util import YamlUtil
from builtins import str
import re
from debug_talk import DebugTalk
from test import Test
class RequestUtil:
def __init__(self,two_node,obj):
self.base_url=YamlUtil().read_config('base',two_node)
self.obj=obj
#替换值的方法
# #(替换url，params,data,json,headers)
# #(string，int,float,list,dict)
def replace_value(self, data):
if data:
# 保存数据类型
data_type = type(data)
# 判断数据类型转换成str
if isinstance(data, dict) or isinstance(data, list):
str_data = json.dumps(data)
else:
str_data = str(data)
for cs 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]
print("old_value:"+old_value)
#反射：通过类的对象和方法字符串调用方法
func_name=old_value[2:old_value.index('(')]
args_value1=old_value[old_value.index('(')+1:old_value.index(')')]
new_value=""
if args_value1 !="" :
args_value2 = args_value1.split(',')
new_value=getattr(self.obj,func_name)(*args_value2)
else:
new_value = getattr(self.obj, func_name)()
str_data = str_data.replace(old_value, str(new_value))
# 还原数据类型
if isinstance(data, dict) or isinstance(data, list):
data = json.loads(str_data)
else:
data = data_type(str_data)
return data
#规范yaml测试用例
def standard_yaml(self,caseinfo):
caseinfo_keys= caseinfo.keys()
#判断一级关键字是否包含：name，request，validate
if "name" in caseinfo_keys and "request" in caseinfo_keys and "validate" in caseinfo_keys:
#判断request下面是否包含：method、url
request_keys=caseinfo["request"].keys()
if "method" in request_keys and "url" in request_keys:
print("yaml基本架构检查通过")
method = caseinfo['request'].pop("method") #pop() 函数用于移除列表中的一个元素，并且返回该元素的值。
url= caseinfo['request'].pop("url")
res = self.send_request(method,url,**caseinfo['request']) #caseinfo需要解包加**
return_text=res.text
return_code = res.status_code
return_json=""
try:
return_json = res.json()
except Exception as e:
print("extract返回的结果不是JSON格式")
# 提取值并写入extract.yaml文件
if "extract" in caseinfo.keys():
for key, value in caseinfo["extract"].items():
if "(.*?)" in value or "(.+?)" in value: # 正则表达式
zz_value = re.search(value, return_text)
if zz_value:
extract_value = {key: zz_value.group(1)}
YamlUtil().write_yaml(extract_value)
else: # jsonpath
js_value = jsonpath.jsonpath(return_json, value)
if js_value:
extract_value = {key: js_value[0]}
YamlUtil().write_yaml(extract_value)
else:
print("在request下必须包含method,url")
else:
print("一级关键字必须包含name,request,validate")
sess= requests.session()
# 统一请求封装
def send_request(self,method,url,**kwargs):
method=str(method).lower() #转换小写
#基础路径的拼接和替换
url= self.base_url + self.replace_value(url)
print(url)
#参数替换
for key,value in kwargs.items():
if key in ['params','data','json','headers']:
kwargs[key]=self.replace_value(value)
elif key == "files":
for file_key, file_path in value.items():
value[file_key] = open(file_path, 'rb')
res = RequestUtil.sess.request(method, url, **kwargs)
print(res.text)
return res参数为本地时间的yaml用例文件：
-
name: 获取城市列表
request:
method: post
url: /jlcloud/simulation/${read_extract_data(data)}/ranAsPlan
json: {"time": "${get_random_time()}"}
headers:
'Content-Type': 'application/json'
'X-Token': ${read_extract_data(token)}
validate: None这里仅使用了随机时间，如果接口包含随机数，可类比随机时间的方法
最后需要在用例上需要创建初始化对象，用到哪个就写哪个，我这里只用到了Test()
import requests
import json
from common.request_util import RequestUtil
from common.yaml_util import YamlUtil
from test import Test
class TestRequest:
pass
@pytest.mark.parametrize("caseinfo",YamlUtil().read_testcase('get_token.yaml'))
def test_login(self,caseinfo):
res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)
#
@pytest.mark.parametrize("caseinfo", YamlUtil().read_testcase('userinfo.yaml'))
def test_userinfo(self, caseinfo):
res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)
# print(res.text)
#
@pytest.mark.parametrize("caseinfo", YamlUtil().read_testcase('city_type.yaml'))
def test_city(self, caseinfo):
res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)
@pytest.mark.parametrize("caseinfo", YamlUtil().read_testcase('simulation.yaml'))
def test_simulation(self, caseinfo):
res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)
@pytest.mark.parametrize("caseinfo", YamlUtil().read_testcase('ranAsPlan.yaml'))
def test_ranAsPlan(self, caseinfo):
res = RequestUtil("base_test_url",Test()).standard_yaml(caseinfo)断言一般有两种，一种是通过返回值的状态码是否等于200，一种是业务的判断，业务判断可以使用两种方式，一种是想等，一种是包含
1、状态断言：返回的状态码是否等于200
2、业务断言：相等的断言、包含的断言
在request.py中进行断言方法的封装：
import jsonpath
import requests
import json
from common.yaml_util import YamlUtil
from builtins import str
import re
from debug_talk import DebugTalk
from test import Test
class RequestUtil:
def __init__(self,two_node,obj):
self.base_url=YamlUtil().read_config('base',two_node)
self.obj=obj
#替换值的方法
# #(替换url，params,data,json,headers)
# #(string，int,float,list,dict)
def replace_value(self, data):
if data:
# 保存数据类型
data_type = type(data)
# 判断数据类型转换成str
if isinstance(data, dict) or isinstance(data, list):
str_data = json.dumps(data)
else:
str_data = str(data)
for cs 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]
print("old_value:"+old_value)
#反射：通过类的对象和方法字符串调用方法
func_name=old_value[2:old_value.index('(')]
args_value1=old_value[old_value.index('(')+1:old_value.index(')')]
new_value=""
if args_value1 !="" :
args_value2 = args_value1.split(',')
new_value=getattr(self.obj,func_name)(*args_value2)
else:
new_value = getattr(self.obj, func_name)()
str_data = str_data.replace(old_value, str(new_value))
# 还原数据类型
if isinstance(data, dict) or isinstance(data, list):
data = json.loads(str_data)
else:
data = data_type(str_data)
return data
#规范yaml测试用例
def standard_yaml(self,caseinfo):
caseinfo_keys= caseinfo.keys()
#判断一级关键字是否包含：name，request，validate
if "name" in caseinfo_keys and "request" in caseinfo_keys and "validate" in caseinfo_keys:
#判断request下面是否包含：method、url
request_keys=caseinfo["request"].keys()
if "method" in request_keys and "url" in request_keys:
print("yaml基本架构检查通过")
method = caseinfo['request'].pop("method") #pop() 函数用于移除列表中的一个元素，并且返回该元素的值。
url= caseinfo['request'].pop("url")
res = self.send_request(method,url,**caseinfo['request']) #caseinfo需要解包加**
return_text=res.text
return_code = res.status_code
return_json=""
try:
return_json = res.json()
except Exception as e:
print("extract返回的结果不是JSON格式")
# 提取值并写入extract.yaml文件
if "extract" in caseinfo.keys():
for key, value in caseinfo["extract"].items():
if "(.*?)" in value or "(.+?)" in value: # 正则表达式
zz_value = re.search(value, return_text)
if zz_value:
extract_value = {key: zz_value.group(1)}
YamlUtil().write_yaml(extract_value)
else: # jsonpath
js_value = jsonpath.jsonpath(return_json, value)
if js_value:
extract_value = {key: js_value[0]}
YamlUtil().write_yaml(extract_value)
#断言：
self.assert_result(caseinfo['validate'],return_json,return_code)
else:
print("在request下必须包含method,url")
else:
print("一级关键字必须包含name,request,validate")
sess= requests.session()
# 统一请求封装
def send_request(self,method,url,**kwargs):
method=str(method).lower() #转换小写
#基础路径的拼接和替换
url= self.base_url + self.replace_value(url)
print(url)
#参数替换
for key,value in kwargs.items():
if key in ['params','data','json','headers']:
kwargs[key]=self.replace_value(value)
elif key == "files":
for file_key, file_path in value.items():
value[file_key] = open(file_path, 'rb')
res = RequestUtil.sess.request(method, url, **kwargs)
print(res.text)
return res
#断言
def assert_result(self,yq_result,sj_result,return_code):
all_flag = 0
for yq in yq_result:
for key,value in yq.items():
print(key,value)
if key=="equals":
flag=self.equals_assert(value,return_code,sj_result)
all_flag =all_flag + flag
elif key == 'contains':
flag=self.contains_assert(value,sj_result)
all_flag = all_flag + flag
else:
print("框架暂不支持此段断言方式")
assert all_flag==0
# 相等断言
def equals_assert(self,value,return_code,sj_result):
flag=0
for assert_key,assert_value in value.items():
print(assert_key,assert_value)
if assert_key=="status_code": #状态断言
assert_value==return_code
if assert_value!=return_code:
flag=flag+1
print("断言失败，返回的状态码不等于%s"%assert_value)
else:
lists=jsonpath.jsonpath(sj_result,'$..%s'%assert_key)
if lists:
if assert_value not in lists:
flag=flag+1
print("断言失败："+assert_key+"不等于"+str(assert_value))
else:
flag = flag + 1
print("断言失败：返回的结果不存在："+assert_key)
return flag
# 包含断言
def contains_assert(self,value,sj_result):
flag=0
if value not in str(sj_result):
flag = flag + 1
print("断言失败：返回的结果中不包含："+value)
return flag
然后在yaml文件中，输入断言的内容，例如我这个登录接口断言了返回值code是否等于200，以及是否包含data
完成后，让我们run起来，结果可以看到：
