• Pytest接口自动化测试实战演练

    结合单元测试框架pytest+数据驱动模型+allure

    目录

    api： 存储测试接口

    conftest.py :设置前置操作

                 目前前置操作：1、获取token并传入headers，2、获取命令行参数给到环境变量,指定运行环境

    commmon：存储封装的公共方法

             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.int :配置文件

    requirement.txt: 本地python包（pip install -r requirements.txt 安装项目中所有python包）

    run_main.py: 项目运行文件

    结构设计

    1.每一个接口用例组合在一个测试类里面生成一个py文件

    2.将每个用例调用的接口封装在一个测试类里面生成一个py文件

    3.将测试数据存放在yml文件中通过parametrize进行参数化，实现数据驱动

    4.通过allure生成测试报告

    代码展示

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

    1. '''
    2. Code description:服务相关接口
    3. Create time: 2020/12/3
    4. Developer: 叶修
    5. '''
    6. import os
    7. from common.http_requests import HttpRequests
    8.  
    9.  
    10. class Api_Auth_Service(object):
    11.  
    12.     def __init__(self):
    13.         self.headers = HttpRequests().headers
    14.  
    15.     def api_home_service_list(self):
    16.         # 首页服务列表
    17.         # url = "http://192.168.2.199:9092/v1/auth/auth_service/findAuthService"
    18.         url = os.environ["host"] + "/v1/auth/auth_service/findAuthService"  # 读取conftest.py文件地址进行拼接
    19.         response = HttpRequests().get(url, headers=self.headers, verify=False)
    20.         # print(response.json())
    21.         return response
    22.  
    23.     def get_service_id(self):
    24.         #获取银行卡三要素认证服务id
    25.         url = "http://192.168.2.199:9092/v1/auth/auth_service/findAuthService"
    26.         #url = os.environ["host"] + "/v1/auth/auth_service/findAuthService"  # 读取conftest.py文件地址进行拼接
    27.         response = HttpRequests().get(url,headers=self.headers)
    28.         #print(response.json()['data'][0]['service_list'][0]['id'])
    29.         service_id = response.json()['data'][0]['service_list'][1]['id']
    30.         return service_id
    31.  
    32.     def api_service_info(self,serviceId='0b6cf45bec757afa7ee7209d30012ce1',developerId=''):
    33.         #服务详情
    34.         body = {
    35.             "serviceId" :serviceId,
    36.             "developerId":developerId
    37.         }
    38.         url = "http://192.168.2.199:9092/v1/auth/auth_service/findServiceDetail"
    39.         #url = os.environ["host"] + "/v1/auth/auth_service/findServiceDetail"#读取conftest.py文件地址进行拼接
    40.         response = HttpRequests().get(url,headers=self.headers,params = body,verify=False)
    41.         #print(response.json())
    42.         return response
    43.  
    44.     def api_add_or_update_service(self,api_param_req,api_param_res,description,error_code,icon,id,interface_remarks,
    45.                               name,product_info,request_method,sample_code,sort,type,url):
    46.         #服务添加或者更新
    47.         body={
    48.                 "api_param_req": api_param_req,
    49.                 "api_param_res": api_param_res,
    50.                 "description": description,
    51.                 "error_code": error_code,
    52.                 "icon": icon,
    53.                 "id": id,
    54.                 "interface_remarks": interface_remarks,
    55.                 "name": name,
    56.                 "product_info": product_info,
    57.                 "request_method": request_method,
    58.                 "sample_code": sample_code,
    59.                 "sort": sort,
    60.                 "type": type,
    61.                 "url": url,
    62.         }
    63.         #url = "http://192.168.2.199:9092/v1/auth/auth_service/insertOrUpdateService"
    64.         url = os.environ["host"] + "/v1/auth/auth_service/insertOrUpdateService"  # 读取conftest.py文件地址进行拼接
    65.         response = HttpRequests().post(url,json=body,headers=self.headers,verify=False)
    66.         return response
    67.  
    68.     def api_add_service_price(self,id,max_number,money,service_id,small_number):
    69.         #服务价格添加
    70.         body = {
    71.             "id": id,
    72.             "max_number": max_number,
    73.             "money": money,
    74.             "service_id": service_id,
    75.             "small_number": small_number
    76.         }
    77.         # url = "http://192.168.2.199:9092/v1/auth/auth_service/insertServicePrice"
    78.         url = os.environ["host"] + "/v1/auth/auth_service/insertServicePrice"  # 读取conftest.py文件地址进行拼接
    79.         response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)
    80.         return response
    81.  
    82.     def api_apply_service(self,developer_id,service_id):
    83.         #申请服务
    84.         body ={
    85.             "developer_id": developer_id,
    86.             "service_id": service_id
    87.         }
    88.         # url = "http://192.168.2.199:9092/v1/auth/auth_service/applyService"
    89.         url = os.environ["host"] + "/v1/auth/auth_service/applyService"  # 读取conftest.py文件地址进行拼接
    90.         response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)
    91.         return response
    92.  
    93.  
    94. if __name__ == '__main__':
    95.     #Auth_Service().api_home_service_list()
    96.     Api_Auth_Service().get_service_id()
    97.     #Auth_Service().api_service_info()

     api/get_token.py#获取登录token

    1. '''
    2. Code description：获取token
    3. Create time：2020-12-03
    4. Developer：叶修
    5. '''
    6. import os
    7.  
    8. import urllib3
    9. from common.http_requests import HttpRequests
    10.  
    11.  
    12. class Get_Token(object):
    13.  
    14.     def get_token(self,account='****',password='****'):
    15.         #url = "http://192.168.2.199:9092/v1/auth/developer/accountLogin"
    16.         url = os.environ["host"]+"/v1/auth/developer/accountLogin"
    17.         body = {
    18.             "account": account,
    19.             "password": password,
    20.         }
    21.         urllib3.disable_warnings()
    22.         r = HttpRequests().post(url, json=body,verify=False)
    23.         #print(r.json())
    24.         token = r.json()['data']['token']
    25.         params = {
    26.             "access_token": token
    27.         }
    28.         HttpRequests().params.update(params)#更新token到session
    29.         return token
    30.  
    31. if __name__ == '__main__':
    32.     print(Get_Token().get_token())

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

    1. '''
    2. Code description: 服务详情
    3. Create time: 2020/12/3
    4. Developer: 叶修
    5. '''
    6. import sys
    7. import allure
    8. import pytest
    9. from common.logger import Log
    10. from common.read_yaml import ReadYaml
    11. from api.api_auth_service.api_auth_service import Api_Auth_Service
    12.  
    13. testdata = ReadYaml("auth_service.yml").get_yaml_data()  # 读取数据
    14.  
    15.  
    16. @allure.feature('服务详情')
    17. class Test_Service_Info(object):
    18.     log = Log()
    19.  
    20.     @pytest.mark.process
    21.     @pytest.mark.parametrize('serviceId,developerId,expect', testdata['service_info'],
    22.                              ids=['服务详情'])
    23.     def test_service_info(self,serviceId,developerId,expect):
    24.         self.log.info('%s{%s}' % ((sys._getframe().f_code.co_name,'------服务详情接口-----')))
    25.         with allure.step('获取服务id'):
    26.             serviceId = Api_Auth_Service().get_service_id()
    27.         with allure.step('服务详情'):
    28.             msg = Api_Auth_Service().api_service_info(serviceId,developerId)
    29.         self.log.info('%s:%s' % ((sys._getframe().f_code.co_name, '获取请求结果：%s' % msg.json())))
    30.         # 断言
    31.         assert msg.json()["result_message"] == expect['result_message']
    32.         assert msg.json()['result_code'] == expect['result_code']
    33.         assert 'url' in msg.json()['data']

     conftest.py

    1. '''
    2. Code description:配置信息
    3. Create time: 2020/12/3
    4. Developer: 叶修
    5. '''
    6. import os
    7. import pytest
    8. from api.get_token import Get_Token
    9. from common.http_requests import HttpRequests
    10.  
    11.  
    12. @pytest.fixture(scope="session")
    13. def get_token():
    14.     '''前置操作获取token并传入headers'''
    15.     Get_Token().get_token()
    16.     if not HttpRequests().params.get("access_token", ""):#没有get到token，跳出用例
    17.         pytest.skip("未获取token跳过用例")
    18.     yield HttpRequests().req
    19.     HttpRequests().req.close()
    20.  
    21. def pytest_addoption(parser):
    22.     parser.addoption(
    23.         "--cmdhost", action="store", default="http://192.168.1.54:32099",
    24.         help="my option: type1 or type2"
    25.     )
    26. @pytest.fixture(scope="session",autouse=True)
    27. def host(request):
    28.     '''获取命令行参数'''
    29.     #获取命令行参数给到环境变量
    30.     #pytest --cmdhost 运行指定环境
    31.     os.environ["host"] = request.config.getoption("--cmdhost")
    32.     print("当前用例运行测试环境:%s" % os.environ["host"])

     common/connect_mysql.py

    1. '''
    2. Code description: 配置连接数据库
    3. Create time: 2020/12/3
    4. Developer: 叶修
    5. '''
    6. import pymysql
    7.  
    8. dbinfo = {
    9.     "host":"******",
    10.     "user":"root",
    11.     "password":"******",
    12.     "port":31855
    13. }
    14. class DbConnect():
    15.     def __init__(self,db_conf,database=""):
    16.         self.db_conf = db_conf
    17.         #打开数据库
    18.         self.db = pymysql.connect(database = database,
    19.                                   cursorclass = pymysql.cursors.DictCursor,
    20.                                   **db_conf)
    21.         #使用cursor()方式获取操作游标
    22.         self.cursor = self.db.cursor()
    23.  
    24.     def select(self,sql):
    25.         #sql查询
    26.         self.cursor.execute(sql)#执行sql
    27.         results = self.cursor.fetchall()
    28.         return results
    29.  
    30.     def execute(self,sql):
    31.         #sql 删除 提示 修改
    32.         try:
    33.             self.cursor.execute(sql)#执行sql
    34.             self.db.commit()#提交修改
    35.         except:
    36.             #发生错误时回滚
    37.             self.db.rollback()
    38.  
    39.     def close(self):
    40.         self.db.close()#关闭连接
    41.  
    42. def select_sql(select_sql):
    43.     '''查询数据库'''
    44.     db = DbConnect(dbinfo,database='auth_platform')
    45.     result = db.select(select_sql)
    46.     db.close()
    47.     return result
    48.  
    49. def execute_sql(sql):
    50.     '''执行SQL'''
    51.     db = DbConnect(dbinfo,database='auth_platform')
    52.     db.execute(sql)
    53.     db.close()
    54.  
    55. if __name__ == '__main__':
    56.     sql = "SELECT * FROM auth_platform.auth_service where name='四要素认证'"
    57.     sel = select_sql(sql)[0]['name']
    58.     print(sel)

     common/http_requests.py

    1. '''
    2. Code description: 封装自己的请求类型
    3. Create time: 2020/12/3
    4. Developer: 叶修
    5. '''
    6.  
    7. import requests
    8.  
    9.  
    10. # 定义一个HttpRequests的类
    11. class HttpRequests(object):
    12.  
    13.     req = requests.session()#定义session会话
    14.     # 定义公共请求头
    15.     headers = {
    16.         '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',
    17.         'cookie':''
    18.     }
    19.     params = {
    20.         'access_token':''
    21.     }
    22.     # 封装自己的get请求,获取资源
    23.     def get(self, url='', params='', data='', headers=None, cookies=None,stream=None,verify=None):
    24.         response = self.req.get(url,params=params,data=data,headers=headers,cookies=cookies,stream=stream,verify=verify)
    25.         return response
    26.  
    27.     # 封装自己的post方法，创建资源
    28.     def post(self, url='', params='',data='', json='', headers=None, cookies=None,stream=None,verify=None):
    29.         response = self.req.post(url,params=params,data=data,json=json,headers=headers,cookies=cookies,stream=stream,verify=verify)
    30.         return response
    31.  
    32.     # 封装自己的put方法，更新资源
    33.     def put(self, url='', params='', data='', headers=None, cookies=None,verify=None):
    34.         response = self.req.put(url, params=params, data=data, headers=headers, cookies=cookies,verify=verify)
    35.         return response
    36.  
    37.     # 封装自己的delete方法，删除资源
    38.     def delete(self, url='', params='', data='', headers=None, cookies=None,verify=None):
    39.         response = self.req.delete(url, params=params, data=data, headers=headers, cookies=cookies,verify=verify)
    40.         return response

     common/logger.py

    1. '''
    2. Code description: 封装输出日志文件
    3. Create time: 2020/12/3
    4. Developer: 叶修
    5. '''
    6. import logging,time
    7. import os
    8. import getpathinfo
    9.  
    10. path = getpathinfo.get_path()#获取本地路径
    11. log_path = os.path.join(path,'logs')# log_path是存放日志的路径
    12. # 如果不存在这个logs文件夹，就自动创建一个
    13. if not os.path.exists(log_path):os.mkdir(log_path)
    14.  
    15. class Log():
    16.     def __init__(self):
    17.         #文件的命名
    18.         self.logname = os.path.join(log_path,'%s.log'%time.strftime('%Y_%m_%d'))
    19.         self.logger = logging.getLogger()
    20.         self.logger.setLevel(logging.DEBUG)
    21.         #日志输出格式
    22.         self.formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    23.  
    24.     def __console(self,level,message):
    25.         #创建一个fileHander，用于写入本地
    26.         fh = logging.FileHandler(self.logname,'a',encoding='utf-8')
    27.         fh.setLevel(logging.DEBUG)
    28.         fh.setFormatter(self.formatter)
    29.         self.logger.addHandler(fh)
    30.  
    31.         #创建一个StreamHandler,用于输入到控制台
    32.         ch = logging.StreamHandler()
    33.         ch.setLevel(logging.DEBUG)
    34.         ch.setFormatter(self.formatter)
    35.         self.logger.addHandler(ch)
    36.  
    37.         if level == 'info':
    38.             self.logger.info(message)
    39.         elif level == 'debug':
    40.             self.logger.debug(message)
    41.         elif level == 'warning':
    42.             self.logger.warning(message)
    43.         elif level == 'error':
    44.             self.logger.error(message)
    45.         #避免日志重复
    46.         self.logger.removeHandler(fh)
    47.         self.logger.removeHandler(ch)
    48.         #关闭打开文件
    49.         fh.close()
    50.  
    51.     def debug(self,message):
    52.         self.__console('debug',message)
    53.  
    54.     def info(self,message):
    55.         self.__console('info',message)
    56.  
    57.     def warning(self,message):
    58.         self.__console('warning',message)
    59.  
    60.     def error(self,message):
    61.         self.__console('error',message)
    62.  
    63. if __name__ == '__main__':
    64.     log = Log()
    65.     log.info('测试')
    66.     log.debug('测试')
    67.     log.warning('测试')
    68.     log.error('测试')

     common/read_save_data.py

    1. '''
    2. Code description: 读取保存数据
    3. Create time: 2020/12/8
    4. Developer: 叶修
    5. '''
    6. import os
    7. import yaml
    8. import getpathinfo
    9. class Read_Save_Date():
    10.     def __init__(self):
    11.         path = getpathinfo.get_path()#获取本地路径
    12.         self.head_img_path = os.path.join(path,'data','save_data')+"/"+'head_img_path.txt'# head_img_path文件地址
    13.         self.order_id_path = os.path.join(path, 'data', 'save_data') + "/" + 'order_id.txt'  # order_id.txt文件地址
    14.  
    15.     def get_head_img_path(self):
    16.         # 获取head_img_path
    17.         with open(self.head_img_path, "r", encoding="utf-8")as f:
    18.             return f.read()
    19.  
    20.     def get_order_id(self):
    21.         # 获取order_id
    22.         with open(self.order_id_path, "r", encoding="utf-8")as f:
    23.             return f.read()
    24.  
    25. if __name__ == '__main__':
    26.     print(Read_Save_Date().get_head_img_path())
    27.     print(Read_Save_Date().get_order_id())

     common/read_yaml.py

    1. '''
    2. Code description: 读取yml文件测试数据
    3. Create time: 2020/12/3
    4. Developer: 叶修
    5. '''
    6. import os
    7. import yaml
    8. import getpathinfo
    9. class ReadYaml():
    10.     def __init__(self,filename):
    11.         path = getpathinfo.get_path()#获取本地路径
    12.         self.filepath = os.path.join(path,'data','test_data')+"/"+filename#拼接定位到data文件夹
    13.  
    14.     def get_yaml_data(self):
    15.         with open(self.filepath, "r", encoding="utf-8")as f:
    16.             # 调用load方法加载文件流
    17.             return yaml.load(f,Loader=yaml.FullLoader)
    18.  
    19. if __name__ == '__main__':
    20.     data = ReadYaml("auth_service.yml").get_yaml_data()['add_or_update_service']
    21.     print(data)

    data/

     data/test_data/auth_service.yml

    1. home_service_list:
    2.   - [{'result_code': '0', 'result_message': '处理成功'}]
    3. service_info:
    4.   - ['','',{'result_code': '0', 'result_message': '处理成功'}]
    5. add_or_update_service:
    6.   - [['1','1','1','1','1','123456','1','测试','1','1','1','1','1','1'],{'result_code': '0', 'result_message': '处理成功'}]
    7. add_service_price:
    8.   - ['123456789','10','0','','0',{'result_code': '0', 'result_message': '处理成功'}]
    9. apply_service:
    10.   - ['','',{'result_code': '0', 'result_message': '处理成功'}]

     logs/

     report/

     getpathinfo.py

    1. '''
    2. Code description:配置文件路径
    3. Create time: 2020/12/3
    4. Developer: 叶修
    5. '''
    6. import os
    7.  
    8. def get_path():
    9.     # 获取当前路径
    10.     curpath = os.path.dirname(os.path.realpath(__file__))
    11.     return curpath
    12.  
    13. if __name__ == '__main__':# 执行该文件，测试下是否OK
    14.     print('测试路径是否OK,路径为：', get_path())

    pytest.ini

    1. #pytest.ini
    2. [pytest]
    3. markers = process
    4. addopts = -p no:warnings
    5. #addopts = -v --reruns 1 --html=./report/report.html --self-contained-html
    6. #addopts = -v --reruns 1 --alluredir ./report/allure_raw
    7. #addopts = -v -s -p no:warnings --reruns  1 --pytest_report ./report/Pytest_Report.html

    requirements.txt

    1. allure-pytest==2.8.18
    2. allure-python-commons==2.8.18
    3. BeautifulReport==0.1.3
    4. beautifulsoup4==4.9.3
    5. ddt==1.4.1
    6. Faker==4.18.0
    7. Flask==1.1.1
    8. httpie==1.0.3
    9. httplib2==0.9.2
    10. HttpRunner==1.5.8
    11. py==1.9.0
    12. PyMySQL==0.10.1
    13. pytest==6.1.1
    14. pytest-base-url==1.4.2
    15. pytest-cov==2.10.1
    16. pytest-forked==1.3.0
    17. pytest-html==2.1.1
    18. pytest-instafail==0.4.2
    19. pytest-metadata==1.10.0
    20. pytest-mock==3.3.1
    21. pywin32==228
    22. PyYAML==5.3.1
    23. requests==2.22.0
    24. requests-oauthlib==1.3.0
    25. requests-toolbelt==0.9.1

    run_main.py

    1. '''
    2. Code description: 运行主流程测试用例
    3. Create time: 2020/11/5
    4. Developer: 叶修
    5. '''
    6. import os
    7. import pytest
    8. if __name__ == '__main__':
    9.     pytest.main(['-m','process', '-s','--alluredir', 'report/tmp'])#-m运行mark标记文件
    10.     os.system('allure generate report/tmp -o report/html --clean') # /report/tmp 为存放报告的源文件目录

    最后感谢每一个认真阅读我文章的人，礼尚往来总是要有的，这些资料，对于【软件测试】的朋友来说应该是最全面最完整的备战仓库，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走：

    这些资料，对于【软件测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴上万个测试工程师们走过最艰难的路程，希望也能帮助到你！ 

  • 相关阅读:
    打工毁一生，创业治百病，零经验零成本创业三个月收益10W+
    IDEA创建父子项目
    docker三剑客compose+machine+swarm小结
    微软外服工作札记②——聊聊微软的知识管理服务平台和一些编程风格
    java毕业生设计在线玩具租赁系统计算机源码+系统+mysql+调试部署+lw
    [附源码]Python计算机毕业设计SSM家政服务预约小程序（程序+LW)
    SQL必需掌握的100个重要知识点：检索数据
    C primer plus学习笔记 —— 5、指针
    内卷下的智能投影行业，未来何去何从？
    常见视频传输接口
  • 原文地址：https://blog.csdn.net/weixin_71807218/article/details/136680269