Jenkins+Python3+Pytest+Allure 接口

Jenkins生成的allure报告

allure报告1
allure报告2

Jenkins配置

1. 路径：Jenkins主页 --->Manage Jenkins--->Manage Plugins--->选中可选插件，右上角过滤一下，安装两个插件。【Allure Jenkins Plugin、HTML Publisher plugin】

插件

 安装完成之后，重启Jenkins。[http://localhost:8080/restart](http://localhost:8080/restart)

2. • 路径：Jenkins主页--->构建执行状态--->设置

     
     设置节点1

   • 配置环境变量：节点属性--->Environment variables，分别填写Allure、JDK、Python目录下Scripts文件夹的键值

     
     设置节点2

3. 新建项目：Jenkins主页--->新建Item，选择自由风格。
   配置：

   项目配置1
   命令：

C:
cd \Users\ziyun\PycharmProjects\api_demo\Test_ziyun
del/f/s/q C:\Users\ziyun\PycharmProjects\api_demo\Test_ziyun\target\allure-results
del/f/s/q C:\Users\ziyun\PycharmProjects\api_demo\Test_ziyun\target\allure-reports
call pytest -s -q --alluredir=target/allure-results
exit 0

项目配置2

指定报告存放位置。
要注意文件生成的目录要保持一致，否则Allure自动生成报告时会报错。

项目配置3
项目配置4

Python框架部分

1. 使用Pytest命名规则，总共分为接口回归用例和测试用例两部分。

   
   代码构成
2. 回归用例部分：

• pytest.mark.parametrize参数化+allure标记，实现方法的封装

import allure
import pytest
import json
import requests
from Common import DataBase
from Common import Request, Assert, read_excel

request = Request.Request()
assertions = Assert.Assertions()
head = {}

url = 'https://.com'

excel_list = read_excel.read_excel_sheet('./document/回归用例.xls', 'DNC系统')
ids_list = []
for i in range(len(excel_list)):
    # 删除excel_list中每个小list的最后一个元素,并赋值给ids_pop
    ids_pop = excel_list[i].pop()
    # 将ids_pop添加到 ids_list 里面
    ids_list.append(ids_pop)


@pytest.mark.ddd
@allure.feature("回归测试")
@allure.severity("blocker")
@allure.issue("https://.com")
class Test_info:

    @allure.story("用户登录")
    @allure.testcase("https://.com",name="正常登录")
    def test_login(self):
        login_resp = request.post_request(url='https://.com' + '/factoryPortal/login',
                                          json={"ssoId": "柒柒", "password": "123456", "userType": ""})
        assertions.assert_code(login_resp.status_code, 200)
        login_resp_json = login_resp.json()
        assertions.assert_in_text(login_resp_json['message'], '成功')
        # 提取token
        data_token = login_resp_json['data']['token']
        # 重新赋值全局变量 head\token_data
        global head
        global token_data
        token_data = data_token
        head = {'token': token_data}
        print(token_data)

    @pytest.mark.parametrize('ff,address,data,key,msg,sql_data', excel_list, ids=ids_list)
    def test_personnel(self, ff, address, data, key, msg, sql_data):
        if sql_data is None or sql_data == '':
            if ff == 'get':
                rp_resp = request.get_request(url=url + address,
                                              params=data, headers=head)
                assertions.assert_code(rp_resp.status_code, 200)
                rp_resp_json = rp_resp.json()
                assertions.assert_in_text(rp_resp_json['message'], msg)
            elif ff == 'post':
                rp_resp = request.post_request(url=url + address,
                                               json=json.loads(data), headers=head)
                assertions.assert_code(rp_resp.status_code, 200)
                rp_resp_json = rp_resp.json()
                assertions.assert_in_text(rp_resp_json['message'], msg)
            elif ff == 'put':
                rp_resp = requests.put(url + address,
                                       json=json.loads(data), headers=head)
                assertions.assert_code(rp_resp.status_code, 200)
                rp_resp_json = rp_resp.json()
                assertions.assert_in_text(rp_resp_json['message'], msg)
        elif sql_data is not None or sql_data != '':
            split = sql_data.split(",", 1)
            db_ = str(split[0])
            sql_ = str(split[1])
            if ff == 'get':
                select_data = DataBase.mysql_select_hg(db=db_, sql=sql_)
                for j in select_data:
                    w = select_data[j]
                all_data = "{}={}".format(key, w)
                rp_resp = request.get_request(url=url + address,
                                              params=data + all_data, headers=head)
                assertions.assert_code(rp_resp.status_code, 200)
                rp_resp_json = rp_resp.json()
                assertions.assert_in_text(rp_resp_json['message'], msg)
            if ff == 'post':
                select_data = DataBase.mysql_select_hg(db=db_, sql=sql_)
                v = 0
                for i in select_data:
                    v1 = select_data[i]
                    v = v1
                v_data = {key: v}
                rp_resp = request.post_request(url=url + address,
                                               json={**json.loads(data), **v_data}, headers=head)
                assertions.assert_code(rp_resp.status_code, 200)
                rp_resp_json = rp_resp.json()
                assertions.assert_in_text(rp_resp_json['message'], msg)
            if ff == 'put':
                select_data = DataBase.mysql_select_hg(db=db_, sql=sql_)
                v = 0
                for i in select_data:
                    v1 = select_data[i]
                    v = v1
                v_data = {key: v}
                rp_resp = requests.put(url + address,
                                       json={**json.loads(data), **v_data}, headers=head)
                assertions.assert_code(rp_resp.status_code, 200)
                rp_resp_json = rp_resp.json()
                assertions.assert_in_text(rp_resp_json['message'], msg)

3. run模块

import pytest
import os


if __name__ == '__main__':
    pytest.main(["-sq",
                 "--alluredir", "./allure-results"])
    os.system(r"allure generate --clean allure-results -o allure-report")

4. pytest 命名规则。

• 新建python类型文件夹，文件夹命名test_开头
• 文件夹内新建.py文件，命名test_开头
• 新建class类，命名Test_开头
• 类下新建方法，命名test_开头

5. 封装连接数据库方法

import pymysql.cursors
import sys

def mysql_select(db, sql, attribute):
    try:
        # 连接mysql数据库
        connection = pymysql.connect(host='MYSQL', port=3306, user='', password='', db=db,  # 'ziyun-dnc',
                                     cursorclass=pymysql.cursors.DictCursor)
        # 通过cursor创建游标,进行每一步操作
        cursor = connection.cursor()
        # 发起请求
        # sql = 'SELECT * FROM `app_package` where app_pk_name = "ws测试产品";'
        cursor.execute(sql)
        # 获取数据
        res = cursor.fetchone()
        # print(res)
        # 赋值变量 parm
        parm = str(res[attribute])
        # 关闭光标对象
        cursor.close()
        # 关闭数据库连接
        connection.close()
        # 返回参数
        return parm
    except Exception as e:
        print("ERROR!!!\n执行sql失败，将自动关闭数据库连接。\t错误信息为：{}".format(e))
        sys.exit()

def mysql_select_hg(db, sql):
    try:
        # 连接mysql数据库
        connection = pymysql.connect(host='MYSQL', port=3306, user='', password='', db=db,  # 'ziyun-dnc',
                                     cursorclass=pymysql.cursors.DictCursor)
        # 通过cursor创建游标,进行每一步操作
        cursor = connection.cursor()
        # 发起请求
        # sql = 'SELECT * FROM `app_package` where app_pk_name = "ws测试产品";'
        cursor.execute(sql)
        # 获取数据
        res = cursor.fetchone()
        # print(res)
        # 赋值变量 parm
        # parm = str(res[attribute])
        # 关闭光标对象
        cursor.close()
        # 关闭数据库连接
        connection.close()
        # 返回参数
        return res
    except Exception as e:
        print("ERROR!!!\n执行sql失败，将自动关闭数据库连接。\t错误信息为：{}".format(e))
        sys.exit()

6. 回归用例excel模板

   
   用例模板

配置完成后，Jenkins界面点击构建

开始构建
测试报告详情