refactor(运维标准版):重构测试框架并优化登录流程

- 重新组织项目目录结构
- 新增浏览器初始化函数
- 优化登录流程，支持不同登录类型
- 更新配置文件读取方式
- 修复部分测试用例中的路径问题

README.md

 # 一、环境运行
-   - python3.10
+   - python3.10.5
    - 需要手动安装库：
      - pip install hytest 
      - pip install selenium

运维集控/项目测试/运维标准版/cases/01登录模块/__st__.py

@@ -4,8 +4,8 @@ sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '.
 from 运维集控.项目测试.运维标准版.lib.base import *
 
 
-# def suite_setup():
-#     wd = GSTORE['wd']
+def suite_setup():
+    wd = GSTORE['wd']
 
 # def suite_teardown():
 #     INFO('进行清除操作')

运维集控/项目测试/运维标准版/cases/01登录模块/管理员登录.py

@@ -3,8 +3,17 @@ from win32trace import flush
 
 from 运维集控.项目测试.运维标准版.lib.base import *
 
-#构建当前项目路径
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', '..')))
+# 获取当前文件所在目录
+current_dir = os.path.dirname(os.path.abspath(__file__))
+# 向上回溯合理层级，假设目标路径为当前文件的上级4级目录
+target_dir = os.path.normpath(os.path.join(current_dir, '..', '..', '..', '..', '..'))
+# 判断路径是否存在再添加，防止无效路径引发后续问题
+if os.path.isdir(target_dir):
+    if target_dir not in sys.path:
+        sys.path.append(target_dir)
+else:
+    raise FileNotFoundError(f"目标路径 {target_dir} 不存在，请检查路径层级是否正确。")
+
 # 构建 CSV 文件的绝对路径
 csv_path = os.path.abspath(
     os.path.join(os.path.dirname(__file__), '..', '..', 'testdata', '01登录模块', '登录信息.csv'))
@@ -25,7 +34,7 @@ class User_Login:
         # 记录当前测试步骤：用户登录
         STEP(1, f'{username} 用户登录测试')
         # 使用从 CSV 读取的数据进行登录
-        user_login(username, password, captcha)
+        admin_login(username, password, captcha)
 
         #记录当前测试步骤：输出提示内容
         STEP(2, f'预期提示内容为：{info}')

运维集控/项目测试/运维标准版/cases/07类型标签/01新增标签.py

@@ -10,7 +10,7 @@ from 运维集控.项目测试.运维标准版.lib.base import *
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', '..')))
 # 构建 CSV 文件的绝对路径
 csv_path = os.path.abspath(
-    os.path.join(os.path.dirname(__file__), '..', '..', 'testdata', '03类型标签', '新增标签.csv'))
+    os.path.join(os.path.dirname(__file__), '..', '..', 'testdata', '07类型标签', '新增标签.csv'))
 
 class AreagroupAdd:
     tag = ['新增类型标签']

运维集控/项目测试/运维标准版/cases/09协议管理/__st__.py

-import sys
-import os
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', '..')))
-from 运维集控.项目测试.运维标准版.lib.base import *
-
-def suite_setup():
-    wd = GSTORE['wd']
-    # user_login("admin@pgy", "ub@123456", "csba")
-    # enter_system()
-    # enter_manage()
-    enter_protocol_manage()
-
-# def suite_teardown():
-#     INFO('进行清除操作')
-#     wd = GSTORE['wd']
-#     wd.quit()
-#

运维集控/项目测试/运维标准版/cases/__st__.py

@@ -7,8 +7,8 @@ from 运维集控.项目测试.运维标准版.lib.base import *
 
 def suite_setup():
     wd = GSTORE['wd']
-    open_browser()
-
+    browser_init("运维标准版")
+    # admin_login("admin@pgy", "ub@123456", "csba")
 
 def suite_teardown():
     INFO('进行清除操作')

运维集控/项目测试/运维标准版/config/config.json

+{
+    "_comment_exhibit_unified_platform": "运维标准版URL",
+    "运维标准版": "https://192.168.5.218:8443/#/login",
+    "初始化网页": "https://www.baidu.com/"
+}
\ No newline at end of file

