接口自动化测试框架(用例自动生成)

项目说明

  • 本框架是一套基于pytest+requests+Python3.7+yaml+Allure+Jenkins+docker而设计的数据驱动接口自动化测试框架,pytest 作为执行器,本框架无需你使用代码编写用例,那你可能会担心万一有接口之间相互依赖,或者说需要登入的token等之类的接口,该如何编写用例呢,在这里告诉你们本框架已经完美解决此问题,所有的一切将在yaml中进行!!本框架实现了在yaml中进行接口用例编写,接口依赖关联,接口断言(支持正则校验、json校验、全等校验、数据库校验等,支持组合多种不同的校验类型来验证),自定义测试用例运行顺序,还有很重要的一点,实现了类jmeter函数助手的功能,譬如生成MD5、SHA1、随机定长字符串、时间戳等,只需要你在yaml中使用特殊的写法$Function(arg)$,就能够使用这些函数啦,此外在测试执行过程中,还可以 对失败用例进行多次重试,其重试次数和重试时间间隔可自定义;而且可以根据实际需要扩展接口协议,目前已支持http接口和webservice接口

技术栈

  • requests
  • suds-py3
  • Allure
  • pytest
  • pytest-html
  • yaml
  • logging
  • Jenkins
  • docker
  • 函数助手

环境部署

  • 命令行窗口执行pip install -r requirements.txt 安装工程所依赖的库文件

  • 解压allure-commandline-2.12.1.zip到任意目录中

  • 打开\allure-2.12.1\bin文件夹,会看到allure.bat文件,将此路径添加到系统环境变量path下,这样cmd任意目录都能执行了

  • 在cmd下执行 allure --version ,返回版本信息,allure即安装成功

  • 进入 \Lib\site-packages\allure 下面的utils文件,修改成以下代码:

  • 点我免费领取全套软件测试(自动化测试)视频资料(备注“csdn000”)

  •   for suitable_name in suitable_names:
          # markers.append(item.get_marker(suitable_name))
          markers.append(item.get_closest_marker(suitable_name))
    

目的是解决pytest运行产生的以下错误:
_pytest.warning_types.RemovedInPytest4Warning: MarkInfo objects are deprecated as they contain merged marks which are hard to deal with correctly.

框架流程图与目录结构图及相关说明

1、框架流程图如下

2、代码目录结构图如下

目录结构说明

  • config ===========> 配置文件
  • common ===========> 公共方法封装,工具类等
  • pytest.ini ==========> pytest的主配置文件,可以改变pytest的默认行为,如运行方式,默认执行用例路径,用例收集规则,定义标记等
  • log ==========> 日志文件
  • report ==========> 测试报告
  • tests ===========> 待测试相关文件,比如测试用例和用例数据等
  • conftest.py ============> 存放测试执行的一些fixture配置,实现环境初始化、数据共享以及环境还原等
  • requirements.txt ============> 相关依赖包文件
  • Main.py =============> 测试用例总执行器
  • RunTest_windows.bat ============> 测试启动按钮

conftest.py配置说明

  • conftest.py文件名字是固定的,不可以做任何修改
  • 不需要import导入conftest.py,pytest用例会自动识别该文件,若conftest.py文件放在根目录下,那么conftest.py作用于整个目录,全局调用
  • 在不同的测试子目录也可以放conftest.py,其作用范围只在该层级以及以下目录生效
  • 所有目录内的测试文件运行前都会先执行该目录下所包含的conftest.py文件
  • conftest.py文件不能被其他文件导入

conftest.py与fixture结合

conftest文件实际应用中需要结合fixture来使用,如下

  • conftest中fixture的scope参数为session时,那么整个测试在执行前会只执行一次该fixture
  • conftest中fixture的scope参数为module时,那么每一个测试文件执行前都会执行一次conftest文件中的fixture
  • conftest中fixture的scope参数为class时,那么每一个测试文件中的测试类执行前都会执行一次conftest文件中的fixture
  • conftest中fixture的scope参数为function时,那么所有文件的测试用例执行前都会执行一次conftest文件中的fixture

conftest应用场景

  • 测试中需共用到的token
  • 测试中需共用到的测试用例数据
  • 测试中需共用到的配置信息
  • 结合 yield 语句,进行运行前环境的初始化和运行结束后环境的清理工作,yield前面的语句相当于unitest中的setup动作,yield后面的语句相当于unitest中的teardown动作,不管测试结果如何,yield后面的语句都会被执行。
  • 当fixture超出范围时(即fixture返回值后,仍有后续操作),通过使用yield语句而不是return,来将值返回(因为return后,说明该函数/方法已结束,return后续的代码不会被执行),如下:

@pytest.fixture(scope="module")
def smtpConnection():
    smtp_connection = smtplib.SMTP("smtp.gmail.com", 587, timeout=5)
    yield smtp_connection  # 返回 fixture 值smtp_connection
    print("teardown smtp")
    smtp_connection.close()

无论测试的异常状态如何,print和close()语句将在模块中的最后一个测试完成执行时执行。

  • 可以使用with语句无缝地使用yield语法(with语句会自动释放资源)

@pytest.fixture(scope="module")
def smtpConnection():
    with smtplib.SMTP("smtp.gmail.com", 587, timeout=5) as smtp_connection: 
        yield smtp_connection  # 返回smtp_connection对象值

测试结束后, 连接将关闭,因为当with语句结束时,smtp_connection对象会自动关闭。

关联详解

  • 公共关联池:意思就是你可以存储接口的响应值到参数池中,以便后续接口使用;同时也可以在测试执行前,制造一些公共关联值,存储到参数池中,供所有的接口使用;
  • 在yaml测试用例中,可通过填写关联键relevance提取响应字段的键值对到参数池中;当提取单个关联值时,关联键relevance的值为字符串,形如relevance: positon;当提取多个关联值时,关联键relevance的值为列表,同时也可提取响应信息中嵌套字典里的键值对;
  • 引用已经存储的关联值:在下个接口入参中使用形如 ${key}$ 的格式,即可提取参数池中的key对应的value,当然你必须保证关联池中已经存储过该key。

函数助手详解

  • 说明:函数助手是来自Jmeter的一个概念,有了它意味着你能在yaml测试用例或者其他配置文件中使用某些函数动态的生成某些数据,比如随机定长字符、随机定长整型数据、随机浮点型数据、时间戳(10位和13位)、md5加密、SHA1、SHA256、AES加解密等等,引用的格式为 $Function(arg)$
  • 目前支持的函数助手:
  • $MD5(arg)$ =========》 md5加密字符串,arg为待加密的字符串,可传入多个字符串拼接后加密,多个字符串之间用逗号隔开,形如$MD5(asd, we57hk)$
  • $SHA1(arg)$ ==========》 SHA1加密字符串,arg为待加密的字符串,可传入多个字符串拼接后加密,多个字符串之间用逗号隔开
  • $SHA256(arg)$ ==========》 SHA256加密字符串,arg为待加密的字符串,可传入多个字符串拼接后加密,多个字符串之间用逗号隔开
  • $DES(arg, key)$ ==========》 DES加密字符串,arg为待加密的字符串
  • $AES(arg, key, vi)$ ==========》 AES加密字符串,arg为待加密的字符串
  • $RandomString(length)$ =========》 生成定长的随机字符串(含数字或字母),length为字符串长度
  • $GetTime(time_type=now,layout=13timestamp,unit=0,0,0,0,0)$ =========》 生成定长的时间戳,time_type=now表示获取当前时间,layout=13timestamp表示时间戳位数为13位,unit为间隔的时间差

代码设计与功能说明

1、定义运行配置文件 runConfig.yml

该文件主要控制测试的执行方式、模块的功能开关、测试用例的筛选、邮件的配置以及日志的配置,具体如下:


runConfig.yml配置信息

