python+requests接口自动化测试

思路和框架

正常的接口测试流程
确定测试接口的工具 —> 配置需要的接口参数 —> 进行测试 —> 检查测试结果（有的需要数据库辅助） —> 生成测试报告（html报告）
根据这样的过程来一步步搭建我们的框架。在这个过程中，我们需要做到业务和数据的分离，这样才能灵活，达到我们写框架的目的。
接下来，我们来进行结构的划分。

• common：存放一些共通的方法
• result：执行过程中生成的文件夹，里面存放每次测试的结果
• testCase：用于存放具体的测试case
• gitignore：Git忽略提交规则（不是必要）
• testFile：存放测试过程中用到的文件，包括上传的文件，测试用例以及数据库的sql语句
• caselist：txt文件，配置每次执行的case名称
• config：配置一些常量，例如数据库的相关信息，接口的相关信息等
• readConfig： 用于读取config配置文件中的内容
• runAll：用于执行case
• token.txt：本地换持有token

config

配置一些常量，例如数据库的相关信息，接口的相关信息等
config.ini内容如下

[EMAIL]
mail_host = smtp.qq.com
mail_user = xxx@qq.com
mail_pass = xxx
mail_port = 465
mail_replyto = xxx@ruguoapp.com
sender = xxx@qq.com
receiver = xxx@qq.com
subject = "测试报告"
content = "今天测试一下python是否可以发送邮件成功。"
user = xx
on_off = on

[HTTP]
scheme =https
base_url =xxx
timeout = 10.0

[DATABASE]
host = localhost
username = root
password = root
port = 3306
database = test

readConfig.py

用来读取config配置文件中的内容，定义的方法，根据名称取对应的值
内容如下

import os
import codecs
import configparser

ProDir = os.path.split(os.path.realpath(__file__))[0]  # 该文件的绝对路径
configPath = os.path.join(ProDir, "config.ini")


class ReadConfig:
    def __init__(self):
        fd = open(configPath)
        data = fd.read()

        #  remove BOM
        if data[:3] == codecs.BOM_UTF8:
            data = data[3:]
            file = codecs.open(configPath, "w")
            file.write(data)
            file.close()
        fd.close()

        self.cf = configparser.ConfigParser()
        self.cf.read(configPath)

    def get_email(self, name):
        value = self.cf.get("EMAIL", name)
        return value

    def get_http(self, name):
        value = self.cf.get("HTTP", name)
        return value
        
    def get_db(self, name):
            value = self.cf.get("DATABASE", name)
            return value

if __name__ == '__main__':  # 测试一下，我们读取配置文件的方法是否可用
    print('HTTP中的base_url值为：', ReadConfig().get_http('base_url'))

common

存放一些共通的方法

conmmontest：操作本地case
configEmail：发送邮件
configHttp：配置接口文件
getHeades：获取请求头
HTMLTestRunner：测试报告
Token：token
Log：输出的日志的所有操作，主要是对输出格式的规定，输出等级的定义以及其他一些输出的定义等等
url：接口列表

Log

对于这个log文件呢，我给它单独启用了一个线程，这样在整个运行过程中，我们在写log的时候也会比较方便，看名字大家也知道了，这里就是我们对输出的日志的所有操作了，主要是对输出格式的规定，输出等级的定义以及其他一些输出的定义等等。总之，你想对log做的任何事情，都可以放到这里来。

import os
import logging
import threading

localReadConfig = readConfig.ReadConfig()

class Log:
    logging.basicConfig()

    def __init__(self):
        global logPath, resultPath, path, logger_name
        path = readConfig.ProDir
        resultPath = os.path.join(path, "result")
        logger_name = 'logs'

        if not os.path.exists(resultPath):  # 判断该文件是否存在
            os.mkdir(resultPath)  # 创建目录

        logPath = os.path.join(resultPath, str(datetime.now().strftime("%Y%m%d%H%M%S")))
        if not os.path.exists(logPath):
            os.mkdir(logPath)

        self.logger = logging.getLogger(logger_name)  # 获得一个logger对象，默认是root
        self.logger.setLevel(logging.INFO)  # 设定INFO级别，所有等级大于等于INFO的信息都会输出

        # 定义handler
        handler = logging.FileHandler(os.path.join(logPath, "output.log"))  # 向文件output.log输出日志信息
        # 定义格式
        formatter = logging.Formatter(
            '[%(asctime)s]  <%(levelname)s>  %(message)s')  # 定义日志输出格式('%(asctime)s] %(levelname)s [%(funcName)s: %(filename)s, %(lineno)d] %(message)s')
        handler.setFormatter(formatter)  # 选择一个格式

        self.logger.addHandler(handler)  # 增加指定的handler