运维集控/项目测试/运维标准版/lib/base.py

@@ -11,9 +11,9 @@ import hmac
 import hashlib
 import base64
 import time
-import win32api
-import win32con
-import win32gui
+# # import win32api
+# import win32con
+# import win32gui
 import logging
 from hytest import *
 import pandas as pd
@@ -27,54 +27,157 @@ from selenium.webdriver.support import expected_conditions as EC
 from selenium.common import TimeoutException, ElementNotInteractableException, NoSuchElementException
 from selenium.webdriver.common.keys import Keys
 from time import sleep
+from selenium.webdriver.chrome.service import Service
+
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 # 打开浏览器，忽略ssh警告
-def open_browser():
-    INFO('打开默认浏览器')
-    # 更改显示屏分辨率为1920x1080
-    # success = change_resolution(1280, 1024)
-    success = change_resolution(1920, 1080)
-    edge_options = Options()
-    edge_options.add_argument('--ignore-certificate-errors')
-    edge_options.add_argument('--disable-blink-features=AutomationControlled')
-    edge_options.add_argument('--allow-insecure-localhost')
-    wd = webdriver.Edge(options=edge_options)
-    GSTORE['wd'] = wd
-    # wd.get('https://rms.ubainsyun.com/#/login')
-    wd.get('https://192.168.5.218:8443/#/login')
-    wd.maximize_window()
-    wd.implicitly_wait(10)
+def browser_init(login_type):
+    """
+    初始化浏览器设置和实例。
+
+    此函数旨在创建并配置一个Chrome浏览器实例，包括设置Chrome选项以排除不必要的日志，
+    并尝试打开特定的登录页面。任何初始化过程中出现的错误都会被捕获并记录。
+
+    参数:
+    login_type (str): 指定登录类型，根据不同的登录类型选择不同的URL。
+
+    返回:
+    无
+    """
+    # 标记初始化过程的开始
+    INFO("'----------' 正在初始化浏览器 '----------'")
+
+    # 创建Chrome选项实例，用于配置浏览器行为
+    options = webdriver.ChromeOptions()
+    # 添加实验性选项，排除某些命令行开关以减少输出日志
+    options.add_experimental_option('excludeSwitches', ['enable-Logging'])
+    # 忽略证书错误，允许在本地主机上运行时不安全
+    options.add_argument('--ignore-certificate-errors')
+    # 禁用自动化控制特征检测，避免被网站识别为自动化流量
+    options.add_argument('--disable-blink-features=AutomationControlled')
+    # 允许不安全的本地主机运行，通常用于开发和测试环境
+    options.add_argument('--allow-insecure-localhost')
+    options.add_argument('--disable-gpu')             # 禁用 GPU 渲染
+    options.add_argument('--disable-software-rasterizer')  # 禁用软件光栅化器
+    # 使用无痕窗口
+    # options.add_argument('--incognito')
+
+    # 使用webdriver_manager自动下载并管理chromedriver
+    #driver_path = ChromeDriverManager().install()
+    #service = ChromeService(driver_path)
+    # 手动指定ChromeDriver的路径
+    service = Service(r'E:\ubains-module-test\drivers\chromedriver.exe')
+    try:
+        # 创建WebDriver实例
+        wd = webdriver.Chrome(service=service, options=options)
+        # 设置隐式等待时间为10秒，以允许元素加载
+        wd.implicitly_wait(10)
+
+        # 获取登录URL
+        login_url = get_login_url_from_config(login_type)
+        # 打开对应类型的登录页面
+        wd.get(login_url)
+        # 最大化浏览器窗口
+        wd.minimize_window()  # 强制恢复窗口（即使不是最小化也无副作用）
+        time.sleep(1)  # 等待恢复完成
+        wd.maximize_window()  # 再次尝试最大化
+
+        # 将WebDriver实例存储在全局存储器中，以便后续使用
+        GSTORE['wd'] = wd
+        # 标记初始化过程完成
+        INFO("'----------' 浏览器初始化完成 '----------'")
+        return wd
+    except Exception as e:
+        # 捕获并记录初始化过程中的任何异常
+        logging.error(f"浏览器初始化失败：{e}")
+        raise  # 主动抛出异常，防止后续继续执行
+
+# 从配置项config中获取登录URL
+def get_login_url_from_config(login_type):
+    """
+    从配置文件中读取登录URL。
+
+    参数:
+    login_type (str): 指定登录类型，根据不同的登录类型选择不同的URL。
+
+    返回:
+    str: 对应的登录URL。
+    """
+    # 检查 login_type 是否为空或 None
+    if not login_type:
+        raise ValueError("login_type 不能为空")
+
+    # 获取当前脚本的绝对路径
+    current_dir = os.path.dirname(os.path.abspath(__file__))
+    # 构建配置文件的绝对路径，指向 ubains-module-test 目录下的 config.json
+    config_path = os.path.abspath(os.path.join(current_dir, '..', 'config', 'config.json'))
+    # 规范化路径，防止路径遍历攻击
+    config_path = os.path.normpath(config_path)
+    print(f"current_dir: {current_dir}")
+    print(f"config_path: {config_path}")
+
+    # 记录配置文件路径以便调试，对路径进行脱敏处理
+    logging.info(f"配置文件路径: {os.path.basename(config_path)}")
+
+    # 检查文件是否存在
+    if not os.path.exists(config_path):
+        # 如果配置文件不存在，则抛出异常
+        raise FileNotFoundError(f"配置文件 {config_path} 不存在")
+
+    try:
+        # 读取配置文件
+        with open(config_path, 'r', encoding='utf-8') as config_file:
+            # 将配置文件内容解析为 JSON 格式
+            config = json.load(config_file)
+            # 根据 login_type 获取对应的登录 URL
+            login_url = config.get(login_type)
+            # 记录正在打开的登录页面类型和 URL
+            logging.info(f"正在打开 {login_type} 的登录页面：{login_url}")
+    except IOError as e:
+        # 处理文件读取异常
+        raise IOError(f"读取配置文件失败: {e}")
+    except json.JSONDecodeError as e:
+        # 处理 JSON 解析异常
+        raise json.JSONDecodeError(f"解析配置文件失败: {e}")
+
+    # 检查是否成功获取到 URL
+    if not login_url:
+        # 如果未找到对应的 URL，则抛出异常
+        raise ValueError(f"未找到对应的 URL 配置项: {login_type}")
+
+    # 返回登录 URL
+    return login_url
 
 # 定义调整屏幕分辨率，仅虚拟机电脑环境跑定时任务需要使用。
