整理下整个接口自动化框架的思路
主要用到的插件：pytest，allure-pytest，yaml

项目框架：

思维导图：

开始搭建框架

第一步： Conf 配置

创建config.ini文件：根据需要来去设计类别和类别下的元素

创建和编写config.py 文件：自定义一些get，set，add方法，用来读取，编辑，添加配置文件中的数据

import configparser
from Common import Log
import osproDir = os.path.split(os.path.realpath(__file__))[0]
configPath = os.path.join(proDir, "config.ini")class Read_config:def __init__(self):self.config = configparser.ConfigParser()self.config.read(configPath)def get_global(self, param):value = self.config.get('global_paras', param)return valuedef get_mail(self, param):value = self.config.get('mail', param)return valuedef get_database(self, param):value = self.config.get('database', param)return valuedef get_conf(self, section, param):value = self.config.get(section, param)return valuedef set_conf(self, section, value, text):"""配置文件修改:param section::param value::param text::return:"""self.config.set(section, value, text)with open(configPath, "w+") as f:return self.config.write(f)def add_conf(self, section_name):"""添加类别到配置环境里:param section_name::return:"""self.config.add_section(section_name)with open(configPath, "w+") as f:return self.config.write(f)

第二步：编写测试用例

安装插件：

pip install PyYAML

说明：

• 创建测试用例.yaml : 里面的结构可以自由设计，只需要了解以下yaml文件的基本语法就好了

编写读取yaml文件的方法：将方法写在通用类Utils里面

import yamldef read_data_from_file(file_name):"""读取yaml文件的内容:param file_name"""f = open(rootPath + '/data/' + file_name, encoding='utf-8')res_json = yaml.load(f, Loader=yaml.FullLoader)  # 添加loader参数是为了去掉load warningreturn res_json

第三步：封装request和assertion方法

安装request插件：

pip install request

创建Customize_request: 请求前的数据处理自行设计，请求后，获取返回的body，耗时，状态码，放入到字典中，该方法返回一个字典

"""
封装request"""import requests
from Common import Token
from Common.Utils import Utils
from Conf import Config
from Common import Logclass Request(Utils):def __init__(self, env):""":param env:"""self.env = envself.config = Config.Read_config()self.log = Log.MyLog()self.t = Token.Token()def post_request(self, url, data):"""Post请求:param url::param data::return:"""# post 请求try:response = requests.post(url=request_url, params=data)except Exception as e:print('%s%s' % ('Exception url: ', request_url))print(e)return ()# time_consuming为响应时间，单位为毫秒time_consuming = response.elapsed.microseconds / 1000# time_total为响应时间，单位为秒time_total = response.elapsed.total_seconds()response_dicts = dict()response_dicts['code'] = response.status_codetry:response_dicts['body'] = response.json()except Exception as e:print(e)response_dicts['body'] = ''response_dicts['text'] = response.textresponse_dicts['time_consuming'] = time_consumingresponse_dicts['time_total'] = time_totalreturn response_dicts

创建Cutomize_assertion方法：

"""
封装Assert方法"""
from Common import Log
import json
import traceback
from Conf import Configclass Assertions:def __init__(self):self.log = Log.MyLog()@staticmethoddef assert_status_code(status_code, expected_code):"""验证response状态码:param status_code::param expected_code::return:"""try:assert status_code == expected_codereturn Trueexcept Exception:log_error(traceback.format_exc(), status_code, expected_code)raise@staticmethoddef assert_single_item(single_item, expected_results):"""验证response body中任意属性的值:param single_item::param expected_results::return:"""try:assert single_item == expected_resultsreturn Trueexcept Exception:log_error(traceback.format_exc(), single_item, expected_results)raise@staticmethoddef assert_in_text(body, expected_results):"""验证response body中是否包含预期字符串:param body::param expected_results::return:"""text = json.dumps(body, ensure_ascii=False)try:# print(text)assert expected_results in textreturn Trueexcept Exception:log_error(traceback.format_exc(), text, expected_results)raise@staticmethoddef assert_items(d_body, expected_results):"""验证body里面的items是否符合期望需保证expected_results中是属性，在body中都能找到:param d_body: 一个dict/json:param expected_results: 一个dict/json:return:"""for key, value in expected_results.items():try:if key in d_body.keys():assert d_body[key] == valuereturn Trueelse:return Falseexcept Exception:log_error(traceback.format_exc(), d_body[key], value)raise@staticmethoddef assert_time(actual_time, expected_time):"""验证response body响应时间小于预期最大响应时间,单位：毫秒:param actual_time::param expected_time::return:"""try:assert actual_time < expected_timereturn Trueexcept Exception:log_error(traceback.format_exc(), actual_time, expected_time)raisedef log_error(e, actual, expected):Log.MyLog.error(str(e))Log.MyLog.error('actual results is %s, expected results is %s ' % (actual, expected))config = Config.Read_config()config.set_conf('results', 'final_results', 'False')