# 自动生成测试用例开关,0 -关, 1 -开,根据接口数据自动生成测试用例和单接口执行脚本; 2 -开,根据手工编写的测试用例,自动生成单接口执行脚本
writeCase_switch: 0
# 本次自动生成的测试用例归属的功能模块(项目名称/功能模块)比如: /icmc/pushes ;若不填,则默认不归类
ProjectAndFunction_path: /icmc/pushes
# 扫描用例路径(相对于TestCases的相对路径),以生成执行脚本;若不填,则默认扫描所有的测试用例(只有自动生成测试用例开关为 2 时,此字段才有效),如 /icmc/pushes
scan_path:
# 执行接口测试开关,0 -关, 1 -开
runTest_switch: 1

# 从上往下逐级筛选
# 待执行项目 (可用表达式:not、and、or)(项目名最好唯一,若多个项目或测试名的前缀或后缀相同,则也会被检测到;检测规则为“包含”)
Project: tests

# 待执行接口,可运行单独接口(填接口名),可运行所有接口(None或者空字符串时,即不填),挑选多接口运行可用表达式:not、and、or ,如 parkinside or GetToken or company
markers:

# 本次测试需排除的产品版本(列表),不填,则默认不排除
product_version:

# 本次测试执行的用例等级(列表),不填,则默认执行所有用例等级;可选['blocker', 'critical', 'normal', 'minor', 'trivial']
case_level:
- blocker
- critical
- normal
- minor

# isRun开关,0 -关, 1 -开 ;关闭时,则用例中的is_run字段无效,即会同时执行is_run为 False 的测试用例
isRun_switch: 1

# 用例运行间隔时间(s)
run_interval: 0

# 本轮测试最大允许失败数,达到最大失败数时,则会立即结束当前测试
maxfail: 20

# 测试结束后,显示执行最慢用例数(如:3,表示显示最慢的三条用例及持续时间)
slowestNum: 3

# 失败重试次数,0表示不重试
reruns: 1
# 失败重试间隔时间(s)
reruns_delay: 0.1

#发送测试报告邮件开关, 0 -关, 1 -开 
emailSwitch: 0
#邮件配置
#发件邮箱
smtp_server: smtp.126.com
server_username:XXXX@126.com
server_pwd: XXXXX
#收件人(列表)
msg_to:
- XXX@163.com
- XXX@qq.com

#邮件主题
msg_subject: '[XX项目][测试环境-develop][jira号][接口自动化测试报告]'

#日志级别(字典),由高到低: CRITICAL 、 ERROR 、 WARNING 、 INFO 、 DEBUG
log:
    backup: 5
    console_level: INFO           #控制台日志级别
    file_level: DEBUG              #文件日志级别
    pattern: '%(asctime)s - %(filename)s [line:%(lineno)2d] - %(levelname)s: %(message)s'

2、接口配置文件 apiConfig.ini


[host]
host = 127.0.0.1:12306
MobileCodeWS_host = ws.webxml.com.cn
WeatherWebService_host = www.webxml.com.cn

[header]
header1 = {"Content-Type": "application/json"}
header2 = {"Content-Type": "application/json;charset=UTF-8"}
header3 = {"Content-Type": "application/json", "description": "$RandomString(10)$",
            "timestamp": "$GetTime(time_type=now,layout=13timestamp,unit=0,0,0,0,0)$", "sign": "$SHA1(${description}$, ${timestamp}$)$"}

[MySqlDB]
host = localhost
port = 3306
user = root
pwd = root
db = course
charset = utf8
  • 可以针对不同的项目,配置不同的host、header等,通过不同的命名区分,如header1、header2,在yaml测试用例文件中,通过变量名引用即可,比如${host}$${header1}$
  • 在该接口配置文件里的字段值,可以调用函数助手的功能,引用相关函数,比如header3,在其字段值里即引用了函数RandomStringtimestamp产生需要的值,并将值拼接在一起,然后再用加密函数SHA1加密后,传给sign。

3、测试用例的设计

测试用例以yaml格式的文件保存,简洁优雅,表达力又强,用例直接反映了接口的定义、请求的数据以及期望的结果,且将测试用例中的公共部分提取出来,平时只需维护测试数据和期望结果,维护成本低。
yaml测试用例的数据格式如下:

  • http类型接口

# 用例基本信息
test_info:
      # 用例标题,在报告中作为一级目录显示,用接口路径倒数第二个字段名作为标题
      title: parkinside
      # 用例所属产品版本,不填则为None
      product_version: icm_v1.0
      # 用例等级,优先级,包含blocker, critical, normal, minor,trivial几个不同的等级
      case_level: blocker
      # 请求的域名,可写死,也可写成模板关联host配置文件,也可写在用例中
      host: ${host}$
      # 请求地址 选填(此处不填,每条用例必填)
      address: /${api}$
      # 请求头 选填(此处不填,每条用例必填,如有的话)
      headers: ${header1}$
      # 请求协议
      http_type: http
      # 请求类型
      request_type: POST
      # 参数类型
      parameter_type: json
      # 是否需要获取cookie
      cookies: False
      # 是否为上传文件的接口
      file: False
      # 超时时间
      timeout: 20
      # 运行顺序,当前接口在本轮测试中的执行次序,1表示第一个运行,-1表示最后一个运行
      run_order: 1

# 前置条件,case之前需关联的接口,与test_case类似,关联接口可写多个 
premise:
      - test_name: 获取token    # 必填
        info: 正常获取token值  # 选填
        address: /GetToken   # 请求接口
        http_type: http             # 请求协议
        request_type: GET          # 请求方式
        parameter_type:    # 参数类型,默认为params类型
        headers: {}                # 请求头
        timeout: 10                 # 超时时间
        parameter:                 # 可填实际传递参数,若参数过多,可保存在相应的参数文件中,用test_name作为索引
              username: "admin"
              password: "123456"    
        file: False                 # 是否上传文件,默认false,若上传文件接口,此处为文件相对路径 bool or string
        relevance:  # 关联的键 list or string ;string时,直接写在后面即可;可提取多个关联值,以列表形式,此处提取的关联值可用于本模块的所有用例

# 测试用例 
test_case:
    - test_name: parkinside_1
      # 用例ID,第一条用例必填,从1开始递增
      case_id: 1
      # 是否运行用例,不运行为 False ,空值或其它值则运行
      is_run:
      # 用例描述
      info: parkinside test
      # 参数保存在单独文件中时,可通过文件路径引入参数
      parameter: data_parkinside.json

      # 校验列表(期望结果)
      check:                                          # 这里为多种校验方式组合
            - check_type: json                          # 校验类型,这里为json校验
	      expected_code: 200                        # 期望状态码
	      expected_result: result_parkinside.json   # 期望结果保存在单独文件中时,可通过文件路径引
            
	    - check_type: datebase_check                # 校验类型,这里为数据库校验
              execute_sql: ${common_sql}$               # 校验时查询数据库的sql语句,可通过变量引入配置中的sql语句,也可以直接写sql语句
	      result_row_num: 1                         # 校验sql语句执行后,数据库返回的结果行数
	      expected_result:
		  - '"name": "zhangsan"'                  # 校验数据库的字段和对应的值,可校验多个字段值   
		  - '"age": 18'	

      # 关联作用范围,True表示全局关联,其他值或者为空表示本模块关联
      global_relevance:
      # 关联键,此处提取的关联值可用于本模块后续的所有用例
      relevance:
            - userID
            - vpl

    - test_name: parkinside_2
      # 第二条用例
      case_id: 2
      is_run:
      info: parkinside
      # 请求的域名
      host: 127.0.0.1:12306
      # 请求协议
      http_type: http
      # 请求类型
      request_type: POST
      # 参数类型
      parameter_type: json
      # 请求地址
      address: /parkinside
      # 请求头
      headers:
            Content-Type: application/json
      # 请求参数
      parameter:
            sign: ${sign}$    # 通过变量引用关联值
            vpl: AJ3585
      # 是否需要获取cookie
      cookies: False
      # 是否为上传文件的接口
      file: False
      # 超时时间
      timeout: 20

      # 校验列表
      check:
            - check_type: Regular_check   #正则校验,多项匹配
              expected_code: 200
              expected_result:
                  - '"username": "wuya'      
                  - '"Parking_time_long": "20小时18分钟"'
                  - '"userID": 22'
                  - '"Parking fee": "20\$"'
	    - check_type: datebase_check                       # 校验类型,这里为数据库校验
              execute_sql: 'select name,age,sex from user'     # 校验时查询数据库的sql语句,可以直接写sql语句,也可以通过变量引入
	      result_row_num: 3                                # 仅校验查询数据库返回结果的行数时,期望结果可不填
	      expected_result:

      global_relevance:
      # 关联键
      relevance:

    - test_name: parkinside_3
      # 第三条用例
      case_id: 3
      # 是否运行用例
      is_run:
      # 用例描述
      info: parkinside
      # 请求参数
      parameter:
            vpl: ${vpl}$
            userID: ${userID}$

      # 校验列表
      check:
            expected_result: result_parkinside.json     # 期望结果保存在单独文件中时,可通过文件路径引入
            check_type: only_check_status
            expected_code: 400

      global_relevance:
      # 关联键
      relevance:
  • webservice类型接口1

