写一篇关于ddt数据驱动的自动化测试

keyword.py

#! encoding:utf-8

"""
author:Jechen

time:2020/08/14

"""

import pandas,json,os,openpyxl,requests,time
from loguru import logger
from public import device,deviceName,gateway,deviceContrl
from multiprocessing import Process


File = open('././conf/config.json','r',encoding='UTF-8')
file = json.load(File)


# 通过字典中的value获取key,返回list
def get_keys(d,value):
    key = [k for k, v in d.items() if v == value]
    return key


# 返回数据表格值
def get_file_and_returnValues(test_file_dir_path):
    for root , dirs, files in os.walk(test_file_dir_path):
        for name in files:
            if name.endswith(".xlsx"):
                pass
            else:
                os.remove(os.path.join(root, name))
    key_list = []
    test_file_list =  os.listdir(test_file_dir_path)
    for fileName in test_file_list:
        os.path.join('{}{}'.format(test_file_list, fileName))
        wb = openpyxl.load_workbook(test_file_dir_path + fileName)
        sheet_names_list = wb.sheetnames
        for sheet_name in sheet_names_list:
            if sheet_name == "配置文件":
                pass
            else:
                pd = pandas.read_excel(io=test_file_dir_path + fileName, sheet_name=sheet_name,keep_default_na=False)
                for rows in pd.index.values:
                    d = dict()
                    # 用例编号
                    d['key1'] = pd.iloc[rows, 0]
                    ID = str(d['key1']).replace("\n","").replace(" ","").replace("，",",")

                    # 测试类型
                    d['key2'] = pd.iloc[rows, 1]
                    Subject = str(d['key2']).replace("\n","").replace(" ","").replace("，",",")

                    # 测试点
                    d['key3'] = pd.iloc[rows, 2]
                    Point = str(d['key3']).replace("\n", "").replace(" ","").replace("，",",")

                    # 用例标题
                    d['key4'] = pd.iloc[rows, 3]
                    Title = str(d['key4']).replace("\n", "").replace(" ","").replace("，",",")

                    # 设备MID
                    list_mid = []
                    d['key5'] = pd.iloc[rows, 4]
                    textTestDevice = str(d['key5']).replace("\n", "").replace(" ","").replace("，",",")
                    list_mid.append(textTestDevice)
                    str_mid = str(list_mid).replace("['","{").replace("']","}")
                    TestDevice = eval(str(str_mid))

                    # 设备条件
                    list_deviceSetUp = []
                    d['key6'] = pd.iloc[rows, 5]
                    textdeviceSetUp = str(d['key6']).replace("\n", "").replace(" ","").replace("，",",")
                    list_deviceSetUp.append(textdeviceSetUp)
                    str_deviceSetUp = str(list_deviceSetUp).replace("['", "{").replace("']", "}")
                    deviceSetUp = eval(str(str_deviceSetUp))

                    # 网关条件
                    list_gatwaySetUp = []
                    d['key7'] = pd.iloc[rows, 6]
                    textgatwaySetUp = str(d['key7']).replace("\n", "").replace(" ","").replace("，",",")
                    list_gatwaySetUp.append(textgatwaySetUp)
                    str_gatwaySetUp = str(list_gatwaySetUp).replace("['", "{").replace("']", "}")
                    gatwaySetUp = eval(str(str_gatwaySetUp))

                    # 操作步骤
                    list_Setpe = []
                    d['key8'] = pd.iloc[rows, 7]
                    textSetpe = str(d['key8']).replace("\n","").replace(" ","").replace("，",",")
                    list_Setpe.append(textSetpe)
                    str_Setpe = str(list_Setpe).replace("['", "{").replace("']", "}")
                    Setpe = eval(str(str_Setpe))

                    # 期望输出队列信息
                    list_Queue = []
                    d['key9'] = pd.iloc[rows, 8]
                    textQueue = str(d['key9']).replace("\n", "").replace(" ","").replace("，",",")
                    list_Queue.append(textQueue)
                    str_Queue = str(list_Queue).replace("['", "{").replace("']", "}")
                    Queue = eval(str(str_Queue))
                    key_list.append([rows,ID,Subject,Point,Title,TestDevice,deviceSetUp,gatwaySetUp,Setpe,Queue])

                return key_list


