266 lines
13 KiB
Python
266 lines
13 KiB
Python
from framework.config.settings import Config
|
||
from framework.core.logger import get_logger
|
||
from framework.business.login_page import LoginPage
|
||
from framework.business.file_manager_page import FileManagerPage
|
||
from framework.business.dev_machine_page import DevMachinePage
|
||
from framework.business.cloud_desktop_page import CloudDesktopPage
|
||
|
||
# 引入抽离出来的业务场景
|
||
from framework.scripts.file_system_scenario import run_full_file_lifecycle
|
||
from framework.scripts.compute_resource_scenario import run_dev_machine_lifecycle
|
||
from framework.scripts.desktop_lifecycle import run_cloud_desktop_lifecycle
|
||
from framework.scripts.mirror_assets import run_mirror_assets_lifecycle
|
||
from framework.scripts.three_d_generation_scenario import run_3d_generation_lifecycle
|
||
from framework.scripts.three_d_assets_scenario import run_3d_assets_lifecycle
|
||
from framework.scripts.quantization_scenario import run_quantization_lifecycle
|
||
from framework.scripts.monkey_scenario import run_monkey_testing
|
||
from framework.business.mirror_assets_page import MirrorAssetsPage
|
||
from framework.business.three_d_generation_page import ThreeDGenerationPage
|
||
from framework.business.three_d_assets_page import ThreeDAssetsPage
|
||
from framework.business.quantization_page import QuantizationPage
|
||
from framework.scripts.quantization_scenario import run_quantization_lifecycle
|
||
|
||
import time
|
||
import os
|
||
import json
|
||
|
||
logger = get_logger("DataManagementRunner")
|
||
|
||
class DataManagement:
|
||
"""符合 PO 模式的场景指挥官"""
|
||
|
||
def __init__(self, headless=False):
|
||
self.ui = LoginPage(headless=headless)
|
||
self.page = None
|
||
self.fm = None
|
||
self.dm = None
|
||
self.cd = None
|
||
self.ma = None
|
||
self.tg = None
|
||
self.ta = None
|
||
self.qp = None
|
||
self.results = [] # 记录结构化测试结果
|
||
|
||
def start(self):
|
||
"""启动浏览器并初始化组件"""
|
||
self.page = self.ui.start()
|
||
# 初始化页面原子对象
|
||
self.fm = FileManagerPage(self.page)
|
||
self.dm = DevMachinePage(self.page)
|
||
self.cd = CloudDesktopPage(self.page)
|
||
self.ma = MirrorAssetsPage(self.page)
|
||
self.tg = ThreeDGenerationPage(self.page)
|
||
self.ta = ThreeDAssetsPage(self.page)
|
||
self.qp = QuantizationPage(self.page)
|
||
|
||
def login(self, user, pwd):
|
||
"""执行登录流程"""
|
||
return self.ui.login(user, pwd)
|
||
|
||
def _safe_screenshot(self, name):
|
||
"""支持统一存放路径的截图"""
|
||
base_dir = os.environ.get("ROBOGO_SCREENSHOTS_DIR", ".")
|
||
task_id = os.environ.get("ROBOGO_TASK_ID", "local")
|
||
|
||
# 统一命名规范: {task_id}_{name}.png
|
||
filename = f"{task_id}_{name}" if ".png" in name else f"{task_id}_{name}.png"
|
||
target_path = os.path.join(base_dir, filename)
|
||
|
||
try:
|
||
# 增加对 page 状态的活跃检查 (包含 context 检查)
|
||
if not self.page or self.page.is_closed():
|
||
return
|
||
if not self.page.context or self.page.context.browser is None:
|
||
return
|
||
|
||
# 使用 self.page 截图（注意：这里使用的是 BasePage 注入的 playwirght 页面）
|
||
self.page.screenshot(path=target_path, full_page=True, timeout=5000)
|
||
logger.info(f"✨ 最终状态截图已保存: {target_path}")
|
||
except Exception as e:
|
||
# 忽略由于浏览器/上下文已关闭导致的截图失败
|
||
err_msg = str(e).lower()
|
||
if "closed" in err_msg or "not open" in err_msg:
|
||
logger.info("ℹ️ 浏览器环境已销毁，跳过末尾截图存档")
|
||
else:
|
||
logger.warning(f"⚠️ 截图存档过程出错: {e}")
|
||
|
||
def record_result(self, name, desc, expected, status, duration):
|
||
"""记录一个测试用例的结果"""
|
||
self.results.append({
|
||
"name": name,
|
||
"desc": desc,
|
||
"expected": expected,
|
||
"status": status, # 'PASS', 'FAIL', 'SKIP'
|
||
"duration": f"{duration:.2f}s"
|
||
})
|
||
|
||
def save_results(self):
|
||
"""将结构化结果保存到文件"""
|
||
task_id = os.environ.get("ROBOGO_TASK_ID", "local")
|
||
reports_dir = os.environ.get("ROBOGO_REPORTS_DIR", "platform_reports")
|
||
os.makedirs(reports_dir, exist_ok=True)
|
||
target_path = os.path.join(reports_dir, f"{task_id}_results.json")
|
||
try:
|
||
with open(target_path, 'w', encoding='utf-8') as f:
|
||
json.dump(self.results, f, ensure_ascii=False, indent=2)
|
||
logger.info(f"📊 结构化数据已保存: {target_path}")
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ 结构化数据保存失败: {e}")
|
||
|
||
def run_all_scenarios(self):
|
||
"""
|
||
场景指挥：依次执行所有的业务流
|
||
每个场景独立 try-except，一个失败不阻塞后续场景
|
||
"""
|
||
env_name = os.environ.get("ROBOGO_ENV", "PROD")
|
||
scope = os.environ.get("ROBOGO_SCOPE", "all")
|
||
logger.info(f"🚦 启动场景编排 - 环境: {env_name} | 策略: {scope}")
|
||
|
||
# 动态定义本次要运行的模块
|
||
active_modules = []
|
||
if scope == "all":
|
||
active_modules = ["file", "dev", "cloud", "mirror", "quant" , "3d"]
|
||
elif scope == "smoke":
|
||
active_modules = ["file"] # 核心资源生命周期
|
||
elif scope == "core":
|
||
active_modules = ["cloud","dev", "mirror", "quant"] # 业务闭环
|
||
else:
|
||
active_modules = ["cloud"] # 默认兜底只跑云桌面
|
||
|
||
errors = []
|
||
|
||
# 1. 执行文件系统场景
|
||
if "file" in active_modules:
|
||
start_t = time.time()
|
||
try:
|
||
run_full_file_lifecycle(self.fm, Config.FOLDER_NAME)
|
||
self._safe_screenshot("file_system_final.png")
|
||
logger.info("✅ 文件系统场景通过")
|
||
self.record_result("file_system_lifecycle", "文件系统全生命周期测试", "文件夹创建、上传、重命名、删除全链路闭环", "PASS", time.time()-start_t)
|
||
except Exception as e:
|
||
logger.error(f"❌ 文件系统场景失败: {e}")
|
||
self._safe_screenshot("file_system_error.png")
|
||
errors.append(f"文件系统: {e}")
|
||
self.record_result("file_system_lifecycle", "文件系统全生命周期测试", "文件夹创建、上传、重命名、删除全链路闭环", "FAIL", time.time()-start_t)
|
||
|
||
# 2. 执行开发机场景
|
||
if "dev" in active_modules:
|
||
start_t = time.time()
|
||
try:
|
||
run_dev_machine_lifecycle(self.dm)
|
||
self._safe_screenshot("dev_machine_final.png")
|
||
logger.info("✅ 开发机场景通过")
|
||
self.record_result("dev_machine_lifecycle", "开发机全生命周期测试", "开发机申请、启动、关机、销毁全流程验证", "PASS", time.time()-start_t)
|
||
except Exception as e:
|
||
logger.error(f"❌ 开发机场景失败: {e}")
|
||
self._safe_screenshot("dev_machine_error.png")
|
||
errors.append(f"开发机: {e}")
|
||
self.record_result("dev_machine_lifecycle", "开发机全生命周期测试", "开发机申请、启动、关机、销毁全流程验证", "FAIL", time.time()-start_t)
|
||
|
||
# 3. 执行云桌面场景
|
||
if "cloud" in active_modules:
|
||
start_t = time.time()
|
||
try:
|
||
run_cloud_desktop_lifecycle(self.cd)
|
||
self._safe_screenshot("cloud_desktop_final.png")
|
||
logger.info("✅ 云桌面场景通过")
|
||
self.record_result("cloud_desktop_lifecycle", "云桌面全生命周期测试", "桌面创建、连接、保存镜像、关机、删除全流程验证", "PASS", time.time()-start_t)
|
||
except Exception as e:
|
||
logger.error(f"❌ 云桌面场景失败: {e}")
|
||
self._safe_screenshot("cloud_desktop_error.png")
|
||
errors.append(f"云桌面: {e}")
|
||
self.record_result("cloud_desktop_lifecycle", "云桌面全生命周期测试", "桌面创建、连接、保存镜像、关机、删除全流程验证", "FAIL", time.time()-start_t)
|
||
|
||
# 4. 执行镜像资产场景
|
||
if "mirror" in active_modules:
|
||
start_t = time.time()
|
||
try:
|
||
run_mirror_assets_lifecycle(self.ma, self.cd)
|
||
self._safe_screenshot("mirror_assets_final.png")
|
||
logger.info("✅ 镜像资产场景通过")
|
||
self.record_result("mirror_assets", "镜像资产巡检", "镜像列表加载及基本信息验证", "PASS", time.time()-start_t)
|
||
except Exception as e:
|
||
logger.error(f"❌ 镜像资产场景失败: {e}")
|
||
self._safe_screenshot("mirror_assets_error.png")
|
||
errors.append(f"镜像资产: {e}")
|
||
self.record_result("mirror_assets", "镜像资产巡检", "镜像列表加载及基本信息验证", "FAIL", time.time()-start_t)
|
||
|
||
# 5. 执行 3D 生成场景 -> 获取资产名 -> 执行归档场景
|
||
if "3d" in active_modules:
|
||
start_t = time.time()
|
||
try:
|
||
asset_name = run_3d_generation_lifecycle(self.tg)
|
||
self._safe_screenshot("3d_generation_final.png")
|
||
|
||
if asset_name:
|
||
# 联动：将生成的资产归档到数据中心
|
||
run_3d_assets_lifecycle(self.ta, asset_name)
|
||
self._safe_screenshot("3d_assets_archive_final.png")
|
||
logger.info("✅ 3D 生成与归档全链路已通过")
|
||
self.record_result("3d_lifecycle", "3D生成与资产归档", "通过 AIGC 生成 3D 模型并成功归档到资产中心", "PASS", time.time()-start_t)
|
||
else:
|
||
logger.warning("⚠️ 3D 生成未产生有效资产名，跳过归档场景")
|
||
self.record_result("3d_lifecycle", "3D生成与资产归档", "生成资产名为空", "SKIP", time.time()-start_t)
|
||
except Exception as e:
|
||
logger.error(f"❌ 3D 链路失败: {e}")
|
||
self._safe_screenshot("3d_chain_error.png")
|
||
errors.append(f"3D链路: {e}")
|
||
self.record_result("3d_lifecycle", "3D生成与资产归档", "存在执行错误", "FAIL", time.time()-start_t)
|
||
|
||
# 6. 执行量化工具场景
|
||
if "quant" in active_modules:
|
||
start_t = time.time()
|
||
try:
|
||
run_quantization_lifecycle(self.qp)
|
||
self._safe_screenshot("quantization_final.png")
|
||
logger.info("✅ 量化工具场景通过")
|
||
self.record_result("quantization", "量化工具效能测试", "执行模型量化脚本并验证输出一致性", "PASS", time.time()-start_t)
|
||
except Exception as e:
|
||
logger.error(f"❌ 量化工具场景失败: {e}")
|
||
self._safe_screenshot("quantization_error.png")
|
||
errors.append(f"量化工具: {e}")
|
||
self.record_result("quantization", "量化工具效能测试", "执行模型量化脚本并验证输出一致性", "FAIL", time.time()-start_t)
|
||
|
||
# 7. 全局 Monkey 稳定性打底测试
|
||
if "monkey" in active_modules:
|
||
start_t = time.time()
|
||
try:
|
||
logger.info("🚀 开启全面压测: 注入大范围 Monkey 测试策略...")
|
||
run_monkey_testing(self.page, action_count=50)
|
||
self._safe_screenshot("monkey_pass.png")
|
||
logger.info("✅ 全链路存活：Monkey 压路机测试无严重崩溃阻断")
|
||
self.record_result("monkey_testing", "Monkey 稳定性压测", "50 次随机点击/交互注入，验证系统抗崩溃能力", "PASS", time.time()-start_t)
|
||
except Exception as e:
|
||
logger.error(f"❌ 稳定性告警: Monkey 测试导致了非预期崩溃或抛错: {e}")
|
||
self._safe_screenshot("monkey_error.png")
|
||
errors.append(f"Monkey测试异常: {e}")
|
||
self.record_result("monkey_testing", "Monkey 稳定性压测", "50 次随机点击/交互注入，验证系统抗崩溃能力", "FAIL", time.time()-start_t)
|
||
|
||
# 汇总
|
||
if errors:
|
||
summary = " | ".join(errors)
|
||
logger.error(f"❌ {len(errors)} 个场景失败: {summary}")
|
||
raise Exception(f"{len(errors)} 个场景失败: {summary}")
|
||
|
||
logger.info(f"🎉 Robogo {env_name} 环境 {scope} 巡检执行圆满完成！")
|
||
|
||
def run(self, user, pwd):
|
||
"""主入口"""
|
||
try:
|
||
self.start()
|
||
if not self.login(user, pwd):
|
||
return
|
||
|
||
# 开始执行指挥任务
|
||
self.run_all_scenarios()
|
||
|
||
finally:
|
||
self.save_results()
|
||
self.ui.stop()
|
||
|
||
if __name__ == "__main__":
|
||
account = input("账号: ")
|
||
password = input("密码: ")
|
||
|
||
dm = DataManagement(headless=False)
|
||
dm.run(account, password) |