第四步：编写log文件（在网上随便复制的）

"""
封装log方法
"""import logging
import os
import timeLEVELS = {'debug': logging.DEBUG,'info': logging.INFO,'warning': logging.WARNING,'error': logging.ERROR,'critical': logging.CRITICAL
}logger = logging.getLogger()
level = 'default'def create_file(filename):path = filename[0:filename.rfind('/')]if not os.path.isdir(path):os.makedirs(path)if not os.path.isfile(filename):fd = open(filename, mode='w', encoding='utf-8')fd.close()else:passdef set_handler(levels):if levels == 'error':logger.addHandler(MyLog.err_handler)logger.addHandler(MyLog.handler)def remove_handler(levels):if levels == 'error':logger.removeHandler(MyLog.err_handler)logger.removeHandler(MyLog.handler)def get_current_time():return time.strftime(MyLog.date, time.localtime(time.time()))class MyLog:path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))log_file = path+'/Log/log.log'err_file = path+'/Log/err.log'logger.setLevel(LEVELS.get(level, logging.NOTSET))create_file(log_file)create_file(err_file)date = '%Y-%m-%d %H:%M:%S'handler = logging.FileHandler(log_file, encoding='utf-8')err_handler = logging.FileHandler(err_file, encoding='utf-8')@staticmethoddef debug(log_meg):set_handler('debug')logger.debug("[DEBUG " + get_current_time() + "]" + log_meg)remove_handler('debug')@staticmethoddef info(log_meg):set_handler('info')logger.info("[INFO " + get_current_time() + "]" + log_meg)remove_handler('info')@staticmethoddef warning(log_meg):set_handler('warning')logger.warning("[WARNING " + get_current_time() + "]" + log_meg)remove_handler('warning')@staticmethoddef error(log_meg):set_handler('error')logger.error("[ERROR " + get_current_time() + "]" + log_meg)remove_handler('error')@staticmethoddef critical(log_meg):set_handler('critical')logger.error("[CRITICAL " + get_current_time() + "]" + log_meg)remove_handler('critical')if __name__ == "__main__":MyLog.debug("This is debug message")MyLog.info("This is info message")MyLog.warning("This is warning message")MyLog.error("This is error")MyLog.critical("This is critical message")

第五步：编写测试用例脚本

安装pytest和allure-pytest：

pip install pytest 
pip install allure-pytest 

系统安装allure：

brew install  allure （mac系统）

测试用例脚本：

import os
import sys
import pytest
import allure# 避免使用命令行运行时，找不到自定义的module，需要添加下module的绝对路径
curPath = os.path.abspath(os.path.dirname(__file__))
rootPath = os.path.split(curPath)[0]
sys.path.append(rootPath)print(sys.path)
from Common.Utils import Utils
from Common.Customize_request import Request
from Common.Customize_assertion import Assertions
from Conf import Config@pytest.mark.parametrize('case', Utils.read_data_from_file('user_details.yaml'))
@allure.feature('Personal Center')
@allure.severity('normal')
@allure.story('obtain the user personal information')
def test_user_details(case):# 动态定制每个测试用例的titleallure.dynamic.title(case['case_id'])allure.dynamic.description(case['title'])config = Config.Read_config()r = Request(config.get_global('current_environment'))test = Assertions()response = r.post_request(case['url'], case['data'])print(response)assert test.assert_status_code(response['code'], case['returns']['code'])assert test.assert_single_item(response['body']['data']['yogoId'], case['returns']['validator']['yogoId'])assert test.assert_in_text(response['body'], 'msg')assert test.assert_items(response['body']['data'], case['returns']['validator'])if __name__ == "__main__":pytest.main(["-s", "test_user_details.py"])

说明：

• pytest的参数化，从每一个yaml文件中读取的，都是测试用例的集合，通过pytest的参数化，循坏运行每个测试用例；
• 定义allure里的的参数，来生成定制化的测试报告；
• 对请求返回的结果进行验证

第六步：运行测试用例

运行单个测试用例，把下面的代码，复制到单个测试用例脚本下，就可以实时看到结果：

if __name__ == "__main__":pytest.main(["-s", "test_user_details.py"])

安装插件（运行失败可设置重新运行N次）：

pip install pytest-rerunfailures

运行多个测试用例，创建Run.py文件：