现在，我们创建了上面的Log类，在__init__初始化方法中，我们进行了log的相关初始化操作。具体的操作内容，注释已经写得很清楚了，这样，log的基本格式已经定义完成了，至于其他的方法，就靠大家自己发挥了，毕竟每个人的需求也不同，我们就只写普遍的共用方法啦。接下来，就是把它放进一个线程内了，请看下面的代码：

class MyLog:
    log = None
    mutex = threading.Lock()  # 多线程

    def __init__(self):
        pass

    @staticmethod
    def get_log():
        if MyLog.log is None:
            MyLog.mutex.acquire()  # 获取互斥锁后，进程只能在释放锁后下个进程才能进来
            MyLog.log = Log()
            MyLog.mutex.release()  # 互斥锁必须被释放掉

        return MyLog.log

configHttp.py

配置接口文件，主要内容如下

import requests
import readConfig as readConfig
from common import getHeaders
from common.Log import MyLog as Log


localReadConfig = readConfig.ReadConfig()


class ConfigHttp:

    def __init__(self):
        global scheme, host, timeout
        scheme = localReadConfig.get_http("scheme")
        host = localReadConfig.get_http("base_url")
        timeout = localReadConfig.get_http("timeout")
        self.log = Log.get_log()
        self.logger = self.log.get_logger()
        self.params = {}
        self.headers = None
        self.data = {}
        self.url = None
        self.files = None

    def set_url(self, url):
        self.url = scheme +'://'+host+url

    def set_headers(self):
        """
        set headers
        """
        self.headers = getHeaders.get_headers()

    def set_local_headers(self):
        """
        set local headers
        """
        self.headers = getHeaders.local_headers()

    def set_params(self, param):
        """
        set params
        :param param:
        :return:
        """
        self.params = param

    def set_data(self, data):
        """
        set data
        :param data:
        :return:
        """
        self.data = data

    def get(self):
        """
        defined get method
        :return:
        """
        try:
            response = requests.get(self.url, headers=self.headers, params=self.params, data=self.data)
            if response.status_code == 200:
                return response
            elif response.status_code == 401:
                getHeaders.refresh_tokens()
                response = requests.get(self.url, headers=self.headers, params=self.params, data=self.data)
                return response
            else:
                return response
        except TimeoutError:
            self.logger.error("Time out!")
            return None

    def post(self):
        """
        defined post method
        :return:
        """
        try:
            response = requests.post(self.url, headers=self.headers, params=self.params, data=self.data)
            if response.status_code == 200:
                return response
            elif response.status_code == 401:
                getHeaders.refresh_tokens()
                response = requests.post(self.url, headers=self.headers, params=self.params, data=self.data)
                return response
            else:
                return response
        except TimeoutError:
            self.logger.error("Time out!")
            return None

    def post_with_file(self):
        """
        defined post method
        :return:
        """
        try:
            response = requests.post(self.url, headers=self.headers, data=self.data, files=self.files, timeout=float(timeout))
            return response
        except TimeoutError:
            self.logger.error("Time out!")
            return None

    def post_with_json(self):
        """
        defined post method
        :return:
        """
        try:
            response = requests.post(self.url, headers=self.headers, json=self.data, timeout=float(timeout))
            return response
        except TimeoutError:
            self.logger.error("Time out!")
            return None

用python自带的requests来进行接口测试，，拿get和post两个方法来说吧。（平时用的最多的就是这两个方法了，其他方法，大家可以仿照着自行扩展）

• get方法

    接口测试中见到最多的就是get方法和post方法，其中，getget方法在传递参数后，url的格式是这样的：http://接口地址?key1=value1&key2=value2，
  

