合并接口默认参数与用例参数:
一、获取请求的各种信息,如请求方式、url、接口、参数、header
在tasks.py里Run类下新增一个方法simple_run,入参是case_id,代码如下:
def simple_run(self, case_id):
"""
1、获取请求的各种信息,如请求方式、url、接口、参数、header
2、是否有${}格式,需要替换,如{'username': '${username}', 'password': '123456'},需要把${username}替换成真实的username
3、缓存
4、报告
5、更新用例状态
:param case_id:
:return:
"""
# 获取用例的对象
case_obj = models.Case.objects.get(id=case_id)
# 域名
host = case_obj.project.host
# 接口
uri = case_obj.interface.uri
# 请求方式
# method = case_obj.method # case_obj.method返回数字
method = case_obj.get_method_display() # 用get_method_display()拿到具体的请求方式
# 参数
# 1、接口默认参数,从interface表里看出case_obj.interface.params取出来是json串,json串要转成字典
interface_params = json.loads(case_obj.interface.params) if case_obj.interface.params else {}
# 2、用例的参数(以用例的为基准)
case_params = json.loads(case_obj.params) if case_obj.params else {}
# header
# 1、接口的header
interface_header = json.loads(case_obj.interface.headers) if case_obj.interface.headers else {}
# 2、用例的header
case_header = json.loads(case_obj.headers) if case_obj.headers else {}
# is_json
is_json = case_obj.is_json
# json数据
json_data = json.loads(case_obj.json) if case_obj.json else {}
# 合并header并替换,将用例的header覆盖接口的
interface_header.update(case_header)
# 合并参数并替换,参数有json和k-v两种,需要根据is_json状态来判断,将用例的参数覆盖接口的参数
if is_json:
interface_params.update(json_data)
else:
interface_params.update(case_params)
logger.debug('host-->%s' % host)
logger.debug('uri-->%s' % uri)
logger.debug('method-->%s' % method)
logger.debug('interface_params-->%s' % interface_params)
logger.debug('case_params-->%s' % case_params)
logger.debug('interface_header-->%s' % interface_header)
logger.debug('case_header-->%s' % case_header)
logger.debug('is_json-->%s' % is_json)
logger.debug('json_data-->%s' % json_data)
然后在run_case里来调,run_case里新增的代码如下图:
点击用例管理页面的登录用例后的运行按钮,然后在celery_tasks异步任务里看到打印的各种信息,没有问题
二、是否有${}格式,需要替换
如果interface_params是这种格式{'username': 'ssj', 'password': '${password}'},需要把${password}替换成真实的password,在simple_run方法里的logger.debug前调一下data = self.data_replace(interface_params),还要导入re,用到正则,接下来写data_replace方法,新增代码如下:
def data_replace(self, data):
data = {'username': 'ssj', 'password': '${password}'}
for k, v in data.items():
keys = re.findall(r'${(.*?)}', str(v))
if keys:
# 如果keys存在,代表有需要替换的参数 ['password'],全局参数进行替换
value = models.Parameter.objects.get(name=keys[0]).value
data[k] = value
logger.debug('替换后的data是-->%s' % data)
return data
重启celery_tasks异步任务,点击用例管理页面的登录用例后的运行按钮,在日志里看到打印出:替换后的data是-->{'username': 'ssj', 'password': '123456'},再调用一下header,header = self.data_replace(interface_header),这样header和data都取到了,接下来下载http.py和response.py文件,把它们拷贝到utils目录下,response.py里如果没有安装jsonpath,要安装一下,然后在tasks.py里导入一下from utils.http import MyRequest,然后实例化MyRequest,输入代码
response = MyRequest(url=host + uri, method=method, data=data, is_json=is_json, headers=header).results和logger.debug('接口返回值-->%s' % response),再把之前的logger.debug修改一下,修改后的代码如下:
点击用例管理页面的登录用例后的运行按钮,在日志里打印出:
通过日志可以看出,接口已经被调用了,只不过username的长度不符合,修改前端登录用例的参数还是如此,最后发现data_replace方法有问题,写死了一个data = {'username': 'ssj', 'password': '${password}'},把这个干掉再重启服务,再次运行,打印出正确的token、user和user_id,如下图:
三、校验
用例运行成功就该做校验了,接下来在tasks.py文件里导入from utils.response import ResponseCheck,接下来输入如下代码:
# 验证接口返回值
check = case_obj.check
res = ResponseCheck(check, response)
# 用例结果状态
status = res.status
# 原因
reason = res.reason
logger.debug('用例状态-->%s' % status)
logger.debug('原因-->%s' % reason)
重启异步任务,点击用例管理页面的登录用例后的运行按钮,在日志里打印出:
修改全局参数里的username和password的值,再次点击用例管理页面的登录用例后的运行按钮,在日志里打印出:
用例状态1是用例通过,用例状态999是用例失败
四、获取缓存字段并set到redis中
根据上图,登录用例的缓存字段是token,接下来获取缓存字段,根据用例结果状态status判断,如果status==1,代表用例通过,才能获取到缓存字段,如果缓存字段存在,导入from utils.response import ResponseCheck,ResponseCheck类里有个get_response_value(key, response)方法,该方法的作用是根据传入的key从实际结果里面取到value,获取到value后把它缓存到redis数据库里,下图代码里的cache_field就是缓存字段的key:token,前提是导入from utils import const和import django_redis,把self.redis = django_redis.get_redis_connection()定义到Run类下的__init__(self)方法里,get_redis_connection()里可以连接settings.py文件里具体的redis数据库,大师兄是连接到redis2,我是用default,新增代码如下图:
在utils目录下的const.py里新增cache_prefix = 'earth:cache:',重启异步任务,然后点击用例管理页面的登录用例后的运行按钮,在日志里打印出:
去redis数据库里查缓存信息,如下图:
redis里的和日志里打出的缓存字段一模一样,这是缓存一个字段的,如果多个字段要改一下代码,首先编辑登录用例,把缓存字段改成token, user_id ,点击确定按钮,保存成功,然后在tasks.py里的Run类中增加set_cache方法,传一个cache和response,如果cache_field为真时,直接调用该方法,这里的代码改成这样,如下图:
把之前写的代码拷到set_cache方法里,以逗号分割cache,然后循环caches,把item里的空格替换为空,set_cache中的代码如下:
def set_cache(self, cache, response):
caches = cache.split(',') # 缓存多个字段,按照,分割,返回一个list
for item in caches:
item = item.replace(' ', '') # 缓存字段前后可能有空格
res_cache_value = ResponseCheck.get_response_value(item, response)
logger.debug('从返回值中获取的缓存字段-->%s' % res_cache_value)
self.redis.set(const.cache_prefix + item, res_cache_value, const.token_expire)
重启异步任务以及项目自己的服务(python manage.py runserver),点击登录用例后面的运行按钮,提示操作成功,在异步任务的日志里看到如下:
再去redis里看写进去的key和value,如下图:
两者一模一样
五、从缓存中取字段
接下来修改data_replace方法,增加判断在缓存里取,data_replace方法修改后如下:
def data_replace(self, data):
for k, v in data.items():
keys = re.findall(r'${(.*?)}', str(v))
if keys:
# 如果keys存在,代表有需要替换的参数 ['password'],全局参数进行替换
parameter_obj = models.Parameter.objects.filter(name=keys[0]).first()
if parameter_obj: # 如果全局参数存在,则取全局参数,否则取缓存中的
data[k] = parameter_obj.value # value在parameter表里
else:
cache_value = self.redis.get(const.cache_prefix + keys[0]) # keys[0]是${}里的参数
data[k] = cache_value
logger.debug('替换后的data是-->%s' % data)
return data
再打印一下header,如下图:
重启异步任务,点击用例管理页面的获取项目信息后的运行按钮,在日志里打印出:
所有能打印的都打印出来了,从缓存中获取字段也成功了,登录用例是两个缓存字段(token和user_id),获取项目信息是一个缓存字段(msg),一共三个缓存字段,在日志里可以看到,如下图:
因为set到redis里,从redis里也可以看到缓存字段,如下图: