接口自动化实现及封装

St_rive

416人浏览 · 2026-08-17 09:30:53

St_rive · 2026-08-17 09:30:53 发布

接口自动化实现及封装

一个简单的接口自动化场景：

用户登录商城-对id=4的商品进行支付购买-生成订单-查询该订单

# order.py
import requests
from requests.packages import urllib3

urllib3.disable_warnings()
# 购买商品生成订单

login_api = 'https://mall.jctesting.cn/api/account/login'
data_1 = {"account":"15200000001","password":"111111.","client":5}
res_1 = requests.post(url=login_api,json=data_1,verify=False)
login_res = res_1.json()
token = login_res['data']['token']
print(f"login_api:{login_res}")

header = {"token":token}

orderBuy_api = 'https://mall.jctesting.cn/api/order/buy'
data_2 = {"action":"submit","goods":[{"item_id":4,"num":1}],"use_integral":0,"address_id":"","remark":""}
res_2 = requests.post(url=orderBuy_api,json=data_2,headers=header,verify=False)
orderBuy_res = res_2.json()
order_id = orderBuy_res['data']['order_id']

pcPrepay_api = 'https://mall.jctesting.cn/api/payment/pcPrepay'
data_3 = {"order_id":order_id,"pay_way":3,"order_source":5}
res_3 = requests.post(url=pcPrepay_api, json=data_3, headers=header, verify=False)
pcPrepay_res = res_3.json()
print(pcPrepay_res)

orderList_api = 'https://mall.jctesting.cn/api/order/lists'
orderIdList = list()
res_4 = requests.get(url=orderList_api, headers=header, verify=False)
orderList_res = res_4.json()
print(orderList_res)
for order in orderList_res['data']['list']:
    orderIdList.append(order['id'])
    print(orderIdList)

把上述代码按【登录获取token】-【支付下单】-【查询订单】功能进行函数封装

# order.py
import requests
from requests.packages import urllib3

urllib3.disable_warnings()
# 购买商品生成订单

def get_token():
	login_api = 'https://mall.jctesting.cn/api/account/login'
	data_1 = {"account": "15200000001", "password": "111111.", "client": 5}
	res_1 = requests.post(url=login_api, json=data_1, verify=False)
	login_res = res_1.json()
	token = login_res['data']['token']
	print(f"login_api:{login_res}")
	return token

def payOrder(goodsId=4):
	token = get_token()
	orderBuy_api = 'https://mall.jctesting.cn/api/order/buy'
	header = {"token": token}
	data_2 = {"action":"submit","goods":[{"item_id":goodsId,"num":1}],"use_integral":0,"address_id":"","remark":""}
	res_2 = requests.post(url=orderBuy_api,json=data_2,headers=header,verify=False)
	orderBuy_res = res_2.json()
	order_id = orderBuy_res['data']['order_id']

	pcPrepay_api = 'https://mall.jctesting.cn/api/payment/pcPrepay'
	data_3 = {"order_id":order_id,"pay_way":3,"order_source":5}
	res_3 = requests.post(url=pcPrepay_api, json=data_3, headers=header, verify=False)
	pcPrepay_res = res_3.json()
	print(pcPrepay_res)
	return pcPrepay_res

def get_orderIdList():
	token = get_token()
	orderList_api = 'https://mall.jctesting.cn/api/order/lists'
	header = {"token": token}
	orderIdList = list()
	res_4 = requests.get(url=orderList_api, headers=header, verify=False)
	orderList_res = res_4.json()
	print(orderList_res)
	for order in orderList_res['data']['list']:
		orderIdList.append(order['id'])
	print(orderIdList)
	return orderIdList


if __name__ == '__main__':
	order = payOrder(4)
	orderIdList = get_orderIdList()

新建一个test_开头的py模块，编写测试用例（支付完成后的订单id是否在订单列表中）

# test_payOrder.py
import pytest

from pay_order import *

def test_pay_order():
	pay_order = payOrder(goodsId=4)
	order_id = pay_order['data']['order_id']
	order_idList = get_orderIdList()
	pytest.assume(order_id in order_idList)


if __name__ == '__main__':
	pytest.main(['-vs',test_pay_order()])

在这里，上述代码完成了一个最初始形态的接口场景式自动化，完成了用户进行通过余额进行下单支付生成订单的场景；

继续对order.py进行，封装，把该模块下所有关于订单的业务操作都规为一个类 class Order

# order.py
import requests
from requests.packages import urllib3

urllib3.disable_warnings()
class Order:
	def payOrder(self,goodsId=4):
		token = get_token()
		orderBuy_api = 'https://mall.jctesting.cn/api/order/buy'
		header = {"token": token}
		data_2 = {"action":"submit","goods":[{"item_id":goodsId,"num":1}],"use_integral":0,"address_id":"","remark":""}
		res_2 = requests.post(url=orderBuy_api,json=data_2,headers=header,verify=False)
		orderBuy_res = res_2.json()
		order_id = orderBuy_res['data']['order_id']

		pcPrepay_api = 'https://mall.jctesting.cn/api/payment/pcPrepay'
		data_3 = {"order_id":order_id,"pay_way":3,"order_source":5}
		res_3 = requests.post(url=pcPrepay_api, json=data_3, headers=header, verify=False)
		pcPrepay_res = res_3.json()
		print(pcPrepay_res)
		return pcPrepay_res


	def get_orderIdList(self):
		token = get_token()
		orderList_api = 'https://mall.jctesting.cn/api/order/lists'
		header = {"token": token}
		orderIdList = list()
		res_4 = requests.get(url=orderList_api, headers=header, verify=False)
		orderList_res = res_4.json()
		print(orderList_res)
		for order in orderList_res['data']['list']:
			orderIdList.append(order['id'])
		print(orderIdList)
		return orderIdList


if __name__ == '__main__':
	order = Order().payOrder(4)
	orderIdList = Order().get_orderIdList()

把 get_token( ) 放入 base_op.py中规为BaseOP类，BaseOP 类中主要存放所有业务模块类中需要重复用到的方法，BaseOP类可作为父类被其业务模块类继承

#base_op.py
import requests

class BaseOP:

	def get_token(self):
		login_api = 'https://mall.jctesting.cn/api/account/login'
		data_1 = {"account": "15200000001", "password": "111111.", "client": 5}
		res_1 = requests.post(url=login_api, json=data_1, verify=False)
		login_res = res_1.json()
		token = login_res['data']['token']
		print(f"login_api:{login_res}")
		return token

#order.py
import requests
from requests.packages import urllib3

from base_op import BaseOP

urllib3.disable_warnings()
# 购买商品生成订单
class Order(BaseOP):
	def payOrder(self,goodsId=4):
		token = self.get_token()
		orderBuy_api = 'https://mall.jctesting.cn/api/order/buy'
		header = {"token": token}
		data_2 = {"action":"submit","goods":[{"item_id":goodsId,"num":1}],"use_integral":0,"address_id":"","remark":""}
		res_2 = requests.post(url=orderBuy_api,json=data_2,headers=header,verify=False)
		orderBuy_res = res_2.json()
		order_id = orderBuy_res['data']['order_id']

		pcPrepay_api = 'https://mall.jctesting.cn/api/payment/pcPrepay'
		data_3 = {"order_id":order_id,"pay_way":3,"order_source":5}
		res_3 = requests.post(url=pcPrepay_api, json=data_3, headers=header, verify=False)
		pcPrepay_res = res_3.json()
		print(pcPrepay_res)
		return pcPrepay_res

	def get_orderIdList(self):
		token = self.get_token()
		orderList_api = 'https://mall.jctesting.cn/api/order/lists'
		header = {"token": token}
		orderIdList = list()
		res_4 = requests.get(url=orderList_api, headers=header, verify=False)
		orderList_res = res_4.json()
		print(orderList_res)
		for order in orderList_res['data']['list']:
			orderIdList.append(order['id'])
		print(orderIdList)
		return orderIdList


if __name__ == '__main__':
	order = Order().payOrder(4)
	orderIdList = Order().get_orderIdList()

order.py模块中 Order类通过继承BaseOP，通过self.get_token（）直接调用BaseOP类中的get_token方法获取token参数；

接下来把order.py模块中的接口请求相关代码独立出来，放入py_page文件夹中，在py_page中新建一个base_api.py模块存放接口请求的封装

#py_page/base_api.py
import os
import traceback
import requests
import yaml
from requests.packages import urllib3
from string import Template


from common.log import Logger

logging = Logger(__name__).get_logger()
class BaseApi:
    urllib3.disable_warnings()
    def request_api(self,method, api, params=None, data=None, json=None, headers=None, res_type='json', verify=False)->dict:
        '''
        基于requests的接口请求封装
        :param method: 接口请求方式get、post
        :param api: 接口请求地址
        :param params: 用于get请求中传入请求数据，接收一个字典类型
        :param data: 用于post请求中传入表单数据，接收一个字典类型
        :param json: post请求中用于传入json格式的请求数据，接收一个字典类型
        :param headers: 用于构造请求头信息，接收一个字典类型
        :param res_type: 指定响应报文类型json、text
        :param verify: 接收布尔值，默认False，即不对https请求SSL验证
        :return: 接口响应状态码，以及接口响应报文，返回一个字典类型
        '''
        host = "https://mall.jctesting.cn"
        api_url = host+api
        res_dict = dict()
        if method == 'get':
            try:
                res = requests.get(url=api_url, params=params, headers=headers, verify=verify)
                if res_type == 'json':
                    res_content = res.json()
                else:
                    res_content = res.text
                res_dict['status_code'] = res.status_code
                res_dict['content'] = res_content
                return res_dict
            except Exception as e:
                error_info = traceback.format_exc()
                logging.error(f"接口::{api}::请求出错::{e},错误详情:{error_info}")

        elif method == 'post':
            try:
                res = requests.post(url=api_url, data=data, json=json, headers=headers, verify=verify)
                if res_type == 'json':
                    res_content = res.json()
                else:
                    res_content = res.text
                res_dict['status_code'] = res.status_code
                res_dict['content'] = res_content
                return res_dict
            except Exception as e:
                error_info = traceback.format_exc()
                logging.error(f"接口::{api}::请求出错::{e},错误详情:{error_info}")


    def run_requests(self,yaml_path,api_name,**kwargs):
        '''
        用于读取yaml_api中的接口数据，实现接口请求
        :param yaml_path: yaml文件路径
        :param api_name: yaml文件中的接口名称
        :param kwargs: 发起接口请求的自定义参数，即实现从外部传入参数到请求体中
        :return: 接口响应状态码，以及接口响应报文，返回一个字典类型
        '''
        try:
            with open(yaml_path,"r",encoding="UTF-8") as f:
                yaml_datas = yaml.safe_load(f)  # 获取整个yaml全部数据，返回一个字典对象
                dic_api_datas = yaml_datas[f'{api_name}'] # 根据接口名字获取该接口的请求数据，返回一个字典对象
                yaml_api_datas = yaml.dump(dic_api_datas) # 把接口的请求数据转换成yaml格式，返回一个字符串对象
                for k in kwargs.keys(): # 对传入参数数据类型为字符串的进行处理，防止模板匹配与yaml转换被转换成非字符串类型
                    if isinstance(kwargs[k], str):
                        kwargs[k] = f"'{kwargs[k]}'"
                temp_api_datas = Template(yaml_api_datas).substitute(kwargs) # 通过Template模板，把字典对象kwargs作为参数传入yaml_api_datas中
                api_datas = yaml.safe_load(temp_api_datas) # 把匹配到参数的yaml格式字符串对象再次通过yaml加载的方式转换成一个字典对象
                print(f"api_datas:{api_datas}")
                logging.info(f"读取YAML文件::{yaml_path}::请求的接口名称::{api_name}::传入接口参数::{kwargs}::接口请求参数api_datas::{api_datas}")
                request = self.request_api(method=api_datas['method'],api=api_datas['api'],headers=api_datas['headers'],
                                           params=api_datas['params'],data=api_datas['data'],json=api_datas['json'])
                logging.info(f"接口::{api_datas['api']}::响应数据为::{request}")
                return request
        except Exception as e:
            error_info = traceback.format_exc()
            logging.error(f"读取YAML文件:{yaml_path}，并发起接口请求过程中出错:{e}::错误详情:{error_info}")

    @classmethod
    def get_path(cls,relative_path):
        '''
        文件的路径拼接，返回一个绝对路径
        :param relative_path: 文件的相对路径
        :return: 文件的绝对路径
        '''
        file_path = os.path.abspath(__file__)
        dir_path = os.path.split(file_path)[0]
        far_path = os.path.dirname(dir_path)
        path = far_path + relative_path
        return path

把接口请求数据独立出来，通过yaml_api文件夹独立存放，通过BaseApi中的run_requests，读取yaml文件获取接口数据，传入request_api( )中完成接口请求

#yaml_api/confirmOrder_page.yaml
confirmOrder_buy: # 确认订单页面提交订单接口
  api: /api/order/buy
  method: post
  headers: {"token":$token}
  params: ~
  data: ~
  json: {"action":"submit","goods":[{"item_id":$goodsId,"num":1}],"use_integral":0,"address_id":"","remark":""}

 payment_pcPrepay: # 确认订单支付接口(pay_way:3 余额支付）
  api: /api/payment/pcPrepay
  method: post
  headers: { "token": $token }
  params: ~
  data: ~
  json: {"order_id":$order_id,"pay_way":$pay_way,"order_source":5}

YAML文件中的$为变量占位符，在run_requests( )方法中首先读取YAML中的数据，然后把传入的参数通过Template方法完成对$变量的匹配成有效的请求参数，传入到request_api方法中发起接口请求；这样就形成了接口数据与请求代码相互分离的基础请求封装；

在py_page文件夹内，继续按页面新建py模块，每一个py模块就是一个页面类，这个类中包含该页面的所有可请求的接口，每个接口封装成独立的方法，这样就形成了页面接口层，每个类都继承BaseApi，实现对run_requests方法的调用完成接口的请求封装；

#py_page/confirmOrder_page.py
from py_page.base_api import BaseApi
from py_page.registerLogin_page import RegisterLoginPageApi


class ConfirmOrderPageApi(BaseApi):

	path = BaseApi.get_path('/yaml_api/confirmOrder_page.yaml')
	token = RegisterLoginPageApi().get_token()

	# 新增收货人信息
	def add_address(self,name,phone,address):
		add_address_res = self.run_requests(self.path,'addAddress', token=self.token,name=name,phone=phone,address=address)
		return add_address_res

	# 确认并提交订单
	def confirmOrder_buy(self,itemId):
		confirmOrder_buy_res = self.run_requests(self.path, 'confirmOrder_buy', token=self.token, itemId=itemId)
		return confirmOrder_buy_res




if __name__ == '__main__':
	pass

现在修改base_op.py与order.py，并把这两个模块放入operation文件夹中形成业务逻辑层，通过调用py_page中的页面接口方法实现对应业务逻辑，实现业务与接口的分层方便接口复用；

因为需要调用登录接口实现用户登录因此，在修改base_op.py前按照PO分层思想，商城登录是一个单独的页面，因此在yaml_api中新建一个registerlogin_page.yaml存放登录和注册接口信息；

#yaml_api/registerlogin_page.yaml
login:
  api: /api/account/login
  method: post
  headers: ~
  params: ~
  data: ~
  json: {"account": $account,"password": $password,"client": 5}

register:
  api: /api/account/register
  method: post
  headers: ~
  params: ~
  data: ~
  json: {"mobile":$mobile,"password":$password,"code":"","client":5}

在py_page中新建一个registerLogin_page.py注册登录页面接口调用模块

#py_page/registerLogin_page.py
from py_page.base_api import BaseApi

class RegisterLoginPageApi(BaseApi):

	def __init__(self,path="./yaml_api/registerlogin_page.yaml",token=None):
		self.path = path
		self.token = token

	def login_api(self,account,password):
		login = self.run_requests(self.path,'login', account=account,password=password)
		return login

	def register_api(self,mobile,password):
		register = self.run_requests(self.path,'register', mobile=mobile,password=password)
		return register

在operation业务层新建registerLogin.py实现登录注册功能

#operation/registerLogin.py
from py_page.registerLogin_page import RegisterLoginPageApi

class RegisterLogin:

	def login(self,account,password):
		login_res = RegisterLoginPageApi().login_api(account,password)
		return login_res


	def register(self,mobile,password):
		register_res = RegisterLoginPageApi().register_api(mobile,password)
		return register_res

小结一下从接口数据-业务实现：yaml_api -> py_page -> operation 即对应接口数据层，页面接口层，业务逻辑层；

因此在base_op.py实现登录BaseOP修改如下

#operation/base_op.py
from operation.registerLogin import RegisterLogin