对于requests提供的get方法，有几个常用的参数：
url：显而易见，就是接口的地址url啦
headers：定制请求头（headers），例如：content-type = application/x-www-form-urlencoded
params：用于传递测试接口所要用的参数，这里我们用python中的字典形式（key：value）进行参数的传递。
timeout：设置接口连接的最大时间（超过该时间会抛出超时错误）

举个栗子：

url=‘http://api.shein.com/v2/member/logout’
header={‘content-type’： application/x-www-form-urlencoded}
param={‘user_id’： 123456,‘email’： 123456@163.com}
timeout=0.5
requests.get（url, headers=header, params=param, timeout=timeout）

• post方法
  与get方法类似，只要设置好对应的参数，post方法中的参数，我们不在使用params进行传递，而是改用data进行传递了。

常用的返回值的操作。
text：获取接口返回值的文本格式
json()：获取接口返回值的json()格式
status_code：返回状态码（成功为：200）
headers：返回完整的请求头信息（headers[‘name’]：返回指定的headers内容）
encoding：返回字符编码格式
url：返回接口的完整url地址

举个栗子：

url=‘http://api.shein.com/v2/member/login’
header={‘content-type’： application/x-www-form-urlencoded}
data={‘email’： 123456@163.com,‘password’： 123456}
timeout=0.5
requests.post（url, headers=header, data=data, timeout=timeout）

以上这些，就是常用的方法啦，大家可自行取之。

关于失败请求抛出异常，我们可以使用“raise_for_status()”来完成，那么，当我们的请求发生错误时，就会抛出异常。在这里提醒下各位朋友，如果你的接口，在地址不正确的时候，会有相应的错误提示（有时也需要进行测试），这时，千万不能使用这个方法来抛出错误，因为python自己在链接接口时就已经把错误抛出，那么，后面你将无法测试期望的内容。而且程序会直接在这里当掉，以错误来计。

common.py

内容如下

import readConfig
import os
from xlrd import open_workbook
from common import configHttp
from common.Log import MyLog as Log
import json

localReadConfig = readConfig.ReadConfig()
testFilePath = os.path.join(readConfig.ProDir,'testFile')
localConfigHttp = configHttp.ConfigHttp()
log = Log.get_log()
logger = log.get_logger()

caseNo = 0


def get_value_from_return_json(response_json, name1, name2):
    """
    get value by key
    :param response_json:
    :param name1:
    :param name2:
    :return:
    """
    group = response_json[name1]
    value = group[name2]
    return value

def get_value_dict_keys(dict):
    """
    :param dict:
    :return:
    """
    list = [i for i in dict]
    return list


def show_return_msg(response):
    """
    show msg detail
    :param response:
    :return:
    """
    url = response.url
    msg = response.text
    code=str(response.status_code)
    print("请求地址："+url)
    print("请求结果："+code)
    print("请求返回值："+'\n'+json.dumps(json.loads(msg), ensure_ascii=False, sort_keys=True, indent=4))
    # print('Response HTTP Response Body:', json.dumps(self.response.json(), indent=2, sort_keys=True, ensure_ascii=False))

    # indent: 缩进空格数，indent = 0输出为一行
    # sort_keys = True: 将json结果的key按ascii码排序
    # ensure_ascii = False: 不确保ascii码，如果返回格式为utf - 8包含中文，不转化为\u...
# ****************************** read testCase excel ********************************


# 从excel文件中读取测试用例
def get_xls_case(xls_name, sheet_name):
    """
    get interface data from xls file
    :return:
    """
    cls = []
    # 获取用例文件路径
    xls_path = os.path.join(testFilePath, 'case', xls_name)
    # 打开用例Excel
    file = open_workbook(xls_path)
    # 获得打开Excel的sheet
    sheet = file.sheet_by_name(sheet_name)
    # 获取这个sheet内容行数
    rows = sheet.nrows
    for i in range(rows):#根据行数做循环
        if sheet.row_values(i)[0] != u'case_name':#如果这个Excel的这个sheet的第i行的第一列不等于case_name那么我们把这行的数据添加到cls[]
            cls.append(sheet.row_values(i))
    return cls