# 获取数据表格测试设备mid
def get_test_deviceMid(test_file_dir_path):
    for root , dirs, files in os.walk(test_file_dir_path):
        for name in files:
            if name.endswith(".xlsx"):
                pass
            else:
                os.remove(os.path.join(root, name))
    test_file_list =  os.listdir(test_file_dir_path)
    for fileName in test_file_list:
        os.path.join('{}{}'.format(test_file_list, fileName))
        wb = openpyxl.load_workbook(test_file_dir_path + fileName)
        sheet_names_list = wb.sheetnames
        for sheet_name in sheet_names_list:
            if sheet_name == "配置文件":
                pass
            else:
                pd = pandas.read_excel(io=test_file_dir_path + fileName, sheet_name=sheet_name,keep_default_na=False)
                for rows in pd.index.values:
                    d = dict()
                    list_mid = []
                    d['key5'] = pd.iloc[rows, 4]
                    textTestDevice = str(d['key5']).replace("\n", "").replace(" ","").replace("，",",")
                    list_mid.append(textTestDevice)
                    str_mid = str(list_mid).replace("['","{").replace("']","}")
                    TestDevice = eval(str(str_mid))
                    if TestDevice == {}:
                        pass
                    else:
                        return TestDevice


# 测试初始化
def test_setUp(test_title):
    if str(test_title) == "初始队列":
        # 关闭智控总开关，所有设备退出智控
        device.modify(mid=file['母管压力变送器'], code=file['母管压力参数码'], value='7')
        enableArray = file['enableArray']
        enableArrayText = str(enableArray).replace('false', "False")
        enableArrayData = eval(str(enableArrayText))
        url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
        body = {"groupId": file['groupID'], "jointEnable": False, "enableArray": enableArrayData}
        response = requests.post(url=url, json=body, headers=None, verify=False)
        try:
            text = str(response.text).replace("false", "False").replace("true", "True").replace("null", "None")
            jsonData = eval(str(text))
            assert jsonData['ret'] == 1
            logger.info('所有设备退出智控成功！')
        except AssertionError:
            logger.error('所有设备退出智控失败！')
            logger.warning('返回信息：{}'.format(response.text))
        logger.info('停机所有设备...')
        devicemid = file['devicemid']
        for mid in devicemid:
            # 解锁全部设备
            device.controlByDevice(mid=str(mid),type=6)
            # 所有设备在线
            device.controlByDevice(mid=str(mid),type=8)
            # 全部设备关机
            deviceContrl.deviceStop(mid=str(mid))
    else:
        pass


# 设备条件
def test_device_setUp(deviceSetUp):
    getMid = get_test_deviceMid(file['excelPath'])
    if deviceSetUp == {}:
        pass
    else:
        jsonData = eval(str(deviceSetUp))
        for state in jsonData.keys():
            if state == '停机':
                get_device_type = jsonData.get('停机')
                jointEnable_device_list = []
                for device_type in get_device_type:
                    mid = getMid.get(device_type)
                    device.controlByDevice(mid=str(mid), type=2)
                    jointEnable_device_list.append({"enable": True, "mid": str(mid)})
                gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                gatwaybodyTrue = {"groupId": file['groupID'],"enableArray": jointEnable_device_list}
                r = requests.post(url=gatway_url, json=gatwaybodyTrue, headers=None, verify=False)
                text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                gatway_data = eval(str(text))
                assert gatway_data['ret'] == 1


            if state == '运行':
                get_device_type = jsonData.get('运行')
                jointEnable_device_list = []
                for device_type in get_device_type:
                    mid = getMid.get(device_type)
                    device.controlByDevice(mid=str(mid), type=1)
                    jointEnable_device_list.append({"enable": True, "mid": str(mid)})
                gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                gatwaybodyTrue = {"groupId": file['groupID'], "enableArray": jointEnable_device_list}
                r = requests.post(url=gatway_url, json=gatwaybodyTrue, headers=None, verify=False)
                text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                gatway_data = eval(str(text))
                assert gatway_data['ret'] == 1

            if state == '卸载':
                get_device_type = jsonData.get('卸载')
                jointEnable_device_list = []
                for device_type in get_device_type:
                    mid = getMid.get(device_type)
                    device.controlByDevice(mid=str(mid), type=4)
                    jointEnable_device_list.append({"enable": True, "mid": str(mid)})
                gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                gatwaybodyTrue = {"groupId": file['groupID'], "enableArray": jointEnable_device_list}
                r = requests.post(url=gatway_url, json=gatwaybodyTrue, headers=None, verify=False)
                text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                gatway_data = eval(str(text))
                assert gatway_data['ret'] == 1

            if state == '加载':
                get_device_type = jsonData.get('加载')
                jointEnable_device_list = []
                for device_type in get_device_type:
                    mid = getMid.get(device_type)
                    device.controlByDevice(mid=str(mid),type=3)
                    jointEnable_device_list.append({"enable": True, "mid": str(mid)})
                gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                gatwaybodyTrue = {"groupId": file['groupID'], "enableArray": jointEnable_device_list}
                r = requests.post(url=gatway_url, json=gatwaybodyTrue, headers=None, verify=False)
                text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                gatway_data = eval(str(text))
                assert gatway_data['ret'] == 1

            if state == '非智控停机':
                get_device_type = jsonData.get('非智控停机')
                jointEnable_device_list = []
                for device_type in get_device_type:
                    mid = getMid.get(device_type)
                    device.controlByDevice(mid=str(mid), type=2)
                    jointEnable_device_list.append({"enable": False, "mid":mid})
                gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                gatwaybodyTrue = {"groupId": file['groupID'],"enableArray": jointEnable_device_list}
                r = requests.post(url=gatway_url, json=gatwaybodyTrue, headers=None, verify=False)
                text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                gatway_data = eval(str(text))
                assert gatway_data['ret'] == 1

            if state == '非智控运行':
                get_device_type = jsonData.get('非智控运行')
                jointEnable_device_list = []
                for device_type in get_device_type:
                    mid = getMid.get(device_type)
                    device.controlByDevice(mid=str(mid), type=1)
                    jointEnable_device_list.append({"enable": False, "mid":mid})
                gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                gatwaybodyTrue = {"groupId": file['groupID'],"enableArray": jointEnable_device_list}
                r = requests.post(url=gatway_url, json=gatwaybodyTrue, headers=None, verify=False)
                text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                gatway_data = eval(str(text))
                assert gatway_data['ret'] == 1