# 用例基本信息
test_info:
      # 用例标题
      title: MobileCodeWS_getMobileCodeInfo
      # 用例所属产品版本
      product_version: icm_v5.0
      # 用例等级
      case_level: normal
      # 请求的域名,可写死,也可写成模板关联host配置文件,也可写在用例中
      host: ${MobileCodeWS_host}$
      # 请求地址 选填(此处不填,每条用例必填)
      address: /WebServices/MobileCodeWS.asmx?wsdl
      # 请求头 选填(此处不填,每条用例必填,如有的话)
      headers:
      # 请求协议
      http_type: http
      # 请求类型
      request_type: SOAP
      # webservice接口里的函数名
      function_name: getMobileCodeInfo
      # 参数类型(get请求一般为params,该值可不填)
      parameter_type:
      # 是否需要获取cookie
      cookies: False
      # 是否为上传文件的接口
      file: False
      # 超时时间(s),SOAP默认超时连接为90s
      timeout: 100
      # 运行顺序
      run_order:

# 前置条件,case之前需关联的接口
premise:

# 测试用例
test_case:
    - test_name: getMobileCodeInfo_1
      # 用例ID
      case_id: 1
      is_run:
      # 用例描述
      info: getMobileCodeInfo test
      # 请求参数
      parameter:
	      mobileCode: "18300000000"
          userID: ""

      # 校验列表
      check:
          check_type: equal
          expected_result: result_getMobileCodeInfo.json
          expected_code:

      global_relevance:
      # 关联键
      relevance:

    - test_name: getMobileCodeInfo_2
      case_id: 2
      is_run:
      info: getMobileCodeInfo test
      # 请求参数
      parameter:
          mobileCode: "18300000000"
          userID: ""

      # 校验列表
      check:
          check_type: equal
          expected_result: result_getMobileCodeInfo.json
          expected_code:

      global_relevance:
      # 关联键
      relevance:

    - test_name: getMobileCodeInfo_3
      case_id: 3
      is_run: 
      info: getMobileCodeInfo test
      # 请求参数
      parameter:
          mobileCode: "18300000000"
          userID: ""

      # 校验列表
      check:
        check_type: Regular
        expected_result:
            - '18300000000:广东'
            - '深圳 广东移动全球通卡'
        expected_code:

      global_relevance:
      # 关联键
      relevance:

    - test_name: getMobileCodeInfo_4
      case_id: 4
      is_run:
      info: getMobileCodeInfo test
      parameter:
          mobileCode: "18300000000"
          userID: ""

      # 校验列表 
      check:
          check_type: no_check
          expected_code:
          expected_result:

      global_relevance:
      # 关联键
      relevance:
  • webservice类型接口2

# 用例基本信息
test_info:
      # 用例标题
      title: MobileCodeWS_getMobileCodeInfo
      # 用例所属产品版本
      product_version: icm_v5.0
      # 用例等级
      case_level: normal
      # 请求的域名,可写死,也可写成模板关联host配置文件,也可写在用例中
      host: ${WeatherWebService_host}$
      # 请求地址 选填(此处不填,每条用例必填)
      address: /WebServices/WeatherWebService.asmx?wsdl
      # 请求过滤地址
      filter_address: http://WebXml.com.cn/
      # 请求头 选填(此处不填,每条用例必填,如有的话)
      headers:
      # 请求协议
      http_type: http
      # 请求类型
      request_type: soap_with_filter
      # webservice接口里的函数名
      function_name: getSupportCity
      # 参数类型
      parameter_type:
      # 是否需要获取cookie
      cookies: False
      # 是否为上传文件的接口
      file: False
      # 超时时间(s),SOAP默认超时连接为90s
      timeout: 100
      # 运行顺序
      run_order:

# 前置条件,case之前需关联的接口
premise:

# 测试用例
test_case:
    - test_name: getSupportCity_1
      # 用例ID
      case_id: 1
      is_run:
      # 用例描述
      info: getSupportCity test
      # 请求参数
      parameter:
          byProvinceName: "四川"

      # 校验列表
      check:
          check_type: Regular
          expected_result:
              - '成都 (56294)'
              - '广元 (57206)'
          expected_code:

      global_relevance:
      # 关联键
      relevance:

    - test_name: getSupportCity_2
      case_id: 2
      is_run:
      info: getSupportCity test
      parameter:
          byProvinceName: "四川"

      # 校验列表
      check:
          check_type: no_check   #不校验结果
          expected_code:
          expected_result:

      global_relevance:
      # 关联键
      relevance:
  • 当该接口的参数数据较多时,为维护方便,可将其保存在一个单独的json文件中,比如上面用例中的data_parkinside.json,就是保存该接口参数数据的一个文件,与测试用例文件在同一个目录下。测试执行时,通过解析该json文件中的test_name字段,获取属于自身用例的参数,参数文件的内容格式如下:

[
  {
    "test_name": "parkinside_1",
    "parameter": {
      "token": "asdgfhh32456asfgrsfss",
	  "vpl": "AJ3585"
    }
  },
  {
    "test_name": "parkinside_3",
    "parameter": {
      "vpl": "AJ3585"
    }
  }
]

该json文件保存了两条用例的参数,通过用例名parkinside_1获取到第一条用例的参数,通过用例名parkinside_3获取到第三条用例的参数(json参数文件中的用例名需与yaml用例文件中的用例名一致)。

  • 当该接口的期望结果较长时,为维护方便,可将其保存在一个单独的json文件中,比如上面用例中的result_parkinside.json,就是保存该接口期望结果的一个文件,与测试用例文件在同一目录下。测试执行时,通过解析该json文件中的test_name字段,获取属于自身用例的期望结果,期望结果文件的内容格式如下:

[
  {
      "json":
          {
            "vplInfo":
            {
              "userID":22,
              "username":"wuya",
              "vpl":"京AJ3585"
            },
            "Parking_time_long":"20小时18分钟",
            "Parking fee":"20$"
          },
      "test_name": "parkinside_1"
  }
]

该json文件保存了一条用例的期望结果,通过用例parkinside_1获取到第一条用例的期望结果(json文件中的用例名需与yaml用例文件中的用例名一致)。

  • 若该接口的测试用例需要引用函数或者变量,则可先在一个单独的relevance.ini关联配置文件中,定义好相关的变量和函数名,并进行拼接,后续可通过变量名,引入测试用例中,比如上面用例中的 ${sign}$ ,就是引用了关联配置文件中的 sign 变量值,relevance.ini关联配置文件的内容格式如下:

[relevance]
nonce=$RandomString(5)$
timestamp = $GetTime(time_type=now,layout=13timestamp,unit=0,0,0,0,0)$
sign = $SHA1(asdh, ${nonce}$, ${timestamp}$)$