上面就是common主要内容。主要利用xlrd来操作excel文件，注意啦，我们是用excel文件来管理测试用例的。

configEmail.py

发送邮件

# coding:utf-8
import os
import smtplib
from email.mime.text import MIMEText
from email.header import Header

from email.mime.application import MIMEApplication  #主要类型的MIME消息对象应用
from email.mime.multipart import MIMEMultipart
from email.mime.image import MIMEImage
from datetime import datetime
import threading
import readConfig as readConfig
from common.Log import MyLog

localReadConfig = readConfig.ReadConfig()


class Email:
    def __init__(self):
        global host, user, password, port, sender, title,receivers

        sender = localReadConfig.get_email("sender")  # 发件人
        receivers = ['8463299@qq.com']  # 收件人
        # receivers = ['test@163.com','test@vip.qq.com']  # 接收多个邮件，可设置为你的QQ邮箱或者其他邮箱
        host = localReadConfig.get_email("mail_host")# 设置服务器
        port = localReadConfig.get_email("mail_port")  # 设置服务器
        user = localReadConfig.get_email("mail_user")# QQ邮件登录名称
        password = localReadConfig.get_email("mail_pass")# QQ邮箱的授权码

        title = localReadConfig.get_email("subject")#邮件主题

        # 定义邮件主题
        date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        self.subject = "接口测试报告" + " " + date

        self.log = MyLog.get_log()
        self.logger = self.log.get_logger()

        self.msg = MIMEMultipart('related')

    def config_header(self):
        """
        defined email header include subject, sender and receiver
        :return:
        """
        self.msg['Subject'] = Header(self.subject)  # 邮件主题
        self.msg['From'] = Header(sender)  # 发件人
        self.msg['To'] = Header(str(";".join(receivers)))  # 收件人

    def config_content(self):
        """
        write the content of email
        :return:
        """
        self.config_file_html_img()
        self.config_file()

    def config_file_html_img(self):
        file_path = os.path.join(readConfig.ProDir, 'testFile', 'emailStyle.html')  # 文件路径
        with open(file_path, 'rb') as fp:  # 读取文件内容
            msg_text = MIMEText(fp.read(), 'html', 'utf-8')  # 创建Text对象，包括文本内容
        self.msg.attach(msg_text)  # 构建HTML格式的邮件内容

        image2_path = os.path.join(readConfig.ProDir, 'testFile', 'img', 'smile.jpg')  # 图片路径
        self.msg.attach(self.add_image(image2_path, '<image2>'))  # 构建HTML格式的邮件内容

    def config_file_html(self):
        report_file_path = self.log.get_report_file_path()
        with open(report_file_path, encoding='utf-8') as f:  # 打开html报告
            email_body = f.read()  # 读取报告内容
        self.msg = MIMEMultipart()  # 混合MIME格式
        self.msg.attach(MIMEText(email_body, 'html', 'utf-8'))

    def config_file(self):

        if self.check_file():
            report_folder_path = self.log.get_result_folder_path()
            filename = [os.path.join(report_folder_path, 'output.log'),
                        os.path.join(report_folder_path, 'report.html')]
            for tmp in filename:
                with open(tmp, 'rb') as f:
                    attach_files = MIMEApplication(f.read())
                    attach_files.add_header('Content-Disposition', 'attachment', filename=tmp)
                    self.msg.attach(attach_files)

    def add_image(self,src, img_id):
        # xml中添加图片
        with open(src, 'rb') as f:
            msg_image = MIMEImage(f.read())  # 读取图片内容
        msg_image.add_header('Content-ID', img_id)  # 指定文件的Content-ID,<img>,在HTML中图片src将用到
        return msg_image

    def check_file(self):
        """
        check test report
        :return:
        """
        report_path = self.log.get_report_file_path()
        if os.path.isfile(report_path) and not os.stat(report_path) == 0:
            return True
        else:
            return False

    def send_email(self):
        """
        send email
        :return:
        """
        global smtp
        self.config_content()
        self.config_header()
        try:
            smtp = smtplib.SMTP_SSL(host,port)
            smtp.login(user, password)
            smtp.sendmail(sender, receivers, self.msg.as_string())
            self.logger.info("测试报告已通过电子邮件发送给开发人员")
        except Exception as ex:
            self.logger.error(str(ex))
            return "邮件发送失败"
        finally:
            smtp.quit()

