# -*-coding:utf-8 -*- # # --------------------------------------------------------------------------- # ProjectName: test62 # FileName: test_bpm.py # Author: lao_zhao # Datetime: 2024/9/4 15:46 # Description: # # --------------------------------------------------------------------------- import pytest from apiAutoTest_v30 import log from apiAutoTest_v30.common.read_excel import ReadExcel from apiAutoTest_v30.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}")