test_bpm.py

# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName:   test62
# FileName:      conftest.py
# Author:       wen
# Datetime:     2024/8/31 下午4:32
# Description:
# 
# ---------------------------------------------------------------------------
import allure
import pytest

from Test_3 import log
from Test_3.common.read_excel import ReadExcel
from Test_3.test_case.test_demo import namespace

# 创建ReadExcel对象
excel = ReadExcel(namespace)


class TestBPM:
    @allure.epic("BPM系统——demo")
    # 设置用例的参数化
    @pytest.mark.parametrize("module_name, api_name, title, level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key", excel.get_data())
    def test_bpm(self, fix_req, fix_db, module_name, api_name, title, level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key):
        # print(url, method, mime, case_data, expect_data, sql_type, sql_data, update_key)
        allure.dynamic.feature(module_name)
        allure.dynamic.story(api_name)
        allure.dynamic.title(title)
        allure.dynamic.severity(level)
        # 判断sql语句的类型是否为delete，如果是，先使用DB类对象调用delete方法执行删除的sql语句，删除完成后，再发送请求
        if sql_type == "delete":
            # 使用DB类对象调用delete方法执行删除语句,fix_db自定义固件在用例中就是DB类对象
            fix_db.delete(sql_data["delete"])
        # 判断sql语句的类型是否为select，如果是，先使用DB类对象调用select方法执行查询的sql语句，获取查询结果后，将查询的结果更新到用例数据中，再发送请求
        elif sql_type == "select":
            # 使用DB类对象调用select方法执行查询语句,fix_db自定义固件在用例中就是DB类对象
            select_result = fix_db.select(sql_data["select"])
            # 更新用例数据中的key
            case_data[update_key] = select_result
        # 判断sql语句的类型是否为delete|select，或者select|delete
        elif sql_type == "delete|select" or sql_type == "select|delete":
            # 使用DB类对象调用delete方法执行删除语句,fix_db自定义固件在用例中就是DB类对象
            fix_db.delete(sql_data["delete"])
            # 使用DB类对象调用select方法执行查询语句,fix_db自定义固件在用例中就是DB类对象
            select_result = fix_db.select(sql_data["select"])
            # 更新用例数据中的key
            case_data[update_key] = select_result
        # 使用RequestMethod类对象调用request_all方法发送请求,fix_req自定义固件在用例中就是RequestMethod类对象
        res = fix_req.request_all(case_url=url, case_method=method, case_mime=mime, case_data=case_data)

        try:
            # 断言
            for key in expect_data.keys():
                # 断言期望数据key对应的值和服务器返回的key对应的值是否相等，相等则断言成功，否则断言失败
                assert expect_data[key] == res.json().get(key)
        except AssertionError:
            # 断言失败后
            log.error(f"断言失败，用例数据为：{case_data}，服务器返回数据为：{res.text}，期望数据为：{expect_data}")
            raise AssertionError("断言失败")
        else:
            # 断言成功后
            log.error(f"断言成功，用例数据为：{case_data}，服务器返回数据为：{res.text}，期望数据为：{expect_data}")


if __name__ == '__main__':
    bpm = TestBPM()
    # bpm.bpm_demo()