import os
import time
from Common import Log
from Conf import Config
from Common import Email
from Common.Utils import Utilscurrent_path = os.path.abspath(os.path.dirname(__file__))
rootPath = os.path.split(current_path)[0]if __name__ == '__main__':Utils.remove_dir_and_its_files(current_path + '/outputs')# 获取当前的时间current_time = time.strftime('%Y%m%d%H%M%S')log = Log.MyLog()config = Config.Read_config()log.info('初始化配置文件： ' + Config.configPath)config.set_conf('results', 'final_results', 'True')# 运行测试用例，并生成测试报告文件 xml格式# 重新运行上一次失败的case,重试2次: --reruns 2print(current_path)os.system('pytest --alluredir=outputs/results/results_' + current_time)os.system('allure generate outputs/results/results_' + current_time + ' -o outputs/reports_' + current_time)

到此，基础的框架就完成了

进阶设置：

邮件通知：

"""
封装发送邮件的方法"""
import smtplib
import time
from email.header import Header
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from Common.Utils import Utilsfrom Common import Log
from Conf import Configclass SendMail(Utils):def __init__(self):super().__init__()self.config = Config.Read_config()self.log = Log.MyLog()def send_mail(self):# 第三方服务mail_host = self.config.get_mail('mail_host')  # 设置服务器mail_user = self.config.get_mail('mail_user')  # 用户名mail_pass = self.config.get_mail('mail_pass')  # 口令sender = self.config.get_mail('sender')receivers = self.config.get_mail('receiver')# msg = MIMEMultipart()body = 'Hi，all\n接口自动化测试完毕，失败case如下：\n'with open(self.projectDir + "/fail_cases.txt", 'r', encoding='utf-8') as f:while True:line = f.readline()body += line.strip() + '\n'if not line:breakmessage = MIMEMultipart()tm = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))message['Subject'] = Header("接口自动化测试报告" + "_" + tm, 'utf-8')message['From'] = Header("herby he", 'utf-8')  # 邮件里展示用户名message['To'] = receiversmessage.attach(MIMEText(body, 'plain', 'utf-8'))att1 = MIMEText(open(self.projectDir + '/Log/err_mail.log', 'rb').read(), 'base64', 'utf-8')att1["Content-Type"] = 'application/octet-stream'# 这里的filename可以任意写，写什么名字，邮件中显示什么名字att1["Content-Disposition"] = 'attachment; filename="log.txt"'message.attach(att1)try:smtp_obj = smtplib.SMTP()smtp_obj.connect(mail_host, 25)  # 25 为 SMTP 端口号smtp_obj.login(mail_user, mail_pass)smtp_obj.sendmail(sender, receivers, message.as_string())print('邮件发送成功，请查收')self.log.info('邮件发送成功')except Exception as e:print(e)print('无法发送邮件')self.log.error('无法发送邮件')if __name__ == '__main__':mail = SendMail()mail.send_mail()

说明：

• 设置邮件内容，邮件格式等

连接数据库：

"""
连接数据库
"""from Common.Utils import Utils
from Conf import Config
from Common import Log
import pymysql
import tracebackclass Database(Utils):def __init__(self):super().__init__()self.config = Config.Read_config()self.log = Log.MyLog()self.host = self.config.get_database('host')self.user = self.config.get_database('user')self.password = self.config.get_database('password')self.db = self.config.get_database('db')self.port = int(self.config.get_database('port'))def connect_db(self):"""连接数据库:return:"""cur = ''try:conn = pymysql.connect(host=self.host,user=self.user,port=self.port,password=self.password,db=self.db,charset='utf8')cur = conn.cursor()except Exception:print('Fail to connect the database')print(traceback.format_exc())self.log.error(traceback.format_exc())return curdef fetch_data_from_db(self, table_name, param, condition):"""查询数据库并返回所有结果记录:param table_name: 表名称:param param: 需要获取的字段名称:param condition: 需填写完整的查询语句，可为空字符；注意填写的参数类型（str， int）准确填写:return:"""cur = self.connect_db()list_dic_data = []# 执行查询语句sql_s = 'select ' + param + ' from ' + table_name + ' ' + conditionprint(sql_s)# 返回记录数量res = cur.execute(sql_s)print(res)# 获取表字段名和表数据table_param = [item[0] for item in cur.description]all_records = cur.fetchall()# 断开数据库连接cur.close()# 将数据组成一个dictfor i in all_records:dic_data = {}for j in range(len(table_param)):dic_data[table_param[j]] = i[j]list_dic_data.append(dic_data)return list_dic_dataif __name__ == '__main__':d = Database()c = d.connect_db()sql_sentence = 'select * from t_user 'results = c.execute(sql_sentence)print(results)print(c.description)print(d.fetch_data_from_db('t_user','*','where mobile = \'11111111111\''))

