Python3+Pytest 接口自动化测试全方案设计与开发（部分笔记）

课程链接： https://www.boxuegu.com/promote/detail-1484.html

第一部分：项目介绍及框架规划

接口测试框架流程

image.png

接口测试代码结构

image.png

第二部分：接口自动化框架编写

1. Requests 的封装

import requests
from utils.LogUtil import my_log
#1、创建封装get方法
def requests_get(url,headers):
#2、发送requests get请求
    r = requests.get(url,headers = headers)
#3、获取结果相应内容
    code = r.status_code
    try:
        body = r.json()
    except Exception as e:
        body = r.text
#4、内容存到字典
    res = dict()
    res["code"] = code
    res["body"] = body
#5、字典返回
    return res

#post方法封装
#1、创建post方法
def requests_post(url,json=None,headers=None):
#2、发送post请求
    r= requests.post(url,json=json,headers=headers)
#3、获取结果内容
    code = r.status_code
    try:
        body = r.json()
    except Exception as e:
        body = r.text
#4、内容存到字典
    res = dict()
    res["code"] = code
    res["body"] = body
#5、字典返回
    return res

#重构
#1、创建类
class Request:
#2、定义公共方法
    def __init__(self):
        self.log = my_log("Requests")
    def requests_api(self,url,data = None,json=None,headers=None,cookies=None,method="get"):

        if method =="get":
            #get请求
            self.log.debug("发送get请求")
            r = requests.get(url, data = data, json=json, headers=headers,cookies=cookies)
        elif method == "post":
            #post请求
            self.log.debug("发送post请求")
            r = requests.post(url,data = data,  json=json, headers=headers,cookies=cookies)

        #2. 重复的内容，复制进来
        #获取结果内容
        code = r.status_code
        try:
            body = r.json()
        except Exception as e:
            body = r.text
        #内容存到字典
        res = dict()
        res["code"] = code
        res["body"] = body
        #字典返回
        return res

#3、重构get/post方法
    #get
    #1、定义方法
    def get(self,url,**kwargs):
    #2、定义参数
        #url,json,headers,cookies,method
    #3、调用公共方法
        return self.requests_api(url,method="get",**kwargs)

    def post(self,url,**kwargs):
    #2、定义参数
        #url,json,headers,cookies,method
    #3、调用公共方法
        return self.requests_api(url,method="post",**kwargs)

2.Yaml 文件的封装

import os
import yaml
#1、创建类
class YamlReader:
#2、初始化，文件是否存在
    def __init__(self,yamlf):
        if os.path.exists(yamlf):
            self.yamlf = yamlf
        else:
            raise FileNotFoundError("文件不存在")
        self._data = None
        self._data_all = None
#3、yaml读取
    #单个文档读取
    def data(self):
        #第一次调用data，读取yaml文档，如果不是，直接返回之前保存的数据
        if not self._data:
            with open(self.yamlf,"rb") as f:
                self._data = yaml.safe_load(f)
        return self._data
    #多个文档读取
    def data_all(self):
        #第一次调用data，读取yaml文档，如果不是，直接返回之前保存的数据
        if not self._data_all:
            with open(self.yamlf,"rb") as f:
                self._data_all = list(yaml.safe_load_all(f))
        return self._data_all

3.日志文件的封装

import logging
from config import Conf
import datetime,os
from config.Conf import ConfigYaml
#定义日志级别的映射
log_l = {
    "info": logging.INFO,
    "debug": logging.DEBUG,
    "warning": logging.WARNING,
    "error": logging.ERROR

}
#1、创建类
class Logger:
#2、定义参数
    #输出文件名称，Loggername，日志级别
    def __init__(self,log_file,log_name,log_level):
        self.log_file = log_file #扩展名 配置文件
        self.log_name = log_name #参数
        self.log_level = log_level # 配置文件
#3、编写输出控制台或文件
        # 设置logger名称
        self.logger = logging.getLogger(self.log_name)
        # 设置log级别
        self.logger.setLevel(log_l[self.log_level]) #logging.INFO
        #判断handlers是否存在
        if not self.logger.handlers:
            # 输出控制台
            fh_stream = logging.StreamHandler()
            fh_stream.setLevel(log_l[self.log_level])
            formatter = logging.Formatter('%(asctime)s %(name)s %(levelname)s %(message)s ')
            fh_stream.setFormatter(formatter)
            # 写入文件
            fh_file = logging.FileHandler(self.log_file)
            fh_file.setLevel(log_l[self.log_level])
            fh_file.setFormatter(formatter)

            # 添加handler
            self.logger.addHandler(fh_stream)
            self.logger.addHandler(fh_file)

