前言

小白个人研究成果，仅供参考

一、准备工作

1. 建好项目及相关文件夹

common - 通用类，用于存放所有可以共用的类，例如写入日志、连接数据库，连接接口等
result - 用于存放所有结果，例如生成的日志文件以及生成的html测试报告
testCase - 存放所有测试用例
testFile - 存放相关文档
没有存放在文件夹里面的还有
caselist.txt - 存放所有要执行的用例的路径
config.ini - 配置文件
readConfig.py - 读取配置文件config.ini
runAll.py - 测试入口

2. 准备好相关文档

testFile 文件夹下：

case_demo.xls 用于存放所有要执行测试用例路径（这个可以后面写，但先讲）
内容如下：
包含 case_name：用例名称、method：请求方式，url：接口路径，params：参数，code：响应码， meaasge：响应内容
sql.xml 存放所有要执行的sql

<page>
	<database name="db_demo_org">
	   <table name="t_organization_info">
		   <sql id="selectByCity">SELECT * FROM t_organization_info WHERE city = %s</sql>
		   <sql id="selectByTypeAndCity">SELECT * FROM t_organization_info WHERE org_code_type = %s and city = %s</sql>
	   </table>
	</database>
	<database name="demo">
	   <table name="t_case">
		   <sql id="selectCaseAll">SELECT * FROM t_case</sql>
	   </table>
	</database>
</page>

最外层文件：
caselist.txt 存放要执行的测试用例路径
在这里插入图片描述
config.ini 存放配置文件，例如数据库配置、邮箱配置、http请求配置等
email 参数讲解：
code：是邮件授权码（不是密码哦，具体参考https://baijiahao.baidu.com/s?id=1552315463915496&wfr=spider&for=pc）
smpt：是邮箱服务器地址，我用的是qq邮箱
sender：发送者
recerives ：接收者，可以写多个
subject ：邮件标题

[DATABASE]
host = 127.0.0.1
port = 3306
name = root
pwd = 123456
database = db_demo_org

[DATABASE_CASE]
host = 127.0.0.1
port = 3306
name = root
pwd = 123456
database = demo

[HTTP]
url = http://localhost:8086/quota/
timeout = 3

[EMAIL]
smtp = smtp.qq.com
sender = 145xxxxxx@qq.com
recerives = 227xxxxx0@qq.com
code = xxxxxxxxxqxoifaa
subject = 接口自动化测试报告
content = 您好，附件是接口自动化测试报告，请查收。
on_off = 1

二、代码部分

1. readConfig.py读取config.ini配置文件

创建readConfig.py

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import os
import codecs
import configparser

# 获取当前py文件路径的上一层
localDir = os.path.split(os.path.realpath(__file__))[0]
# 拼接成config.ini完整路径
configPath = os.path.join(localDir, 'config.ini')


class ReadConfig:
    """读取config.ini文件"""
    def __init__(self):
        # 打开文件
        config_file = open(configPath, encoding='UTF-8')
        data = config_file.read()

        # 判断是否带BOM文件
        if data[:3] == codecs.BOM_UTF8:
            data = data[3:]
            file = codecs.open(configPath, 'w', encoding='UTF-8')
            # 改写文件
            file.write(data)
            file.close()
        config_file.close()

        # 创建configparser实例
        self.cf = configparser.ConfigParser()
        # 读取配置文件
        self.cf.read(configPath, encoding='UTF-8')

    # 获取配置文件email的参数
    def get_email(self, name):
        return self.cf.get('EMAIL', name)

    # 获取配置文件http的参数
    def get_http(self, name):
        return self.cf.get('HTTP', name)

    # 获取配置文件database的参数
    def get_database(self, name):
        return self.cf.get('DATABASE', name)

    # 获取配置文件database_case的参数
    def get_database_case(self, name):
        return self.cf.get('DATABASE_CASE', name)

2. 创建公用类

common文件夹下（以下几个py文件我就不分先后顺序介绍了）：