上面配置中的nonce变量,引用了随机函数RandomString,该随机函数产生长度为5的随机数,这些函数的定义都已封装在functions模块中,在这里只需要通过对应的函数名,并存入参数即可引用相关函数。变量timestamp引用了时间戳函数,在这里将生成一个13位的时间戳,并传给变量timestamp。变量sign则是引用了加密函数SHA1,这里将会把字符串asdh、变量nonce的值和变量timestamp的值先拼接起来,然后再将拼接好的字符串传给加密函数SHA1加密。然后即可在用例中引用变量sign,如下:


  # 请求参数
  parameter:
        sign: ${sign}$    # 通过变量引用关联值
        vpl: AJ3585
  • 若该接口的测试用例的期望结果中,需要引用变量来传递SQL语句,则可先在一个单独的sql_check.ini关联配置文件中,定义好相关的变量,并赋予SQL,后续可通过变量名,将SQL语句引入测试用例中,比如上面用例中的 ${common_sql}$,就是引用了关联配置文件中的 common_sql 变量值,这里可以定义一些共用的sql语句,避免冗余,方便维护,relevance.ini关联配置文件的内容格式如下:

[relevance]
common_sql=select name,age,sex from user where id=1
parkinside_6_sql=select name,age,sex from user where id=2

4、单接口用例执行脚本

单接口测试用例执行脚本,由程序根据yaml格式的测试用例文件自动生成,并根据相应yaml格式的测试用例文件所在的路径生成当前用例执行脚本的保存路径,且该用例执行脚本平时不需要人工维护,如下是接口parkinside的执行脚本test_parkinside.py的格式:


# -*- coding: utf-8 -*-

import allure
import pytest
import time
from Main import root_path, case_level, product_version, run_interval
from common.unit.initializeYamlFile import ini_yaml
from common.unit.initializePremise import ini_request
from common.unit.apiSendCheck import api_send_check
from common.unit.initializeRelevance import ini_relevance
from common.unit import setupTest

case_path = root_path + "/tests/TestCases/parkinsideApi"
relevance_path = root_path + "/common/configModel/relevance"
case_dict = ini_yaml(case_path, "parkinside")


@allure.feature(case_dict["test_info"]["title"])
class TestParkinside:

    @pytest.fixture(scope="class")
    def setupClass(self):
        """
        :rel: 获取关联文件得到的字典
        :return:
        """
        self.rel = ini_relevance(case_path, 'relevance')     #获取本用例初始公共关联值
        self.relevance = ini_request(case_dict, case_path, self.rel)   #执行完前置条件后,得到的本用例最新全部关联值
        return self.relevance, self.rel


    @pytest.mark.skipif(case_dict["test_info"]["product_version"] in product_version,
                        reason="该用例所属版本为:{0},在本次排除版本{1}内".format(case_dict["test_info"]["product_version"], product_version))
    @pytest.mark.skipif(case_dict["test_info"]["case_level"] not in case_level,
                        reason="该用例的用例等级为:{0},不在本次运行级别{1}内".format(case_dict["test_info"]["case_level"], case_level))
    @pytest.mark.run(order=case_dict["test_info"]["run_order"])
    @pytest.mark.parametrize("case_data", case_dict["test_case"], ids=[])
    @allure.severity(case_dict["test_info"]["case_level"])
    @pytest.mark.parkinside
    @allure.story("parkinside")
    @allure.issue("http://www.bugjira.com")  # bug地址
    @allure.testcase("http://www.testlink.com")  # 用例连接地址
    def test_parkinside(self, case_data, setupClass):
        """
        测试接口为:parkinside
        :param case_data: 测试用例
        :return:
        """
        self.relevance = setupTest.setupTest(relevance_path, case_data, setupClass)

        # 发送测试请求
        api_send_check(case_data, case_dict, case_path, self.relevance)
        time.sleep(run_interval)


if __name__ == '__main__':

    import subprocess
    subprocess.call(['pytest', '-v'])

5、封装请求协议apiMethod.py


def post(header, address, request_parameter_type, timeout=8, data=None, files=None, cookie=None):
    """
    post请求
    :param header: 请求头
    :param address: 请求地址
    :param request_parameter_type: 请求参数格式(form_data,raw)
    :param timeout: 超时时间
    :param data: 请求参数
    :param files: 文件路径
    :return:
    """
    if 'form_data' in request_parameter_type:
        for i in files:
            value = files[i]
            if '/' in value:
                file_parm = i
                files[file_parm] = (os.path.basename(value), open(value, 'rb'))
        enc = MultipartEncoder(
            fields=files,
            boundary='--------------' + str(random.randint(1e28, 1e29 - 1))
        )
        header['Content-Type'] = enc.content_type

        response = requests.post(url=address, data=enc, headers=header, timeout=timeout, cookies=cookie)

    elif 'data' in request_parameter_type:
        response = requests.post(url=address, data=data, headers=header, timeout=timeout, files=files, cookies=cookie)

    elif 'json' in request_parameter_type:
        response = requests.post(url=address, json=data, headers=header, timeout=timeout, files=files, cookies=cookie)

    try:
        if response.status_code != 200:
            return response.status_code, response.text
        else:
            return response.status_code, response.json()
    except json.decoder.JSONDecodeError:
        return response.status_code, ''
    except simplejson.errors.JSONDecodeError:
        return response.status_code, ''
    except Exception as e:
        logging.exception('ERROR')
        logging.error(e)
        raise


def get(header, address, data, timeout=8, cookie=None):
    """
    get请求
    :param header: 请求头
    :param address: 请求地址
    :param data: 请求参数
    :param timeout: 超时时间
    :return:
    """
    response = requests.get(url=address, params=data, headers=header, timeout=timeout, cookies=cookie)
    if response.status_code == 301:
        response = requests.get(url=response.headers["location"])
    try:
        return response.status_code, response.json()
    except json.decoder.JSONDecodeError:
        return response.status_code, ''
    except simplejson.errors.JSONDecodeError:
        return response.status_code, ''
    except Exception as e:
        logging.exception('ERROR')
        logging.error(e)
        raise


def put(header, address, request_parameter_type, timeout=8, data=None, files=None, cookie=None):
    """
    put请求
    :param header: 请求头
    :param address: 请求地址
    :param request_parameter_type: 请求参数格式(form_data,raw)
    :param timeout: 超时时间
    :param data: 请求参数
    :param files: 文件路径
    :return:
    """
    if request_parameter_type == 'raw':
        data = json.dumps(data)
    response = requests.put(url=address, data=data, headers=header, timeout=timeout, files=files, cookies=cookie)
    try:
        return response.status_code, response.json()
    except json.decoder.JSONDecodeError:
        return response.status_code, ''
    except simplejson.errors.JSONDecodeError:
        return response.status_code, ''
    except Exception as e:
        logging.exception('ERROR')
        logging.error(e)
        raise


def delete(header, address, data, timeout=8, cookie=None):
    """
    delete请求
    :param header: 请求头
    :param address: 请求地址
    :param data: 请求参数
    :param timeout: 超时时间
    :return:
    """
    response = requests.delete(url=address, params=data, headers=header, timeout=timeout, cookies=cookie)

    try:
        return response.status_code, response.json()
    except json.decoder.JSONDecodeError:
        return response.status_code, ''
    except simplejson.errors.JSONDecodeError:
        return response.status_code, ''
    except Exception as e:
        logging.exception('ERROR')
        logging.error(e)
        raise


