本文共 12289 字，大约阅读时间需要 40 分钟。

pytest+数据驱动模型+allure测试框架实践指南

目录

• api：存储测试接口
• conftest.py：设置前置操作
  • 前置操作：1. 获取token并传入headers，2. 获取命令行参数给到环境变量，指定运行环境
• common：存储封装的公共方法
  • connect_mysql.py：连接数据库
  • http_requests.py：封装自己的请求方法
  • logger.py：封装输出日志文件
  • read_yaml.py：读取yaml文件测试用例数据
  • read_save_data.py：读取保存的数据文件
• case：存放所有的测试用例
• data：存放测试需要的数据
• save_data：存放接口返回数据、接口下载文件
• test_data：存放测试用例依赖数据
• upload_data：存放上传接口文件
• logs：存放输出的日志文件
• report：存放测试输出报告
• getpathinfo.py：封装项目测试路径
• pytest.ini：配置文件
• requirements.txt：本地python包（pip install -r requirements.txt 安装项目中所有python包）
• run_main.py：项目运行文件

结构设计

代码展示

api/api_service.py（需要测试的一类接口）

'''服务相关接口Create time: 2020/12/3Developer: 叶修'''import osfrom common.http_requests import HttpRequestsclass Api_Auth_Service(object):    def __init__(self):        self.headers = HttpRequests().headers    def api_home_service_list(self):        # 首页服务列表        # url = "http://192.168.2.199:9092/v1/auth/auth_service/findAuthService"        url = os.environ["host"] + "/v1/auth/auth_service/findAuthService"        response = HttpRequests().get(url, headers=self.headers, verify=False)        return response    def get_service_id(self):        # 获取银行卡三要素认证服务id        # url = "http://192.168.2.199:9092/v1/auth/auth_service/findAuthService"        # url = os.environ["host"] + "/v1/auth/auth_service/findAuthService"        response = HttpRequests().get(url, headers=self.headers)        service_id = response.json()['data'][0]['service_list'][1]['id']        return service_id    def api_service_info(self, serviceId='0b6cf45bec757afa7ee7209d30012ce1', developerId=''):        # 服务详情        body = {            "serviceId": serviceId,            "developerId": developerId        }        # url = "http://192.168.2.199:9092/v1/auth/auth_service/findServiceDetail"        # url = os.environ["host"] + "/v1/auth/auth_service/findServiceDetail"        response = HttpRequests().get(url, headers=self.headers, params=body, verify=False)        return response    def api_add_or_update_service(self, api_param_req, api_param_res, description, error_code, icon, id, interface_remarks, name, product_info, request_method, sample_code, sort, type, url):        # 服务添加或者更新        body = {            "api_param_req": api_param_req,            "api_param_res": api_param_res,            "description": description,            "error_code": error_code,            "icon": icon,            "id": id,            "interface_remarks": interface_remarks,            "name": name,            "product_info": product_info,            "request_method": request_method,            "sample_code": sample_code,            "sort": sort,            "type": type,            "url": url        }        # url = "http://192.168.2.199:9092/v1/auth/auth_service/insertOrUpdateService"        url = os.environ["host"] + "/v1/auth/auth_service/insertOrUpdateService"        response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)        return response    def api_add_service_price(self, id, max_number, money, service_id, small_number):        # 服务价格添加        body = {            "id": id,            "max_number": max_number,            "money": money,            "service_id": service_id,            "small_number": small_number        }        # url = "http://192.168.2.199:9092/v1/auth/auth_service/insertServicePrice"        url = os.environ["host"] + "/v1/auth/auth_service/insertServicePrice"        response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)        return response    def api_apply_service(self, developer_id, service_id):        # 申请服务        body = {            "developer_id": developer_id,            "service_id": service_id        }        # url = "http://192.168.2.199:9092/v1/auth/auth_service/applyService"        url = os.environ["host"] + "/v1/auth/auth_service/applyService"        response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)        return response

case/test_service_info.py（上面接口某一测试用例）

