test_bpm.py

# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName:   test62
# FileName:      test_bpm.py
# Author:       lao_zhao
# Datetime:     2024/9/4 15:46
# Description:
# 
# ---------------------------------------------------------------------------
import pytest

from apiAutoTest_v3 import log
from apiAutoTest_v3.common.read_excel import ReadExcel
from apiAutoTest_v3.test_case.test_lao_zhang import namespace

excel = ReadExcel(namespace)


class TestBPM:

    @pytest.mark.parametrize("url, method, mime, case_data, expect_data, sql_type, sql_data, update_key", excel.get_data())
    def test_bpm(self, fix_db, fix_req, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key):

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

        # 判断sql语句类型是否为select
        elif sql_type == "select":
            # 使用DB类对象调用select方法执行查询的sql语句， 并获取查询结果，DB类对象===fix_db自定义固件
            select_result = fix_db.select(sql_data['select'])
            # 将查询结果更新到用例数据中
            case_data[update_key] = select_result

        # 判断sql语句的类型是否为select|delete 或者为 delete|select
        elif sql_type == "select|delete" or sql_type == "delete|select":
            # 使用DB类对象调用delete方法执行删除的sql语句，DB类对象===fix_db自定义固件
            fix_db.delete(sql_data['delete'])

            # 使用DB类对象调用select方法执行查询的sql语句， 并获取查询结果，DB类对象===fix_db自定义固件
            select_result = fix_db.select(sql_data['select'])
            # 将查询结果更新到用例数据中
            case_data[update_key] = select_result

        # 使用RequestMethod类对象发送请求
        res = fix_req.request_all(req_method=method, req_url=url, req_mime=mime, case_data=case_data)

        # 断言
        try:
            for key in expect_data.keys():
                assert expect_data[key] == res.json().get(key)
        except AssertionError:
            log.error(f"断言失败，接口url为:{url}, 用例数据:{case_data}, 期望数据:{expect_data}, 服务器返回数据:{res.text}")
            raise AssertionError("断言失败")
        else:
            log.info(f"断言成功，接口url为:{url}, 用例数据:{case_data}, 期望数据:{expect_data}, 服务器返回数据:{res.text}")