def save_cookie(header, address, request_parameter_type, timeout=8, data=None, files=None, cookie=None):
    """
    保存cookie信息
    :param header: 请求头
    :param address: 请求地址
    :param timeout: 超时时间
    :param data: 请求参数
    :param files: 文件路径
    :return:
    """
    cookie_path = root_path + '/common/configModel/relevance/cookie.ini'
    if 'data' in request_parameter_type:
        response = requests.post(url=address, data=data, headers=header, timeout=timeout, files=files, cookies=cookie)

    elif 'json' in request_parameter_type:
        response = requests.post(url=address, json=data, headers=header, timeout=timeout, files=files, cookies=cookie)

    try:
        if response.status_code != 200:
            return response.status_code, response.text

        else:
            re_cookie = response.cookies.get_dict()
            cf = Config(cookie_path)
            cf.add_section_option('relevance', re_cookie)
            for i in re_cookie:
                values = re_cookie[i]
                logging.debug("cookies已保存,结果为:{}".format(i+"="+values))
            return response.status_code, response.json()

    except json.decoder.JSONDecodeError:
        return response.status_code, ''
    except simplejson.errors.JSONDecodeError:
        return response.status_code, ''
    except Exception as e:
        logging.exception('ERROR')
        logging.error(e)
        raise
    ……………………

6、封装方法apiSend.py:处理测试用例,拼接请求并发送


def send_request(data, project_dict, _path, relevance=None):
    """
    封装请求
    :param data: 测试用例
    :param project_dict: 用例文件内容字典
    :param relevance: 关联对象
    :param _path: case路径
    :return:
    """
    logging.info("="*100)
    try:
        # 获取用例基本信息
        get_header =project_dict["test_info"].get("headers")
        get_host = project_dict["test_info"].get("host")
        get_address = project_dict["test_info"].get("address")
        get_http_type = project_dict["test_info"].get("http_type")
        get_request_type = project_dict["test_info"].get("request_type")
        get_parameter_type = project_dict["test_info"].get("parameter_type")
        get_cookies = project_dict["test_info"].get("cookies")
        get_file = project_dict["test_info"].get("file")
        get_timeout = project_dict["test_info"].get("timeout")
    except Exception as e:
        logging.exception('获取用例基本信息失败,{}'.format(e))

    try:
        # 如果用例中写了headers关键字,则用用例中的headers值(若该关键字没有值,则会将其值置为none),否则用全局headers
        get_header = data["headers"]
    except KeyError:
        pass
    try:
        # 替换成用例中相应关键字的值,如果用例中写了host和address,则使用用例中的host和address,若没有则使用全局传入的默认值
        get_host = data["host"]
    except KeyError:
        pass
    try:
        get_address = data["address"]
    except KeyError:
        pass
    try:
        get_http_type = data["http_type"]
    except KeyError:
        pass
    try:
        get_request_type = data["request_type"]
    except KeyError:
        pass
    try:
        get_parameter_type = data["parameter_type"]
    except KeyError:
        pass
    try:
        get_cookies = data["cookies"]
    except KeyError:
        pass
    try:
        get_file = data["file"]
    except KeyError:
        pass
    try:
        get_timeout = data["timeout"]
    except KeyError:
        pass

    Cookie = None

    header = get_header
    if get_header:
        if isinstance(get_header, str):
            header = confManage.conf_manage(get_header, "header")  # 处理请求头中的变量
            if header == get_header:
                pass

            else:
                var_list = re.findall('\$.*?\$', header)
                header = literal_eval(header)  # 将字典类型的字符串,转成字典
                # 处理请求头中的变量和函数
                if var_list:
                    # 将关联对象里的键值对遍历出来,并替换掉字典值中的函数
                    rel = dict()
                    for key, value in header.items():
                        rel[key] = replace_random(value)
                    header = rel
                    logging.debug("替换请求头中的函数处理结果为:{}".format(header))

                    str_header = str(header)
                    var_list = re.findall('\${.*?}\$', str_header)
                    if var_list:
                        # 用自身关联对象里的变量值,替换掉自身关联对象里的变量
                        header = replaceRelevance.replace(header, header)

                        str_header = str(header)
                        var_list = re.findall('\$.*?\$', str_header)
                        if var_list:
                            # 再次将关联对象里的键值对遍历出来,并替换掉字典值中的函数
                            rel = dict()
                            for key, value in header.items():
                                rel[key] = replace_random(value)
                            header = rel
                        else:
                            pass
                    else:
                        pass
                else:
                    pass
        else:
            pass
    else:
        pass
    logging.debug("请求头处理结果为:{}".format(header))

    if get_cookies is True:
        cookie_path = root_path + "/common/configModel/relevance"
        Cookie = ini_relevance(cookie_path, 'cookie')   # 为字典类型的字符串
        logging.debug("cookie处理结果为:{}".format(Cookie))
    else:
        pass

    parameter = readParameter.read_param(data["test_name"], data["parameter"], _path, relevance)    #处理请求参数(含参数为文件的情况)
    logging.debug("请求参数处理结果:{}".format(parameter))

    get_address = str(replaceRelevance.replace(get_address, relevance))  # 处理请求地址中的变量
    logging.debug("请求地址处理结果:{}".format(get_address))

    get_host = str(confManage.conf_manage(get_host, "host"))   # host处理,读取配置文件中的host
    logging.debug("host处理结果:{}".format(get_host))
    if not get_host:
        raise Exception("接口请求地址为空 {}".format(get_host))
    logging.info("请求接口:{}".format(data["test_name"]))
    logging.info("请求地址:{}".format((get_http_type + "://" + get_host + get_address)))
    logging.info("请求头: {}".format(header))
    logging.info("请求参数: {}".format(parameter))

    # 通过get_request_type来判断,如果get_request_type为post_cookie;如果get_request_type为get_cookie
    if get_request_type.lower() == 'post_cookie':
        with allure.step("保存cookie信息"):
            allure.attach("请求接口:", data["test_name"])
            allure.attach("用例描述:", data["info"])
            allure.attach("请求地址", get_http_type + "://" + get_host + get_address)
            allure.attach("请求头", str(header))
            allure.attach("请求参数", str(parameter))

        result = apiMethod.save_cookie(header=header, address=get_http_type + "://" + get_host + get_address,
                                  request_parameter_type=get_parameter_type,
                                  data=parameter,
                                  cookie=Cookie,
                                  timeout=get_timeout)

    elif get_request_type.lower() == 'post':
        logging.info("请求方法: POST")
        if get_file:
            with allure.step("POST上传文件"):
                allure.attach("请求接口:",data["test_name"])
                allure.attach("用例描述:", data["info"])
                allure.attach("请求地址", get_http_type + "://" + get_host + get_address)
                allure.attach("请求头", str(header))
                allure.attach("请求参数", str(parameter))

            result = apiMethod.post(header=header,
                                    address=get_http_type + "://" + get_host + get_address,
                                    request_parameter_type=get_parameter_type,
                                    files=parameter,
                                    cookie=Cookie,
                                    timeout=get_timeout)
        else:
            with allure.step("POST请求接口"):
                allure.attach("请求接口:", data["test_name"])
                allure.attach("用例描述:", data["info"])
                allure.attach("请求地址", get_http_type + "://" + get_host + get_address)
                allure.attach("请求头", str(header))
                allure.attach("请求参数", str(parameter))
            result = apiMethod.post(header=header,
                                    address=get_http_type + "://" + get_host + get_address,
                                    request_parameter_type=get_parameter_type,
                                    data=parameter,
                                    cookie=Cookie,
                                    timeout=get_timeout)
    elif get_request_type.lower() == 'get':
        with allure.step("GET请求接口"):
            allure.attach("请求接口:", data["test_name"])
            allure.attach("用例描述:", data["info"])
            allure.attach("请求地址", get_http_type + "://" + get_host + get_address)
            allure.attach("请求头", str(header))
            allure.attach("请求参数", str(parameter))
            logging.info("请求方法: GET")
        result = apiMethod.get(header=header,
                               address=get_http_type + "://" + get_host + get_address,
                               data=parameter,
                               cookie=Cookie,
                               timeout=get_timeout)
    elif get_request_type.lower() == 'put':
        logging.info("请求方法: PUT")
        if get_file:
            with allure.step("PUT上传文件"):
                allure.attach("请求接口:", data["test_name"])
                allure.attach("用例描述:", data["info"])
                allure.attach("请求地址", get_http_type + "://" + get_host + get_address)
                allure.attach("请求头", str(header))
                allure.attach("请求参数", str(parameter))
            result = apiMethod.put(header=header,
                                    address=get_http_type + "://" + get_host + get_address,
                                    request_parameter_type=get_parameter_type,
                                    files=parameter,
                                    cookie=Cookie,
                                    timeout=get_timeout)
        else:
            with allure.step("PUT请求接口"):
                allure.attach("请求接口:", data["test_name"])
                allure.attach("用例描述:", data["info"])
                allure.attach("请求地址", get_http_type + "://" + get_host + get_address)
                allure.attach("请求头", str(header))
                allure.attach("请求参数", str(parameter))
            result = apiMethod.put(header=header,
                                    address=get_http_type + "://" + get_host + get_address,
                                    request_parameter_type=get_parameter_type,
                                    data=parameter,
                                    cookie=Cookie,
                                    timeout=get_timeout)
    elif get_request_type.lower() == 'delete':
        with allure.step("DELETE请求接口"):
            allure.attach("请求接口:", data["test_name"])
            allure.attach("用例描述:", data["info"])
            allure.attach("请求地址", get_http_type + "://" + get_host + get_address)
            allure.attach("请求头", str(header))
            allure.attach("请求参数", str(parameter))
        logging.info("请求方法: DELETE")
        result = apiMethod.delete(header=header,
                               address=get_http_type + "://" + get_host + get_address,
                               data=parameter,
                               cookie=Cookie,
                               timeout=get_timeout)
    …………………………
    else:
        result = {"code": False, "data": False}
        logging.info("没有找到对应的请求方法!")
    logging.info("请求接口结果:\n {}".format(result))
    return result