# 网关初始化
def test_gatway_setUp(gatwaySetUp):
    if gatwaySetUp == {}:
        pass
    else:
        GatwayData = eval(str(gatwaySetUp))
        gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
        jointEnable = GatwayData.get('智控状态')
        if str(jointEnable) == "开启":
            gatwaybodyTrue = {"groupId": file['groupID'],
                           "jointEnable": True,
                           "upLimitPress": GatwayData.get('母管上限压力'),
                           "endPress": GatwayData.get('母管下限压力'),
                           "pressDrop": "",
                           "safePress": "",
                           "startInterval": GatwayData.get('启动时间间隔'),
                           "stopInterval": GatwayData.get('停机时间间隔')
                           }
            r = requests.post(url=gatway_url, json=gatwaybodyTrue, headers=None, verify=False)
            try:
                text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                gatway_data = eval(str(text))
                assert gatway_data['ret'] == 1
                logger.info('网关条件初始化成功！')
            except AssertionError:
                logger.error('网关条件初始化失败！')
                logger.error('请求报文：{}'.format(gatwaybodyTrue))
                logger.warning('返回信息：{}'.format(r.text))
        elif str(jointEnable) == "关闭":
            gatwaybodyFalse = {"groupId": file['groupID'],
                           "jointEnable": False,
                           "upLimitPress": GatwayData.get('母管上限压力'),
                           "endPress": GatwayData.get('母管下限压力'),
                           "pressDrop": "",
                           "safePress": "",
                           "startInterval": GatwayData.get('启动时间间隔'),
                           "stopInterval": GatwayData.get('停机时间间隔')
                           }
            r = requests.post(url=gatway_url, json=gatwaybodyFalse, headers=None, verify=False)
            try:
                text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                gatway_data = eval(str(text))
                assert gatway_data['ret'] == 1
                logger.info('网关条件初始化成功！')
            except AssertionError:
                logger.error('网关条件初始化失败！')
                logger.error('请求报文：{}'.format(gatwaybodyFalse))
                logger.warning('返回信息：{}'.format(r.text))

        for val in GatwayData.keys():
            if val == str('瞬时流量'):
                Instantaneous_flow = GatwayData.get(str(val))
                device.modify(mid='1020070060',code='Instantaneous_flow',value=str(Instantaneous_flow))

            if val == str('累计流量'):
                TOTAL = GatwayData.get(str(val))
                device.modify(mid='1020070060',code='TOTAL',value=str(TOTAL))