configDB.py 操作数据库

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import MySQLdb  # python3安装命令：pip3 install Mysqlclien
from readConfig import ReadConfig
from common.myLog import MyLog

myConfig = ReadConfig()


class MyDB:
    """数据库"""
    global host, name, pwd, port, database, config
    # 获取配置文件中的参数
    host = myConfig.get_database("host")
    name = myConfig.get_database("name")
    pwd = myConfig.get_database("pwd")
    port = myConfig.get_database("port")
    database = myConfig.get_database("database")
    config = {
        'host': str(host),
        'user': name,
        'password': pwd,
        'port': int(port),
        'db': database,
        'charset': 'utf8'
    }

    def __init__(self):
        self.logger = MyLog.get_log().logger
        self.db = None
        self.cursor = None

    def connect_db(self):
        """连接数据库"""
        try:
            self.db = MySQLdb.connect(**config)
            self.cursor = self.db.cursor()
            print('数据库已连接成功')
        except ConnectionError as e:
            self.logger.error(str(e))

    def execute_sql(self, sql, params):
        """执行sql并返回操作游标"""
        try:
            self.connect_db()
            self.cursor.execute(sql, params)
            self.db.commit()
            return self.cursor
        except RuntimeError as e:
            self.db.rollback()

    def get_all(self, cursor):
        """获取全部数据"""
        value = cursor.fetchall()
        return value

    def get_one(self, cursor):
        """获取第一条数据"""
        value = cursor.fetchone()
        return value

    def close_db(self):
        """关闭数据库"""
        self.db.close()
        print('数据库已关闭')

configDBCase.py 还是操作数据库，但我这里是另一个数据库专门处理测试用例的（本文读取测试用例一共用了两种方式 1-读取数据库，2-读取xls文档）

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import MySQLdb  # python3安装命令：pip3 install Mysqlclien
from readConfig import ReadConfig
from common.myLog import MyLog

myConfig = ReadConfig()


class MyDBCase:
    """测试用例数据库"""
    global host, name, pwd, port, database, config
    # 获取配置文件中的参数
    host = myConfig.get_database_case("host")
    name = myConfig.get_database_case("name")
    pwd = myConfig.get_database_case("pwd")
    port = myConfig.get_database_case("port")
    database = myConfig.get_database_case("database")
    config = {
        'host': str(host),
        'user': name,
        'password': pwd,
        'port': int(port),
        'db': database,
        'charset': 'utf8'
    }

    def __init__(self):
        """初始化数据"""
        self.logger = MyLog.get_log().logger
        self.db = None
        self.cursor = None

    def connect_db(self):
        """连接数据库"""
        try:
            self.db = MySQLdb.connect(**config)
            self.cursor = self.db.cursor()
            print('数据库已连接成功')
        except ConnectionError as e:
            self.logger.error(str(e))

    def execute_sql(self, sql, params):
        """执行sql并返回操作游标"""
        try:
            self.connect_db()
            self.cursor.execute(sql, params)
            self.db.commit()
            return self.cursor
        except RuntimeError as e:
            self.db.rollback()

    def get_all(self, cursor):
        """返回全部数据"""
        value = cursor.fetchall()
        return value

    def get_one(self, cursor):
        """返回第一条数据"""
        value = cursor.fetchone()
        return value

    def close_db(self):
        """关闭数据库连接"""
        self.db.close()
        print('数据库已关闭')

configEmail.py 处理邮件

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import time
import yagmail
from readConfig import ReadConfig
from common.myLog import MyLog

myConfig = ReadConfig()