7、测试结果断言封装checkResult.py


def check_json(src_data, dst_data):
    """
    校验的json
    :param src_data: 检验内容
    :param dst_data: 接口返回的数据
    :return:
    """
    if isinstance(src_data, dict):
        for key in src_data:
            if key not in dst_data:
                raise Exception("JSON格式校验,关键字%s不在返回结果%s中" % (key, dst_data))
            else:
                this_key = key
                if isinstance(src_data[this_key], dict) and isinstance(dst_data[this_key], dict):
                    check_json(src_data[this_key], dst_data[this_key])
                elif isinstance(type(src_data[this_key]), type(dst_data[this_key])):
                    raise Exception("JSON格式校验,关键字 %s 与 %s 类型不符" % (src_data[this_key], dst_data[this_key]))
                else:
                    pass
    else:
        raise Exception("JSON校验内容非dict格式")


def check_result(test_name, case, code, data, _path, relevance=None):
    """
    校验测试结果
    :param test_name: 测试名称
    :param case: 测试用例
    :param code: HTTP状态
    :param data: 返回的接口json数据
    :param relevance: 关联值对象
    :param _path: case路径
    :return:
    """
    # 不校验结果
    if case["check_type"] == 'no_check':
        with allure.step("不校验结果"):
            pass

    # json格式校验
    elif case["check_type"] == 'json':
        expected_result = case["expected_result"]
        if isinstance(case["expected_result"], str):
            expected_result = readExpectedResult.read_json(test_name, expected_result, _path, relevance)
        with allure.step("JSON格式校验"):
            allure.attach("期望code", str(case["expected_code"]))
            allure.attach('期望data', str(expected_result))
            allure.attach("实际code", str(code))
            allure.attach('实际data', str(data))
        if int(code) == case["expected_code"]:
            if not data:
                data = "{}"
            check_json(expected_result, data)
        else:
            raise Exception("http状态码错误!\n {0} != {1}".format(code, case["expected_code"]))

    # 只校验状态码
    elif case["check_type"] == 'only_check_status':
        with allure.step("校验HTTP状态"):
            allure.attach("期望code", str(case["expected_code"]))
            allure.attach("实际code", str(code))
            allure.attach('实际data', str(data))
        if int(code) == case["expected_code"]:
            pass
        else:
            raise Exception("http状态码错误!\n {0} != {1}".format(code, case["expected_code"]))

    # 完全校验
    elif case["check_type"] == 'entirely_check':
        expected_result = case["expected_result"]
        if isinstance(case["expected_result"], str):
            expected_result = readExpectedResult.read_json(test_name, expected_result, _path, relevance)
        with allure.step("完全校验结果"):
            allure.attach("期望code", str(case["expected_code"]))
            allure.attach('期望data', str(expected_result))
            allure.attach("实际code", str(code))
            allure.attach('实际data', str(data))
        if int(code) == case["expected_code"]:
            result = operator.eq(expected_result, data)
            if result:
                pass
            else:
                raise Exception("完全校验失败! {0} ! = {1}".format(expected_result, data))
        else:
            raise Exception("http状态码错误!\n {0} != {1}".format(code, case["expected_code"]))

    # 正则校验
    elif case["check_type"] == 'Regular_check':
        if int(code) == case["expected_code"]:
            try:
                result = ""
                if isinstance(case["expected_result"], list):
                    with allure.step("正则校验"):
                        for i in case["expected_result"]:
                            result = re.findall(i.replace("\"","\'"), str(data))
                            allure.attach('正则校验结果\n',str(result))
                        allure.attach('实际data', str(data))
                else:
                    result = re.findall(case["expected_result"].replace("\"", "\'"), str(data))
                    with allure.step("正则校验"):
                        allure.attach("期望code", str(case["expected_code"]))
                        allure.attach('正则表达式', str(case["expected_result"]).replace("\'", "\""))
                        allure.attach("实际code", str(code))
                        allure.attach('实际data', str(data))
                        allure.attach(case["expected_result"].replace("\"", "\'") + '校验完成结果',
                                      str(result).replace("\'", "\""))
                if not result:
                    raise Exception("正则未校验到内容! {}".format(case["expected_result"]))
            except KeyError:
                raise Exception("正则校验执行失败! {}\n正则表达式为空时".format(case["expected_result"]))
        else:
            raise Exception("http状态码错误!\n {0} != {1}".format(code, case["expected_code"]))

    # 数据库校验
    elif case["check_type"] == "datebase_check":
        if isinstance(case["result_row_num"],int) and case["result_row_num"] >= 0:
            get_sel_sql = case["execute_sql"]
            conf_path = os.path.join(_path, 'sql_check.ini')
            sel_sql = conf_manage(get_sel_sql, 'relevance', conf_path)
            DB_obj = MySqlDB()
            sel_result = DB_obj.sel_operation(sel_sql)
            result_row_num = len(sel_result)
            if result_row_num == case["result_row_num"]:
                if result_row_num == 1:
                    try:
                        if isinstance(case["expected_result"], list):
                            with allure.step("数据库校验"):
                                for i in case["expected_result"]:
                                    # 需将待匹配数据中的括号替换成转义的普通括号,否则会匹配失败,括号会被当成正则表达式的一部分
                                    result = re.findall(i.replace("\"", "\'").replace("(", "\(").replace(")", "\)"),
                                                        str(sel_result))
                                    allure.attach('数据库校验结果\n', str(result))
                                    if not result:
                                        allure.attach('数据库实际返回结果', str(sel_result))
                                        raise Exception("数据库未校验到期望内容! {}".format(i))
                                allure.attach('数据库实际返回结果', str(sel_result))
                        else:
                            result = re.findall(
                                case["expected_result"].replace("\"", "\'").replace("(", "\(").replace(")", "\)"),
                                str(sel_result))
                            with allure.step("数据库校验"):
                                allure.attach('数据库校验结果\n', str(result))
                                allure.attach('数据库实际返回结果', str(sel_result))
                            if not result:
                                raise Exception("数据库未校验到期望内容! {}".format(case["expected_result"]))
                    except KeyError:
                        raise Exception("数据库校验执行失败! {}\n正则表达式为空时".format(case["expected_result"]))
                else:
                    with allure.step("数据库返回结果行数校验"):
                        allure.attach('期望行数', str(case["result_row_num"]))
                        allure.attach('实际行数', str(result_row_num))
            else:
                raise Exception("返回的结果行数不对!\n 实际行数:{0} != 期望行数:{1}".format(result_row_num, case["result_row_num"]))
        else:
            raise Exception("用例中的结果行数result_row_num填写格式不对!")

    else:
        raise Exception("无该校验方式:{}".format(case["check_type"]))