class BaseOP:

	def get_token(self):
		login = RegisterLogin().login(account='15200000001', password='111111.')
		token = login['content']['data']['token']
		return token

同样在order.py中因为需要调用订单列表接口实现获取订单，按照分层思想 yaml_api -> py_page -> operation构建数据与方法

在yaml_api中新建orderList_page.yaml

#yaml_api/orderList_page.yaml
order_list: # 订单列表
  api: /api/order/lists
  method: get
  headers: {"token":$token}
  params: ~
  data: ~
  json: ~

在py_page中新建orderList_page.py

#py_page/orderList_page.py
from py_page.base_api import BaseApi

class OrderListApi(BaseApi):

	def __init__(self,path="./yaml_api/orderList_page.yaml",token=None):
		self.path = path
		self.token = token

	def orderList_api(self):
		orderList = self.run_requests(self.path,'order_list', token=self.token)
		return orderList

在operation中order.py实现业务实现如下

#operation/order.py
from requests.packages import urllib3

from operation.base_op import BaseOP
from py_page.confirmOrder_page import ConfirmOrderPageApi
from py_page.orderList_page import OrderListApi

urllib3.disable_warnings()
# 购买商品生成订单
class Order(BaseOP):

	def __init__(self):
		self.token = self.get_token()

	def payOrder(self,goodsId=4,pay_way=3):
		# 提交订单
		res_1 = ConfirmOrderPageApi(token=self.token).confirmOrder_buy_api(goodsId)
		order_id = res_1['content']['data']['order_id']
		# 选择支付方式进行支付
		ConfirmOrderPageApi(token=self.token).payment_pcPrepay_api(order_id=order_id,pay_way=pay_way)
		# 支付完成返回订单详情
		res_2 = ConfirmOrderPageApi(token=self.token).confirmOrder_detail_api(id=order_id)
		return res_2


	def get_orderIdList(self):
		id_list = list()
		res = OrderListApi(token=self.token).orderList_api()
		data_list = res['content']['data']['list']
		for order in data_list:
			id_list.append(order['id'])
		return id_list

到此，场景：用户登录商城-对id=4的商品进行支付购买-生成订单-查询该订单已经按PO设计模式形成了代码分层；

接下来基于该分层基础上，直接对operation（业务功能）调用实现测试用例编写；

新建test文件夹，在该文件夹下新建以test_开头的测试模块，在模块中按类编写测试用例

#test/test_payOrder.py
import pytest

from operation.order import Order

class TestOrder:
	def test_pay_order(self):
		pay_order = Order().payOrder(goodsId=4,pay_way=3)
		order_id = pay_order['data']['order_id']
		order_idList = Order().get_orderIdList()
		pytest.assume(order_id in order_idList)

目录结构

为框架增加日志

在项目根目录增加log.py模块

import logging
import time
import os

class Logger:
    def __init__(self,name,CmdLevel=logging.INFO, FileLevel=logging.INFO):
        # 创建一个 logger
        self.logger = logging.getLogger(name)
        # 设置Log等级总开关
        self.logger.setLevel(CmdLevel)
        # 创建 handler，用于日志写入
        currTime = time.strftime("%Y-%m-%d")
        log_path = os.path.abspath(os.path.join(os.getcwd(), "")) + '/Logs/'
        log_name = log_path + currTime + '.log'
        fh = logging.FileHandler(log_name, mode='a')  # mode = 'a' 为在原日志上追加，'w'为覆盖 输出日志到文件
        fh.setLevel(FileLevel) # 输出到 file 的日志等级
        ch = logging.StreamHandler() # 输出日志到控制台
        ch.setLevel(FileLevel) # 输出到控制台的日志等级
        # 定义日志handler的输出格式
        fmt = logging.Formatter('%(asctime)s - %(filename)s:[%(lineno)s] - [%(levelname)s] - %(message)s')
        fh.setFormatter(fmt)
        ch.setFormatter(fmt)
        # 添加 logger 到 handler里面
        self.logger.addHandler(fh)
        self.logger.addHandler(ch)

在base_api.py模块中引入日志，把每个接口请求及响应都以日志info形式记录到项目根目录Logs文件夹（注：在项目根目录下创建Logs文件夹）

import yaml
from requests.packages import urllib3
import json
from string import Template

from log import Logger

