# !/usr/bin/python # -*- coding: utf-8 -*- import unittest,allure,ddt,time,json,requests from configs.globalObj import LOGGER from configs.globalParams import * from YiMao.config.ApiInfo_web import * from commons.FileHandler import ReadExcel,Txt from commons.AssertLib import _assert_result ExlFilePath = GlobalPath['YMDataPath'] + '/YMHT_ApiTestCase_0.0.1.xlsx' ExcelContent = ReadExcel().get_data(FilePath=ExlFilePath, SheetName='02-门店管理') @ddt.ddt class Test_StoreManagementModule(unittest.TestCase): def setUp(self) -> None: self.timestamp = int(time.time()) @allure.story('[管理平台] 门店管理模块') @ddt.data(*ExcelContent) def test_store_management(self, data): '''@Date:: 2024/5/20 @Desc:: 管理后台 -> 门店管理模块接口 ''' ''' step01：获取测试用例数据 ''' allure.dynamic.title(data['CaseName']) allure.dynamic.description("描述：门店管理模块接口自动化测试") caseModule_c = data['BusinessModule'] # 用例编号 caseNO_c = data['NO.'] # 用例编号 caseName_c = data['CaseName'] # 用例名称 requestType_c = str(data['RequestType']) # 请求方式 expStaCode_c = data['ExpStatusCode'] # 预期的状态码 expRespJson_c = eval(data['RespJson']) # 预期的返回体 LOGGER.info(f">>> 开始执行->{caseModule_c}接口自动化测试 <<<\n") LOGGER.info(f"---用例编号为：{caseNO_c}") LOGGER.info(f"---用例名称为：{caseName_c}") LOGGER.info(f"---请求方式为：{requestType_c}") # LOGGER.info(f"---请求Header为：{HEADERS}") LOGGER.info(f"---预期接口返回码为：{expStaCode_c}") LOGGER.info(f"---预期返回体为：{expRespJson_c}\n") ''' step02：请求接口 ''' if data['BusinessModule'] == '删除门店系统参数': time.sleep(0.5) sysSettingID = 97 storeId = 65 api_url = api_env[ProCfgData['ExecutionEnv']] + ManagementPlatformApi_zh[data['BusinessModule']] # 请求地址 last_api_url = api_url.replace("{StoreID}", str(sysSettingID)) LOGGER.info(f"---请求接口为：{last_api_url}") payload_c = json.dumps(eval(data['Payload']), indent=4, ensure_ascii=False) # 请求体为json格式 LOGGER.info(f"---请求体为：{payload_c}\n") response = requests.request(method=requestType_c, url=last_api_url, headers=HEADERS, data=payload_c) rspJson = response.json() rspJsonFormat = json.dumps(response.json(), indent=4, ensure_ascii=False) LOGGER.info(f"---接口返回状态码为:: {response.status_code}") LOGGER.info(f"---接口返回体为:: {rspJsonFormat}\n") elif data['BusinessModule'] == '修改与删除促销商品白名单': GoodsWhiteListIDList = Txt().read_txt("GoodsWhiteListID.txt") LOGGER.info(f"---促销商品白名单ID列表为：{GoodsWhiteListIDList}") if requestType_c == "DELETE": for i in range(len(GoodsWhiteListIDList)): if GoodsWhiteListIDList[i] not in [None, '']: api_url = api_env[ProCfgData['ExecutionEnv']] + ManagementPlatformApi_zh[ data['BusinessModule']] # 请求地址 last_api_url = api_url.replace("{GoodsWhiteListID}", str(GoodsWhiteListIDList[i])) LOGGER.info(f"---请求接口为：{last_api_url}") payload_c = json.dumps(eval(data['Payload']), indent=4, ensure_ascii=False) # 请求体为json格式 LOGGER.info(f"---请求体为：{payload_c}\n") time.sleep(1) response = requests.request(method=requestType_c, url=last_api_url, headers=HEADERS, data=payload_c) rspJson = response.json() rspJsonFormat = json.dumps(response.json(), indent=4, ensure_ascii=False) LOGGER.info(f"---接口返回状态码为:: {response.status_code}") LOGGER.info(f"---接口返回体为:: {rspJsonFormat}\n") else: api_url = api_env[ProCfgData['ExecutionEnv']] + ManagementPlatformApi_zh[data['BusinessModule']] # 请求地址 last_api_url = api_url.replace("{GoodsWhiteListID}", str(GoodsWhiteListIDList[-2])) LOGGER.info(f"---请求接口为：{last_api_url}") payload_c = json.dumps(eval(data['Payload']), indent=4, ensure_ascii=False) # 请求体为json格式 LOGGER.info(f"---请求体为：{payload_c}\n") response = requests.request(method=requestType_c, url=last_api_url, headers=HEADERS, data=payload_c) rspJson = response.json() rspJsonFormat = json.dumps(response.json(), indent=4, ensure_ascii=False) LOGGER.info(f"---接口返回状态码为:: {response.status_code}") LOGGER.info(f"---接口返回体为:: {rspJsonFormat}\n") elif data['BusinessModule'] == '修改与删除重量放通白名单': WeightWhiteListIDList = Txt().read_txt("WeightWhiteListID.txt") LOGGER.info(f"---删除重量放通白名单ID列表为：{WeightWhiteListIDList}") if requestType_c == "DELETE": for j in range(len(WeightWhiteListIDList)): if WeightWhiteListIDList[j] not in [None, '']: api_url = api_env[ProCfgData['ExecutionEnv']] + ManagementPlatformApi_zh[ data['BusinessModule']] # 请求地址 last_api_url2 = api_url.replace("{WeightWhiteListID}", str(WeightWhiteListIDList[j])) LOGGER.info(f"---请求接口为：{last_api_url2}") payload_c = json.dumps(eval(data['Payload']), indent=4, ensure_ascii=False) # 请求体为json格式 LOGGER.info(f"---请求体为：{payload_c}\n") time.sleep(0.5) response = requests.request(method=requestType_c, url=last_api_url2, headers=HEADERS, data=payload_c) rspJson = response.json() rspJsonFormat = json.dumps(response.json(), indent=4, ensure_ascii=False) LOGGER.info(f"---删除第 {j+1} 条数据，接口返回状态码为:: {response.status_code}") LOGGER.info(f"---第 {j+1} 条数据，接口返回体为:: {rspJsonFormat}\n") else: api_url = api_env[ProCfgData['ExecutionEnv']] + ManagementPlatformApi_zh[data['BusinessModule']] # 请求地址 last_api_url = api_url.replace("{WeightWhiteListID}", str(WeightWhiteListIDList[0])) LOGGER.info(f"---请求接口为：{last_api_url}") payload_c = json.dumps(eval(data['Payload']), indent=4, ensure_ascii=False) # 请求体为json格式 LOGGER.info(f"---请求体为：{payload_c}\n") response = requests.request(method=requestType_c, url=last_api_url, headers=HEADERS, data=payload_c) rspJson = response.json() rspJsonFormat = json.dumps(response.json(), indent=4, ensure_ascii=False) LOGGER.info(f"---接口返回状态码为:: {response.status_code}") LOGGER.info(f"---接口返回体为:: {rspJsonFormat}\n") else: api_url = api_env[ProCfgData['ExecutionEnv']] + ManagementPlatformApi_zh[data['BusinessModule']] # 请求地址 LOGGER.info(f"---请求接口为：{api_url}") payload_c = json.dumps(eval(data['Payload']), indent=4, ensure_ascii=False) # 请求体为json格式 LOGGER.info(f"---请求体为：{payload_c}\n") response = requests.request(method=requestType_c, url=api_url, headers=HEADERS, data=payload_c) rspJson = response.json() rspJsonFormat = json.dumps(response.json(), indent=4, ensure_ascii=False) LOGGER.info(f"---接口返回状态码为:: {response.status_code}") LOGGER.info(f"---接口返回体为:: {rspJsonFormat}\n") if data['BusinessModule'] == '新建促销商品白名单': time.sleep(0.5) goodsWhiteListID = rspJson["data"]['id'] writeResult = Txt().append_write_txt("GoodsWhiteListID.txt", str(goodsWhiteListID)) if not writeResult: LOGGER.error(f"---写入 新建促销商品白名单ID 失败！") elif data['BusinessModule'] == '新建重量放通白名单': time.sleep(0.5) weightWhiteListID = rspJson["data"]['id'] writeResult = Txt().append_write_txt("WeightWhiteListID.txt", str(weightWhiteListID)) if not writeResult: LOGGER.error(f"---写入 新建重量放通白名单ID 失败！") ''' step03: 断言处理 ''' assert _assert_result(rspJson, expRespJson_c, LOGGER) == True, f"---断言失败！" LOGGER.info(f">>> {data['BusinessModule']}接口请求 & 断言执行结束 <<<\n\n") if __name__ == '__main__': unittest.main(verbosity=2)