Python3+Requests+Excel完整接口自动化框架

import json
import requests

requests.packages.urllib3.disable_warnings()

class RunMethod:
    def post_main(self,url,data,header=None):
        res = None
        if header != None:
            res = requests.post(url=url,data=data,headers=header,verify=False)
        else:
            res = requests.post(url=url,verify=False)
        return res.json()

    def get_main(self,data=None,header=None):
        res = None
        if header != None:
            res = requests.get(url=url,params=data,verify=False)
        else:
            res = requests.get(url=url,verify=False)
        return res.json()

    def run_main(self,method,header=None):
        res = None
        if method == ‘Post‘:
            res = self.post_main(url,header)
        else:
            res = self.get_main(url,header)
        return json.dumps(res,indent=2,sort_keys=True,ensure_ascii=False)


if __name__ == ‘__main__‘:
    url = ‘http://httpbin.org/post‘
    data = {
        ‘cart‘: ‘11‘
    }
    run = RunMethod()
    run_test = run.run_main(method="Post",url=url,data=data)
    print(run_test)

class global_val:
    Id = ‘0‘
    request_name = ‘1‘
    url = ‘2‘
    run = ‘3‘
    request_way = ‘4‘
    header = ‘5‘
    case_depend = ‘6‘
    data_depend = ‘7‘
    field_depend = ‘8‘
    data = ‘9‘
    expect = ‘10‘
    result = ‘11‘


def get_id():
    """获取case_id"""
    return global_val.Id


def get_request_name():
    """获取请求模块名称"""
    return global_val.request_name


def get_url():
    """获取请求url"""
    return global_val.url


def get_run():
    """获取是否运行"""
    return global_val.run


def get_run_way():
    """获取请求方式"""
    return global_val.request_way


def get_header():
    """获取是否携带header"""
    return global_val.header


def get_case_depend():
    """case依赖"""
    return global_val.case_depend


def get_data_depend():
    """依赖的返回数据"""
    return global_val.data_depend


def get_field_depend():
    """数据依赖字段"""
    return global_val.field_depend


def get_data():
    """获取请求数据"""
    return global_val.data


def get_expect():
    """获取预期结果"""
    return global_val.expect


def get_result():
    """获取返回结果"""
    return global_val.result

from util.operation_excel import OperationExcel
from base.runmethod import RunMethod
from data.get_data import GetData
from jsonpath_rw import jsonpath,parse
import json


class DependentData:
    """解决数据依赖问题"""

    def __init__(self,case_id):
        self.case_id = case_id
        self.opera_excel = OperationExcel()
        self.data = GetData()

    def get_case_line_data(self):
        """
        通过case_id去获取该case_id的整行数据
        :param case_id: 用例ID
        :return:
        """
        rows_data = self.opera_excel.get_row_data(self.case_id)
        return rows_data

    def run_dependent(self):
        """
        执行依赖测试，获取结果
        :return:
        """
        run_method = RunMethod()
        row_num = self.opera_excel.get_row_num(self.case_id)
        request_data = self.data.get_data_for_json(row_num)
        # header = self.data.is_header(row_num)
        method = self.data.get_request_method(row_num)
        url = self.data.get_request_url(row_num)
        res = run_method.run_main(method,request_data)
        return json.loads(res)

    def get_data_for_key(self,row):
        """
        根据依赖的key去获取执行依赖case的响应然后返回
        :return:
        """
        depend_data = self.data.get_depend_key(row)
        response_data = self.run_dependent()
        return [match.value for match in parse(depend_data).find(response_data)][0]

from util.operation_excel import OperationExcel
from data import data_config
from util.operation_json import OperationJson


