提交 107af0b5 authored 作者: 陈泽健's avatar 陈泽健

refactor(预定系统): 重构代码并优化文件路径处理

- 优化了文件路径的处理方式,使用相对路径和os.path.join方法
- 重构了issue_send_and_upload函数,增加了异常处理机制
- 修改了定时任务执行时间
- 删除了未使用的测试文件
- 统一了类名命名方式
上级 75628d60
......@@ -753,7 +753,7 @@ import subprocess
from venv import logger
def app_init(device_ip, port=5555, adb_path=r"C:\Program Files\androidsdk\androidsdk\platform-tools\adb.exe"):
def app_init(device_ip, port=5555, adb_path=r"C:\Program Files\androidsdk\platform-tools\adb.exe"):
"""
初始化 ADB 连接并检查设备状态
......@@ -767,6 +767,9 @@ def app_init(device_ip, port=5555, adb_path=r"C:\Program Files\androidsdk\androi
返回:
- True: 连接成功且设备可用
- False: 连接失败或设备不可用
- adb_path: r"C:\Program Files\androidsdk\androidsdk\platform-tools\adb.exe"(公司电脑adb路径)
- adb_path:r"C:\Program Files\androidsdk\platform-tools\adb.exe"(自动化虚拟机adb路径)
- adb_path:r"E:\androidsdk\androidsdk\platform-tools"(云桌面adb路径)
"""
# 标记初始化过程
INFO("---------- ADB 连接初始化开始 ----------")
......
......@@ -271,50 +271,90 @@ def issue_send_and_upload(wd, issue_num, issue_name):
输入议题名称以及上传议题文件。
参数:
wd: WebDriver实例,用于操作浏览器。
issue_num: 需要上传的议题文件数量。
issue_name: 会议议题的名称。
"""
wd: WebDriver实例,用于操作浏览器
issue_num: 需要上传的议题文件数量
issue_name: 会议议题的名称
# 议题文件的路径列表
异常:
FileNotFoundError: 当议题文件不存在时抛出
Exception: 其他上传过程中可能出现的异常
"""
try:
# 获取当前文件的绝对路径
current_file_path = os.path.abspath(__file__)
# 获取当前文件的父级目录(即Base目录)
parent_dir = os.path.dirname(current_file_path)
# 获取项目根目录(Base目录的上一级)
project_root = os.path.dirname(parent_dir)
# 构建议题文件相对路径(相对于项目根目录)
issue_file_path = [
r"C:\PycharmData\ubains-module-test\预定系统\reports\issue_file\5.164Scan 安全报告.pdf",
r"C:\PycharmData\ubains-module-test\预定系统\reports\issue_file\议题图片.png"
os.path.join(project_root, "reports", "issue_file", "5.164Scan 安全报告.pdf"),
os.path.join(project_root, "reports", "issue_file", "议题图片.png")
]
# 打印议题文件路径
INFO(f"项目根目录: {project_root}")
INFO(f"议题目录: {os.path.join(project_root, 'reports', 'issue_file')}")
# 打印并输入议题名称
INFO(f"输入议题名称:{issue_name}")
safe_send_keys((By.XPATH, f"(//input[@placeholder='请输入会议议题'])[1]"), f"{issue_name}", wd)
# 点击【上传文件】按钮以开始上传议题文件
INFO("点击【上传文件】按钮")
safe_click((By.XPATH, f"(//div[@class='topicsHandleButton uploadFile'][contains(text(),'上传文件(0)')])[1]"), wd)
safe_click((By.XPATH, f"(//div[@class='topicsHandleButton uploadFile'][contains(text(),'上传文件(0)')])[1]"),
wd)
sleep(2)
# 遍历每个议题文件进行上传
uploaded_files = 0
for i in range(issue_num):
try:
# 检查文件是否存在
if not os.path.exists(issue_file_path[i]):
INFO(f"文件 {issue_file_path[i]} 不存在,跳出函数")
return
raise FileNotFoundError(f"议题文件不存在: {issue_file_path[i]}")
# 定位【选择文件】按钮
upload_button = wd.find_element(By.XPATH, '//*[@id="global-uploader-btn"]/input')
sleep(2)
upload_button = WebDriverWait(wd, 10).until(
EC.presence_of_element_located((By.XPATH, '//*[@id="global-uploader-btn"]/input'))
)
# 选择议题文件上传
upload_button.send_keys(issue_file_path[i])
INFO(f"正在上传文件: {os.path.basename(issue_file_path[i])}")
uploaded_files += 1
# 等待文件上传完成
sleep(15)
except FileNotFoundError as e:
logging.error(f"文件上传失败: {str(e)}")
continue
except Exception as e:
logging.error(f"上传文件时发生异常: {str(e)}")
continue
if uploaded_files == 0:
raise Exception("未成功上传任何议题文件")
# 截取上传完成后的屏幕日志
SELENIUM_LOG_SCREEN(wd, "75%", "Exhibit_Inspect", "Meeting_Message", "添加议题文件")
sleep(20)
sleep(5)
# 点击【确定】按钮完成上传
logging.info("点击【确定】按钮完成上传")
safe_click((By.XPATH,"//div[@aria-label='会议文件上传']//div[@class='el-dialog__footer']//div//span[contains(text(),'确定')]"),wd)
INFO("点击【确定】按钮完成上传")
safe_click(
(By.XPATH,
"//div[@aria-label='会议文件上传']//div[@class='el-dialog__footer']//div//span[contains(text(),'确定')]"),
wd
)
sleep(2)
except Exception as e:
logging.error(f"议题上传过程中出现错误: {str(e)}")
raise # 重新抛出异常让调用者处理
# 清除输入框函数
def input_clear(element_locator, wd):
......@@ -593,6 +633,8 @@ def clear_columns_in_xlsx(xlsx_file_path, sheet_name=None, columns_to_clear=None
logging.info(f"文件 {xlsx_file_path} 已保存,指定列的单元格值已清空。")
except Exception as e:
logging.error(f"保存文件时出错: {e}")
# if __name__ == "__main__":
# # 示例调用
# clear_columns_in_xlsx(r'D:\GithubData\自动化\ubains-module-test\预定系统\测试数据\长安大学测试用例.xlsx',
......
import cv2
import os
def image_matching(floder_name,picture_name):
# 使用绝对路径
path1 = os.path.abspath(r"C:\information_meeting_on.png")
path2 = os.path.abspath(r"C:\information_brithday_on.png")
if not os.path.exists(path1):
print(f"文件不存在: {path1}")
if not os.path.exists(path2):
print(f"文件不存在: {path2}")
# 加载图片
image1 = cv2.imread(path1, cv2.IMREAD_GRAYSCALE)
image2 = cv2.imread(path2, cv2.IMREAD_GRAYSCALE)
# 检查图片是否加载成功
if image1 is None:
print("图片1加载失败,请检查路径或文件是否存在")
if image2 is None:
print("图片2加载失败,请检查路径或文件是否存在")
# 如果图片加载成功,继续执行特征匹配
if image1 is not None and image2 is not None:
# 初始化ORB检测器
orb = cv2.ORB_create()
# 检测关键点和描述符
keypoints1, descriptors1 = orb.detectAndCompute(image1, None)
keypoints2, descriptors2 = orb.detectAndCompute(image2, None)
# 使用BFMatcher进行匹配
bf = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=True)
matches = bf.match(descriptors1, descriptors2)
# 根据匹配点数量判断相似性
matches_count = len(matches)
print(f"匹配点数量: {matches_count}")
if matches_count > 400: # 阈值可以根据需求调整
print("图片相似")
else:
print("图片不相似")
if __name__ == "__main__":
panduan()
\ No newline at end of file
......@@ -6,7 +6,7 @@ from hytest import *
# 配置日志记录
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
class Exhibition_hall_Control:
class ExhibitionhallControl:
"""
执行指令:
1.cd 预定系统
......
......@@ -24,7 +24,7 @@ csv_file_path = os.path.join(current_dir, '../../../测试数据/展厅巡检/
# 定义时间格式
time_format = "%H:%M"
class Exhibition_hall_inspection:
class ExhibitionHallInspection:
"""
执行指令:
1.cd 预定系统
......
......@@ -7,7 +7,7 @@ from hytest import *
# 配置日志记录
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
class Exhibition_hall_NoPaperinspection_000x:
class ExhibitionHallNoPaperinSpection:
"""
执行指令:
1.cd 预定系统
......
from pywinauto import Application, Desktop
from pywinauto.timings import wait_until_passes, TimeoutError
import os
import logging
# 确保日志配置是第一个被执行的语句
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
def check_path():
path = r"D:\腾讯会议\WeMeet\WeMeetApp.exe"
if not os.path.exists(path):
logging.error(f"路径 {path} 不存在")
return False
if not os.access(path, os.X_OK):
logging.error(f"路径 {path} 不可执行")
return False
logging.info(f"路径 {path} 存在且可执行")
return True
def find_window(app, title_re):
try:
window = app.window(title_re=title_re)
logging.info(f"找到窗口: {window.window_text()}")
return window
except Exception as e:
logging.error(f"查找窗口时出错: {e}")
return None
def print_all_window_titles():
desktop = Desktop(backend="uia")
for window in desktop.windows():
logging.info(f"窗口标题: {window.window_text()}")
def tx_run():
logging.debug("tx_run 函数开始执行")
# 启动腾讯会议
try:
app = Application(backend="uia").start(r"D:\腾讯会议\WeMeet\WeMeetApp.exe")
logging.debug("启动腾讯会议完成")
except Exception as e:
logging.error(f"启动腾讯会议时出错: {e}")
return
# 打印所有窗口标题
print_all_window_titles()
# 等待腾讯会议窗口出现
try:
wait_until_passes(60, 1, lambda: app.window(title_re=".*腾讯会议.*").exists())
logging.info("腾讯会议窗口已找到")
except TimeoutError:
logging.error("腾讯会议窗口未在规定时间内出现")
return
# 连接到腾讯会议窗口
main_window = find_window(app, ".*腾讯会议.*")
if main_window is None:
return
# 打印窗口中的所有控件信息
main_window.print_control_identifiers()
# 点击“加入会议”按钮
try:
join_button = main_window.child_window(title="加入会议", control_type="Button")
logging.info(join_button.window_text()) # 打印按钮标题以验证控件是否找到
join_button.click()
logging.info("点击“加入会议”按钮完成")
except Exception as e:
logging.error(f"点击“加入会议”按钮时出错: {e}")
return
# 等待加入会议窗口出现
try:
wait_until_passes(60, 1, lambda: app.window(title_re=".*加入会议.*").exists())
logging.info("加入会议窗口已找到")
except TimeoutError:
logging.error("加入会议窗口未在规定时间内出现")
return
# 输入会议号
meeting_id = "123456789" # 替换为实际的会议号
try:
meeting_id_edit = main_window.child_window(title="会议号", control_type="Edit")
logging.info(meeting_id_edit.window_text()) # 打印编辑框标题以验证控件是否找到
meeting_id_edit.type_keys(meeting_id)
logging.info("输入会议号完成")
except Exception as e:
logging.error(f"输入会议号时出错: {e}")
return
# 点击“加入”按钮
try:
join_meeting_button = main_window.child_window(title="加入", control_type="Button")
logging.info(join_meeting_button.window_text()) # 打印按钮标题以验证控件是否找到
join_meeting_button.click()
logging.info("点击“加入”按钮完成")
except Exception as e:
logging.error(f"点击“加入”按钮时出错: {e}")
return
# 等待会议界面出现
try:
wait_until_passes(60, 1, lambda: app.window(title_re=".*腾讯会议.*").exists())
logging.info("会议界面已找到")
except TimeoutError:
logging.error("会议界面未在规定时间内出现")
return
# 静音麦克风
try:
mute_button = main_window.child_window(title="麦克风", control_type="Button")
logging.info(mute_button.window_text()) # 打印按钮标题以验证控件是否找到
mute_button.click()
logging.info("静音麦克风完成")
except Exception as e:
logging.error(f"静音麦克风时出错: {e}")
return
# 退出会议
try:
end_meeting_button = main_window.child_window(title="结束会议", control_type="Button")
logging.info(end_meeting_button.window_text()) # 打印按钮标题以验证控件是否找到
end_meeting_button.click()
logging.info("结束会议完成")
except Exception as e:
logging.error(f"结束会议时出错: {e}")
return
logging.debug("tx_run 函数执行完毕")
if __name__ == "__main__":
try:
if check_path():
tx_run()
except Exception as e:
logging.error(f"程序执行时出错: {e}")
......@@ -6,13 +6,13 @@ from time import sleep
# 配置日志记录
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
class Exhibition_hall_Txinspection_000x:
class ExhibitionHallTxinspection:
"""
执行指令:
1.cd 预定系统
2.hytest --report_title 会议预约测试报告 --report_url_prefix http://nat.ubainsyun.com:31133 --tag 腾讯会议
"""
tags = ['展厅巡检','腾讯会议']
tags = ['腾讯会议']
def teststeps(self):
"""
......
......@@ -34,7 +34,7 @@ class LanzhouSinopecProject:
name = self.name
# 刷新页面
wd.refresh()
# wd.refresh()
wd.refresh()
if "议题申报" in name:
......@@ -71,6 +71,8 @@ class LanzhouSinopecProject:
SELENIUM_LOG_SCREEN(wd, "75")
elif element_type == "SwitchWindow":
# 将字符转换为int类型
element_value = int(element_value)
wd.switch_to.window(wd.window_handles[element_value])
sleep(2)
SELENIUM_LOG_SCREEN(wd, "75")
......
此差异已折叠。
此差异已折叠。
此差异已折叠。
......@@ -68,10 +68,10 @@ def start_workers(num_workers):
start_workers(3)
# 定时执行预定系统测试任务
schedule.every().day.at("10:00").do(run_task, run_automation_test, report_title="预定系统测试报告", report_url_prefix="http://nat.ubainsyun.com:31134", test_case="JSON测试", ding_type="标准版巡检")
schedule.every().day.at("10:00").do(run_task, run_automation_test, report_title="预定系统测试报告", report_url_prefix="http://nat.ubainsyun.com:31133", test_case="JSON测试", ding_type="标准版巡检")
# 定时执行兰州中石化项目测试任务
schedule.every().day.at("06:00").do(run_task, run_automation_test, report_title="兰州中石化项目测试报告", report_url_prefix="http://nat.ubainsyun.com:31135", test_case="兰州中石化项目", ding_type="标准版巡检")
schedule.every().day.at("15:11").do(run_task, run_automation_test, report_title="兰州中石化项目测试报告", report_url_prefix="http://nat.ubainsyun.com:31135", test_case="兰州中石化项目", ding_type="标准版巡检")
# 定时执行展厅巡检任务
# schedule.every().day.at("07:45").do(run_task, run_automation_test, report_title="展厅巡检测试报告", report_url_prefix="http://nat.ubainsyun.com:31136", test_case="展厅巡检", ding_type="展厅巡检")
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论