Python接口自动化（初版）

原创

Meccer

修改于 2022-03-12 11:40:54

89100

代码可运行

文章被收录于专栏：测试学习之路测试学习之路

运行总次数：0

代码可运行

框架设计

pytest+requests+allure

目录结构

utils                  ——工具类
data                   ——用例文件存放目录
conf                   ——配置类
scripts                ——测试用例驱动
TestCase               ——测试用例
start.py               ——执行脚本
pytest.ini             ——pytest配置文件

添加配置文件

配置文件总是项目中必不可少的部分！

将固定不变的信息集中在固定的文件中

settings.py

项目中都应该有一个文件对整体的配置进行管理，我也在这个python项目中设置了此文件。

在项目conf目录创建settings.py文件，所有的配置信息写在这个文件里面。

import os
import datetime

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

# 关于Excel配置表中测试用例默认配置
FILE_NAME = "本地接口测试用例.xlsx"
FILES_PATH = os.path.join(BASE_DIR, "data", FILE_NAME)


#----初始化template cookies dict

COOKIES_DICT = {}

# -----allure报告相关

ALLURE_COMMAND = "allure generate {from_json_path} -o {save_to_path} --clean".format(
    from_json_path=os.path.join(BASE_DIR, "report", "json_result"),
    save_to_path=os.path.join(BASE_DIR, "report", "allure_result")
)

# ---日志相关
# 日志级别
LOG_LEVEL = "debug"
LOG_STREAM_LEVEL = "debug"  # 屏幕输出流
LOG_FILE_LEVEL = "info"  # 文件输出流

# 日志文件命名
LOG_FILE_NAME = os.path.join(BASE_DIR, "logs", datetime.datetime.now().strftime("%Y-%m-%d") + ".log")



# 邮件相关
# 第三方SMTP服务
MAIL_HOST = "smtp.qq.com"  # 设置服务器
# 邮件相关
# 第三方SMTP服务
MAIL_HOST = "smtp.qq.com"  # 设置服务器
MAIL_USER = "test@qq.com"  # 用户名
MAIL_TOKEN = "hhhhhhhhm"  # 授权码
# 设置收件人和发件人
SENDER = "test@qq.com"
RECEIVERS = ["test@qq.com", "test@163.com"]  # 接收邮箱可以设置你的qq或者其它邮箱

# 邮件主题
THEME = "接口测试报告"
# 正文内容
SEND_CONTENT = "详情看附件"
# 附件的file name
SEND_FILE_NAME = "allure_report.zip"

excel工具类

用来读取Excel中的用例数据。

我们在utils目录中新建一个文件

import xlrd
from conf import settings


class ExcelOperate:
    def __init__(self, file_path, sheet_by_index=0):
        self.book = xlrd.open_workbook(file_path)
        self.sheet = self.book.sheet_by_index(sheet_by_index)

    def get_excel(self):
        """
        获取Excel表中接口测试用例
        :return: 构造好的测试用例数据,格式：[{'case_num': '', 'title ': '', 'url': '',}]
        dict()
        """
        title = self.sheet.row_values(0)    #获取excel中的标题栏，
        print(type(title))
        case_suite=[dict(zip(title, self.sheet.row_values(row))) for row in range(1, self.sheet.nrows)]   #将标题和遍历取出的值一一对应，形成多个字典，放在用例列表中。
        return case_suite

用python的xlrd 模块对Excel文件进行了读取。再按照制定格式输出用例。

核心驱动类

用于读取用例中所有和请求相关的数据，并且处理依赖和写入cookies


"""
处理请求
ExcelHandler模块中读取测试用例,对用例字段进去处理，然后发送请求获取响应结果
"""
import json
import re

import requests

from conf import settings
from jsonpath_rw import parse  # 第三方插件，需要pip install ,作用json数据提取
# jsonpath_rw官网：https://github.com/kennknowles/python-jsonpath-rw


from utils.ExcelHandler import ExcelOperate

from utils.LogHandler import logger