class Email:
    """邮件处理"""
    def __init__(self):
        global smtp, sender, recerives, code, title, content
        # 读取配置文件中的数据
        smtp = myConfig.get_email('smtp')
        sender = myConfig.get_email('sender')
        recerives = myConfig.get_email('recerives')
        code = myConfig.get_email('code')
        title = myConfig.get_email('subject')
        content = myConfig.get_email('content')
        # 获取当前时间
        date = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
        self.subject = title + date
        self.log = MyLog.get_log()
        self.logger = self.log.logger
        pass

    def send_email(self, accessory):
        """发送邮件"""
        # 参数：发送者，邮件授权码（不是密码哦，具体参考https://baijiahao.baidu.com/s?id=1552315463915496&wfr=spider&for=pc），邮箱服务器地址
        yag = yagmail.SMTP(user=sender, password=code, host=smtp)
        # 发送邮件，参数：接收者，标题，内容，附件
        yag.send(recerives, self.subject, content, accessory)
        # 关闭
        yag.close()

configHttp.py 请求接口

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import requests
import json
# 引入写好的几个类
from readConfig import ReadConfig
from common.myLog import MyLog

# 实例化ReadConfig类
myConfig = ReadConfig()


class ConfigHttp:
    """请求接口类"""
    # 全局变量token
    token = {}

    def __init__(self):
        """初始化"""
        global timeout
        timeout = myConfig.get_http('timeout')
        self.logger = MyLog.get_log().logger
        self.url = ''
        self.headers = {}
        self.params = {}
        self.data = {}
        self.files = {}
        pass

    def set_headers(self, headers):
        """设置参数"""
        self.headers = headers

    def set_params(self, params):
        self.params = params

    def set_data(self, data):
        self.data = data

    def set_files(self, files):
        self.files = files

    def set_url(self, url):
        self.url = url

    def get_requests(self):
        try:
            return requests.get(url=myConfig.get_http('url') + self.url, params=self.params, headers=self.headers, timeout=float(timeout))
        except TimeoutError:
            self.logger.error('请求超时')
            return None

    def post_requests(self):
        """post请求方式"""
        try:
            return requests.post(url=myConfig.get_http('url') + self.url, json=self.data, files=self.files, headers=self.headers, timeout=float(timeout))
        except TimeoutError:
            self.logger.error('请求超时')
            return None

myLog.py 处理日志

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import os
import time
import logging
import threading
import readConfig


logPath = ''
resultPath = ''
localDir = ''


class Log:
    """日志"""
    def __init__(self):
        """初始化"""
        global logPath, resultPath, localDir
        # 获取readConfig.py中的localDir参数
        localDir = readConfig.localDir
        # 拼接result文件夹完整路径
        resultPath = os.path.join(logPath, "result")
        # 判断文件是否存在，否则创建
        if not os.path.exists(resultPath):
            os.mkdir(resultPath)
        # 创建日志文件
        date = time.strftime('%Y%m%d', time.localtime())
        logPath = os.path.join(resultPath, date)
        if not os.path.exists(logPath):
            os.mkdir(logPath)
        # 创建logger实例
        self.logger = logging.getLogger()
        # 定义logger级别
        self.logger.setLevel(logging.INFO)

        # 自定义Handle日志规则，如果是logging.StreamHandler则是输出到控制台，FileHandle则是输出到文件
        handler = logging.FileHandler(os.path.join(logPath, 'output.log'))

        # 创建日志格式实例：%(asctime)s：打印日志的时间, %(levelname)s：打印日志级别的名称, %(message)s：打印日志信息
        formater = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

        # handler自定义格式
        handler.setFormatter(formater)

        # logger日志加载handle实例
        self.logger.addHandler(handler)

        pass


class MyLog:
    log = None
    # 创建线程锁
    threadLock = threading.Lock()

    def __init__(self):
        pass

    # 申明静态方法，无需实例化
    @staticmethod
    def get_log():
        if MyLog.log is None:
            # 开启锁
            MyLog.threadLock.acquire()
            # 调用Log类
            MyLog.log = Log()
            # 释放锁
            MyLog.threadLock.release()
        return MyLog.log

    pass

readCase.py 读取xml、xls文档

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import os
from xlrd import open_workbook  # 下载参考：https://blog.csdn.net/ZMJ_566/article/details/89360918
from common.configHttp import ConfigHttp
from common.myLog import MyLog
import readConfig
from xml.etree import ElementTree

