""" UI 集成测试 - case_mix：清空规则、设置规则并执行四种同步策略 运行前准备: 1. 启动Docker服务: docker compose up -d 2. 确保服务运行在 http://localhost:8888 运行: pytest tests/test_ui_case_mix.py --headed # 显示浏览器 pytest tests/test_ui_case_mix.py # 无头模式 """ from enum import Enum from pathlib import Path import shutil import time from playwright.sync_api import Page, expect BASE_URL = "http://localhost:8888" PROJECT_ROOT = Path(__file__).parent.parent OUTPUT_DIR = PROJECT_ROOT / "output_playlists" / "case_mix" EXPECTED_DIR = PROJECT_ROOT / "test_res" class SyncStrategy(str, Enum): """同步策略枚举""" LOCAL_OVERWRITE = "LOCAL_OVERWRITE" CLOUD_OVERWRITE = "CLOUD_OVERWRITE" MERGE_LOCAL = "MERGE_LOCAL" MERGE_CLOUD = "MERGE_CLOUD" def _handle_connection_modal(page: Page): """处理登录模态框：如果存在则关闭""" # 检查模态框是否存在 (根据 ConnectionModal.tsx 的结构) # 模态框通常有一个全屏的遮罩层 modal_overlay = page.locator("div.fixed.inset-0.z-50") if modal_overlay.is_visible(): print("检测到登录模态框，尝试关闭...") # 尝试找到关闭按钮 (通常在右上角，包含 X 图标) # 在 ConnectionModal.tsx 中，关闭按钮在 Header 里 close_button = modal_overlay.locator("button").filter(has=page.locator("svg")).last if close_button.is_visible(): close_button.click() else: # 如果找不到关闭按钮，尝试按 ESC page.keyboard.press("Escape") page.wait_for_timeout(500) # 等待模态框关闭动画 def _open_strategy_selector(page: Page): """打开策略选择器下拉菜单""" # 1. 先处理可能遮挡的登录模态框 _handle_connection_modal(page) # 2. 检查下拉菜单是否已经打开 # 下拉菜单的特征类名 dropdown = page.locator("div.absolute.top-14") if dropdown.is_visible(): return # 已经打开，无需操作 # 3. 查找并点击策略选择器按钮 # 使用 title 属性定位更准确 (StrategySelector.tsx 中定义了 title="Current Strategy: ...") strategy_button = page.locator("button[title^='Current Strategy']") if strategy_button.count() == 0: # 备用定位方式：查找包含特定图标的圆形按钮 # 注意：页面上可能有多个按钮，需要小心 # 策略按钮在中间，且包含 ChevronDown 小图标 strategy_button = page.locator("button").filter(has=page.locator("svg")).nth(1) # nth(0) 可能是 Header 里的连接按钮 if strategy_button.is_visible(): strategy_button.click() page.wait_for_timeout(300) # 等待下拉菜单动画完成 else: print("警告: 无法找到策略选择器按钮") def _clear_all_rules(page: Page): """清空所有正则规则""" _open_strategy_selector(page) # 等待下拉菜单打开 dropdown = page.locator("div.absolute.top-14") expect(dropdown).to_be_visible() # 查找并点击所有删除按钮 delete_buttons = page.locator("button[title='Delete Rule']") while delete_buttons.count() > 0: try: delete_buttons.first.click() page.wait_for_timeout(100) except Exception: break # 关闭下拉菜单 page.keyboard.press("Escape") page.wait_for_timeout(200) def _normalize_playlist_lines(file_path: Path) -> list[str]: """读取播放列表并返回规范化曲目路径列表（忽略注释与空行）""" if not file_path.exists(): return [] with open(file_path, "r", encoding="utf-8") as f: lines = [ line.strip() for line in f if line.strip() and not line.startswith("#") ] return lines def _compare_playlists(actual: Path, expected: Path) -> tuple[bool, str]: """对比实际输出与期望结果，返回 (是否匹配, 差异描述)""" actual_lines = _normalize_playlist_lines(actual) expected_lines = _normalize_playlist_lines(expected) if actual_lines == expected_lines: return True, "" # 生成差异报告 diff_lines = [] diff_lines.append(f"实际曲目数: {len(actual_lines)}, 期望曲目数: {len(expected_lines)}") only_actual = set(actual_lines) - set(expected_lines) only_expected = set(expected_lines) - set(actual_lines) if only_actual: diff_lines.append(f"仅在实际输出中: {only_actual}") if only_expected: diff_lines.append(f"仅在期望结果中: {only_expected}") return False, "\n".join(diff_lines) def test_case_mix_run_all_modes(page: Page): """ 1) 清空当前正则规则 2) 填写并保存 case_mix 所用规则 3) 依次执行四种同步策略，每次同步后立即验证输出并与期望对比 """ # 导航到首页并确认加载（端口为 8080） page.goto(BASE_URL + "/") page.wait_for_load_state("networkidle") page.wait_for_timeout(1000) # 等待React应用初始化 expect(page).to_have_url(BASE_URL + "/") # 处理可能出现的登录模态框 _handle_connection_modal(page) # 1. 清空规则 _clear_all_rules(page) # 2. 添加并保存 case_mix 所用规则（顺序很重要） rules = [ (r"^\\\\koha9-nas\\koha9-nas\\Music", r"N:\\Music"), # UNC 到盘符 (r"^/mnt/music", r"N:\\Music"), # Linux 挂载到盘符 (r"(?i)^N:\\MUSIC", r"N:\\Music"), # 大小写规范化 (r"/", r"\\"), # 斜杠统一为反斜杠（需在映射后） ] # 打开策略选择器 _open_strategy_selector(page) # 添加规则 for pattern, replacement in rules: # 点击 "Add Rule" 按钮 add_button = page.locator("button[title='Add Rule']") if add_button.count() == 0: # 如果没有"Add Rule"按钮,尝试使用带文本的按钮 add_button = page.locator("button:has-text('Add Rule')") add_button.click() page.wait_for_timeout(200) # 填写最后一组输入框 pattern_inputs = page.locator("input[placeholder='Regex Pattern']") replacement_inputs = page.locator("input[placeholder='Replacement']") pattern_inputs.last.fill(pattern) replacement_inputs.last.fill(replacement) page.wait_for_timeout(100) # 保存规则 - 点击 "Save Changes" 按钮 save_button = page.locator("button:has-text('Save Changes')") expect(save_button).to_be_enabled() save_button.click() page.wait_for_timeout(500) # 等待保存完成 # 验证保存成功 - 检查toast通知 toast = page.locator("div:has-text('Regex preprocessing rules have been saved')") if toast.count() > 0: expect(toast.first).to_be_visible() # 关闭下拉菜单 page.keyboard.press("Escape") page.wait_for_timeout(300) # 3. 依次执行四种同步模式，每次执行后立即验证 # 策略名称映射: UI中的策略值 -> 测试用例名称 strategy_mappings = [ (SyncStrategy.LOCAL_OVERWRITE, "Local Overwrite", "case_mix_local_force.m3u"), (SyncStrategy.CLOUD_OVERWRITE, "Cloud Overwrite", "case_mix_remote_force.m3u"), (SyncStrategy.MERGE_LOCAL, "Two-way Merge (Local Priority)", "case_mix_merge_local_primary.m3u"), (SyncStrategy.MERGE_CLOUD, "Two-way Merge (Cloud Priority)", "case_mix_merge_remote_primary.m3u"), ] # 准备初始 Base（每次测试前恢复） initial_base_content = """#EXTM3U N:\\Music\\Anime\\New PANTY & STOCKING with GARTERBELT\\Theme of New PANTY & STOCKING\\02. Reckless - Theme of New PANTY & STOCKING.flac N:\\Music\\Anime\\CITY THE ANIMATION\\Hello\\01. Hello - Hello.flac N:\\Music\\音ゲー\\USAO\\MEMORIES\\15. Empty Room (Srav3R Remix) - MEMORIES.flac """ for strategy_value, strategy_label, expected_file in strategy_mappings: print(f"\n==== 执行同步策略: {strategy_label} ====") # 恢复初始 Base（避免前次同步影响） base_next_path = OUTPUT_DIR / "base_next.m3u8" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) with open(base_next_path, "w", encoding="utf-8") as f: f.write(initial_base_content) print(f"已恢复初始 Base: {base_next_path}") # 选择策略 - 打开下拉菜单 _open_strategy_selector(page) # 点击对应的策略选项 - 更精确的定位 # 找到包含策略名称的可点击div (class包含cursor-pointer) strategy_option = page.locator("div.cursor-pointer").filter(has_text=strategy_label) expect(strategy_option.first).to_be_visible() strategy_option.first.click() page.wait_for_timeout(500) # 等待策略保存 # 验证策略选择成功的toast toast = page.locator(f"div:has-text('Selected strategy \"{strategy_label}\" has been saved')") if toast.count() > 0: expect(toast.first).to_be_visible(timeout=3000) # 关闭下拉菜单 page.keyboard.press("Escape") page.wait_for_timeout(300) # 执行同步 - 通过API触发同步操作 # 新UI需要显式调用同步API import requests sync_response = requests.post( f"{BASE_URL}/api/sync", json={"mode": None} # 使用当前配置的策略 ) assert sync_response.status_code == 200, f"同步API调用失败: {sync_response.text}" print(f"同步API响应: {sync_response.json()}") time.sleep(1) # 确保文件写入完成 # 验证输出文件生成 local_result = OUTPUT_DIR / "local_result.m3u8" remote_result = OUTPUT_DIR / "remote_result.m3u8" base_next = OUTPUT_DIR / "base_next.m3u8" assert local_result.exists(), f"{strategy_label}: local_result.m3u8 未生成" assert remote_result.exists(), f"{strategy_label}: remote_result.m3u8 未生成" assert base_next.exists(), f"{strategy_label}: base_next.m3u8 未生成" # 与期望结果对比（使用 local_result.m3u8 作为主要输出） expected_path = EXPECTED_DIR / expected_file match, diff = _compare_playlists(local_result, expected_path) # 备份当前输出以便后续检查 backup_dir = OUTPUT_DIR / f"backup_{strategy_value}" backup_dir.mkdir(exist_ok=True) shutil.copy(local_result, backup_dir / "local_result.m3u8") shutil.copy(remote_result, backup_dir / "remote_result.m3u8") shutil.copy(base_next, backup_dir / "base_next.m3u8") print(f"输出已备份到: {backup_dir}") # 断言匹配 assert match, f"{strategy_label} 输出与期望不符:\n{diff}" print(f"✓ {strategy_label} 验证通过") print("\n==== 全部四种策略测试通过 ====")