class RequestsOperate:

    def __init__(self, current_case, all_excel_data_list):
        """
        :param current_case: 用例列表中的一条条单独测试用例(字典形式),用来给request构造发送请求
        :param all_excel_data_list: 全部用例列表
        """
        self.current_case = current_case
        self.all_excel_data_list = all_excel_data_list

    def get_response_msg(self):
        """
        发送请求并且获取结果
        :return:
        """
        return self._send_msg()

    def _send_msg(self):
        """发请求"""
        logger().info("正在向{0}发送请求,{1}".format(self.current_case.get("url"), self.current_case))
        response = requests.request(
            method=self.current_case.get("method"),
            url=self.current_case.get("url"),
            data=self._check_request_data(),
            json=self._check_request_json(),
            params=self._check_request_params(),
            cookies=self._check_request_cookies(),
            headers=self._check_request_headers(),
        )
        self._write_cookies(response)
        return json.loads(self.current_case.get("except")), response.json()

    def _check_request_headers(self):
        """
        校验请求头，做携带cookies 和 数据依赖的问题
        {
            'user':'${}$',
            'testfan-id':'ca447223-876e-46ba-9e45-f775335dfcbe'
        }
        :return:
        """
        headers = self.current_case.get("headers")
        if headers:
            return self._operate_re_msg(headers)
        else:
            return {}

    def _write_cookies(self, response):
        """ 监测响应结果中是否含有cookies，有就保存起来"""
        for item in self.all_excel_data_list:
            if item.get("case_num") == self.current_case.get("case_num"):
                item["temporary_response_cookies"] = response.cookies.get_dict()
                if response.headers.get("Content-Type").lower() == "application/json":
                    item["temporary_response_json"] = response.json()
                # 如果去请求头中取数据都存一份
                item["temporary_request_headers"] = self.current_case.get("headers")
                item["temporary_request_data"] = self.current_case.get("data")
                item["temporary_request_json"] = self.current_case.get("json")
                item["temporary_request_params"] = self.current_case.get("params")
                # 所有去响应头中取值的都存一份
                item["temporary_response_headers"] = response.headers
                print(response.cookies.get_dict())



    def _check_request_data(self):
        """处理请求的data参数，检查是否有依赖"""
        data = self.current_case.get("data")
        if data:
            return self._operate_re_msg(data)
        else:
            return {}

    def _check_request_json(self):
        """处理请求的data参数，检查是否有依赖"""
        data = self.current_case.get("json")
        if data:
            return json.loads(data)
        else:
            return {}

    def _check_request_params(self):
        """处理get请求的参数，检查是否有依赖"""
        params = self.current_case.get("params")
        if params:
            return json.loads(params)
        else:
            return {}

    def _check_request_cookies(self):
        """处理请求中的cookies"""
        cookies_case_num = self.current_case.get("cookies")
        if cookies_case_num:  # 当前接口需要cookies
            for item in self.all_excel_data_list:
                if item.get("case_num") == cookies_case_num:
                    return item.get("temporary_response_cookies", {})
        else:
            return {}

    def _operate_re_msg(self, parameter):
        """
        正则校验,数据依赖的字段
        :param parameter: 各种参数，如：data,header,params
        :return:
        """
        # 使用re 将提取依赖字段 {"testfan-token":"${neeo_001>response_json>data}$"}
        if isinstance(parameter, dict):
            json.dumps(parameter)
        pattern = re.compile("\${(.*?)}\$")  # 定义规则
        rule_list = pattern.findall(parameter)  # 按照规则匹配
        if rule_list:  # 该参数有数据依赖要处理
            for rule in rule_list:
                case_num, params, json_path = rule.split(">")
                for line in self.all_excel_data_list:
                    print(line)
                    if line.get("case_num") == case_num:
                        temp_data = line.get("temporary_{0}".format(params))
                        if isinstance(temp_data, str):
                            temp_data = json.loads(temp_data)
                        match_list = parse(json_path).find(temp_data)
                        print(match_list)
                        if match_list:
                            match_data = [v.value for v in match_list][0]
                        # 将提取出来的值替换到原来规则,
                parameter = re.sub(pattern=pattern, repl=match_data, string=parameter, count=1)
            return json.loads(parameter)
        else:
            if isinstance(parameter, str):
                parameter = json.loads(parameter)
            return parameter