8、共享模块conftest.py(初始化测试环境,制造测试数据,并还原测试环境)


import allure
import pytest
from common.configModel import confRead
from Main import root_path
from common.unit.initializeYamlFile import ini_yaml
from common.unit.initializeRelevance import ini_relevance
from common.unit.apiSendCheck import api_send_check
from common.configModel.confRead import Config
import logging
import os

conf_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config", "apiConfig.ini")
case_path = root_path + "/tests/CommonApi/loginApi"
relevance_path = root_path + "/common/configModel/relevance"

@pytest.fixture(scope="session", autouse=True)
def setup_env():
    # 定义环境;定义报告中environment
    Host = confRead.Config(conf_path).read_apiConfig("host")
    allure.environment(测试环境="online", hostName=Host["host"], 执行人="XX", 测试项目="线上接口测试")


case_dict = ini_yaml(case_path, "login")

# 参数化 fixture
@pytest.fixture(scope="session", autouse=True, params=case_dict["test_case"])
def login(request):
    # setup
    """
    :param request: 上下文
    :param request.param: 测试用例
    :return:
    """
    # 清空关联配置
    for i in ["GlobalRelevance.ini", "ModuleRelevance.ini"]:
        relevance_file = os.path.join(relevance_path, i)
        cf = Config(relevance_file)
        cf.add_conf("relevance")

    logging.info("执行全局用例依赖接口,初始化数据!")
    relevance = ini_relevance(relevance_path, "ModuleRelevance")
    if request.param["case_id"] == 1:
        relevance = ini_relevance(case_path, "relevance")

    logging.info("本用例最终的关联数据为:{}".format(relevance))
    # 发送测试请求
    api_send_check(request.param, case_dict, case_path, relevance)
    logging.info("初始化数据完成!")
    yield
    # teardown
	# 还原测试环境部分代码
	……
	……
    logging.info("本轮测试已结束,正在还原测试环境!")

9、测试执行总入口Main.py(收集测试用例,批量执行并生成测试报告)


import os
import shutil
import subprocess
import pytest
import logging
from common.unit.initializeYamlFile import ini_yaml
from common.utils.logs import LogConfig
from common.script.writeCase import write_case
from common.script.writeCaseScript import write_caseScript
from common.utils.formatChange import formatChange
from common.utils.emailModel.runSendEmail import sendEailMock

root_path = os.path.split(os.path.realpath(__file__))[0]
xml_report_path = root_path + "\\report\\xml"
detail_report_path = root_path + "\\report\\detail_report"
summary_report_path = root_path + "\\report\\summary_report\\summary_report.html"
runConf_path = os.path.join(root_path, "config")

# 获取运行配置信息
runConfig_dict = ini_yaml(runConf_path, "runConfig")

case_level = runConfig_dict["case_level"]
if not case_level:
    case_level = ["blocker", "critical", "normal", "minor", "trivial"]
else:
    pass

product_version = runConfig_dict["product_version"]
if not product_version:
    product_version = []
else:
    pass

isRun_switch = runConfig_dict["isRun_switch"]
run_interval = runConfig_dict["run_interval"]
writeCase_switch = runConfig_dict["writeCase_switch"]

ProjectAndFunction_path = runConfig_dict["ProjectAndFunction_path"]
if not ProjectAndFunction_path:
    ProjectAndFunction_path = ""
else:
    pass

scan_path = runConfig_dict["scan_path"]
if not scan_path:
    scan_path = ""
else:
    pass

runTest_switch = runConfig_dict["runTest_switch"]
reruns = str(runConfig_dict["reruns"])
reruns_delay = str(runConfig_dict["reruns_delay"])
log = runConfig_dict["log"]