#1、初始化参数数据
#日志文件名称，日志文件级别
#日志文件名称 = logs目录 + 当前时间+扩展名
#log目录
log_path = Conf.get_log_path()
#当前时间
current_time = datetime.datetime.now().strftime("%Y-%m-%d")
#扩展名
log_extension = ConfigYaml().get_conf_log_extension()
logfile = os.path.join(log_path,current_time+log_extension)
#print(logfile)
#日志文件级别
loglevel = ConfigYaml().get_conf_log()
#print(loglevel)
#2、对外方法，初始log工具类，提供其它类使用
def my_log(log_name = __file__):
    return Logger(log_file=logfile,log_name=log_name,log_level=loglevel).logger

if __name__ == "__main__":
    my_log().debug("this is a debug")

4.pytest的应用

之前总结的文档： https://www.jianshu.com/p/981c32b65de1

image.png

[pytest]
# 命令行参数----空格分隔，可添加多个命令行参数 -所有参数均为插件包的参数
addopts = -s -reruns 1 --html=../report/report.html
# 测试路径----当前目录下的scripts文件夹 -可自定义
testpaths = ../scripts
# 搜索文件名----当前目录下的scripts文件夹下，以test_开头，以.py结尾的所有文件 -可自定义
python_files = test_*.py
# 搜索测试类名----当前目录下的scripts文件夹下，以Test_开头的类 -可自定义
python_classes = Test_*
# 搜索测试方法名----当前目录下的scripts文件夹下，以Test_开头的类内，以test_开头的方法 -可自定义
python_functions = test_*

5.结果断言

from utils.LogUtil import my_log
import json
#1、定义封装类
class AssertUtil:
#2、初始化数据，日志
    def __init__(self):
        self.log = my_log("AssertUtil")
#3、code相等
    def assert_code(self,code,expected_code):
        """
        验证返回状态码
        :param code:
        :param expected_code:
        :return:
        """
        try:
            assert int(code) == int(expected_code)
            return True
        except:
            self.log.error("code error,code is %s,expected_code is %s"%(code,expected_code))

            raise
#4、body相等
    def assert_body(self,body,expected_body):
        """
        验证返回结果内容相等
        :param body:
        :param expected_body:
        :return:
        """
        try :
            assert body == expected_body
            return True
        except:
            self.log.error("body error,body is %s,expected_body is %s"%(body,expected_body))
            raise
#5、body包含
    def assert_in_body(self,body,expected_body):
        """
        验证返回结果是否包含期望的结果
        :param body:
        :param expected_body:
        :return:
        """
        try:
            body = json.dumps(body)
            print(body)
            assert expected_body in body
            return True
        except:
            self.log.error("不包含或者body是错误，body is %s,expected_body is %s"%(body,expected_body))
            raise

from utils.LogUtil import my_log
import pymysql
#1、创建封装类
class Mysql:
#2、初始化数据，连接数据库，光标对象
    def __init__(self,host,user,password,database,charset="utf8",port=3306):
        self.log = my_log()
        self.conn = pymysql.connect(
            host=host,
            user=user,
            password=password,
            database=database,
            charset=charset,
            port=port
            )
        self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
#3、创建查询、执行方法
    def fetchone(self,sql):
        """
        单个查询
        :param sql:
        :return:
        """
        self.cursor.execute(sql)
        return self.cursor.fetchone()

    def fetchall(self,sql):
        """
        多个查询
        :param sql:
        :return:
        """
        self.cursor.execute(sql)
        return self.cursor.fetchall()

    def exec(self,sql):
        """
        执行
        :return:
        """
        try:
            if self.conn and self.cursor:
                self.cursor.execute(sql)
                self.conn.commit()
        except Exception as ex:
            self.conn.rollback()
            self.log.error("Mysql 执行失败")
            self.log.error(ex)
            return False
        return True

#4、关闭对象
    def __del__(self):
        #关闭光标对象
        if self.cursor is not None:
            self.cursor.close()
        #关闭连接对象
        if self.conn is not None:
            self.cursor.close()

if __name__ == "__main__":
    mysql = Mysql("211.103.136.242",
                  "test",
                  "test123456","meiduo",
                  charset="utf8",
                  port=7090)
    #res = mysql.fetchall("select username,password from tb_users")
    res = mysql.exec("update tb_users set first_name='python' where username = 'python'")
    print(res)
    #1、创建db_conf.yml, db1,db2
    #2、编写数据库基本信息
    #3、重构Conf.py
    #4、执行