class GetData:
    """获取excel数据"""

    def __init__(self):
        self.opera_excel = OperationExcel()

    def get_case_lines(self):
        """获取excel行数，即case的个数"""
        return self.opera_excel.get_lines()

    def get_is_run(self,row):
        """获取是否执行"""
        flag = None
        col = int(data_config.get_run())
        run_model = self.opera_excel.get_cell_value(row,col)
        if run_model == ‘yes‘:
            flag = True
        else:
            flag = False
        return flag

    def is_header(self,row):
        """
        是否携带header
        :param row: 行号
        :return:
        """
        col = int(data_config.get_header())
        header = self.opera_excel.get_cell_value(row,col)
        if header != ‘‘:
            return header
        else:
            return None

    def get_request_method(self,row):
        """
        获取请求方式
        :param row: 行号
        :return:
        """
        # col 列
        col = int(data_config.get_run_way())
        request_method = self.opera_excel.get_cell_value(row,col)
        return request_method

    def get_request_url(self,row):
        """
        获取url
        :param row: 行号
        :return:
        """
        col = int(data_config.get_url())
        url = self.opera_excel.get_cell_value(row,col)
        return url

    def get_request_data(self,row):
        """
        获取请求数据
        :param row:行号
        :return:
        """
        col = int(data_config.get_data())
        data = self.opera_excel.get_cell_value(row,col)
        if data == ‘‘:
            return None
        return data

    def get_data_for_json(self,row):
        """
        通过关键字拿到data数据
        :param row:
        :return:
        """
        opera_json = OperationJson()
        request_data = opera_json.get_data(self.get_request_data(row))
        return request_data

    def get_expcet_data(self,row):
        """
        获取预期结果
        :param row:
        :return:
        """
        col = int(data_config.get_expect())
        expect = self.opera_excel.get_cell_value(row,col)
        if expect == "":
            return None
        else:
            return expect

    def write_result(self,row,value):
        """
        写入结果数据
        :param row:
        :param col:
        :return:
        """
        col = int(data_config.get_result())
        self.opera_excel.write_value(row,col,value)

    def get_depend_key(self,row):
        """
        获取依赖数据的key
        :param row:行号
        :return:
        """
        col = int(data_config.get_data_depend())
        depend_key = self.opera_excel.get_cell_value(row,col)
        if depend_key == "":
            return None
        else:
            return depend_key

    def is_depend(self,row):
        """
        判断是否有case依赖
        :param row:行号
        :return:
        """
        col = int(data_config.get_case_depend())  # 获取是否存在数据依赖列
        depend_case_id = self.opera_excel.get_cell_value(row,col)
        if depend_case_id == "":
            return None
        else:
            return depend_case_id

    def get_depend_field(self,row):
        """
        获取依赖字段
        :param row:
        :return:
        """
        col = int(data_config.get_field_depend())
        data = self.opera_excel.get_cell_value(row,col)
        if data == "":
            return None
        else:
            return data