# 操作步骤
time_list = []
def test_setpe(Setpe,Title):
    getMid = get_test_deviceMid(file['excelPath'])
    if Setpe == {}:
        pass
    else:
        setpes = eval(str(Setpe))
        for setpe in setpes.keys():
            if setpe == "智控状态":
                join_state = setpes.get(setpe)
                if str(join_state) == "开启":
                    gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                    gatwaybody = {"groupId": file['groupID'], "jointEnable": True}
                    r = requests.post(url=gatway_url, json=gatwaybody, headers=None, verify=False)
                    text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                    gatway_data = eval(str(text))
                    try:
                        assert gatway_data['ret'] == 1
                        logger.info('智控总开关开启成功！')
                    except AssertionError:
                        logger.error('智控总开关开启失败！')
                        logger.warning('请求报文：{}'.format(gatwaybody))
                        logger.warning('返回信息：{}'.format(r.text))
                        break

                if str(join_state) == "关闭":
                    gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                    gatwaybody = {"groupId": file['groupID'], "jointEnable": False}
                    r = requests.post(url=gatway_url, json=gatwaybody, headers=None, verify=False)
                    text = str(r.text).replace("false", "False").replace("true", "True").replace("null", "None")
                    gatway_data = json.loads(str(text))
                    try:
                        assert gatway_data['ret'] == 1
                        logger.info('智控总开关关闭成功！')
                    except AssertionError:
                        logger.error('智控总开关关闭失败！')
                        logger.warning('请求报文：{}'.format(gatwaybody))
                        logger.warning('返回信息：{}'.format(r.text))
                        break

            if setpe == "母管压力":
                device.modify(mid=file["母管压力变送器"],code=file["母管压力参数码"],value=str(setpes.get(setpe)))

            if setpe == "等待时间":
                excel_wait_time = setpes.get(str(setpe))
                logger.info('需要等待{}秒'.format(excel_wait_time))
                if time_list == []:
                    count = 0
                    while True:
                        count += 1
                        time.sleep(1)
                        logger.info("等待时间：{}s".format(count))
                        if count == excel_wait_time:
                            time_list.clear()
                            time_list.append(excel_wait_time)
                            break
                else:
                    if Title == "初始队列":
                        time_list.clear()
                        excel_wait_time = setpes.get(str(setpe))
                        logger.info('需要等待{}秒'.format(excel_wait_time))
                        count = 0
                        while True:
                            count += 1
                            time.sleep(1)
                            logger.info("等待时间：{}s".format(count))
                            if count == excel_wait_time:
                                time_list.append(excel_wait_time)
                                break
                    else:
                        wait = int(excel_wait_time) - int(time_list[0])
                        count = 0
                        while True:
                            count += 1
                            time.sleep(1)
                            logger.info("等待时间：{}s".format(count))
                            if count == wait:
                                time_list.append(excel_wait_time)
                                time_list.remove(time_list[0])
                                break

            if setpe == "在线设备":
                online_device_list = setpes.get(setpe)
                for online_device in online_device_list:
                    for device_name in getMid.keys():
                        if str(online_device) == str(device_name):
                            mid = getMid.get(str(device_name))
                            device.controlByDevice(mid=str(mid), type=8)

            if setpe == "模拟器停机":
                stop_device_list = setpes.get(setpe)
                for stop_device in stop_device_list:
                    for device_name in getMid.keys():
                        if str(stop_device) == str(device_name):
                            mid = getMid.get(str(device_name))
                            device.controlByDevice(mid=str(mid), type=2)

            if setpe == "模拟器运行":
                run_device_list = setpes.get(setpe)
                for run_device in run_device_list:
                    for device_name in getMid.keys():
                        if str(run_device) == str(device_name):
                            mid = getMid.get(str(device_name))
                            device.controlByDevice(mid=str(mid), type=1)

            if setpe == "网关运行":
                run_device_list = setpes.get(setpe)
                for run_device in run_device_list:
                    for device_name in getMid.keys():
                        if str(run_device) == str(device_name):
                            mid = getMid.get(str(device_name))
                            gateway.controlDeviceByGeteway(mid=str(mid),type=1)

            if setpe == "网关停机":
                run_device_list = setpes.get(setpe)
                for run_device in run_device_list:
                    for device_name in getMid.keys():
                        if str(run_device) == str(device_name):
                            mid = getMid.get(str(device_name))
                            gateway.controlDeviceByGeteway(mid=str(mid), type=2)

            if setpe == "离线设备":
                offline_device_list = setpes.get(setpe)
                for offline_device in offline_device_list:
                    for device_name in getMid.keys():
                        if str(offline_device) == str(device_name):
                            mid = getMid.get(str(device_name))
                            device.controlByDevice(mid=str(mid), type=7)

            if setpe == "锁定设备":
                lock_device_list = setpes.get(setpe)
                for lock_device in lock_device_list:
                    for device_name in getMid.keys():
                        if str(device_name) == str(lock_device):
                            mid = getMid.get(str(device_name))
                            device.controlByDevice(mid=str(mid), type=5)

            if setpe == "解锁设备":
                unlock_device_list = setpes.get(setpe)
                for unlock_device in unlock_device_list:
                    for device_name in getMid.keys():
                        if str(device_name) == str(unlock_device):
                            mid = getMid.get(str(unlock_device))
                            device.controlByDevice(mid=str(mid), type=6)

            if setpe == "退出智控":
                unJoin_device_list = setpes.get(setpe)
                for unJoin_device in unJoin_device_list:
                    for device_name in getMid.keys():
                        if str(unJoin_device) == str(device_name):
                            mid = getMid.get(str(device_name))
                            enableArray = []
                            enableArray.append({"enable": False, "mid": str(mid)})
                            gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                            gatway_body = {"groupId": file['groupID'],"enableArray": enableArray}
                            r = requests.post(url=gatway_url, json=gatway_body, headers=None, verify=False)
                            name = deviceName.deviceMid(mid=str(mid))
                            try:
                                text = str(r.text).replace("false","False").replace("true","True").replace("null","None")
                                gatway_data = json.loads(str(text))
                                assert gatway_data['ret'] == 1
                                logger.info('{}退出智控成功！'.format(name))
                            except AssertionError:
                                logger.error('{}退出智控失败！'.format(name))
                                logger.warning('返回信息：{}'.format(r.text))
                                break

            if setpe == "加入智控":
                join_device_list = setpes.get(setpe)
                for join_device in join_device_list:
                    for device_name in getMid.keys():
                        if str(join_device) == str(device_name):
                            mid = getMid.get(str(device_name))
                            enableArray = []
                            enableArray.append({"enable": False, "mid": str(mid)})
                            gatway_url = file['url'] + file['gatewayPort'] + file['gateway']['modifyByGeteway']
                            gatway_body = {"groupId": file['groupID'],"enableArray": enableArray}
                            r = requests.post(url=gatway_url, json=gatway_body, headers=None, verify=False)
                            name = deviceName.deviceMid(mid=str(mid))
                            try:
                                text = str(r.text).replace("false","False").replace("true","True").replace("null","None")
                                gatway_data = json.loads(str(text))
                                assert gatway_data['ret'] == 1
                                logger.info('{}加入智控成功！'.format(name))
                            except AssertionError:
                                logger.error('{}加入智控失败！'.format(name))
                                logger.warning('返回信息：{}'.format(r.text))
                                break

            if setpe == "累计流量":
                TOTAL = setpes.get(str(setpe))
                device.modify(mid='1020070060',code='TOTAL',value=str(TOTAL))

            if setpe == "瞬时流量":
                Instantaneous_flow = setpes.get(str(setpe))
                device.modify(mid='1020070060',code='Instantaneous_flow',value=str(Instantaneous_flow))



