#!/usr/bin/env python3 """管理界面性能与前端约定的回归测试（任务②：点「同步曲库」不再卡住）。 覆盖三条硬要求： 1) `/api/stats` 在 69 万行级别的库上必须快 —— 走 `stats_cache_sec` TTL 缓存（目标 < 300ms）， 且写入 keeps/songs/downloads 后缓存必须立刻失效（不能显示陈旧数字）； 2) 曲库导入/清单比对**必须分块**：长事务不再独占 `Store._lock`， "同步进行中"轮询接口依然能响应（这里用一个并发轮询线程测最大延迟）； 3) 管理页的 JS 结构：一个共享的、防重入的轮询 + 只更新"变化的 DOM 块"， 整页 `innerHTML` 重建只允许出现在"进入页签时渲染骨架"的那几个函数里。 """ import os import re import sys import tempfile import threading import time import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ktv import admin, ui from ktv.config import Config from ktv.store import Store from ktv.web import Request def req(method, path, query=None, body=b""): return Request(method, path, query or {}, {}, body) def _library(n: int) -> list[tuple]: return [(i, f"歌{i}", "国语", "", f"歌手{i % 37}", f"{i}.ts", i % 1000, "") for i in range(1, n + 1)] class CountingStore(Store): """统计真算次数的 Store（用来验证 TTL 缓存真的生效/真的失效）。""" def __init__(self, *a, **k): super().__init__(*a, **k) self.stats_calls = 0 self.keep_stats_calls = 0 def _stats_uncached(self): self.stats_calls += 1 return super()._stats_uncached() def _keep_stats_uncached(self): self.keep_stats_calls += 1 return super()._keep_stats_uncached() class StatsCacheCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.cfg = Config(os.path.join(self.tmp, "c.json")) self.media = os.path.join(self.tmp, "media") self.cfg.update({"data_dir": self.tmp, "media_dir": self.media, "tmp_dir": os.path.join(self.tmp, "t")}) os.makedirs(self.media, exist_ok=True) self.db = os.path.join(self.tmp, "t.db") def tearDown(self): try: self.store.close() except Exception: pass class TestStatsCache(StatsCacheCase): """1) 概览统计：TTL 缓存 + 写入即失效。""" def test_ttl_cache_hits_and_write_invalidates(self): self.store = CountingStore(self.db, "DELETE", stats_cache_sec=5) self.store.import_library(_library(200), "v1") self.store.keep_add([1, 2, 3]) base_stats = self.store.stats_calls base_keeps = self.store.keep_stats_calls for _ in range(5): # TTL 内反复取 → 只真算一次 self.store.stats() self.store.keep_stats() self.assertEqual(self.store.stats_calls - base_stats, 1) self.assertEqual(self.store.keep_stats_calls - base_keeps, 1) self.store.keep_add([4]) # 写 keeps → 缓存立刻失效 self.assertEqual(self.store.keep_stats()["total"], 4) # 不是陈旧的 3 self.assertEqual(self.store.keep_stats_calls - base_keeps, 2) self.store.stats() self.assertEqual(self.store.stats_calls - base_stats, 2) # keeps 的写入也让 stats 重算 self.store.import_library(_library(201), "v2") # 写 songs → 同样失效且数字是新的 self.assertEqual(self.store.stats()["songs"], 201) self.assertEqual(self.store.stats_calls - base_stats, 3) def test_zero_ttl_means_no_cache(self): self.store = CountingStore(self.db, "DELETE", stats_cache_sec=0) self.store.import_library(_library(50), "v1") self.store.stats(); self.store.stats() self.assertEqual(self.store.stats_calls, 2) # 0 = 每次都真查（现有行为不变） def test_api_stats_is_fast_with_cache(self): """69 万行级别的统计：缓存命中后 /api/stats 必须 < 300ms。""" self.store = Store(self.db, "DELETE", stats_cache_sec=0) self.store.import_library(_library(50000), "v1") self.store.keep_add([1, 2, 3]) router = admin.build_router(self.cfg, self.store, None) self.store.stats_cache_sec = 0 self.store.clear_stats_cache() admin._stats_side_cache.clear() t0 = time.perf_counter() resp = router.dispatch(req("GET", "/api/stats")) uncached_ms = (time.perf_counter() - t0) * 1000 self.assertEqual(resp.status, 200) self.store.stats_cache_sec = 4 self.store.clear_stats_cache() admin._stats_side_cache.clear() router.dispatch(req("GET", "/api/stats")) # 第一次：填缓存 t0 = time.perf_counter() resp2 = router.dispatch(req("GET", "/api/stats")) # 第二次：命中缓存 cached_ms = (time.perf_counter() - t0) * 1000 self.assertEqual(resp2.status, 200) self.assertLess(cached_ms, 300.0, f"缓存命中后仍然 {cached_ms:.0f}ms（无缓存时 {uncached_ms:.0f}ms）") self.assertLessEqual(cached_ms, uncached_ms + 5.0) def test_library_snapshot_is_chunked_and_complete(self): self.store = Store(self.db, "DELETE") self.store.import_library(_library(12345), "v1") snap = self.store._library_snapshot(chunk=1000) # 分块读也不许漏行 self.assertEqual(len(snap), 12345) self.assertEqual(snap[12345]["name"], "歌12345") class TestImportDoesNotBlockPolling(StatsCacheCase): """2) 导入/比对分块提交：同步进行中轮询接口仍可响应。""" def test_poll_latency_during_import(self): self.store = Store(self.db, "DELETE", stats_cache_sec=0) self.store.import_library(_library(2000), "v0") lat: list[float] = [] stop = threading.Event() def poller(): while not stop.is_set(): t0 = time.perf_counter() self.store.stats() lat.append((time.perf_counter() - t0) * 1000) time.sleep(0.02) th = threading.Thread(target=poller, name="poll") th.start() try: self.store.import_library(_library(40000), "v1") # 4 万行：分块提交 finally: stop.set() th.join(timeout=5) self.assertGreaterEqual(len(lat), 5, "轮询根本没跑起来") self.assertLess(max(lat), 1000.0, f"导入期间轮询最大延迟 {max(lat):.0f}ms（挡住了就是整页卡住的根源）") self.assertEqual(self.store.stats()["songs"], 40000) class TestUiAsyncContract(unittest.TestCase): """3) 管理页 JS 的结构约定（把"卡住"的两大原因钉死：前端 await + 整页重建）。""" def setUp(self): self.html = ui.PAGE m = re.search(r"", self.html, re.S) self.js = m.group(1) def test_shared_polling_with_reentrancy_guard(self): self.assertIn("setInterval(()=>tick(false), REFRESH_MS)", self.js) self.assertIn("Promise.allSettled(jobs)", self.js) self.assertRegex(self.js, r"if\(tickBusy\) return;") # 防重入 self.assertIn("id=\"autoRefresh\"", self.html) # 自动刷新开关 self.assertIn("id=\"lastUpdate\"", self.html) # 最后更新时间 self.assertIn("touchUpdate()", self.js) def test_action_buttons_do_not_block(self): """动作接口一律走 act()（loading + 顶部提示 + 轮询出结果），不能裸 await。""" for path in ("/api/library/sync", "/api/library/cancel", "/api/download/retry_failed", "/api/upstream/probe", "/api/download/start", "/api/download/stop"): self.assertNotIn(f"await api('{path}'", self.js, f"{path} 不能让前端等长任务") self.assertTrue(f"'{path}'" in self.js, f"页面里没有 {path}") self.assertRegex(self.js, r"async function act\(") self.assertTrue("已发起" in self.js) def test_full_page_rebuild_only_in_skeletons(self): """整页 innerHTML 只允许出现在"进入页签时渲染骨架"的函数里（轮询绝不重建整页）。""" allowed = {"renderOverview", "renderSongs", "renderKeeps", "loadTasks", "renderSettings", "loadSettings", "renderLogs"} for m in re.finditer(r"\$\('#tab-(\w+)'\)\.innerHTML", self.js): func = re.findall(r"\n(?:async )?function (\w+)\(", self.js[:m.start()])[-1] self.assertIn(func, allowed, f"{func}() 里整页重建了 #tab-{m.group(1)}") def test_blocks_are_patched_by_signature(self): self.assertRegex(self.js, r"function patch\(id, sig, html\)") self.assertTrue("SIG[id] === sig" in self.js) # 内容没变就不动 DOM for block in ("ovCards", "ovBatchBody", "ovStorageBody", "upstream", "syncTasks", "syncTasksAll", "dlTasks", "activeRows", "kRows", "sRows", "logPre"): self.assertTrue(f"'{block}'" in self.js, f"没有按块更新 {block}") def test_progress_is_visible(self): """同步有进度条/百分比/已用时；批量有 keeps 计数、百分比、速率与预计剩余。""" for token in ("syncBar", "syncElapsed", "已用时", "keeps_total", "keeps_want", "keeps_done", "keeps_remaining", "预计剩余", "首/小时", "%"): self.assertTrue(token in self.html, f"页面缺少进度元素 {token}") if __name__ == "__main__": unittest.main(verbosity=2)