'''服务详情Create time: 2020/12/3Developer: 叶修'''import sysimport allureimport pytestfrom common.logger import Logfrom common.read_yaml import ReadYamlfrom api.api_auth_service.api_auth_service import Api_Auth_Servicetestdata = ReadYaml("auth_service.yml").get_yaml_data()@allure.feature('服务详情')class Test_Service_Info(object):    log = Log()    @pytest.mark.process    @pytest.mark.parametrize('serviceId,developerId,expect', testdata['service_info'], ids=['服务详情'])    def test_service_info(self, serviceId, developerId, expect):        self.log.info('%s{%s}' % ((sys._getframe().f_code.co_name, '------服务详情接口-----')))        with allure.step('获取服务id'):            serviceId = Api_Auth_Service().get_service_id()        with allure.step('服务详情'):            msg = Api_Auth_Service().api_service_info(serviceId, developerId)        self.log.info('%s:%s' % ((sys._getframe().f_code.co_name, '获取请求结果：%s' % msg.json())))        # 断言        assert msg.json()["result_message"] == expect['result_message']        assert msg.json()['result_code'] == expect['result_code']        assert 'url' in msg.json()['data']

conftest.py

'''配置信息Create time: 2020/12/3Developer: 叶修'''import osimport pytestfrom api.get_token import Get_Tokenfrom common.http_requests import HttpRequests@pytest.fixture(scope="session")def get_token():    '''前置操作获取token并传入headers'''    Get_Token().get_token()    if not HttpRequests().params.get("access_token", ""):        pytest.skip("未获取token跳过用例")    yield HttpRequests().req    HttpRequests().req.close()def pytest_addoption(parser):    parser.addoption(        "--cmdhost", action="store", default="http://192.168.1.54:32099",        help="my option: type1 or type2"    )@pytest.fixture(scope="session", autouse=True)def host(request):    '''获取命令行参数'''    # 获取命令行参数给到环境变量    # pytest --cmdhost 运行指定环境    os.environ["host"] = request.config.getoption("--cmdhost")    print("当前用例运行测试环境:%s" % os.environ["host"])

common/connect_mysql.py

'''配置连接数据库Create time: 2020/12/3Developer: 叶修'''import pymysqldbinfo = {    "host": "******",    "user": "root",    "password": "******",    "port": 31855}class DbConnect():    def __init__(self, db_conf, database=""):        self.db_conf = db_conf        # 打开数据库        self.db = pymysql.connect(database=database, cursorclass=pymysql.cursors.DictCursor, **db_conf)        # 使用cursor()方式获取操作游标        self.cursor = self.db.cursor()    def select(self, sql):        # sql查询        self.cursor.execute(sql)        results = self.cursor.fetchall()        return results    def execute(self, sql):        # sql 删除 提示 修改        try:            self.cursor.execute(sql)            self.db.commit()        except:            # 发生错误时回滚            self.db.rollback()    def close(self):        self.db.close()def select_sql(select_sql):    '''查询数据库'''    db = DbConnect(dbinfo, database='auth_platform')    result = db.select(select_sql)    db.close()    return resultdef execute_sql(sql):    '''执行SQL'''    db = DbConnect(dbinfo, database='auth_platform')    db.execute(sql)    db.close()

common/http_requests.py

'''封装自己的请求类型Create time: 2020/12/3Developer: 叶修'''import requestsimport os# 定义一个HttpRequests的类class HttpRequests(object):    req = requests.session()    # 定义公共请求头    headers = {        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.125 Safari/537.36',        'cookie': ''    }    params = {        'access_token': ''    }    # 封装自己的get请求,获取资源    def get(self, url='', params='', data='', headers=None, cookies=None, stream=None, verify=None):        response = self.req.get(url, params=params, data=data, headers=headers, cookies=cookies, stream=stream, verify=verify)        return response    # 封装自己的post方法，创建资源    def post(self, url='', params='', data='', json='', headers=None, cookies=None, stream=None, verify=None):        response = self.req.post(url, params=params, data=data, json=json, headers=headers, cookies=cookies, stream=stream, verify=verify)        return response    # 封装自己的put方法，更新资源    def put(self, url='', params='', data='', headers=None, cookies=None, verify=None):        response = self.req.put(url, params=params, data=data, headers=headers, cookies=cookies, verify=verify)        return response    # 封装自己的delete方法，删除资源    def delete(self, url='', params='', data='', headers=None, cookies=None, verify=None):        response = self.req.delete(url, params=params, data=data, headers=headers, cookies=cookies, verify=verify)        return response

