python接口自动化第二章

AutomatedTesting

第二节 common目录（封装读取配置文件，参数依赖，封装请求信息）

---

文章目录

• AutomatedTesting
• 前言
• 一、封装读取配置文件
• 二、参数依赖
• 三、封装请求信息
• 总结

---

前言

接口自动化 项目结构图

目录结构大部分都是常用目录
common - 公共方法
config - 配置文件信息
report - 日志/报告存放
runtest - 运行入口
testfile - 用例文件
tools - 工具箱
DDT - 数据驱动参数化
jenkins - 定时运行
实现：读取表格用例文件，jenkins定时运行，实现接口自动化，并做到数据与代码分离。

---

一、封装读取配置文件

代码如下：

"""
@File    :  read_config.py
@Time    :  2022/4/21
@Author  :  Mr_桐
"""

# -*- coding:utf-8 -*-
import sys
import os
import configparser
curPath = os.path.abspath(os.path.dirname(__file__))
rootPath = os.path.split(curPath)[0]
sys.path.append(rootPath)
from common import setting_path


class ReadConfig:

    # 初始化函数：实例化conf对象，读取传入的配置文件
    def __init__(self):
        self.conf = configparser.ConfigParser()
        # 全局变量global.ini路径
        file = setting_path.global_ini
        self.conf.read(file, encoding='utf-8')
    # 获取字符串类型
    def get_config_str(self, section, option):
        return self.conf.get(section, option)
    # 获取布尔类型
    def get_config_boolean(self, section, option):
        return self.conf.getboolean(section, option)
    # 获取整型
    def get_config_int(self, section, option):
        return self.conf.getint(section, option)
    # 获取浮点型
    def get_config_float(self, section, option):
        return self.conf.getfloat(section, option)

二、参数依赖

代码如下：

"""
@File    :  function.py
@Time    :  2022/4/21
@Author  :  Mr_桐
"""

# -*- coding:utf-8 -*-
import jsonpath,codecs,unittest
import sys
import os
curPath = os.path.abspath(os.path.dirname(__file__))
rootPath = os.path.split(curPath)[0]
sys.path.append(rootPath)
from common import setting_path,logger



varibles_file = setting_path.testdata_dir
log = logger.LogInformation()
# 按规则需要提取的返回值写入到变量文件
def extract( re, varibles, rules):

    # 创建空字典，用于存提取数据
    varible_dirs = {}
    # 支持多个，分割规则字段及变量字段
    rules = rules.split(',')
    varibles = varibles.split(',')
    # 利用zip循环取出来规则及变量名，转成字典格式
    for rule,varible in zip(rules, varibles):
        # varible_dir为提取出来的变量
        varible_dir = jsonpath.jsonpath(re, rule)
        # list转换为str,去掉[]'等字符
        varible_dir = str(varible_dir).replace('[','').replace(']','').replace("'","")
        # 写入varible_dirs字典中
        varible_dirs.setdefault(varible,varible_dir)
    # 把varible_dirs字典内容写入变量文件
    with codecs.open(varibles_file , 'a+', encoding='utf-8') as f:
        f.write(str(varible_dirs))
        f.write('\r\n')



# 用value查找对应的键值
def get_keys(re, value):
    # print([k for k, v in re.items() if v == value])
    return [k for k, v in re.items() if v == value]



# 传参进行处理，查找变量文件匹配替换
def find_value(body):

    # 打开变量文件
    try:
        with open(varibles_file, encoding='utf-8') as readf:
            # 倒序循环读取每一列
            for varibles_line in reversed(readf.readlines()):
                # 每列转为字符串
                varibles_line = eval(varibles_line)
                # 循环取出变量文件的键值
                for k, v in varibles_line.items():
                    # 处理key键格式与文档方式相同${varibles}
                    k = str('$' + '{' + str(k) + '}')
                    # 用获取键值方法查找应该替换的变量
                    varibles_keys = get_keys(body, k)
                    # 遍历变量key键
                    for varibles_key in varibles_keys:
                        # 遍历数据body_key建
                        for body_key in body.keys():
                            # 判断变量key键是否等于body_key键,等于则替换成变量中的value值
                            if varibles_key == body_key:
                                body[body_key] = v
                    readf.close()
            return body
    except BaseException as erro:
        log.log_info('错误状态为:{}'.format(erro))
        return body


def skip_test_case(msg):

    if msg == 'skip':
        @unittest.skip('初始化接口跳过用例')
        def skip():
            pass
        skip()
        
if __name__ == '__main__':
	# 测试extract方法
    body_old = {'status': 'success', 'code': '123', 'message': 'ok'}
    extract(body_old, 'status,code', '$.status,$.code')
    
	# 测试find_value方法
    body_new = {"id":'${status}',"result_id":'${code}'}
    find_value(body_new)
    print(body_new)

extract 提取变量名为code和status, 提取格式为json_path方法，记录到varibles_file文件中

find_value 对需要修改的变量进行替换，可以看见输出结果，body_new方法中id的值变量替换成了提取的status字段的值，result_id的值也同样替换成了提取后的code值

三、封装请求信息

代码如下：

"""
@File    :  send_request.py
@Time    :  2022/4/21
@Author  :  Mr_桐
"""

# -*- coding:utf-8 -*-
import sys
import os
import requests,json
curPath = os.path.abspath(os.path.dirname(__file__))
rootPath = os.path.split(curPath)[0]
sys.path.append(rootPath)
from common.function import extract, find_value
from common import setting_path
from common.read_excel import ReadExcel
from common.email_switch import SwitchTrue
from testfile.log_output import Logcase



class SendRequests():
    # 从读取的表格中获取响应的参数作为传递
    def sendRequests(self, s, apiData):
        # ssl警告忽略
        def verify():
            v = True
            return v

        # 请求头信息
        def headers():
            headers = apiData['headers']
            if headers == '':
                headers = None
                return headers
            else:
                headers = dict([line.split(": ", 1)
                                for line in apiData['headers'].split("\n")])
                return headers

        # 请求方式
        def method():
            method = apiData["method"]
            return method

        # 请求地址
        def url():
            host = apiData["host"]
            url = apiData["url"]
            url = host + url
            return url

        # 接口地址
        def body():
            # 请求body
            if apiData["body"] == "":
                body = None
                return body
            else:
                if apiData['is_variable'] == 1:
                    body = eval(apiData["body"])
                    body = json.dumps(body)
                else:
                    body = eval(apiData["body"])
                    body = find_value(body)
                    return body
                return body

        # 返回json结果
        def res():

            re = s.request(
                method=method(),
                url=url(),
                headers=headers(),
                data=body(),
                verify=verify(),
                )

            #  判断是否需要提取
            if apiData['dependent'] == '1':
                extract(re.json(), apiData['variable'], apiData['rule'])
            # 返回状态不等于200时开启邮件发送
            if re.status_code != 200:
                SwitchTrue()
            # 输出日志，body()为提取后的传值
            Logcase(re, apiData, body())
            return re


        return res()

---

总结

欢迎大家多交流，vx：-20206688