logging = Logger(__name__)
class BaseApi:
    urllib3.disable_warnings()
    def request_api(self,method=None,api=None,params=None,data=None,json=None,headers=None,res_type='json',verify=False,**kwargs)->dict:
        host = "https://mall.jctesting.cn"
        api_url = host+api
        res_dict = dict()
        if method == 'get':
            try:
                res = requests.get(url=api_url,params=params,headers=headers,verify=verify)
                if res_type == 'json':
                    res_content = res.json()
                else:
                    res_content = res.text
                res_dict['status_code'] = res.status_code
                res_dict['content'] = res_content
                return res_dict
            except Exception as e:
                logging.logger.error(f"接口::{api}::请求出错::{e}")

        elif method == 'post':
            try:
                res = requests.post(url=api_url,data=data,json=json,headers=headers,verify=verify)
                if res_type == 'json':
                    res_content = res.json()
                else:
                    res_content = res.text
                res_dict['status_code'] = res.status_code
                res_dict['content'] = res_content
                return res_dict
            except Exception as e:
                logging.logger.error(f"接口::{api}::请求出错::{e}")


    def run_requests(self,yaml_path,api_name,**kwargs):
        try:
            with open(yaml_path,"r",encoding="UTF-8") as f:
                yaml_datas = yaml.safe_load(f)
                datas = yaml_datas[f'{api_name}']
                datas_str = json.dumps(datas)
                datas_str_var = Template(datas_str).substitute(kwargs)
                api_datas = json.loads(datas_str_var)
                logging.logger.info(f"读取YAML文件::{yaml_path}::请求的接口名称::{api_name}::传入接口参数::{kwargs}::接口请求参数api_datas::{api_datas}")
                request = self.request_api(method=api_datas['method'],api=api_datas['api'],headers=api_datas['headers'],
                                           params=api_datas['params'],data=api_datas['data'],json=api_datas['json'])
                logging.logger.info(f"接口::{api_datas['api']}::响应数据为::{request}")
                return request
        except Exception as e:
            logging.logger.error(f"读取YAML文件::{yaml_path}出错::{e}")

在测试用例中引用allure和log

import pytest
import allure
from log import Logger

from operation.order import Order

logging = Logger(__name__)
@allure.feature("测试订单逻辑")
class TestOrder:
	@allure.story("测试生成订单")
	@allure.title("用余额支付生成订单")
	def test_pay_order(self,goodsId = 4,pay_way=3):
		with allure.step(f"对商品：{goodsId}进行下单支付"):
			pay_order = Order().payOrder(goodsId=goodsId,pay_way=pay_way)
		with allure.step("获取订单ID"):
			order_id = pay_order['content']['data']['id']
		order_idList = Order().get_orderIdList()
		pytest.assume(order_id in order_idList)


if __name__ == '__main__':
	pytest.main(['-vs','test_apyOrder.py::test_pay_order()'])

在测试框架中引入邮件模块

在项目根目录下新建一个common文件夹用于存放send_mail.py邮件模块

#common/send_mail.py
import smtplib
from email.header import Header
from email.mime.application import MIMEApplication
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

class SendEmail:

    # 初始化服务器信息
    def __init__(self,mail_host, mail_user, mail_pass, sender, receives):
        self.mail_host = mail_host
        self.mail_user = mail_user
        self.mail_pass = mail_pass
        self.sender = sender
        self.receivers = receives

    # 以文本的形式发送邮件
    def make_email_by_text(self,context, subject, from_address, to_address):
        message = MIMEText(context, 'plain', 'UTF-8')
        message['Subject'] = subject  # 邮件标题
        message['From'] = Header(from_address, "utf-8")  # 邮件主体中发送者名称
        message['To'] = Header(to_address, "utf-8")  # 邮件主体中接收者名称
        self.send_email(message)

    # 以文本和附件的形式发送邮件
    def make_email_by_att(self,content,file_path,subject, from_address, to_address):
        message = MIMEMultipart()
        message['Subject'] = subject  # 邮件标题
        message['From'] = Header(from_address, "utf-8")  # 邮件主体中发送者名称
        message['To'] = Header(to_address, "utf-8")  # 邮件主体中接收者名称
        body = MIMEText(content,'plain','utf-8')
        message.attach(body)
        att = MIMEApplication(open(file_path, 'rb').read())
        att.add_header('Content-Disposition','attachment',filename='allure测试报告.zip')
        message.attach(att)
        self.send_email(message)

    # 登录并进行发送
    def send_email(self,message):
        # 进行登录发送
        try:
            smtpobj = smtplib.SMTP_SSL(self.mail_host)
            smtpobj.ehlo(self.mail_host)
            smtpobj.login(self.mail_user,self.mail_pass)
            smtpobj.sendmail(self.sender,self.receivers,message.as_string())
            smtpobj.quit()
            print('success')
        except Exception as e:
            print(f'error: {e}')
            raise e

