[开发测试]python3+requests+unittest接口测试框架

今天来分享一套比较简单的接口测试框架，使用python3 + requests + unittest，使用HTMLTestRunner来生成测试报告。

创建base目录，准备几个基础的工具类，本次被测项目使用比较多的请求方式就是POST和GET，其他请求方式暂不封装

首先在base目录中创建一个日志类，用来记录接口请求信息和测试过程

base_log.py

# coding=utf-8
import logging
import os
import settings

class BaseLogger(object):

    def __init__(self,name):
        """
        初始化logger
        :param name:
        """
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)  # Log等级总开关

    def get_logger(self):
        """
        自定义logger
        :return:
        """
        # 定义handler的输出格式
        formatter = logging.Formatter(settings.LOG_FORMATTER)

        # 创建一个handler，用于写入日志文件
        if settings.ENV == 'test':
            logfile = os.path.join(os.getcwd(),settings.LOG_FILE_NAME)
            file_handler = logging.FileHandler(logfile, mode='w',encoding='utf-8')
            file_handler.setLevel(logging.DEBUG)  # 输出到file的log等级的开关
            file_handler.setFormatter(formatter)
            self.logger.addHandler(file_handler)
            # 创建一个handler，用于输出到控制台
            console_handler = logging.StreamHandler()
            console_handler.setLevel(logging.DEBUG)  # 输出到console的log等级的开关
            console_handler.setFormatter(formatter)
            self.logger.addHandler(console_handler)
        else:
            # 创建一个handler，用于输出到控制台
            console_handler = logging.StreamHandler()
            console_handler.setLevel(logging.DEBUG)  # 输出到console的log等级的开关
            console_handler.setFormatter(formatter)
            self.logger.addHandler(console_handler)
        return self.logger

创建settings.py配置文件，将项目中配置相关的提取到配置文件中，方便后续维护管理

配置文件中日志相关配置:

# -*- coding:utf-8 -*-
import os,time

ENV = 'test'

# 日志配置
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
now_time = time.strftime("%Y_%m_%d_%H_%M_%S")
LOG_DIR_PATH = os.path.join(BASE_DIR,'log')
if not os.path.exists(LOG_DIR_PATH):
    os.makedirs(LOG_DIR_PATH)
LOG_FILE_NAME = '{0}/{1}.log'.format(LOG_DIR_PATH,now_time)
LOG_FORMATTER = "%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s"

然后继续在base目录下创建接口测试基础类

base_api.py

# -*- coding:utf-8 -*-
from base.base_log import BaseLogger
import json, requests
import settings

logger = BaseLogger(__name__).get_logger()


class BaseApi(object):
    """
    接口测试基础类
    """
    url = ''  # 接口相对地址
    base_url = settings.API_BASE_URL  # 定义接口域名

    def __init__(self):
        self.response = None
        self.headers = settings.HEADERS  # 接口请求头

    def api_url(self):
        """
        url拼接，将接口相对地址与域名拼接
        """
        url = '{0}{1}'.format(self.base_url, self.url)
        return url

    def build_base_data(self):
        """
        接口公共参数
        """
        return {
            'version': '1.1.2', 
            'source': '1', 
            'deviceName': 'TestDevice'
        }

    def build_custom_data(self, data):
        """
        接口除了公共参数之外的其他参数，该方法用到时用来重写
        """
        return {}

    def format_data(self, data):
        """
        格式化请求参数，将公共参数与非公共参数合并
        """
        if not data:
            data = {}
        base_params = self.build_base_data()
        custom_params = self.build_custom_data(data)
        data.update(base_params)
        data.update(custom_params)
        return data

    def get(self, data=None):
        """
        请求方式：Get
        """
        logger.info('GET')
        self.response = requests.get(url=self.api_url(), data=json.dumps(self.format_data(data)), headers=self.headers)
        logger.info('request url: {0}'.format(self.api_url()))
        logger.info('request data: {0}'.format(self.format_data(data)))
        logger.info('response: {0}'.format(self.response.text))
        return self.response

    def post(self, data=None):
        """
        请求方式:Post
        """
        logger.info('POST')
        self.response = requests.post(url=self.api_url(), data=self.format_data(data), headers=self.headers)
        logger.info('request url: {0}'.format(self.api_url()))
        logger.info('request data: {0}'.format(self.format_data(data)))
        logger.info('response: {0}'.format(self.response.text))
        return self.response

    def get_status_code(self):
        """
        获取接口网络状态码,可用来判断接口连通性，例如：200，500，404
        """
        if self.response:
            return self.response.status_code

    def get_resp_code(self):
        """
        获取接口响应数据中的code值
        """
        if self.response:
            return json.loads(self.response.content)['code']

    def get_resp_message(self):
        """
        获取接口响应数据中的message值
        """
        if self.response:
            return json.loads(self.response.content)['message']

    def get_resp_data(self):
        """
        获取接口相应数据中的data值
        """
        if self.response:
            return json.loads(self.response.content)['data']