# 获取实时队列信息断言
def get_gatway_Queue_Assert(test_id,queue_info):
    # 断言
    if queue_info == {}:
        pass
    else:
        queue = gateway.getQueue(device_type=1)
        readyqueueList = []
        runqueueList = []
        readyqueuelist_1 = []
        readyqueuelist_2 = []
        readyqueuelist_3 = []
        readyqueuelist_4 = []
        runqueue_1 = []
        runqueue_2 = []
        runqueue_3 = []
        runqueue_4 = []
        Queue = {"待机队列":{"1":readyqueuelist_1,"2":readyqueuelist_2,"3":readyqueuelist_3,"4":readyqueuelist_4},
                 "运行队列":{"1":runqueue_1,"2":runqueue_2,"3":runqueue_3,"4":runqueue_4}}

        # 获取待机队列设备
        try:
            ready__Queue = queue['readyQueue']
            excel_device_mid = get_test_deviceMid(file['excelPath'])
            lst = []
            for level_mid in ready__Queue:
                get_mid_interface = level_mid['mid']
                for deviceName_mid in excel_device_mid:
                    get_mid_excel = excel_device_mid[deviceName_mid]
                    if str(get_mid_interface) == str(get_mid_excel):
                        n = str(level_mid).replace(str(get_mid_interface), str(deviceName_mid), 10)
                        lst.append(n)
            for alldevice in lst:
                alldeviceData = eval(alldevice)
                del alldeviceData['time']
                readyqueueList.append(alldeviceData)
                for readyQueueDevice in readyqueueList:
                    text = eval(str(readyQueueDevice))
                if text['level'] == 1:
                    readyqueuelist_1.append(text['mid'])
                elif text['level'] == 2:
                    readyqueuelist_2.append(text['mid'])
                elif text['level'] == 3:
                    readyqueuelist_3.append(text['mid'])
                elif text['level'] == 4:
                    readyqueuelist_4.append(text['mid'])
        except Exception:
            pass

        l3 = []
        l4 = []
        for device in Queue['待机队列'].values():
            if device == []:
                pass
            else:
                key = get_keys(d=Queue['待机队列'], value=device)
                l3.append(key[0])
                l4.append(device)
        d_readyqueue = dict(zip(l3, l4))

        # 获取运行队列设备
        try:
            run__Queue = queue['runQueue']
            device_mid_text = get_test_deviceMid(file['excelPath'])
            lst_run = []
            for m1 in run__Queue:
                mid1 = m1['mid']
                for m2 in device_mid_text:
                    mid2 = device_mid_text[m2]
                    if str(mid1) == str(mid2):
                        n = str(m1).replace(str(mid1),str(m2),10)
                        lst_run.append(n)
            for val in lst_run:
                j = eval(str(val))
                del j['time']
                runqueueList.append(j)
                for runQueueDevice in runqueueList:
                    textrun = eval(str(runQueueDevice))
                if textrun['level'] == 1:
                    runqueue_1.append(textrun['mid'])
                elif textrun['level'] == 2:
                    runqueue_2.append(textrun['mid'])
                elif textrun['level'] == 3:
                    runqueue_3.append(textrun['mid'])
                elif textrun['level'] == 4:
                    runqueue_4.append(textrun['mid'])
        except Exception:
            pass

        l1 = []
        l2 = []
        for device in Queue['运行队列'].values():
            if device == []:
                pass
            else:
                key = get_keys(d=Queue['运行队列'], value=device)
                l1.append(key[0])
                l2.append(device)
        d_runqueue = dict(zip(l1, l2))
        all_queue = {"待机队列":d_readyqueue,"运行队列":d_runqueue}
        print('网关队列信息：{}'.format(all_queue))
        print('期望队列信息：{}'.format(queue_info))
        logger.info('网关队列信息：{}'.format(all_queue))
        logger.info('期望队列信息：{}'.format(queue_info))
        assert all_queue == queue_info
        logger.info('用例ID：{} 期望队列和网关队列信息一致！'.format(test_id))

