requests顺序执行实现

多步请求封装，执行完一个用例

def requests(self,step_info):
    request_type =step_info['请求方式']
    if request_type=="get":
        result=self.get(step_info)
    elif request_type == "post":
        result =self.post(step_info)
    else:
        result ={'code':1,'result':'请求方式不支持'}
    return result

思路是用 for 循环调用单接口运行方法时，把单接口的返回字典数据结果作为判断依据，当出现 code 不为 0 时，表示失败，此时可以用 break 终止循环

def request_by_step(self,test_steps): #test_step不止一步
    for test_step in test_steps:
        result = self.request(test_step)
        if result['code'] != 0: #code=1 不执行，第一步已经报错
            break
    return result

# -*- coding: utf-8 -*-
#@File ：requests_utils.py
#@Auth ： wwd
#@Time ： 2020/12/8 8:00 下午
import jsonpath
import requests
import json
from utils.config_utils import local_config
class RequestsUtils:
    def __init__(self):
        self.hosts = local_config.HOSTS
        self.session = requests.session()
    def __get(self, requests_info):
        url = self.hosts + requests_info['请求地址']# 取出下面字典中的请求地址。
# {'测试用例编号': 'api_case_01', '测试用例名称': '获取access_token接口测试', '用例执行': '是', '用例步骤': 'step_01', '接口名称': '获取access_token接口', '请求方式': 'get', '请求头部信息': '', '请求地址': '/cgi-bin/token', '请求参数(get)': '{"grant_type":"client_credential","appid":"wx55614004f367f8ca","secret":"65515b46dd758dfdb09420bb7db2c67f"}', '请求参数(post)': '', '取值方式': '无', '取值代码': '', '取值变量': '', '断言类型': 'body_regexp', '期望结果': '"access_token":"(.+?)"'}
        response = self.session.get(url = url,
                                    params = json.loads(requests_info['请求参数(get)']),
                                    headers = requests_info['请求头部信息']
                                    ) #参数名 =参数值
        result = {
            'code':0,
            'response_code':response.status_code,
            'response_reason':response.reason,
            'response_headers':response.headers,
            'response_body':response.text
        }
        return result

    def __post(self, requests_info):
      url = self.hosts + requests_info['请求地址']# 取出下面字典中的请求地址。
# {'测试用例编号': 'api_case_01', '测试用例名称': '获取access_token接口测试', '用例执行': '是', '用例步骤': 'step_01', '接口名称': '获取access_token接口', '请求方式': 'get', '请求头部信息': '', '请求地址': '/cgi-bin/token', '请求参数(get)': '{"grant_type":"client_credential","appid":"wx55614004f367f8ca","secret":"65515b46dd758dfdb09420bb7db2c67f"}', '请求参数(post)': '', '取值方式': '无', '取值代码': '', '取值变量': '', '断言类型': 'body_regexp', '期望结果': '"access_token":"(.+?)"'}
      response = self.session.post(url = url,
                                    #params = json.loads(requests_info['请求参数(get)']),
                                    params= json.loads(requests_info['请求参数(post)'])
                                     ,
                                    json = json.loads(requests_info['请求参数(post)'])
                                    ) #参数名 =参数值
      response.encoding = response.apparent_encoding#防止乱码
      result = {
            'code':0,
            'response_code':response.status_code,
            'response_reason':response.reason,
            'response_headers':response.headers,
            'response_body':response.text
        }
      return result

    def request(self, step_info):
        request_type = step_info['请求方式']
        if request_type == "get":
            result = self.__get(step_info)
        elif request_type == "post":
            result = self.__post(step_info)
        else:
            result = {'code': 1, 'result': '请求方式不支持'}
        return result
# if __name__=='__main__':    # 测试封装的get（）方法
#         req_dict = {'测试用例编号': 'api_case_01', '测试用例名称': '获取access_token接口测试', '用例执行': '是', '用例步骤': 'step_01', '接口名称': '获取access_token接口', '请求方式': 'get', '请求头部信息': '', '请求地址': '/cgi-bin/token', '请求参数(get)': '{"grant_type":"client_credential","appid":"wx55614004f367f8ca","secret":"65515b46dd758dfdb09420bb7db2c67f"}', '请求参数(post)': '', '取值方式': '无', '取值代码': '', '取值变量': '', '断言类型': 'body_regexp', '期望结果': '"access_token":"(.+?)"'}
#         requestsUtils = RequestsUtils()
#         v = requestsUtils.get(req_dict)
#         print(v)
#
# if __name__ == '__main__':# 测试封装的post（）方法
#         req_post_dict = {'测试用例编号': 'api_case_03', '测试用例名称': '删除标签接口测试', '用例执行': '是', '用例步骤': 'step_03', '接口名称': '删除标签接口', '请求方式': 'post', '请求头部信息': '', '请求地址': '/cgi-bin/tags/delete', '请求参数(get)': '{"access_token":"39_ZlzNDPma7qLWpLJ4K0ir_cSahJ_fg9aevBpGvqRp9VNjqRE6hSkBOSUFla-mFjSGKyF-YFx28sM4Ch1rJISPGVSTahZ8l_xQ9M7CnAFoqUfibusAdeOI4lHEIzB6zhXJQHN5b9as9zhcGtSbBYKeAGAEBN"}', '请求参数(post)': '{   "tag":{        "id" : 456   } }'}
#         requestsUtils = RequestsUtils()
#         v = requestsUtils.post( req_post_dict)
#         print( v )

if __name__=='__main__': #不用直接封装get和post获取
        req_dict = {'测试用例编号': 'api_case_01', '测试用例名称': '获取access_token接口测试', '用例执行': '是', '用例步骤': 'step_01', '接口名称': '获取access_token接口', '请求方式': 'get', '请求头部信息': '', '请求地址': '/cgi-bin/token', '请求参数(get)': '{"grant_type":"client_credential","appid":"wx55614004f367f8ca","secret":"65515b46dd758dfdb09420bb7db2c67f"}', '请求参数(post)': '', '取值方式': '无', '取值代码': '', '取值变量': '', '断言类型': 'body_regexp', '期望结果': '"access_token":"(.+?)"'}
        requestsUtils = RequestsUtils()
        v = requestsUtils.request(req_dict)
        print(v)

巴特西

requests顺序执行实现

最新文章

热门文章