settings.py中加入被测接口域名API_BASE_URL和请求头HEADERS

# -*- coding:utf-8 -*-
import os,time

ENV = 'test'

API_BASE_URL = 'http://www.datv.com/v2'
HEADERS = {'content-type': 'application/json; charset=UTF-8'}

# 日志配置
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
now_time = time.strftime("%Y_%m_%d_%H_%M_%S")
LOG_DIR_PATH = os.path.join(BASE_DIR,'log')
if not os.path.exists(LOG_DIR_PATH):
    os.makedirs(LOG_DIR_PATH)
LOG_FILE_NAME = '{0}/{1}.log'.format(LOG_DIR_PATH,now_time)
LOG_FORMATTER = "%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s"

此时目录结构：

创建base_api目录用来定义被测接口的相对地址和入参

比如此时有两个接口，接口信息如下：

接口地址：/sysback/robotmgr/customerLabelGroup/createCustomerLabelGroup?

接口名称：创建标签组

接口请求方式：POST

接口参数（公共参数除外）：

customerLabelList 标签组列表，传入一个列表，例如：[{'labelName':'标签一'},{labelName':'标签二'}]?

groupName 标签组名称，传入字符串

在base_api目录中创建文件create_customer_label_group_api.py

# -*- coding:utf-8 -*-
from base.base_api import BaseApi


class CreateCustomerLabelGroupApi(BaseApi):
    """
    创建标签组接口
    """

    url = '/sysback/robotmgr/customerLabelGroup/createCustomerLabelGroup'

    def build_custom_data(self, data):
        return {
            "customerLabelList": data['customerLabelList'],
            "groupName": data['groupName']
        }

在该文件中，创建被测接口信息类，继承接口测试基础类，方便使用基础类中的response取值方法和请求方式，定义好接口相对地址与入参，此时的入参是除去公共参数之外的参数，重写基础类中的build_custom_data方法来实现与公共参数在接口请求前做拼接

在此处只是定义接口的基本信息，并没有对接口具体的请求参数赋值，我们希望在编写测试用例的时候再给参数赋值，所以此处请求参数的写法以字典的形式在data中取值，后面我们编写case的时候再给接口的data参数传值

接口相对地址：/sysback/robotmgr/customerLabelGroup/getAllCustomerLabelGroup

接口名称：查询标签组

接口请求方式：GET

接口参数（公共参数除外）：无

在base_api目录中创建文件get_customer_label_group_api.py

# -*- coding:utf-8 -*-
from base.base_api import BaseApi


class CustomerLabelGroupGetAllApi(BaseApi):
    """
    查询标签组接口
    """

    url = '/sysback/robotmgr/customerLabelGroup/getAllCustomerLabelGroup'

如果除了公共参数之外没有其他的参数，那么在这里就只定义接口地址就可以了

此时目录结构：

接下来就开始编写这个接口的测试用例

创建test_case目录用来存放测试用例文件

在test_case目录中创建测试文件test_customer_label_group_api.py

# -*- coding:utf-8 -*-
from base_api.create_customer_label_group_api import CreateCustomerLabelGroupApi
from base_api.get_customer_label_group_api import CustomerLabelGroupGetAllApi
from unittest import TestCase
import time


class TestCreateCustomerLabelGroupApi(TestCase):
    """
    测试创建客户标签组
    """

    def setUp(self):
        """
        测试用例前置准备
        """
        now_time = int(time.time())  # 获取当前时间戳
        self.label_name = 'test_label'
        self.group_name = 'test_' + str(now_time)  # 根据时间戳生成客户标签组名称

    def test_create_customer_label_group_success(self):
        """
        测试创建客户标签组成功
        """
        # 调用创建标签组接口
        create_label_group_api = CreateCustomerLabelGroupApi()
        create_label_group_api.post({'groupName': self.group_name, 'customerLabelList': [{'labelName': self.label_name}]})
        create_group_time = int(time.time())  # 记录请求创建接口时间，用户后续校验创建时间是否正确
        # 断言接口网络状态码
        self.assertEqual(create_label_group_api.get_status_code(), 200)
        # 断言接口响应code
        self.assertEqual(create_label_group_api.get_resp_code(), 200)
        # 断言响应中message值
        self.assertEqual(create_label_group_api.get_resp_message(), '创建成功')
        # 获取接口中响应数据中的data
        response_data = create_label_group_api.get_resp_data()
        label_group_uuid = response_data['uuid']
        # 断言响应中标签组的uuid长度为32位
        self.assertEqual(len(label_group_uuid), 32)

        # 调用查询客户标签组接口验证创建成功
        customer_group_list_api = CustomerLabelGroupGetAllApi()
        customer_group_list_api.get()
        self.assertEqual(customer_group_list_api.get_status_code(), 200)
        self.assertEqual(customer_group_list_api.get_resp_code(), 200)
        self.assertEqual(customer_group_list_api.get_resp_message(), 'success')
        # 获取列表中第一条数据进行断言，顺便用索引验证列表排序，新创建的标签组优先展示
        group_list = customer_group_list_api.get_resp_data()['result']
        # 断言客户标签组名称
        self.assertEqual(group_list[0]['groupName'], self.group_name)
        # 断言客户标签组uuid
        self.assertEqual(group_list[0]['uuid'], label_group_uuid)
        # 断言客户标签组中的标签数量
        self.assertEqual(len(group_list[0]['labels']), 1)
        # 断言客户标签组中的标签名称
        self.assertEqual(group_list[0]['labels'][0]['name'], self.label_name)
        # 断言客户标签组创建时间,考虑时间服务器不同，定义时间误差为3秒
        create_ope_time = group_list[0]['createOpeTime']
        time_array = time.strptime(create_ope_time, "%Y-%m-%d %H:%M:%S")
        time_stamp = int(time.mktime(time_array))
        self.assertLessEqual(time_stamp - create_group_time, 3)

    def test_create_customer_label_group_name_null(self):
        """
        测试创建客户标签组名称为空，接口返回message：客户标签名不能为空
        """
        create_label_group_api = CreateCustomerLabelGroupApi()
        create_label_group_api.post({'groupName': None, 'customerLabelList': [{'labelName': self.label_name}]})
        self.assertEqual(create_label_group_api.get_status_code(), 200)
        self.assertEqual(create_label_group_api.get_resp_code(), 500)
        self.assertEqual(create_label_group_api.get_resp_message(), '客户标签组名不能为空')
        self.assertIsNone(create_label_group_api.get_resp_data(),None)


    def tearDown(self):
        # 因setUp中使用秒级时间戳作为客户标签组名称，避免后续用例名称重复，每个用例之间等待1秒
        time.sleep(1)

?简单写了两个测试用例

第一个用例用来验证两个接口的基本功能，能够正常创建，创建完成之后，另外的查询接口能够正常查询，查询出来的数据与创建时传入的数据一致

第二个测试用例用来验证创建标签组接口groupName为空的时候，接口返回的message是否正确，也就是接口的异常情况处理，往往接口异常情况下的message会直接提示给用户

setup()：每个测试函数运行前运行；

teardown()：每个测试函数运行完后执行；

setUpClass()：必须使用@classmethod 装饰器,所有测试用例运行前运行一次；

tearDownClass()：必须使用@classmethod装饰器,所有测试用例运行完后运行一次

也可以对某些用例执行过程中跳过：unittest.skip()，具体使用方法可上网查询:

https://docs.python.org/3/library/unittest.html

此时的目录结构：

?测试用例写好之后，在base目录中创建base_runner.py来创建测试套件与生成测试报告

base_runner.py

# -*- coding:utf-8 -*-
from base.base_log import BaseLogger
import HTMLTestRunner
import unittest,os
import settings

logger = BaseLogger(__name__).get_logger()

class BaseRunner(object):

    def __init__(self,test_dir_path='./test_case'):
        """
        指定测试用例存放路径
        """
        self.test_dir_path = os.path.abspath(test_dir_path) 

    def create_suite(self):
        """
        创建测试套件，并且将用例添加到测试套件
        """
        test_unit = unittest.TestSuite()
        discover = unittest.defaultTestLoader.discover(start_dir=self.test_dir_path, pattern='test*.py')
        for test_suite in discover:
            for test_case in test_suite:
                test_unit.addTest(test_case)
        return test_unit

    def run_tests(self):
        """
        运行测试用例并生成测试报告
        """
        fp = open(settings.REPORT_FILE_NAME,'wb+')
        title = settings.REPORT_TITLE
        description = settings.REPORT_DESCRIPTION
        tester = settings.REPORT_TESTER
        runner = HTMLTestRunner.HTMLTestRunner(stream=fp, title=title, description=description, tester=tester, verbosity=2)
        runner.run(self.create_suite())
        fp.close()

此处使用HTMLTestRunner.py来生成测试报告

下载地址：http://tungwaiyip.info/software/HTMLTestRunner.html

测试报告相关的配置在settings中定义，并且自动创建result目录用来存放测试报告

# -*- coding:utf-8 -*-
import os,time

ENV = 'test'

API_BASE_URL = 'http://www.dwatv.com/v2'
HEADERS = {'content-type': 'application/json; charset=UTF-8'}

# 日志配置
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
now_time = time.strftime("%Y_%m_%d_%H_%M_%S")
LOG_DIR_PATH = os.path.join(BASE_DIR,'log')
if not os.path.exists(LOG_DIR_PATH):
    os.makedirs(LOG_DIR_PATH)
LOG_FILE_NAME = '{0}/{1}.log'.format(LOG_DIR_PATH,now_time)
LOG_FORMATTER = "%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s"

# 测试报告配置
REPORT_DIR_PATH = os.path.join(BASE_DIR,'result')
if not os.path.exists(REPORT_DIR_PATH):
    os.mkdir('./result')
REPORT_FILE_NAME = './result/' + time.strftime("%Y%m%d%H%M%S") + '_result.html'
REPORT_TITLE = '接口自动化测试报告'
REPORT_DESCRIPTION = '用例执行情况详情如下:'
REPORT_TESTER = '测试组'

执行完测试，生成测试报告之后，希望测试报告以邮件的形式发送给相关人员

在base目录下创建文件base_email.py

# -*- coding:utf-8 -*-
from base.base_log import BaseLogger
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
import smtplib, os, settings,time


logger = BaseLogger(__name__).get_logger()

class BaseMail(object):
    """
    发送测试报告邮件类
    """

    def __init__(self):
        self.report_dir_path = settings.REPORT_DIR_PATH # 存放测试报告的路径
        self.report_file = self.get_new_report()

    def get_new_report(self):
        """
        根据时间获取最新测试报告，返回最新测试报告的完整路径
        """
        logger.info('获取最新的测试报告...')
        lists = os.listdir(self.report_dir_path)
        lists.sort(key=lambda fn: os.path.getmtime(self.report_dir_path + '/' + fn))
        file_new = os.path.join(self.report_dir_path, lists[-1])
        logger.info('最新的测试报告完整路径为:{0}'.format(file_new))
        return file_new

    def send_mail(self):
        """
        发送测试报告附件邮件
        :return:
        """
        msg = MIMEMultipart()
        msg['Subject'] = settings.MAIL_HEADER
        msg['From'] = settings.MAIL_FROM
        msg['To'] = settings.MAIL_TO
        msg['Accept-Language'] = 'zh-CN'
        msg["Accept-Charset"] = "ISO-8859-1,utf-8"

        # 测试报告附件的描述
        pure_text = MIMEText('详细测试报告请见附件！',_charset='utf-8')
        msg.attach(pure_text)
        # HTML格式的附件
        html_application = MIMEApplication(open(self.report_file, 'rb').read())
        file_name = '自动化测试报告-{0}.html'.format(time.strftime('%Y-%m-%d')) # 以日期来命名测试报告
        html_application.add_header('Content-Disposition', 'attachment', filename=file_name)
        msg.attach(html_application)
        try:
            # 链接163邮箱服务器
            client = smtplib.SMTP()
            logger.info('链接邮箱服务器...')
            client.connect(settings.MAIL_SERVER)
            # 登录163邮箱
            logger.info('登录邮箱服务器...')
            client.login(settings.MAIL_FROM, settings.MAIL_FROM_PASSWORD)
            # 发送邮件
            client.sendmail(settings.MAIL_FROM, settings.MAIL_TO, msg.as_string())
            # 关闭链接
            client.quit()
            logger.info('邮件发送成功')
        except Exception as error:
            logger.error('邮件发送失败,原因:')
            logger.error(error)

邮箱相关的配置，在settings.py中定义

# -*- coding:utf-8 -*-
import os,time

ENV = 'test'

API_BASE_URL = 'http://www.dwatv.com/v2'
HEADERS = {'content-type': 'application/json; charset=UTF-8'}

# 日志配置
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
now_time = time.strftime("%Y_%m_%d_%H_%M_%S")
LOG_DIR_PATH = os.path.join(BASE_DIR,'log')
if not os.path.exists(LOG_DIR_PATH):
    os.makedirs(LOG_DIR_PATH)
LOG_FILE_NAME = '{0}/{1}.log'.format(LOG_DIR_PATH,now_time)
LOG_FORMATTER = "%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s"

# 测试报告配置
REPORT_DIR_PATH = os.path.join(BASE_DIR,'result')
if not os.path.exists(REPORT_DIR_PATH):
    os.mkdir('./result')
REPORT_FILE_NAME = './result/' + time.strftime("%Y%m%d%H%M%S") + '_result.html'
REPORT_TITLE = '接口自动化测试报告'
REPORT_DESCRIPTION = '用例执行情况详情如下:'
REPORT_TESTER = '测试组'

# 邮件配置
MAIL_SERVER = 'smtp.163.com'
MAIL_FROM = 'tester@163.com'
MAIL_FROM_PASSWORD = '12345***qwer'
MAIL_HEADER = '接口测试执行结果'
MAIL_TO = '225***778@qq.com'

最后在项目主目录下创建文件run_api_test.py

# -*- coding:utf-8 -*-
from base.base_runner import BaseRunner
from base.base_email import BaseMail


if __name__ == '__main__':
    """
    1、运行test_case目录下全部测试用例
    2、生成测试报告并且发送邮件
    """
    BaseRunner(test_dir_path='./test_case').run_tests()
    BaseMail().send_mail()

最终目录结构

写完测试用例之后只需要执行项目主目录下的run_api_test.py文件就可以运行test_case目录下的全部测试用例了，并且生成测试报告发送邮件

进入到项目主目录下，执行以下命令即可：

export PYTHONPATH=.
python3 run_api_test.py

最后，做一个简单的概括

?执行完成后，会自动创建一个result目录，用来存放HTML测试报告

后面抽时间继续给大家分享pytest + allure的一些使用心得，以及WEB端UI自动化测试与APP的UI自动化测试