记录操作日志

日志，大家应该都很熟悉这个名词，就是记录代码中的动作。

在utils目录中新建loggerHandler.py文件。

这个文件就是我们用来在自动化测试过程中记录一些操作步骤的。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import logging

from conf import settings


class LoggerHandler:
    """ 日志操作 """
    _logger_level = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'critical': logging.CRITICAL
    }

    def __init__(self, log_name, file_name, logger_level, stream_level='info', file_level='warning'):
        self.log_name = log_name
        self.file_name = file_name
        self.logger_level = self._logger_level.get(logger_level, 'debug')
        self.stream_level = self._logger_level.get(stream_level, 'info')
        self.file_level = self._logger_level.get(file_level, 'warning')
        # 创建日志对象
        self.logger = logging.getLogger(self.log_name)
        # 设置日志级别
        self.logger.setLevel(self.logger_level)

        if not self.logger.handlers:
            # 设置日志输出流
            f_stream = logging.StreamHandler()
            f_file = logging.FileHandler(self.file_name)
            # 设置输出流级别
            f_stream.setLevel(self.stream_level)
            f_file.setLevel(self.file_level)
            # 设置日志输出格式
            formatter = logging.Formatter(
                "%(asctime)s %(name)s %(levelname)s %(message)s"
            )
            f_stream.setFormatter(formatter)
            f_file.setFormatter(formatter)
            self.logger.addHandler(f_stream)
            self.logger.addHandler(f_file)
        # print(1111111, self.logger.handlers)

    @property
    def get_logger(self):
        return self.logger


def logger(log_name='接口测试'):
    return LoggerHandler(
        log_name=log_name,
        logger_level=settings.LOG_LEVEL,
        file_name=settings.LOG_FILE_NAME,
        stream_level=settings.LOG_STREAM_LEVEL,
        file_level=settings.LOG_FILE_LEVEL,
    ).get_logger


if __name__ == '__main__':
    logger().debug('aaaa')
    logger().info('aaaa')
    logger().warning('aaaa')
    logger().error('aaaa')

在终端中运行该文件，就看到命令行打印出了：

2022-03-12 18:49:59,371 接口测试 DEBUG aaaa
2022-03-12 18:49:59,371 接口测试 INFO aaaa
2022-03-12 18:49:59,371 接口测试 WARNING aaaa
2022-03-12 18:49:59,371 接口测试 ERROR aaaa

然后在项目logs目录下生成了当日的日志文件。

allure报告类