class MyEmail:
    email = None
    mutex = threading.Lock()

    def __init__(self):
        pass

    @staticmethod
    def get_email():

        if MyEmail.email is None:
            MyEmail.mutex.acquire()
            MyEmail.email = Email()
            MyEmail.mutex.release()
        return MyEmail.email


if __name__ == "__main__":
    email = MyEmail.get_email()

HTMLTestRunner.py

HTMLTestRunner.py这个文件从网上下载的，大神写好的，用于生成html格式的测试报告
生成报告如下

runAll.py

这是整个框架运行的入口

import os
import unittest
from common.Log import MyLog as Log
import readConfig as readConfig
from common import HTMLTestRunner
from common.configEmail import MyEmail

localReadConfig = readConfig.ReadConfig()


class AllTest:
    def __init__(self):
        global log, logger, resultPath, on_off
        log = Log.get_log()
        logger = log.get_logger()
        resultPath = log.get_report_file_path()
        on_off = localReadConfig.get_email("on_off")
        self.caseListFile = os.path.join(readConfig.ProDir, "caseList.txt")
        self.caseFile = os.path.join(readConfig.ProDir, "testCase")
        self.caseList = []
        self.email = MyEmail.get_email()

    def set_case_list(self):
        """
        set case list
        :return:
        """
        fb = open(self.caseListFile)
        for value in fb.readlines():
            data = str(value)
            if data != '' and not data.startswith("#"):
                self.caseList.append(data.replace("\n", ""))
        fb.close()

    def set_case_suite(self):
        """
        set case suite
        :return:
        """
        self.set_case_list()
        test_suite = unittest.TestSuite()
        suite_module = []

        for case in self.caseList:
            case_name = case.split("/")[-1]
            print(case_name + ".py")
            discover = unittest.defaultTestLoader.discover(self.caseFile, pattern=case_name + '.py', top_level_dir=None)
            suite_module.append(discover)

        if len(suite_module) > 0:

            for suite in suite_module:
                for test_name in suite:
                    test_suite.addTest(test_name)
        else:
            return None

        return test_suite

    def run(self):
        """
        run test
        :return:
        """
        try:
            suit = self.set_case_suite()
            if suit is not None:
                logger.info("================================== 测试开始 ==================================")
                fp = open(resultPath, 'wb')
                runner = HTMLTestRunner.HTMLTestRunner(stream=fp, title='即刻接口测试报告', description='用例执行情况')
                runner.run(suit)
                fp.close()

            else:
                logger.info("没有需要测试的案例")
        except Exception as ex:
            logger.error(str(ex))
        finally:
            # 通过电子邮件发送测试报告
            if on_off == 'on':
                self.email.send_email()
            elif on_off == 'off':
                logger.info("不向开发人员发送报告电子邮件")
            else:
                logger.info("未知状态")

            logger.info("================================== 测试结束 ==================================")


if __name__ == '__main__':
    obj = AllTest()
    obj.run()

上面贴出了runAll里面的主要部分，首先我们要从caselist.txt文件中读取需要执行的case名称，然后将他们添加到python自带的unittest测试集中，最后执行run()函数，执行测试集。

result

result文件夹会在首次执行case时生成，并且以后的测试结果都会被保存在该文件夹下，同时每次测试的文件夹都是用系统时间命名，里面包含了两个文件，log文件和测试报告

testCase

testCase文件夹下，存放我们写的具体的测试case。注意，所有的case名称都要以test开头来命名，因为unittest在进行测试时会自动匹配testCase文件夹下面所有test开头的.py文件

testFile

testFile文件夹下，放置我们测试时用来管理测试用例的excel文件和用于数据库查询的sql语句的xml文件。

caselist.txt

凡是没有被注释掉的，都是要被执行的case名称啦。在这里写上你要执行的case名称就可以啦。

参考
https://my.oschina.net/u/3041656/blog/820023?p=7
参考源码
https://gitee.com/null_534_6629/interfacetest/tree/master