同步逻辑，正则替换测试。

2025-11-27 05:03:35 +09:00
parent c2b429272f
commit 58709570a9
37 changed files with 2370 additions and 3 deletions
@@ -0,0 +1,169 @@
+"""
+UI 集成测试 - case_mix：清空规则、设置规则并执行四种同步策略
+
+运行:
+    pytest tests/test_ui_case_mix.py --headed  # 显示浏览器
+    pytest tests/test_ui_case_mix.py           # 无头模式
+"""
+
+from pathlib import Path
+import shutil
+import time
+
+from playwright.sync_api import Page, expect
+
+
+BASE_URL = "http://localhost:8080"
+
+PROJECT_ROOT = Path(__file__).parent.parent
+OUTPUT_DIR = PROJECT_ROOT / "dockerapp" / "test_playlists" / "case_mix"
+EXPECTED_DIR = PROJECT_ROOT / "test_res"
+
+
+def _clear_all_rules(page: Page):
+    while page.locator(".rule-row").count() > 0:
+        try:
+            page.locator(".rule-row button[title='删除此规则']").first.click()
+            page.wait_for_timeout(50)
+        except Exception:
+            break
+
+
+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")
+    expect(page).to_have_url(BASE_URL + "/")
+
+    # 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"\\"),                                           # 斜杠统一为反斜杠（需在映射后）
+    ]
+
+    add_button = page.locator("#addRuleBtn")
+    for pattern, replacement in rules:
+        add_button.click()
+        page.wait_for_timeout(100)
+        page.locator("input[name='pattern']").last.fill(pattern)
+        page.locator("input[name='replacement']").last.fill(replacement)
+
+    # 保存
+    page.locator("button:has-text('保存规则')").click()
+    page.wait_for_load_state("networkidle")
+    # 成功提示可选校验
+    success_indicator = page.locator(".alert-success, .text-success, :has-text('已保存'), :has-text('保存规则')")
+    if success_indicator.count() > 0:
+        expect(success_indicator.first).to_be_visible()
+
+    # 3. 依次执行四种同步模式，每次执行后立即验证
+    test_cases = [
+        ("local_force", "case_mix_local_force.m3u"),
+        ("remote_force", "case_mix_remote_force.m3u"),
+        ("merge_local_primary", "case_mix_merge_local_primary.m3u"),
+        ("merge_remote_primary", "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 mode, expected_file in test_cases:
+        print(f"\n==== 执行同步策略: {mode} ====")
+        
+        # 恢复初始 Base（避免前次同步影响）
+        base_next_path = OUTPUT_DIR / "base_next.m3u8"
+        with open(base_next_path, "w", encoding="utf-8") as f:
+            f.write(initial_base_content)
+        print(f"已恢复初始 Base: {base_next_path}")
+        
+        # 选择模式
+        page.select_option("select[name='mode']", value=mode)
+        
+        # 执行同步
+        page.locator("form[action='/sync'] button[type='submit']").click()
+        page.wait_for_load_state("networkidle")
+        
+        # 等待提示出现（如果存在）
+        alert = page.locator(".alert-success, .alert-info, :has-text('同步完成')")
+        if alert.count() > 0:
+            expect(alert.first).to_be_visible()
+        
+        time.sleep(0.5)  # 确保文件写入完成
+        
+        # 验证输出文件生成
+        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"{mode}: local_result.m3u8 未生成"
+        assert remote_result.exists(), f"{mode}: remote_result.m3u8 未生成"
+        assert base_next.exists(), f"{mode}: 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_{mode}"
+        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"{mode} 输出与期望不符:\n{diff}"
+        print(f"✓ {mode} 验证通过")
+
+    print("\n==== 全部四种策略测试通过 ====")