Arthur-Wu committed this file on 2024-11-15

This commit is contained in:
ieemoo
2024-11-15 17:35:57 +08:00
parent c871e12b16
commit 298268c133
20 changed files with 595 additions and 18 deletions

View File

@ -0,0 +1,145 @@
# !/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.ServiceApiInfo 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)