-def change_resolution(width, height):
-    # 获取当前显示器的设备上下文（Device Context, DC）
-    device = win32api.EnumDisplayDevices(None, 0)
-    dm = win32api.EnumDisplaySettings(device.DeviceName, win32con.ENUM_CURRENT_SETTINGS)
-
-    if dm.PelsWidth != width or dm.PelsHeight != height:
-        print(f"Changing resolution to {width}x{height}")
-        dm.PelsWidth = width
-        dm.PelsHeight = height
-
-        # CDS_TEST 是测试模式，如果设置成功则不实际应用更改
-        if win32api.ChangeDisplaySettings(dm, win32con.CDS_TEST) != win32con.DISP_CHANGE_SUCCESSFUL :
-            print("The requested resolution change is not supported.")
-            return False
-
-        # 实际应用新的分辨率设置
-        if win32api.ChangeDisplaySettings(dm, 0) != win32con.DISP_CHANGE_SUCCESSFUL:
-            print("Failed to change resolution.")
-            return False
-
-        print("Resolution changed successfully.")
-        return True
-    else:
-        print("The requested resolution is already set.")
-        return True
+# def change_resolution(width, height):
+#     # 获取当前显示器的设备上下文（Device Context, DC）
+#     device = win32api.EnumDisplayDevices(None, 0)
+#     dm = win32api.EnumDisplaySettings(device.DeviceName, win32con.ENUM_CURRENT_SETTINGS)
+#
+#     if dm.PelsWidth != width or dm.PelsHeight != height:
+#         print(f"Changing resolution to {width}x{height}")
+#         dm.PelsWidth = width
+#         dm.PelsHeight = height
+#
+#         # CDS_TEST 是测试模式，如果设置成功则不实际应用更改
+#         if win32api.ChangeDisplaySettings(dm, win32con.CDS_TEST) != win32con.DISP_CHANGE_SUCCESSFUL :
+#             print("The requested resolution change is not supported.")
+#             return False
+#
+#         # 实际应用新的分辨率设置
+#         if win32api.ChangeDisplaySettings(dm, 0) != win32con.DISP_CHANGE_SUCCESSFUL:
+#             print("Failed to change resolution.")
+#             return False
+#
+#         print("Resolution changed successfully.")
+#         return True
+#     else:
+#         print("The requested resolution is already set.")
+#         return True
 
 # 用户进行登录