localConfigHttp = ConfigHttp()
logger = MyLog.get_log().logger
database = {}


class ReadCase:
    """读取文档"""
    def get_xls(self, xls_name, sheet_name):
        """读取xls文档。xls_name:文档名称，sheet_name:工作表sheet名称 """
        cls = []
        # 拼接地址
        xlsPath = os.path.join(readConfig.localDir, 'testFile', xls_name)
        # 打开文档
        file = open_workbook(xlsPath)
        # 根据工作表名获取工作表信息
        sheet = file.sheet_by_name(sheet_name)
        # 获取工作表总行数
        nrows = sheet.nrows
        # 遍历并忽略第一行表头
        for i in range(1, nrows):
            cls.append(sheet.row_values(i))
        return cls

    def get_xml(self):
        """读取xml文档"""
        if len(database) == 0:
            # 拼接地址
            sql_path = os.path.join(readConfig.localDir, 'testFile', 'sql.xml')
            # 解析xml
            tree = ElementTree.parse(sql_path)
            # 遍历所有database标签
            for db in tree.findall("database"):
                # 获取标签的name属性
                db_name = db.get("name")
                table = {}
                for tb in db:
                    table_name = tb.get("name")
                    sql = {}
                    for data in tb:
                        sql_id = data.get("id")
                        sql[sql_id] = data.text
                    table[table_name] = sql
                database[db_name] = table

    def get_sql(self, database_name, table_name, sql_id):
        """根据数据库名、表名、sql_id获取sql"""
        try:
            self.get_xml()
            sql = database.get(database_name).get(table_name).get(sql_id)
            return sql
        except RuntimeError as e:
            logger.info(e + '：获取的数据库名或表名或sql_id不存在')

3. 测试用例

在这里插入图片描述

testLogin.py 这里只用到了简单的请求参数

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import unittest
from common.configHttp import ConfigHttp
from common.myLog import MyLog

logger = MyLog.get_log().logger
configHttp = ConfigHttp()


class Login(unittest.TestCase):
    """登录"""
    def setUp(self):
        print('*****启动Login用例*****')

    # 这三个用例是我凑数用的
    def test_one(self):
        """凑数用例1"""
        logger.info('test_one Method')

    def test_two(self):
        """凑数用例2"""
        logger.info('test_two Method')

    def test_three(self):
        """凑数用例3"""
        logger.info('test_three Method')

    def test_login(self):
        """登录接口"""
        # 请求参数
        data = {'login': 'admin', 'password': '202cb962axxxxxxxxxxxxxxxxx0'}
        # 接口路径
        url = 'eLogin'
        # 调用configHttp的相关方法
        configHttp.set_url(url)
        configHttp.set_data(data)
        result = configHttp.post_requests()
        result_json = result.json()
        # 判断字段是否存在与json中
        if 'code' in result_json:
            if result_json['code'] == 200 and result_json['message'] == '成功':
                logger.info('登录成功!')
                # 将token设置到configHttp中方便其他类调用
                configHttp.token['token'] = result_json['data']['token']
        pass

    def tearDown(self):
        print('*****销毁Login用例*****')

testOrganizationInfo.py 这里用到了读取xml获取sql并执行，读取xls文档执行接口

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

from common.configDB import MyDB
from common.readCase import ReadCase
from common.myLog import MyLog
from common.configHttp import ConfigHttp

db = MyDB()
readCase = ReadCase()
logger = MyLog().get_log().logger