def batch(CMD):
    output, errors = subprocess.Popen(CMD, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()
    outs = output.decode("utf-8")
    return outs


if __name__ == "__main__":
    try:
        LogConfig(root_path, log)

        if writeCase_switch == 1:
            # 根据har_path里的文件,自动生成用例文件yml和用例执行文件py,若已存在相关文件,则不再创建
            write_case(root_path, ProjectAndFunction_path)

        elif writeCase_switch == 2:
            write_caseScript(root_path, scan_path)

        else:
            logging.info("="*20+"本次测试自动生成测试用例功能已关闭!"+"="*20+"\n")

        if runTest_switch == 1:
            # 清空目录和文件
            email_target_dir = root_path + "/report/zip_report"  # 压缩文件保存路径
            shutil.rmtree(email_target_dir)
            if os.path.exists(summary_report_path):
                os.remove(summary_report_path)
            else:
                pass
            os.mkdir(email_target_dir)

            args = ["-k", runConfig_dict["Project"], "-m", runConfig_dict["markers"], "--maxfail=%s" % runConfig_dict["maxfail"],
                    "--durations=%s" % runConfig_dict["slowestNum"], "--reruns", reruns, "--reruns-delay", reruns_delay,
                    "--alluredir", xml_report_path, "--html=%s" % summary_report_path]

            test_result = pytest.main(args)   # 全部通过,返回0;有失败或者错误,则返回1

            cmd = "allure generate %s -o %s --clean" % (xml_report_path, detail_report_path)
            reportResult = batch(cmd)
            logging.debug("生成html的报告结果为:{}".format(reportResult))

            # 发送report到邮件
            emailFunction = runConfig_dict["emailSwitch"]
            if emailFunction == 1:

                if test_result == 0:
                    ReportResult = "测试通过!"
                else:
                    ReportResult = "测试不通过!"

                # 将字符中的反斜杠转成正斜杠
                fileUrl_PATH = root_path.replace("\\", "/")
                logging.debug("基础路径的反斜杠转成正斜杠为:{}".format(fileUrl_PATH))

                fileUrl = "file:///{}/report/summary_report/summary_report.html".format(fileUrl_PATH)
                logging.info("html测试报告的url为:{}".format(fileUrl))
                save_fn = r"{}\report\zip_report\summary_report.png".format(root_path)
                logging.debug("转成图片报告后保存的目标路径为:{}".format(save_fn))
                formatChange_obj = formatChange()
                formatChange_obj.html_to_image(fileUrl, save_fn)

                email_folder_dir = root_path + "/report/detail_report"  # 待压缩文件夹
                logging.debug("待压缩文件夹为:{}".format(email_folder_dir))
                sendEailMock_obj = sendEailMock()
                sendEailMock_obj.send_email(email_folder_dir, email_target_dir, runConfig_dict, ReportResult, save_fn)

            else:
                logging.info("="*20+"本次测试的邮件功能已关闭!"+"="*20+"\n")

        else:
            logging.info("="*20+"本次运行测试开关已关闭!"+"="*20+"\n")
    except Exception as err:
        logging.error("本次测试有异常为:{}".format(err))

10、结合Allure生成报告

  • 好的测试报告在整个测试框架是至关重要的部分,Allure是一个很好用的报告框架,不仅报告美观,而且方便CI集成。

  • Allure中对严重级别的定义:

    1. Blocker级别:中断缺陷(客户端程序无响应,无法执行下一步操作)
    2. Critical级别:临界缺陷(功能点缺失)
    3. Normal级别:普通缺陷(数值计算错误)
    4. Minor级别:次要缺陷(界面错误与UI需求不符)
    5. Trivial级别:轻微缺陷(必输项无提示,或者提示不规范)
  • Allure报告总览,如图所示:







     

  • 发送到邮件中的测试报告
     

  • 测试执行项目演示

pytest、Allure与Jenkins集成

1、集成环境部署

1、Linux安装docker容器

  • 安装docker容器脚本

      curl -fsSL https://get.docker.com | bash -s docker --mirror Aliyun
    
  • 启动docker

      systemctl start docker
    
  • 通过修改daemon配置文件/etc/docker/daemon.json来使用阿里云镜像加速器

      sudo mkdir -p /etc/docker
      sudo tee /etc/docker/daemon.json <<-'EOF'
      {
        "registry-mirrors": ["https://XXXX.XXXX.aliyuncs.com"]
      }
      EOF
      sudo systemctl daemon-reload
      sudo systemctl restart docker		
    
  • 查看阿里云加速器是否配置成功

      vi  /etc/docker/daemon.json
    

2、安装Jenkins

  • 在 Docker 中安装并启动 Jenkins 的样例命令如下:

      docker run -d -u root \
        --name jenkins-blueocean \
        --restart=always \
        -p 8080:8080 \
        -p 50000:50000 \
        -p 50022:50022 \
        -v /home/jenkins/var:/var/jenkins_home \
        -v /var/run/docker.sock:/var/run/docker.sock \
        -v "$HOME":/home \
        --env JAVA_OPTS="-Duser.timezone=GMT+08" \
        jenkinsci/blueocean
      
      其中的 50000 是映射到 TCP port for JNLP agents 对应的端口,50022 是映射到 SSHD Port。在成功启动 Jenkins 后,可在Jenkins启动页面 http://ip:8080/configureSecurity/ 上设置。
      这两个端口其实不是必须的,只是为了方便通过 SSH 使用 Jenkins 才开启它们。--env:指定Jenkins的运行时区为东八区,默认是零时区;"$HOME":/home是将容器的home目前映射到本地某个目录,比如"$HOME"为container_home,则写成 container_home:/home
    
  • 在此页面打开 SSHD Port 后,运行以下命令即可验证对应的端口值。

      curl -Lv http://ip:8080/login 2>&1 | grep 'X-SSH-Endpoint'
    
  • 把Jenkins容器里的密码粘贴上

      /var/jenkins_home/secrets/initialAdminPassword
    
  • 访问 http://ip:8080 ,安装默认推荐插件

  • 先到admin配置界面,再次修改admin的用户密码

3、allure与jenkins集成

  • jenkins安装插件
    在管理Jenkins-插件管理-可选插件处,搜索allure ,然后安装,如下
    插件名称为Allure Jenkins Plugin,如下图所示:
     

  • jenkins安装allure_commandline(若之前已安装过allure插件,则跳过此步,按第三步进行)
    如果jenkins上有安装maven的话,则此工具安装就比较简单了,打开jenkins的Global Tool Configuration,找到Allure Commandline,选择安装,如下所示:

如果没有安装maven,则需要去jenkins服务器上安装此工具。

  • 点击管理Jenkins,打开jenkins的Global Tool Configuration,找到Allure Commandline
    配置已安装的jdk的JAVA_HOME,如图

  • 配置Allure Commandline,如图

  • 针对Linux上的远程从节点配置:

  1. 配置远程从节点

  2. 将agent.jar下载到该远程节点Linux的某个目录中,然后在agent.jar所在的目录下,执行所生成的节点命令,即可启动节点,将该节点连接到Jenkins。
  • 针对Windows的远程从节点配置:

    1. 配置远程从节点

    2. 在Windows上启动该节点


      将agent.jar下载到该远程节点windows的某个目录中,然后在agent.jar所在的目录下,执行里面的命令,比如java -jar agent.jar -jnlpUrl http://192.168.201.9:8080/computer/win10_jun/slave-agent.jnlp -secret 1db00accef84f75b239febacc436e834b2164615a459f3b7f00f77a14ed51539 -workDir "E:\jenkins_work"
      即可以将该节点连接到Jenkins,如下

    3. 新建job,配置如下,比如保留7天以内的build,并规定最多只保留10个build




      编写构建脚本

      在命令后,换行,写上 exit 0 (加上exit 0表示执行完成退出)
      添加allure report插件

      配置生成的xml路径和生成html报告的路径

  • 设置邮件通知

  1. 安装插件Email Extension
     

  2. 进入jenkins的系统管理-系统设置,进行相关配置


     

  3. 修改Default Content的内容,具体内容如下:

    $PROJECT_NAME - Build # $BUILD_NUMBER - $BUILD_STATUS:
    
    Check console output at ${BUILD_URL}allure/ to view the results.
    


 

  1. 再进入【系统管理-系统设置】拉到最下面,设置问题追踪,在Allure Report下选择增加:

    Key: allure.issues.tracker.pattern
    
    Value: http://tracker.company.com/%s
    
  • 对构建的job添加邮件发送
  1. job配置页面,添加构建后步骤“Editable Email Notification”,如图

  2. 以下可以使用默认配置:



     

  3. 在Default Content中定义邮件正文,模板如下


<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>${ENV, var="JOB_NAME"}-第${BUILD_NUMBER}次构建日志</title>
</head>
 
<body leftmargin="8" marginwidth="0" topmargin="8" marginheight="4" offset="0">
    <table width="95%" cellpadding="0" cellspacing="0" style="font-size: 11pt; font-family: Tahoma, Arial, Helvetica, sans-serif">
        <tr>
            <td>(本邮件由程序自动下发,请勿回复!)</td>
        </tr>
        <tr>
            <td>
                <h2><font color="#FF0000">构建结果 - ${BUILD_STATUS}</font></h2>
            </td>
        </tr>
        <tr>
            <td><br />
                <b><font color="#0B610B">构建信息</font></b>
                <hr size="2" width="100%" align="center" />
            </td>
        </tr>
        <tr><a href="${PROJECT_URL}">${PROJECT_URL}</a>
            <td>
                <ul>
                    <li>项目名称:${PROJECT_NAME}</li>
                    <li>GIT路径:<a href="${GIT_URL}">${GIT_URL}</a></li>                    
                    <li>构建编号:第${BUILD_NUMBER}次构建</li>                    
                    <li>触发原因:${CAUSE}</li>
                    <li>系统的测试报告 :<a href="${PROJECT_URL}${BUILD_NUMBER}/allure">${PROJECT_URL}${BUILD_NUMBER}/allure</a></li><br />
                    <li>构建日志:<a href="${BUILD_URL}console">${BUILD_URL}console</a></li>
                </ul>
            </td>
        </tr>
        <tr>
            <td>
                <b><font color="#0B610B">变更信息:</font></b>
               <hr size="2" width="100%" align="center" />
            </td>
        </tr>
        <tr>
            <td>
                <ul>
                    <li>上次构建成功后变化 :  ${CHANGES_SINCE_LAST_SUCCESS}</a></li>
                </ul>    
            </td>
        </tr>
 <tr>
            <td>
                <ul>
                    <li>上次构建不稳定后变化 :  ${CHANGES_SINCE_LAST_UNSTABLE}</a></li>
                </ul>    
            </td>
        </tr>
        <tr>
            <td>
                <ul>
                    <li>历史变更记录 : <a href="${PROJECT_URL}changes">${PROJECT_URL}changes</a></li>
                </ul>    
            </td>
        </tr>
        <tr>
            <td>
                <ul>
                    <li>变更集:${JELLY_SCRIPT,template="html"}</a></li>
                </ul>    
            </td>
        </tr>

        <hr size="2" width="100%" align="center" />
 
    </table>

</body>
</html>
  • 在Jenkins上启动测试,如图

  • 启动测试产生的过程日志如下

  • 测试构建完成结果如下

  • 构建完成后,Jenkins自动发送的邮件报告如下(未优化前)

  • 构建完成后,Jenkins自动发送的邮件报告如下(优化后)
     

  • CI集成后,产生的Allure报告如下

  • Jenkins中启动测试项目演示

Logo

权威|前沿|技术|干货|国内首个API全生命周期开发者社区

更多推荐