test_case.py

# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName:   test61
# FileName:      test_case.py
# Author:       lao_zhao
# Datetime:     2024/7/12 14:38
# Description:
# 
# ---------------------------------------------------------------------------
import pytest
import allure   # 导入allure
from APIAutoTest_v3_1 import log_decorator
from APIAutoTest_v3_1.common.read_excel import ReadExcel
from APIAutoTest_v3_1.test_case.test_basic import USERNAME


class TestBPM:
    @log_decorator
    @allure.epic("BPM-Basic")
    # @allure.feature("模块名称")
    # @allure.story("接口名称")
    # @allure.title("接口标题")
    # @allure.severity("用例级别")
    @pytest.mark.parametrize("module_name, api_name, level, case_title, case_method, case_url, case_mime, case_data, expect_data, sql_type, sql_data, update_key", ReadExcel(USERNAME).get_data())
    def test_bpm(self, req_fix, db_fix, module_name, api_name, level, case_title, case_method, case_url, case_mime, case_data, expect_data, sql_type, sql_data, update_key):
        # 影响allure报告的输出
        allure.dynamic.feature(module_name)
        allure.dynamic.story(api_name)
        allure.dynamic.title(case_title)
        allure.dynamic.severity(level)

        print(case_method, case_url, case_mime, case_data, expect_data, sql_type, sql_data, update_key)

        # 判断sql语句类型是否为delete，如果是，使用DB类对象调用delete方法执行删除的sql语句
        if sql_type == "delete":
            # 使用DB类对象调用delete方法执行删除的sql语句, DB类对象就是自定义固件db_fix
            db_fix.delete(sql_data)

        # 判断sql语句的类型是否为select，如果是，使用DB类对象调用select方法执行查询语句，并接收查询的结果
        elif sql_type == "select":
            # 使用DB类对象调用select方法执行查询语句，并接收查询的结果
            select_result = db_fix.select(sql_data)
            # 将查询结果更新到用例数据中
            case_data[update_key] = select_result

        # 判断sql语句的类型是否为select|delete 或者 delete|select ，如果是，使用DB类对象执行sql语句
        elif sql_type == "select|delete" or sql_type == "delete|select":
            # 使用DB类对象调用delete方法执行删除的sql语句。
            db_fix.delete(sql_data["delete"])

            # 使用DB类对象调用select方法执行查询的sql语句, 并接收查询的结果
            select_result = db_fix.select(sql_data["select"])
            # 将查询结果更新到用例数据中
            case_data[update_key] = select_result

        result = req_fix.request_all(req_method=case_method, req_url=case_url, req_mime=case_mime, case_data=case_data)
        print(result.text)

        try:
            for key in expect_data:
                assert expect_data[key] == result.json().get(key)
        except AssertionError:
            raise AssertionError("断言失败，描述失败的原因")
        else:
            print("断言成功")