说明：

• 连接数据库
• 封装方法：查询数据库的数据

设置运行用例前和运行用例后的操作

创建 environment.properties 文件：
r
创建 Properties.py 文件：

"""
编辑/读取/添加参数到 环境配置文件（allure报告中的环境参数设定）"""import re
import os
import tempfile
from Common.Log import MyLogclass Properties:def __init__(self, file_name):self.file_name = file_nameself.properties = {}try:with open(self.file_name, 'r') as f:for line in f:line = line.strip()if line.find('=') > 0 and not line.startswith('#'):strs = line.split('=')self.properties[strs[0].strip()] = strs[1].strip()except Exception as e:MyLog.error('something is wrong!!!! when read the file ' + file_name)raise edef has_key(self, key):return key in self.propertiesdef get(self, key, default_value=''):if key in self.properties:return self.properties[key]return default_valuedef put(self, key, value):self.properties[key] = valuereplace_property(self.file_name, key + '=.*', key + '=' + value, True)def replace_property(file_name, from_regex, to_str, append_on_not_exists=True):tmpfile = tempfile.TemporaryFile()if os.path.exists(file_name):with open(file_name, 'r') as r_open:pattern = re.compile(r'' + from_regex)found = Nonefor line in r_open:if pattern.search(line) and not line.strip().startswith('#'):found = Trueline = re.sub(from_regex, to_str, line)tmpfile.write(line.encode())if not found and append_on_not_exists:tmpfile.write(('\n' + to_str).encode())tmpfile.seek(0)content = tmpfile.read()if os.path.exists(file_name):os.remove(file_name)with open(file_name, 'wb') as f_w:f_w.write(content)tmpfile.close()else:print("file %s not found" % file_name)if __name__ == "__main__":file_path = '../environment.properties'props = Properties(file_path)  # 读取文件props.put('www', '111111')  # 修改/添加key=valueprint(props.get('Author'))  # 根据key读取value

创建 conftest.py 文件（该文件的名称是特定的，不可随便更改，pytest会自动去检测这个文件）

from Common.Utils import *
import os
import pytest
from Common.Email import SendMail
from Common.Log import MyLog
from Common.Properties import Properties
from Conf.Config import Read_config
import allure
"""
该文件名称不可修改！
目前单独运行某个case时，会先调用以下方法
"""current_path = os.path.abspath(os.path.dirname(__file__))
# rootPath = os.path.split(current_path)[0]fail_case_txt_path = 'fail_cases.txt'
err_mail_log_path = current_path + '/Log/err_mail.log'properties = Properties(current_path + '/environment.properties')
config = Read_config()@pytest.fixture(scope='package', autouse=True)
def resource():"""这是pytest的装饰器package是module级别的，执行多个.py文件时，只执行一次该方法yield前面的内容 是运行前执行的，相当于testNG的setupyield后面的内容，是运行后执行的，相当于testNG的teardown:return:"""# 清除上次运行的失败case的记录# 清除上此运行的错误日志（仅用于发送个邮箱的日志）Utils.clean_file_content(fail_case_txt_path)Utils.clean_file_content(err_mail_log_path)current_env = config.get_global('current_environment')properties.put('Environment', current_env)properties.put('Endpoint', config.get_conf(current_env, 'endpoint'))properties.put('Author', config.get_conf(current_env, 'tester'))properties.put('Version', config.get_conf(current_env, 'version_code'))@pytest.fixture(scope='package', autouse=True)
def send_mail():yieldif os.path.getsize(err_mail_log_path) != 0:try:mail = SendMail()mail.send_mail()except Exception:MyLog.error('发送邮件失败，请检查邮件配置')raise@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_makereport(item):"""pytest_runtest_makereport 钩子方法when=’setup’ 返回setup 的执行结果when=’call’ 返回call 的执行结果when=’teardown’返回teardown 的执行结果:param item: 运行的case的对象:return:"""outcome = yieldrep = outcome.get_result()setattr(item, "rep_" + rep.when, rep)if rep.when == 'call':if rep.failed:# 如果case运行失败，则获取case的名称并写入一个txt文件中Log.MyLog.error('test case: ' + item.name + '--------' + rep.outcome)with open(fail_case_txt_path, "a+") as f:f.write(item.name + '\n')print('\n%s' % item.name + 'is ' + rep.outcome)

说明：

• 设置运行用例前的操作
• 设置运行用例后的操作
• 该段代码主要是：1. 运行用例前，设置测试报告的一些内容；2. 运行用例时，如果有失败用例，写入一个指定文件中；3. 若有失败用例，则有邮件通知