-def user_login(username, password, captcha):
+def admin_login(username, password, captcha):
     wd = GSTORE['wd']
     INFO(f'输入登录账号: {username}')
     username_input = WebDriverWait(wd, 10).until(
@@ -376,7 +479,8 @@ def dingding_send_message(test_report_url, title, mobile, ding_type):
     # secret = 'SEC928b11659c5fd6476cfa2042edbf56da876abf759289f7e4d3c671fb9a81bf43'
     # 钉钉机器人的 Webhook URL 和密钥（测试环境）
 
-    if ding_type ==  '标准版巡检':
+    VALID_TYPES = ['标准版巡检', '项目功能验证']
+    if ding_type in VALID_TYPES:
         webhook_url = 'https://oapi.dingtalk.com/robot/send?access_token=7fbf40798cad98b1b5db55ff844ba376b1816e80c5777e6f47ae1d9165dacbb4'
         secret = 'SEC610498ed6261ae2df1d071d0880aaa70abf5e67efe47f75a809c1f2314e0dbd6'
     elif ding_type == '展厅巡检':
@@ -406,7 +510,7 @@ def dingding_send_message(test_report_url, title, mobile, ding_type):
     logging.info(f"钉钉机器人Webhook URL: {final_webhook_url}")
 
     # 调用测试结果获取函数
-    open_browser()
+    browser_init("初始化网页")
     wd = GSTORE['wd']
     # print(latest_report)
     test_result = get_test_result(test_report_url, wd)

运维集控/项目测试/运维标准版/ngrok/ngrok.cfg

@@ -2,6 +2,10 @@ server_addr: "ngrok.ubsyun.com:9083"
 trust_host_root_certs: false
 tunnels:
   nat1:
-    remote_port: 31135
+    remote_port: 31136
     proto:
-      tcp: "192.168.5.88:80"
+      tcp: "192.168.5.88:80"
\ No newline at end of file
+  nat2:
+    remote_port: 30135
+    proto:
+      tcp: "192.168.1.66:8081"
\ No newline at end of file

运维集控/项目测试/运维标准版/ngrok/start.bat

-ngrok -config=ngrok.cfg start nat1
+ngrok -config=ngrok.cfg start nat1 nat2
\ No newline at end of file

运维集控/项目测试/运维标准版/定时任务执行.py

@@ -65,14 +65,8 @@ def start_workers(num_workers):
 
 # 启动3个工作线程
 start_workers(3)
-
-# 定义每天定时执行的任务
-# 每天早上07:50执行后台系统设置功能测试
-for day in ['monday', 'tuesday', 'wednesday', 'thursday', 'friday']:
-    schedule.every().__getattribute__(day).at("11:21").do(run_task, run_automation_test, report_title="运维系统测试报告", report_url_prefix="http://nat.ubainsyun.com:31135", ding_type="标准版巡检")
-
-# 调试使用
-#schedule.every().day.at("09:59").do(run_task, run_automation_test, report_title="运维系统脚本调试", report_url_prefix="http://nat.ubainsyun.com:31135",  ding_type="标准版巡检")
+# 运维自动化测试
+schedule.every().day.at("16:05").do(run_task, run_automation_test, report_title="运维系统自动化报告", report_url_prefix="http://nat.ubainsyun.com:32136",  ding_type="标准版巡检")
 
 try:
     # 无限循环，持续检查并执行计划任务