优化MQTT模块增加账号密码输入，输出通过读取测试用例的JSON数据格式进行自动化测试的demo程序。

MQTT通用工具/预定系统-门口屏/安卓信息-消息监听.py

@@ -18,12 +18,17 @@ def on_message(client, userdata, msg):
 if __name__ == "__main__":
     # 5.218测试环境 192.168.5.218 1883
 
-    broker_address = "nat.ubainsyun.com"
-    port = 18685
+    broker_address = "192.168.5.235"
+    port = 1883
+    username = "mqtt@cmdb"
+    password = "mqtt@webpassw0RD"
 
     # 创建 MQTT 客户端实例
     mqtt_client = mqtt.Client()
 
+    # 设置用户名和密码
+    mqtt_client.username_pw_set(username=username, password=password)
+
     # 设置回调函数
     mqtt_client.on_connect = on_connect
     mqtt_client.on_message = on_message

预定系统/Base/app_base.py

@@ -412,12 +412,12 @@ def curtain_control(app_drive,wd):
         curtain_up_button.click()
         sleep(2)
     INFO("请检查窗帘上升状态是否正常")
-    SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "curtain_up")
+    # SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "curtain_up")
     # 截图获取当前中控屏软件窗帘上升的界面
-    get_screenshot_with_retry(app_drive, "Control_Manage", "curtain_up")
+    get_screenshot_with_retry(wd,app_drive, "Control_Manage", "curtain_up")
     sleep(30)
     # # 测试报告中补充窗帘上升的截图
-    SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "curtain_rtsp_up")
+    # SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "curtain_rtsp_up")
     # 通过rtsp流获取当前窗帘的上升效果图
     curtain_rtsp_url = "rtsp://admin:huawei@123@192.168.4.18/LiveMedia/ch1/Media2"
     logging.info("开始捕获RTSP流中的帧")
@@ -437,9 +437,9 @@ def curtain_control(app_drive,wd):
         sleep(2)
     sleep(30)
     INFO("请检查窗帘下降状态是否正常")
-    SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "curtain_down")
+    # SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "curtain_down")
     # 截图获取当前中控屏软件窗帘上升的界面
-    get_screenshot_with_retry(app_drive, "Control_Manage", "curtain_down")
+    get_screenshot_with_retry(wd,app_drive, "Control_Manage", "curtain_down")
     # 测试报告中补充窗帘下降的截图
     SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "curtain_rtsp_down")
     logging.info("开始捕获RTSP流中的帧")
@@ -458,9 +458,9 @@ def air_condition_control(app_drive,wd):
 
     # 这是空调开启的状态显示
     INFO("请检查空调开启的状态是否正常")
-    SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "air_condition_on")
+    # SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "air_condition_on")
 
-    get_screenshot_with_retry(app_drive, "Control_Manage", "air_condition_on")
+    get_screenshot_with_retry(wd,app_drive, "Control_Manage", "air_condition_on")
     sleep(2)
 
     # 点击【关闭空调】按钮
@@ -471,8 +471,8 @@ def air_condition_control(app_drive,wd):
     sleep(20)
     # 这是空调关闭的状态显示
     INFO("请检查空调关闭的状态是否正常")
-    SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "air_condition_off")
-    get_screenshot_with_retry(app_drive, "Control_Manage", "air_condition_off")
+    # SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "air_condition_off")
+    get_screenshot_with_retry(wd,app_drive, "Control_Manage", "air_condition_off")
     sleep(2)
 
 def information_control(app_drive,wd):
@@ -552,8 +552,8 @@ def music_control(app_drive,wd):
     sleep(5)
     # 这是音乐开启播放后的界面显示
     INFO("请检查中控屏软件打开音乐播放后的界面状态显示")