{
  "user": {
    "username": "2222211","password": "123456"
  },"filtrate": {
    "type_id": "2","brand_id": "1","model_id": "111"
  },"search": {
    "page": "1","keyword": "oppo","type": "12"
  },"token": {
    "token": ""
  }

{"data": {"token": "db6f0abee4e5040f5337f5c47a82879"}}

from base.runmethod import RunMethod
from data.get_data import GetData
from util.common_util import CommonUtil
from data.dependent_data import DependentData
# from util.send_email import SendEmail
from util.operation_header import OperationHeader
from util.operation_json import OperationJson


class RunTest:

    def __init__(self):
        self.run_method = RunMethod()
        self.data = GetData()
        self.com_util = CommonUtil()
        # self.send_email = SendEmail()

    def go_on_run(self):
        """程序执行"""
        pass_count = []
        fail_count = []
        res = None
        # 获取用例数
        rows_count = self.data.get_case_lines()
        # 第一行索引为0
        for i in range(1,rows_count):
            is_run = self.data.get_is_run(i)
            if is_run:
                url = self.data.get_request_url(i)
                method = self.data.get_request_method(i)
                request_data = self.data.get_data_for_json(i)
                expect = self.data.get_expcet_data(i)
                header = self.data.is_header(i)
                depend_case = self.data.is_depend(i)

                if depend_case != None:
                    self.depend_data = DependentData(depend_case)
                    # 获取依赖的响应数据
                    depend_response_data = self.depend_data.get_data_for_key(i)
                    # 获取依赖的key
                    depend_key = self.data.get_depend_field(i)
                    # 更新请求字段
                    request_data[depend_key] = depend_response_data
                # 如果header字段值为write则将该接口的返回的token写入到token.json文件，如果为yes则读取token.json文件
                if header == "write":
                    res = self.run_method.run_main(method,request_data)
                    op_header = OperationHeader(res)
                    op_header.write_token()
                elif header == ‘yes‘:
                    op_json = OperationJson("../dataconfig/token.json")
                    token = op_json.get_data(‘data‘)
                    request_data = dict(request_data,**token)  # 把请求数据与登录token合并，并作为请求数据

                    res = self.run_method.run_main(method,request_data)
                else:
                    res = self.run_method.run_main(method,request_data)

                if expect != None:
                    if self.com_util.is_contain(expect,res):
                        self.data.write_result(i,"Pass")
                        pass_count.append(i)
                    else:
                        self.data.write_result(i,res)
                        fail_count.append(i)
                else:
                    print(f"用例ID：case-{i}，预期结果不能为空")

        # 发送邮件
        # self.send_email.send_main(pass_count,fail_count)

        print(f"通过用例数：{len(pass_count)}")
        print(f"失败用例数：{len(fail_count)}")


if __name__ == ‘__main__‘:
    run = RunTest()
    run.go_on_run()

class CommonUtil:
    def is_contain(self,str_one,str_two):
        """
        判断一个字符串是否在另一个字符串中
        :param str_one:
        :param str_two:
        :return:
        """
        flag = None
        if str_one in str_two:
            flag = True
        else:
            flag = False
        return flag

import xlrd
from xlutils.copy import copy


class OperationExcel:
    """操作excel"""

    def __init__(self,file_name=None,sheet_id=None):
        if file_name:
            self.file_name = file_name
            self.sheet_id = sheet_id
        else:
            self.file_name =‘../dataconfig/case1.xls‘
            self.sheet_id = 0
        self.data = self.get_data()

    def get_data(self):
        """
        获取sheets的内容
        :return:
        """
        data = xlrd.open_workbook(self.file_name)
        tables = data.sheets()[self.sheet_id]
        return tables

    def get_lines(self):
        """
        获取单元格行数
        :return:
        """
        tables = self.data
        return tables.nrows

    def get_cell_value(self,col):
        """
        获取单元格数据
        :param row: 行
        :param col: 列
        :return:
        """
        tables = self.data
        cell = tables.cell_value(row,col)
        return cell

    def write_value(self,value):
        """
        回写数据到excel
        :param row:行
        :param col:列
        :param value:值
        :return:
        """
        read_data = xlrd.open_workbook(self.file_name)
        write_data = copy(read_data)
        sheet_data = write_data.get_sheet(0)
        sheet_data.write(row,value)
        write_data.save(self.file_name)

    def get_row_data(self,case_id):
        """
        根据对应的case_id获取对应行的内容
        :param case_id: 用例id
        :return:
        """
        row_num = self.get_row_num(case_id)
        row_data = self.get_row_value(row_num)
        return row_data

    def get_row_num(self,case_id):
        """
        根据case_id获取对应行号
        :param case_id:
        :return:
        """
        num = 0
        cols_data = self.get_cols_data()
        for col_data in cols_data:
            if case_id in col_data:
                return num
            num = num + 1

    def get_row_value(self,row):
        """
         根据行号，找到该行的内容
        :param row:行号
        :return:

        """
        tables = self.data
        row_data = tables.row_values(row)
        return row_data

    def get_cols_data(self,col_id=None):
        """
        获取某一列的内容
        :param col_id:列号
        :return:
        """
        if col_id != None:
            cols = self.data.col_values(col_id)
        else:
            cols = self.data.col_values(0)
        return cols


if __name__ == ‘__main__‘:
    opera = OperationExcel()
    opera.get_data()
    print(opera.get_data().nrows)
    print(opera.get_lines())
    print(opera.get_cell_value(1,2))

import json
from util.operation_json import OperationJson
from base.runmethod import RunMethod
class OperationHeader:

    def __init__(self,response):
        self.response = json.loads(response)

    def get_response_token(self):
        ‘‘‘
        获取登录返回的token
        ‘‘‘
        token = {"data":{"token":self.response[‘data‘][‘token‘]}}
        return token

    def write_token(self):
        op_json = OperationJson()
        op_json.write_data(self.get_response_token())



if __name__ == ‘__main__‘:

    url = "http://xxxxx"

    data = {
        "username": "1111","password": "123456"
    }
    run_method=RunMethod()
    # res = json.dumps(requests.post(url,data).json())
    res=run_method.run_main(‘Post‘,data)
    op = OperationHeader(res)
    op.write_token()

import json


class OperationJson:
    """操作json文件"""

    def __init__(self,file_path=None):
        if file_path==None:
            self.file_path="../dataconfig/data.json"
        else:
            self.file_path=file_path
        self.data = self.read_data()

    def read_data(self):
        """
        读取json文件
        :param file_name:文件路径
        :return:
        """
        with open(self.file_path) as fp:
            data = json.load(fp)
            return data

    def get_data(self,id):
        """根据关键字获取对应数据"""
        return self.data[id]

    # 写入json
    def write_data(self,data):
        with open("../dataconfig/token.json",‘w‘) as fp:
            fp.write(json.dumps(data))


if __name__ == ‘__main__‘:
    # file_path = "../dataconfig/data.json"
    opejson = OperationJson()
    print(opejson.read_data())
    print(opejson.get_data(‘filtrate‘))