由于allure测试报告是一个包含html, js, css 等多种文件的文件夹，要实现测试执行后把测试报告通过邮件打包发送，这里需要在common文件夹下封装一个zip_file.py模块实现对文件夹的封装

import os
import zipfile
from log import Logger

logging = Logger(__name__)
def zip_files(dir_path,zip_path):
	'''
	:param dir_path: 需要压缩的文件夹地址  例：'D:\Buyer_test_code\apipotest_jcmall\report'
	:param zip_path: 压缩后的文件存在地址+压缩文件名  例：'D:\Buyer_test_code\apipotest_jcmall\report_zip\report.zip'
	:return: 0  压缩失败  1 压缩成功
	'''
	try:
		with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as f:
			for dirpath, dirnames, filenames in os.walk(dir_path):
				fpath = dirpath.replace(dir_path, '')
				fpath = fpath and fpath + os.sep or ''
				for filename in filenames:
					f.write(os.path.join(dirpath, filename), fpath+filename)
			logging.logger.info(f"文件：{dir_path} 压缩成功")
	except Exception as e:
		logging.logger.error(f"文件：{dir_path} 压缩失败，错误：{e}")
		return 0
	else:
		return 1

在项目根目录下新建一个程序执行模块run.py，这个模块的主要功能实现：

1、执行命令：python -m pytest ./test/ -s -q --alluredir=./result/ 执行测试用例生成allure测试结果；

2、在allure测试结果基础上生成测试报告：allure generate ./result/ -o ./report/ --clean；

3、调用zip_file.py模块对allure测试报告进行压缩打包，生成 report.zip；

4、进行邮件发送，把测试结果压缩包作为附件进行打包发送；

import os
import yaml
from common.send_mail import SendEmail
from log import Logger
from common.zip_file import zip_files

logging = Logger(__name__)
def run_testCase():
	test_allure = "python -m pytest ./test/ -s -q --alluredir=./result/"
	logging.logger.info(f"run_testCase::正在执行测试命令：{test_allure}")
	try:
		os.system(test_allure)
	except Exception as e:
		logging.logger.error(f"os执行命令{test_allure}错误：{e}")
	else:
		make_report = "allure generate ./result/ -o ./report/ --clean"
		try:
			logging.logger.info(f"run_testCase::测试执行完毕，正在执行生成报告命令：{make_report}")
			os.system(make_report)
		except Exception as e:
			logging.logger.error(f"os执行命令{make_report}错误：{e}")
		else:
			dir_path = r'./report'
			zip_path = r'./report_zip/report.zip'
			zip_report = zip_files(dir_path,zip_path)
			if zip_report:
				file_name = r'./config.yaml'
				file_path = zip_path
				content = "测试执行结束，Allure测试报告在邮件附件中，请使用Pycharm打开查看。"
				Subject = '接口场景式自动化测试报告'  # 邮件标题
				From = "Jie的QQ邮箱"  # 邮件主体中发送者名称
				To = "Jie的163邮箱"  # 邮件主体中接收者名称
				try:
					with open (file_name,"r",encoding="UTF-8") as f:
						config = yaml.safe_load(f)
						config_email = config["Email"]
					send = SendEmail(mail_host=config_email['mail_host'], mail_user=config_email['mail_user'], mail_pass=config_email['mail_pass'],
								 sender=config_email['sender'], receives=config_email['receives'])
					send.make_email_by_att(content,file_path, Subject, From, To)
					logging.logger.info(f"邮件发送成功，邮件标题：{Subject}")
				except Exception as e:
					logging.logger.error(f"读取配置文件::{file_name}::失败::{e}")
					raise e

run_testCase()

到此一个接口自动化测试框架基本完成

完整目录结构如下