common/logger.py

'''封装输出日志文件Create time: 2020/12/3Developer: 叶修'''import loggingimport timeimport osimport getpathinfopath = getpathinfo.get_path()log_path = os.path.join(path, 'logs')# 如果不存在这个logs文件夹，就自动创建一个if not os.path.exists(log_path):    os.mkdir(log_path)class Log():    def __init__(self):        # 文件的命名        self.logname = os.path.join(log_path, '%s.log' % time.strftime('%Y_%m_%d'))        self.logger = logging.getLogger()        self.logger.setLevel(logging.DEBUG)        # 日志输出格式        self.formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')    def __console(self, level, message):        # 创建一个fileHander，用于写入本地        fh = logging.FileHandler(self.logname, 'a', encoding='utf-8')        fh.setLevel(logging.DEBUG)        fh.setFormatter(self.formatter)        self.logger.addHandler(fh)        # 创建一个StreamHandler，用于输入到控制台        ch = logging.StreamHandler()        ch.setLevel(logging.DEBUG)        ch.setFormatter(self.formatter)        self.logger.addHandler(ch)        if level == 'info':            self.logger.info(message)        elif level == 'debug':            self.logger.debug(message)        elif level == 'warning':            self.logger.warning(message)        elif level == 'error':            self.logger.error(message)        # 避免日志重复        self.logger.removeHandler(fh)        self.logger.removeHandler(ch)        # 关闭打开文件        fh.close()    def debug(self, message):        self.__console('debug', message)    def info(self, message):        self.__console('info', message)    def warning(self, message):        self.__console('warning', message)    def error(self, message):        self.__console('error', message)

common/read_save_data.py

'''读取保存数据Create time: 2020/12/8Developer: 叶修'''import osimport yamlimport getpathinfoclass Read_Save_Date():    def __init__(self):        path = getpathinfo.get_path()        self.head_img_path = os.path.join(path, 'data', 'save_data') + "/" + 'head_img_path.txt'        self.order_id_path = os.path.join(path, 'data', 'save_data') + "/" + 'order_id.txt'    def get_head_img_path(self):        with open(self.head_img_path, "r", encoding="utf-8") as f:            return f.read()    def get_order_id(self):        with open(self.order_id_path, "r", encoding="utf-8") as f:            return f.read()

common/read_yaml.py

'''读取yml文件测试数据Create time: 2020/12/3Developer: 叶修'''import osimport yamlimport getpathinfoclass ReadYaml():    def __init__(self, filename):        path = getpathinfo.get_path()        self.filepath = os.path.join(path, 'data', 'test_data') + "/" + filename    def get_yaml_data(self):        with open(self.filepath, "r", encoding="utf-8") as f:            return yaml.load(f, Loader=yaml.FullLoader)

getpathinfo.py

'''配置文件路径Create time: 2020/12/3Developer: 叶修'''import osdef get_path():    # 获取当前路径    curpath = os.path.dirname(os.path.realpath(__file__))    return curpathif __name__ == '__main__':    print('测试路径是否OK,路径为：', get_path())

requirements.txt

allure-pytest==2.8.18allure-python-commons==2.8.18BeautifulReport==0.1.3beautifulsoup4==4.9.3ddt==1.4.1Faker==4.18.0Flask==1.1.1httpie==1.0.3httplib2==0.9.2HttpRunner==1.5.8py==1.9.0PyMySQL==0.10.1pytest==6.1.1pytest-base-url==1.4.2pytest-cov==2.10.1pytest-forked==1.3.0pytest-html==2.1.1pytest-instafail==0.4.2pytest-metadata==1.10.0pytest-mock==3.3.1pywin32==228PyYAML==5.3.1requests==2.22.0requests-oauthlib==1.3.0requests-toolbelt==0.9.1

run_main.py

'''运行主流程测试用例Create time: 2020/11/5Developer: 叶修'''import osimport pytestif __name__ == '__main__':    pytest.main(['-m', 'process', '-s', '--alluredir', 'report/tmp'])# -m运行mark标记文件# -s不捕获异常# --alluredir report/tmp 为存放报告的源文件目录os.system('allure generate report/tmp -o report/html --clean')

转载地址：http://ejafk.baihongyu.com/