class OrganizationInfo(unittest.TestCase):
    """组织机构"""
    def setUp(self):
        logger.info('*****启动OrganizationInfo用例*****')

    # 这里演示一下读取xml并操作数据库
    def test_selectByCity(self):
        """读取xml中的sql并执行"""
        # 连接数据库
        db.connect_db()
        # 读取xml获取sql
        sql = readCase.get_sql('db_demo_org', 't_organization_info', 'selectByCity')
        params = (u'广州市',)
        # 执行sql并返回操作游标
        cursor = db.execute_sql(sql, params)
        # 获取全部数据
        datas = db.get_all(cursor)
        for data in datas:
            print(data)
        # 关闭数据库
        db.close_db()
        logger.info('test_selectByCity Method')

    def test_selectByTypeAndCity(self):
        """读取xml中的sql并执行"""
        # 连接数据库
        db.connect_db()
        # 读取xml获取sql
        sql = readCase.get_sql('db_demo_org', 't_organization_info', 'selectByTypeAndCity')
        params = ('1', u'广州市')
        # 执行sql并返回游标
        cursor = db.execute_sql(sql, params)
        # 获取全部数据
        datas = db.get_all(cursor)
        for data in datas:
            print(data)
        # 关闭数据库
        db.close_db()
        logger.info('test_selectByIdAndCity Method')

    def test_read_xls(self):
        """读取xls文档获取用例并执行"""
        # 读取xls文档获取用例
        datas = readCase.get_xls('case_demo.xls', 'organizarion')
        configHttp = ConfigHttp()
        headers = {'Content-Type': 'application/json', 'Authorization': configHttp.token['token']}
        configHttp.set_headers(headers)
        for data in datas:
            # 设置参数
            configHttp.set_url(data[2])
            if data[1] == 'post':
                # 这里传过来是字符串，需要转成字典dict类型，否则接口报错（我的接口参数是map）
                configHttp.set_data(json.loads(data[3]))
                # 执行post请求
                result = configHttp.post_requests()
            else:
                configHttp.set_params(data[3])
                # 执行get请求
                result = configHttp.get_requests()
            # 数据转json
            result_json = result.json()
            # 判断字段是否存在与json中
            if 'code' in result_json:
                if result_json['code'] != 200:
                    logger.error('接口异常')
                    # 手动抛出用例异常
                    raise RuntimeError('接口异常')
            else:
                logger.error('接口异常')
                # 手动抛出用例异常
                raise RuntimeError('接口异常')

        logger.info('测试organization接口')

    def tearDown(self):
        logger.info('*****销毁OrganizationInfo用例*****')


if __name__ == '__main__':
    OrganizationInfo().test_selectByTypeAndCity()

testUser.py 这里用到了读取数据库执行测试用例

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import unittest
import json
from common.myLog import MyLog
from common.configHttp import ConfigHttp
from common.configDBCase import MyDBCase
from common.readCase import ReadCase

logger = MyLog.get_log().logger
db = MyDBCase()
configHttp = ConfigHttp()
readCase = ReadCase()


class User(unittest.TestCase):
    """用户"""
    # setUp(self) -> None 是 python3的写法，等于cls.func = none
    def setUp(self) -> None:
        logger.info('*****启动user用例*****')

    def test_read_db(self):
        """读取数据库中的测试用例并执行"""
        # 连接数据库
        db.connect_db()
        # 获取xml文件的sql
        sql = readCase.get_sql('demo', 't_case', 'selectCaseAll')
        # 执行sql
        cursor = db.execute_sql(sql, '')
        # 返回全部数据
        datas = db.get_all(cursor)
        # 设置header
        headers = {'Content-Type': 'application/json', 'Authorization': configHttp.token['token']}
        configHttp.set_headers(headers)
        for data in datas:
            # 设置url和参数
            configHttp.set_url(data[3])
            # map类型参数这里需要字符串转字典
            configHttp.set_data(json.loads(data[4]))
            if data[2] == 'post':
                # 执行post请求
                result = configHttp.post_requests()
            else:
                # 执行get请求
                result = configHttp.get_requests()
            # 数据转json
            result_json = result.json()
            if 'code' in result_json:
                if result_json['code'] != data[5]:
                    logger.error('接口异常')
                    # 手动抛出异常
                    raise RuntimeError('接口异常')
            else:
                logger.error('接口异常')
                # 手动抛出异常
                raise RuntimeError('接口异常')
            print(data)
        # 关闭数据库
        db.close_db()
        logger.info('*****销毁test_read_db Method*****')
        pass

    def tearDown(self) -> None:
        logger.info('*****测试user用例*****')