用来定制allure的输出，处理打包发送报告邮件的需求

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
allure报告相关
"""
import os

import zipfile  # 打包
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header

import subprocess  # 执行cmd命令
from conf import settings
from utils.LogHandler import logger


class AllureOperate:

    def get_allure_report(self):
        """生成报告"""
        # os.system(ALLURE_COMMAND) 运行cmd命令，但是不安全
        logger().info("正在生成测试报告......")
        subprocess.call(settings.ALLURE_COMMAND, shell=True)
        logger().info("生成测试报告成功......")

    def check_zip(self):
        """打包"""
        try:
            logger().info("正在打包测试报告......")
            BASE_DIR = os.path.join(settings.BASE_DIR, "report")
            start_zip_dir = os.path.join(BASE_DIR, "allure_result")  # 要压缩文件夹的根路径
            zip_file_name = 'allure_report.zip'  # 为压缩后的文件起个名字
            zip_file_path = os.path.join(BASE_DIR, zip_file_name)
            f = zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED)
            for dir_path, dir_name, file_names in os.walk(start_zip_dir):
                # 要是不replace，就从根目录开始复制
                file_path = dir_path.replace(start_zip_dir, '')
                # 实现当前文件夹以及包含的所有文件
                file_path = file_path and file_path + os.sep or ''
                for file_name in file_names:
                    f.write(os.path.join(dir_path, file_name), file_path + file_name)
            f.close()
            logger().info("打包测试报告完成......")
        except Exception as er:
            logger().error("打包测试报告失败:{0}".format(er))

    def send_mail(self):
        """发送邮件"""
        # 第三方SMTP服务
        mail_host = settings.MAIL_HOST  # 设置服务器
        mail_user = settings.MAIL_USER  # 用户名
        mail_pass = settings.MAIL_TOKEN  # 口令
        # 设置收件人和发件人
        sender = settings.SENDER
        receivers = settings.RECEIVERS  # 接收邮箱可以设置你的qq或者其它邮箱
        # 创建一个带附件的实例对象
        message = MIMEMultipart()
        # 邮箱主题、收件人、发件人
        subject = settings.THEME  # 邮件主题
        message["Subject"] = Header(subject, "utf-8")
        message["From"] = Header("{0}".format(sender), "utf-8")
        message["To"] = Header("{0}".format(";".join(receivers)), "utf-8")
        # 邮件正文内容
        send_content = settings.SEND_CONTENT
        content_obj = MIMEText(send_content, "plain", "utf-8")  # 第一个参数为邮件内容
        message.attach(content_obj)
        # 构造附件
        att = MIMEText(_text=self._get_zip_file(), _subtype="base64", _charset="utf-8")
        att["Content-Type"] = "application/octet-stream"
        att["Content-Disposition"] = 'attachment;filename="{}"'.format(settings.SEND_FILE_NAME)  # 邮件附件中显示什么名字
        message.attach(att)
        try:
            smtp_obj = smtplib.SMTP()
            smtp_obj.connect(mail_host, 25)  # 25 为 SMTP端口号
            smtp_obj.login(mail_user, mail_pass)
            smtp_obj.send_message(sender, receivers, message.as_string())
            smtp_obj.quit()
            logger().info("邮件发送成功")
            print("邮箱发送成功")
        except smtplib.SMTPException as er:
            logger().error("email end error:{0}".format(er))

    def _get_zip_file(self):
        """获取zip文件内容"""
        with open(file=os.path.join(settings.BASE_DIR, "report", "allure_report.zip"), mode="rb") as f:
            return f.read()


if __name__ == '__main__':
    AllureOperate().get_allure_report()

执行用例

以上我们已经编写完成了整个框架和测试用例。

我们新增一个模板用例方法，并利用@pytest.mark.parametrize来进行参数化

#!/usr/bin/env python
# -*- coding: utf-8 -*-


import pytest
import allure
from deepdiff import DeepDiff

from utils.AssertResult import AssertUtil
from utils.ExcelHandler import ExcelOperate
from utils.RequestsHandler import RequestsOperate
from utils.LogHandler import logger
from conf import settings

excel_data_list = ExcelOperate(settings.FILES_PATH, 3).get_excel()


class TestCase:
    @pytest.mark.parametrize("item", excel_data_list)
    def test_case(self, item):
        logger().info("正在进行断言...")
        except_data, result = RequestsOperate(current_case=item, all_excel_data_list=excel_data_list).get_response_msg()
        allure.dynamic.title(item.get("title"))
        allure.dynamic.description(
            "<b style='color:red'>请求的url:</b>{0}<hr />"
            "<b style='color:red'>预期值:</b>{1}<hr />"
            "<b style='color:red'>实际执行结果:</b>{2}<hr />".format(item["url"], item["except"], result)
        )

        assert AssertUtil().assert_in_body(except_data,result)
        logger().info("完成断言,{0}-{1}".format(except_data, result))

接下来就是执行命令脚本：

#!/usr/bin/env python
# -*- coding: utf-8 -*-


import os

import shutil
import pytest

from conf.settings import BASE_DIR
from utils.AllureHandler import AllureOperate

if __name__ == '__main__':
    dir_path = os.path.join(BASE_DIR, "report", "json_result")
    if os.path.isdir(dir_path):
        # 执行用例前，首先清除上一次测试生成的json文件
        shutil.rmtree(dir_path)
        pytest.main()  # 执行测试用例
        # 生成allure测试报告
        allure_obj = AllureOperate()
        allure_obj.get_allure_report()
        # 压缩文件
        allure_obj.check_zip()
        # 执行完毕后发送邮件
        # allure_obj.send_mail()z
    else:
        os.makedirs(dir_path)