-    SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "music_on")
-    get_screenshot_with_retry(app_drive, "Control_Manage", "music_on")
+    # SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "music_on")
+    get_screenshot_with_retry(wd,app_drive, "Control_Manage", "music_on")
 
     # 点击【关闭播放音乐】
     logging.info("尝试定位【关闭播放音乐】按钮元素，并点击按钮")
@@ -563,8 +563,8 @@ def music_control(app_drive,wd):
     sleep(5)
     # 这是音乐关闭播放后的界面显示
     INFO("请检查中控屏软件关闭音乐播放后的界面状态显示")
-    SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "music_off")
-    get_screenshot_with_retry(app_drive, "Control_Manage", "music_off")
+    # SELENIUM_LOG_SCREEN(wd, "50%", "Exhibit_Inspect", "Control_Manage", "music_off")
+    get_screenshot_with_retry(wd,app_drive, "Control_Manage", "music_off")
 
 def command_centre_control(rtsp_url,app_drive,wd):
     open_center_button = app_drive.find_element(AppiumBy.XPATH,
@@ -711,7 +711,7 @@ def app_quit(device_ip,port=5555):
     INFO(f"ADB 连接已断开: {device_address}")
 
 
-def get_screenshot_with_retry(app_drive, module_name, function_name, max_retries=3, retry_delay=5):
+def get_screenshot_with_retry(wd,app_drive, module_name, function_name, max_retries=3, retry_delay=5):
     """
     使用重试机制获取并保存截图。
 
@@ -736,6 +736,9 @@ def get_screenshot_with_retry(app_drive, module_name, function_name, max_retries
     # 构造文件路径
     file_path = os.path.join(target_dir, f"{function_name}.png")
 
+    #截屏
+    SELENIUM_LOG_SCREEN(wd, "75%", "Exhibit_Inspect", "Control_Manage", f"{function_name}")
+
     # 使用循环实现重试机制
     for _ in range(max_retries):
         try:

预定系统/Base/base.py

@@ -433,6 +433,50 @@ def read_csv_data(csv_file_path):
     # 返回包含所有测试用例数据的列表
     return ddt_cases
 
+import openpyxl
+
+def read_xlsx_data(xlsx_file_path):
+    """
+    读取XLSX文件中的数据，并将其转换为一个包含字典的列表，每个字典代表一行测试用例数据。
+
+    参数:
+    xlsx_file_path (str): XLSX文件的路径。
+
+    返回:
+    list: 包含字典的列表，每个字典包含测试用例的名称和参数。
+    """
+    # 打开XLSX文件
+    workbook = openpyxl.load_workbook(xlsx_file_path)
+    # 假设数据在第一个工作表中
+    sheet = workbook.active
+
+    # 读取表头，从第三行开始
+    headers = [cell.value for cell in sheet[2]]
+
+    # 找到表头中名为 'JSON' 的列索引
+    try:
+        json_index = headers.index('JSON')
+    except ValueError:
+        raise ValueError("表头中没有找到名为 'JSON' 的列")
+
+    ddt_cases = []
+    # 遍历XLSX文件中的每一行数据，从第四行开始
+    for row in sheet.iter_rows(min_row=3, values_only=True):
+        # 获取 'JSON' 列的数据
+        json_data = row[json_index]
+        # 将每一行数据转换为字典，其中包含测试用例的名称和参数
+        case = {
+            'name': row[0],
+            'para': json_data
+        }
+        # 将转换后的测试用例数据添加到列表中
+        ddt_cases.append(case)
+    # 日志记录：XLSX文件已读取
+    INFO("XLSX文件已读取")
+    # 返回包含所有测试用例数据的列表
+    return ddt_cases
+
+
 def get_cpu_usage(interval=1):
     """
     获取当前进程的 CPU 占用率。
@@ -1189,3 +1233,152 @@ def kill_ngrok():
 # if __name__ == '__main__':
 #     logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 #     run_http_server()
+
+def user_login(element_locators, username, password, verifycode):
+    """
+    管理员登录函数。
+    该函数通过模拟用户输入用户名、密码和验证码，并点击登录按钮，以实现管理员登录。
+    """
+    # 获取webdriver实例
+    wd = GSTORE['wd']
+
+    # 打印用户名输入信息
+    INFO(f"输入用户名：{username}")
+    # 向用户名输入框发送用户名
+    safe_send_keys(element_locators['username_locator'], f'{username}', wd)
+    sleep(5)
+
+    # 打印密码输入信息
+    INFO(f"输入密码：{password}")
+    # 向密码输入框发送密码
+    safe_send_keys(element_locators['password_locator'], f"{password}", wd)
+    sleep(5)
+
+    # 打印验证码输入信息
+    INFO(f"输入验证码：{verifycode}")
+    # 向验证码输入验证码
+    safe_send_keys(element_locators['verifycode_locator'], f"{verifycode}", wd)
+    sleep(5)
+
+    #点击登录按钮
+    INFO("点击登录按钮")
+    safe_click(element_locators['submitButton_locator'], wd)
+    sleep(5)
+
+def read_xlsx_data(xlsx_file_path):
+    """
+    读取XLSX文件中的数据，并将其转换为一个包含字典的列表，每个字典代表一行测试用例数据。
+
+    参数:
+    xlsx_file_path (str): XLSX文件的路径。
+
+    返回:
+    list: 包含字典的列表，每个字典包含测试用例的名称和参数。
+    """
+    # 打开XLSX文件
+    workbook = openpyxl.load_workbook(xlsx_file_path)
+    # 假设数据在第一个工作表中
+    sheet = workbook.active
+
+    # 读取表头，从第三行开始
+    headers = [cell.value for cell in sheet[3]]
+
+    # 打印表头列名
+    # print(f"表头列名: {headers}")
+
+    # 找到表头中名为 'function'、'Locator_JSON' 和 'Data_JSON' 的列索引
+    try:
+        function_index = headers.index('function')
+        locator_json_index = headers.index('Locator_JSON')
+        data_json_index = headers.index('Data_JSON')
+    except ValueError as e:
+        raise ValueError(f"表头中没有找到所需的列: {e}")
+
+    ddt_cases = []
+    # 遍历XLSX文件中的每一行数据，从第四行开始
+    for row in sheet.iter_rows(min_row=4, values_only=True):
+        # 获取每一列的数据
+        function_name = row[function_index]
+        locator_json_data = row[locator_json_index]
+        data_json_data = row[data_json_index]
+
+        # 打印 Locator_JSON 和 Data_JSON 数据以进行调试
+        # print(f"Locator_JSON 数据: {locator_json_data}")
+        # print(f"Data_JSON 数据: {data_json_data}")
+
+        # 解析 JSON 字符串
+        try:
+            if locator_json_data:
+                parsed_locator_json = json.loads(locator_json_data)
+            else:
+                raise ValueError("Locator_JSON 数据为空")
+        except json.JSONDecodeError:
+            raise ValueError(f"无法解析 Locator_JSON 数据: {locator_json_data}")
+
+        try:
+            if data_json_data:
+                parsed_data_json = json.loads(data_json_data)
+                print(f"这是我想看的{parsed_data_json}")
+            else:
+                raise ValueError("Data_JSON 数据为空")
+        except json.JSONDecodeError:
+            raise ValueError(f"无法解析 Data_JSON 数据: {data_json_data}")
+
+        # 将每一行数据转换为字典，其中包含测试用例的名称、函数名和解析后的 JSON 数据
+        case = {
+            'name': row[0],
+            'function': function_name,
+            'locator_json': parsed_locator_json,
+            'data_json': parsed_data_json
+        }
+        # 将转换后的测试用例数据添加到列表中
+        ddt_cases.append(case)
+
+    # 打印解析后的数据
+    for case in ddt_cases:
+        # print(f"测试用例名称: {case['name']}")
+        # print(f"函数名: {case['function']}")
+        # print(f"Locator JSON: {case['locator_json']}")
+        # print(f"Data JSON: {case['data_json']}")
+
+        # 访问 Locator JSON 中的特定字段
+        locator_json = case['locator_json']
+        type_value = locator_json.get('PageData', {}).get('type')
+        username_locator_value = locator_json.get('PageData', {}).get('username_locator_value')
+        # print(f"Type: {type_value}")
+        # print(f"Username Locator Value: {username_locator_value}")
+
+    # 日志记录：XLSX文件已读取
+    INFO("XLSX文件已读取")
+    # 返回包含所有测试用例数据的列表
+    return ddt_cases
+
+def get_by_enum(type_str):
+    """
+    将字符串类型的定位器类型转换为 selenium.webdriver.common.by.By 枚举类型。
+
+    参数:
+    type_str (str): 定位器类型字符串，例如 'XPATH'。
+
+    返回:
+    selenium.webdriver.common.by.By: 对应的 By 枚举类型。
+    """
+    type_str = type_str.upper()
+    if type_str == 'XPATH':
+        return By.XPATH
+    elif type_str == 'ID':
+        return By.ID
+    elif type_str == 'NAME':
+        return By.NAME
+    elif type_str == 'CLASS_NAME':
+        return By.CLASS_NAME
+    elif type_str == 'CSS_SELECTOR':
+        return By.CSS_SELECTOR
+    elif type_str == 'TAG_NAME':
+        return By.TAG_NAME
+    elif type_str == 'LINK_TEXT':
+        return By.LINK_TEXT
+    elif type_str == 'PARTIAL_LINK_TEXT':
+        return By.PARTIAL_LINK_TEXT
+    else:
+        raise ValueError(f"未知的定位器类型: {type_str}")
\ No newline at end of file

预定系统/README.md

@@ -226,3 +226,5 @@
    - 排查展厅巡检报告中无纸化会议操作的截图显示错误问题，补充上一无纸化同屏流程的结束同屏操作步骤。在初始化函数补充adb连接是否可用的判断，如连接失败，则不进行后续操作。
    - 处理展厅巡检中控屏的信息发布屏流程步骤缺失问题，增加异常重试机制。
    - 封装三个移动端的函数，分别用于设备控制时的异常重试机制以及设备截屏操作，增加相应的异常处理。
+59. 2025-02-26
+   - 优化MQTT模块增加账号密码输入，输出通过读取测试用例的JSON数据格式进行自动化测试的demo程序。
\ No newline at end of file

预定系统/cases/会议主流程/01会议预约/JSON测试.py

+import sys
+import os
+
+# 获取当前脚本的绝对路径
+current_dir = os.path.dirname(os.path.abspath(__file__))
+# 构建预定系统的绝对路径
+预定系统_path = os.path.abspath(os.path.join(current_dir, '..', '..', '..', '..'))
+# 添加路径
+sys.path.append(预定系统_path)
+# 导入模块
+from 预定系统.Base.base import *
+
+# 获取当前脚本所在的目录
+current_dir = os.path.dirname(os.path.abspath(__file__))
+
+# 构建CSV文件的绝对路径
+csv_file_path = os.path.join(current_dir, '../../../测试数据/登录模块/新账号密码登录.xlsx')
+
+
+if __name__ == '__main__':
+    ddt_cases = read_xlsx_data(csv_file_path)
+
+    # 假设我们只处理第一个测试用例
+    if ddt_cases:
+        first_case = ddt_cases[0]
+        function_name = first_case['function']
+        locator_json = first_case['locator_json']
+        data_json = first_case['data_json']
+
+
+        # 提取定位器和数据
+        element_locators = locator_json.get('PageData', {})
+        element_value = data_json.get('testData', {})
+        username_value = element_value.get('username')
+        password_value = element_value.get('password')
+        verifycode_value = element_value.get('verifycode')
+
+        # 构造正确的定位器元组
+        type_value = get_by_enum(element_locators.get('type'))
+        username_locator_value = element_locators.get('username_locator_value')
+        password_locator_value = element_locators.get('password_locator_value')
+        verifycode_locator_value = element_locators.get('verifycode_locator_value')
+        submit_button_locator_value = element_locators.get('submitButton_locator_value')
+
+        # 初始化浏览器
+        browser_init("标准版预定系统")
+
+        # 构造定位器元组
+        username_locator = (type_value, username_locator_value)
+        password_locator = (type_value, password_locator_value)
+        verifycode_locator = (type_value, verifycode_locator_value)
+        submit_button_locator = (type_value, submit_button_locator_value)
+
+        # 调用 user_login 函数
+        user_login({
+            'username_locator': username_locator,
+            'password_locator': password_locator,
+            'verifycode_locator': verifycode_locator,
+            'submitButton_locator': submit_button_locator
+        }, username_value, password_value, verifycode_value)

预定系统/cases/展厅巡检/01展厅中控屏/展厅中控屏.py

@@ -61,10 +61,10 @@ class Exhibition_hall_Control_000x:
 
             # 这是全部灯光关闭后在软件界面上的状态显示
             INFO("请检查中控屏软件的灯光关闭状态显示是否正常")
-            SELENIUM_LOG_SCREEN(wd, "75%", "Exhibit_Inspect", "Control_Manage", "light_all_off")
+            # SELENIUM_LOG_SCREEN(wd, "75%", "Exhibit_Inspect", "Control_Manage", "light_all_off")
 
             # 截图获取当前软件的灯光控制界面
-            get_screenshot_with_retry(app_drive,
+            get_screenshot_with_retry(wd,app_drive,
                                            "Control_Manage", "light_all_off")
             # 调用灯光控制函数
             light_control(app_drive)
@@ -74,7 +74,7 @@ class Exhibition_hall_Control_000x:
             SELENIUM_LOG_SCREEN(wd, "75%", "Exhibit_Inspect", "Control_Manage", "light_all_on")
 
             # 截图获取当前软件的灯光控制界面
-            get_screenshot_with_retry(app_drive,
+            get_screenshot_with_retry(wd,app_drive,
                                       "Control_Manage", "light_all_on")
 
             # 这是灯光开启后的截图

预定系统/cases/展厅巡检/05展厅会议室中控屏/展厅会议室中控屏.py

@@ -12,7 +12,7 @@ class Exhibition_Meeting_Control_0001:
             1.cd 预定系统
             2.hytest --report_title 会议预约测试报告 --report_url_prefix http://192.168.1.225 --tag 展厅会议室中控屏
     """
-    tags = ['展厅会议室中控屏','展厅巡检']
+    tags = ['展厅会议室中控屏','展厅巡检','讯飞转录流程测试']
 
     def teststeps(self):
         """

预定系统/cases/展厅巡检/06讯飞系统巡检/讯飞系统巡检.py

@@ -11,7 +11,7 @@ sys.path.append(预定系统_path)
 from 预定系统.Base.base import *
 
 class XF_Meeting_0001:
-    tags = ['展厅巡检','讯飞测试']
+    tags = ['展厅巡检','讯飞测试','讯飞转录流程测试']
 
     def teststeps(self):
         wd = GSTORE['wd']
@@ -22,7 +22,7 @@ class XF_Meeting_0001:
         sleep(5)
         safe_click((By.XPATH, "//span[contains(text(),'更多操作')]"),wd)
         sleep(5)
-        safe_click((By.CSS_SELECTOR, "ul[id='dropdown-menu-280'] li:nth-child(2)"), wd)
+        safe_click((By.XPATH, "//li[contains(text(),'讯飞纪要')]"), wd)
         sleep(3)
         wd.switch_to.window(wd.window_handles[1])
         sleep(5)