测试用例框架文件

test_main.py

#! encoding:utf-8

"""
author:Jechen

time:2020/08/14

"""
import unittest,json,time
from public.my_ddt import ddt,data,unpack
from public import keyWord
from loguru import logger

File = open('././conf/config.json','r',encoding='UTF-8')
file = json.load(File)

@ddt
class case(unittest.TestCase):


    @data(*keyWord.get_file_and_returnValues(file['excelPath']))
    @unpack
    def test_case(self,rows,ID,Subject,Point,Title,TestDevice,deviceSetUp,gatwaySetUp,Setpe,Queue):
        if Subject == "":
            pass
        else:
            print("测试类型：{}".format(Subject))
            logger.info("测试类型：{}".format(Subject))
        if Point == "":
            pass
        else:
            print("用例测试点：{}".format(Point))
            logger.info("用例测试点：{}".format(Point))
        if ID == "":
            pass
        else:
            print('用例ID：{}'.format(ID))
            logger.info('用例ID：{}'.format(ID))
        if Title == "":
            pass
        else:
            print("用例标题:{}".format(Title))
            logger.info("用例标题:{}".format(Title))

        # 网关智控初始化
        keyWord.test_setUp(test_title=Title)

        # 设备条件初始化
        keyWord.test_device_setUp(deviceSetUp=deviceSetUp)

        # 网关条件初始化
        keyWord.test_gatway_setUp(gatwaySetUp=gatwaySetUp)

        # 用例执行操作步骤
        keyWord.test_setpe(Setpe=Setpe,Title=Title)

        # 用例断言
        time.sleep(3)
        keyWord.get_gatway_Queue_Assert(test_id=ID,queue_info=Queue)


if __name__=="__main__":
    unittest.main()

测试用例数据源：test.xslx

运行结果：

有天大的问题可以滴滴我~