# -*-coding:utf-8 -*- # # --------------------------------------------------------------------------- # ProjectName: test_56 # FileName: test_bpm.py # Author: xxxxxxx # Datetime: 2023/8/7 15:56 # Description: # # --------------------------------------------------------------------------- import pytest import allure import logging from InterfaceAutoTest.common.read_excel import ReadExcel class TestBPM: @allure.epic("BPM系统") @pytest.mark.parametrize("module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence, update_key", ReadExcel().get_data()) def test_bpm(self, req_fix, db_fix, module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence, update_key): allure.dynamic.feature(module) allure.dynamic.story(interface) allure.dynamic.title(title) allure.dynamic.severity(level) # 1.1:判断sql语句的类型是否为delete if sql_type == "delete": # 1.2 使用DB类对象,调用delete方法执行删除的sql语句 db_fix.delete(sql_sentence) # # 1.3 使用RequestsMethod类对象调用request_all方法发送请求 # result = req_fix.request_all(req_method=method, req_url=url, req_mime=mime, case_data=data) # 2.1 判断sql语句的类型是否为select elif sql_type == "select": # 2.2 使用DB类对象,调用select方法执行查询的sql语句,并接收查询结果 select_result = db_fix.select(sql_sentence) # 2.3 将查询结果更新到用例数据中 data[update_key] = select_result # # 2.3 使用RequestsMethod类对象调用request_all方法发送请求 # result = req_fix.request_all(req_method=method, req_url=url, req_mime=mime, case_data=data) # 3.1 判断sql语句的类型是否delete|select或者为select|delete elif sql_type == "delete|select" or sql_type == "select|delete": # 3.2 使用DB类对象,调用delete方法执行删除的sql语句 db_fix.delete(sql_sentence["delete"]) # 3.3 使用DB类对象,调用select方法执行查询的sql语句,并接收查询结果 select_result = db_fix.select(sql_sentence["select"]) # 3.4 将查询结果更新到用例数据中 data[update_key] = select_result # # 3.5 使用RequestsMethod类对象调用request_all方法发送请求 # result = req_fix.request_all(req_method=method, req_url=url, req_mime=mime, case_data=data) # # else: # result = req_fix.request_all(req_method=method, req_url=url, req_mime=mime, case_data=data) result = req_fix.request_all(req_method=method, req_url=url, req_mime=mime, case_data=data) # 断言 try: for key in expect: assert expect[key] == result.json().get(key) except: print("断言失败") logging.error("断言失败, 模块名称为:{},接口名称为:{},用例标题:{},用例为:{},期望结果为:{},服务器返回数据为:{}".format(module, interface, title, str(data), str(expect), result.text)) raise AssertionError("断言失败") else: print("断言成功") if __name__ == '__main__': pytest.main()