"""
#1、导入pymysql包
import pymysql
#2、连接database
conn = pymysql.connect(
    host = "211.103.136.242",
    user = "test",
    password = "test123456",
    database = "meiduo",
    charset = "utf8",
    port =7090

)
#3、获取执行sql的光标对象
cursor = conn.cursor()
#4、执行sql
sql = "select username,password from tb_users"
cursor.execute(sql)
res = cursor.fetchone()
print(res)
#5、关闭对象
cursor.close()
conn.close()"""

6.数据驱动

Excel 读取文件

import os
import xlrd

#目的：参数化，pytest list
#自定义异常
class SheetTypeError:
    pass
#1、验证文件是否存在，存在读取，不存在报错
class ExcelReader:
    def __init__(self,excel_file,sheet_by):
        if os.path.exists(excel_file):
            self.excel_file = excel_file
            self.sheet_by = sheet_by
            self._data=list()
        else:
            raise  FileNotFoundError("文件不存在")
#2、读取sheet方式，名称，索引
    def data(self):
        #存在不读取，不存在读取
        if not self._data:
            workbook = xlrd.open_workbook(self.excel_file)
            if type(self.sheet_by) not in [str,int]:
                raise SheetTypeError("请输入Int or Str")
            elif type(self.sheet_by) == int:
                sheet = workbook.sheet_by_index(self.sheet_by)
            elif type(self.sheet_by) == str:
                sheet = workbook.sheet_by_name(self.sheet_by)
    #3、读取sheet内容
            #返回list，元素:字典
            #格式[{"a":"a1","b":"b1"},{"a":"a2","b":"b2"}]
            #1.获取首行的信息
            title = sheet.row_values(0)
            #2.遍历测试行，与首行组成dict，放在list
                #1 循环，过滤首行，从1开始
            for col in range(1,sheet.nrows):
                col_value = sheet.row_values(col)
                #2 与首组成字典，放list
                self._data.append(dict(zip(title, col_value)))

#4、结果返回
        return self._data

# head = ["a","b"]
# value1 = ["a1","b1"]
# value2 =  ["a2","b2"]
# data_list= list()
# #zip
# data_list.append(dict(zip(head,value1)))
# data_list.append(dict(zip(head,value2)))
# #print(dict(zip(head,value1)))
# #print(dict(zip(head,value2)))
# print(data_list)

if __name__ == "__main__":
    reader = ExcelReader("../data/testdata.xlsx","美多商城接口测试")
    print(reader.data())

发送邮件的封装

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import smtplib
#初始化
#smtp地址，用户名，密码，接收邮件者，邮件标题，邮件内容，邮件附件
class SendEmail:
    def __init__(self,smtp_addr,username,password,recv,
                 title,content=None,file=None):
        self.smtp_addr = smtp_addr
        self.username = username
        self.password = password
        self.recv = recv
        self.title = title
        self.content = content
        self.file = file
#发送邮件方法
    def send_mail(self):
        #MIME
        msg = MIMEMultipart()
        #初始化邮件信息
        msg.attach(MIMEText(self.content,_charset="utf-8"))
        msg["Subject"] = self.title
        msg["From"] = self.username
        msg["To"] = self.recv
        #邮件附件
        #判断是否附件
        if self.file:
        #MIMEText读取文件
            att = MIMEText(open(self.file).read())
        #设置内容类型
            att["Content-Type"] = 'application/octet-stream'
        #设置附件头
            att["Content-Disposition"] = 'attachment;filename="%s"'%self.file
        #将内容附加到邮件主体中
            msg.attach(att)
        #登录邮件服务器
        self.smtp = smtplib.SMTP(self.smtp_addr,port=25)
        self.smtp.login(self.username,self.password)
    #发送邮件
        self.smtp.sendmail(self.username,self.recv,msg.as_string())

if __name__ == "__main__":
    #初始化类(self,smtp_addr,username,password,recv,
            #     title,content=None,file=None):
    from config.Conf import ConfigYaml
    email_info = ConfigYaml().get_email_info()
    smtp_addr = email_info["smtpserver"]
    username = email_info["username"]
    password = email_info["password"]
    recv = email_info["receiver"]
    email = SendEmail(smtp_addr,username,password,recv,"测试")
    email.send_mail()
    #封装公共方法
    #应用测试发送

》

Python3+Pytest 接口自动化测试全方案设计与开发（部分笔记）

第一部分： 项目介绍及框架规划

接口测试框架流程

接口测试代码结构

第二部分：接口自动化框架编写

1. Requests 的封装

2.Yaml 文件的封装

3.日志文件的封装

4.pytest的应用

5.结果断言

6.数据驱动

发送邮件的封装

第一部分：项目介绍及框架规划