if __name__ == '__main__':
    unittest.main

数据库表：

CREATE TABLE `t_case` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) NOT NULL COMMENT '用例名称',
  `method` varchar(255) NOT NULL COMMENT '方法',
  `url` varchar(255) NOT NULL COMMENT '路径',
  `param` varchar(255) NOT NULL COMMENT '参数',
  `code` int(255) DEFAULT NULL COMMENT '编码',
  `message` varchar(255) DEFAULT NULL COMMENT '消息',
  PRIMARY KEY (`id`)
) ENGINE=MyISAM AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4;

表内容：
在这里插入图片描述

3. 执行所有用例

runAll.py

#!/usr/bin/env python 
# -*- coding:utf-8 -*-
import unittest
import os
import time
from HTMLTestRunner import HTMLTestRunner # 这里参考https://blog.csdn.net/notHavaBug/article/details/115457523?spm=1001.2014.3001.5502
import readConfig
from common.myLog import MyLog
from common.configEmail import Email

logger = MyLog.get_log().logger
myConfig = readConfig.ReadConfig()


class RunAll:
    def __init__(self):
        self.caseList = []
        pass

    def set_case_list(self):
        # 读取caselist.txt文档获取用例路径
        caselist_file = os.path.join(readConfig.localDir, 'caselist.txt')
        fb = open(caselist_file, 'r')
        for value in fb.readlines():
            data = str(value)
            # 忽略被注释掉的用例路径
            if data != '' and not data.startswith('#'):
                self.caseList.append(data.replace('\n', ''))

    def set_case_suit(self):
        self.set_case_list()
        # 创建测试套件
        test_suite = unittest.TestSuite()
        suite_model = []
        for case in self.caseList:
            # 根据'/'分割文件夹、文件名
            case_url = case.split('/')
            # 拼接路径
            case_file = os.path.join(os.getcwd(), 'testCase', case_url[0])
            # 划重点！这个地方不要用unittest.defaultTestLoader.discover会因为初始化问题只能执行一次，因为这个我真的找了一整天的原因！！！
            test_loader = unittest.TestLoader()
            # 自动执行所有测试用例
            # pattern支持模糊匹配，例如login下有很多测试用例且都是以test开头，不想一个个都写在caselist.txt中，可以在caselist.txt中只写文件夹名，这里直接写pattern='test*.py'
            discover = test_loader.discover(case_file, pattern=case_url[1] + '.py', top_level_dir=None)
            suite_model.append(discover)

        if len(suite_model) > 0:
            for suite in suite_model:
                for test_name in suite:
                    test_suite.addTest(test_name)
        else:
            return None
        return test_suite

    def run(self):
        flag = 0
        try:
            suit = self.set_case_suit()
            if suit is not None:
                logger.info('*****************开始测试***************')
                date = time.strftime('%Y%m%d-%H%M%S', time.localtime())
                # 拼接html报告存放路径
                resultPath = os.path.join(readConfig.localDir, 'result', date + '_test.html')
                fp = open(resultPath, 'wb')
                # 生成html测试报告
                runner = HTMLTestRunner(stream=fp, title='测试报告', description=date + '测试用例执行情况')
                runner.run(suit)
                flag = 1
            else:
                logger.info('没有测试用例')
        except Exception as e:
            print(e)
            logger.error(str(e))
        finally:
            logger.info('*****************测试结束***************')
            on_off = myConfig.get_email('on_off')
            if int(on_off) == 1 and flag == 1:
                # 调用写好email类发送邮件
                Email().send_email(resultPath)
                logger.info('已发送邮件')


if __name__ == '__main__':
    RunAll().run()