maidong-ktv/selfhost-source/server/tests/test_tasks.py

#!/usr/bin/env python3
"""核验三项需求的专项测试（不启动任何服务）：

  需求① 同步数据库列举清单 + 更新时**比对清单**（新增/上游删除/变更）
  需求② 同步任务**有控制**（启动/取消/状态）且**有任务清单**（同步任务 + 下载任务）
  需求③ 歌曲源可配**多个**并**自动选择**（健康度排序 + 失败冷却）
"""
import os
import sys
import tempfile
import time
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from ktv import admin
from ktv.config import Config
from ktv.resolver import Resolver
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 rows(*specs):
    return [(tid, name, lang, "", singers, f"{tid}.ts", hot, "") for tid, name, lang, singers, hot in specs]


class StoreCase(unittest.TestCase):
    """带"显式关连接"的夹具（同 test_table/test_keeps）：否则 ResourceWarning 会污染验收输出。"""

    def tearDown(self):
        try:
            self.store.conn.close()
        except Exception:
            pass


class TestRequirement1Diff(StoreCase):
    """需求①：清单列举 + 后续更新比对清单。"""

    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.store = Store(os.path.join(self.tmp, "t.db"))

    def test_first_sync_lists_all_as_added(self):
        d = self.store.diff_and_import(rows((1, "A", "国语", "s1", 10), (2, "B", "粤语", "s2", 20)), "v1")
        self.assertEqual((d["total"], d["added"], d["removed"], d["changed"]), (2, 2, 0, 0))
        self.assertEqual(self.store.library_meta()["songs"], 2)

    def test_second_sync_reports_added_removed_changed(self):
        self.store.diff_and_import(rows((1, "A", "国语", "s1", 10), (2, "B", "粤语", "s2", 20)), "v1")
        # v2：① 新增 3；② 删除 2；③ 1 改名
        d = self.store.diff_and_import(rows((1, "A改", "国语", "s1", 10), (3, "C", "英语", "s3", 30)), "v2")
        self.assertEqual(d["added"], 1, d)
        self.assertEqual(d["removed"], 1, d)
        self.assertEqual(d["changed"], 1, d)
        self.assertEqual(d["added_sample"], [3])
        self.assertEqual(d["removed_sample"], [2])
        # 上游删除的歌曲不物理删除，只标记（保留下载登记）
        removed_row = self.store.q1("SELECT removed_at FROM songs WHERE tid=2")
        self.assertGreater(removed_row["removed_at"], 0)
        self.assertEqual(self.store.stats()["songs_removed_upstream"], 1)

    def test_reappearing_song_clears_removed_flag(self):
        self.store.diff_and_import(rows((1, "A", "国语", "s1", 10)), "v1")
        self.store.diff_and_import(rows(), "v2")
        self.assertEqual(self.store.stats()["songs_removed_upstream"], 1)
        self.store.diff_and_import(rows((1, "A", "国语", "s1", 10)), "v3")
        self.assertEqual(self.store.stats()["songs_removed_upstream"], 0)


class TestRequirement2Tasks(StoreCase):
    """需求②：同步任务的清单与控制 + 下载任务清单。"""

    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.cfg = Config(os.path.join(self.tmp, "c.json"))
        self.cfg.update({"media_dir": os.path.join(self.tmp, "media"), "data_dir": self.tmp})
        self.store = Store(os.path.join(self.tmp, "t.db"))
        self.store.import_library(rows((1, "A", "国语", "s", 5), (2, "B", "粤语", "s", 9)), "v1")

    def test_sync_task_lifecycle_and_list(self):
        tid = self.store.sync_task_start("v1")
        self.assertGreater(tid, 0)
        self.assertEqual(self.store.running_sync_task()["state"], "running")
        self.store.sync_task_finish(tid, "done", "v2",
                                   {"total": 2, "added": 1, "removed": 0, "changed": 0})
        tasks = self.store.sync_tasks(10)
        self.assertEqual(len(tasks), 1)
        self.assertEqual(tasks[0]["state"], "done")
        self.assertEqual(tasks[0]["added"], 1)
        self.assertEqual(tasks[0]["old_version"], "v1")
        self.assertEqual(tasks[0]["new_version"], "v2")
        detail = self.store.sync_task_get(tid)
        self.assertIn("sample", detail)
        self.assertIsNone(self.store.running_sync_task())

    def test_sync_task_failed_records_error(self):
        tid = self.store.sync_task_start("v1")
        self.store.sync_task_finish(tid, "failed", error="登录失败")
        self.assertEqual(self.store.sync_tasks(1)[0]["error"], "登录失败")

    def test_download_task_list_filters(self):
        self.store.enqueue([1, 2])
        self.assertEqual(self.store.download_tasks()["total"], 2)
        self.assertEqual(self.store.download_tasks("pending")["total"], 2)
        self.store.claim_one(1)
        self.store.finish(1, True, bytes_=123)
        self.assertEqual(self.store.download_tasks("done")["total"], 1)
        self.assertEqual(self.store.download_tasks("pending")["total"], 1)
        items = self.store.download_tasks("done")["items"]
        self.assertEqual(items[0]["tid"], 1)
        self.assertEqual(items[0]["name"], "A")

    def test_auto_enqueue_mode_removed_explicit_tids_kept(self):
        """补队策略：`mode=auto`（按热度自己挑歌）已删除；补队只由维护任务按歌曲表做。

        显式传 `tids` 的入队路径必须保留（下载队列页的"下载选中/本页"还在用）。
        """
        self.store.keep_add([2], user="test")
        cands = self.store.keep_next_batch(1, "hot", True)
        self.assertEqual([c["tid"] for c in cands], [2])          # 补队候选只来自歌曲表
        r = admin.build_router(self.cfg, self.store, None)
        import json
        # 不传 tids（老脚本用 mode=auto）→ 明确拒绝，不再按热度自己挑歌
        d = json.loads(r.dispatch(req("POST", "/api/download/enqueue",
                                      body=b'{"mode":"auto","limit":1}')).body)
        self.assertFalse(d["ok"], d)
        # 显式 tids → 正常入队
        d = json.loads(r.dispatch(req("POST", "/api/download/enqueue",
                                      body=b'{"tids":[2]}')).body)
        self.assertTrue(d["ok"], d)
        self.assertEqual(d["added"], 1)
        # 不在歌曲表里的歌（tid=1）永不入队：候选只有歌曲表里的 [2]
        self.assertEqual([c["tid"] for c in self.store.keep_next_batch(5, "hot", True)], [2])
        # 已就绪的不再被选中
        self.store.claim_one(2)
        self.store.finish(2, True, bytes_=1)
        self.store.keep_mark_done(2, 1, "")
        self.assertEqual(self.store.keep_next_batch(5, "hot", True), [])

    def test_admin_task_endpoints(self):
        self.store.enqueue([1])
        r = admin.build_router(self.cfg, self.store, None)
        import json
        d = json.loads(r.dispatch(req("GET", "/api/download/tasks", {"state": "pending"})).body)
        self.assertEqual(d["total"], 1)
        d2 = json.loads(r.dispatch(req("GET", "/api/library/tasks")).body)
        self.assertIn("items", d2)
        # 没有进行中的同步 → 取消应明确拒绝
        d3 = json.loads(r.dispatch(req("POST", "/api/library/cancel")).body)
        self.assertFalse(d3["ok"])


class TestRegressionsFromDeploy(StoreCase):
    """2026-10-10 部署实测暴露的两个缺陷的回归测试。"""

    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.store = Store(os.path.join(self.tmp, "t.db"))
        self.store.import_library(rows((1, "A", "国语", "s", 1)), "v")

    def test_retry_stops_at_max_attempts(self):
        """缺陷①：失败任务被无限立刻重领 → 2 分钟打上游 504 次取址。"""
        self.store.enqueue([1])
        got = []
        for _ in range(6):
            ok = self.store.claim_one(1, max_attempts=3, backoff_sec=0)
            got.append(ok)
            if ok:
                self.store.finish(1, False, error="boom")
        self.assertEqual(got, [True, True, True, False, False, False])
        self.assertEqual(self.store.q1("SELECT state FROM downloads WHERE tid=1")["state"], "failed")
        self.assertEqual(self.store.next_pending(5, "hot", True, 3, 0), [])

    def test_backoff_does_not_block_fresh_task(self):
        """缺陷①附带：退避窗口不能挡住刚入队的任务。"""
        self.store.enqueue([1])
        self.assertTrue(self.store.claim_one(1, max_attempts=3, backoff_sec=3600))
        self.store.finish(1, False, error="x")
        self.assertFalse(self.store.claim_one(1, max_attempts=3, backoff_sec=3600))   # 重试被退避挡住
        self.assertTrue(self.store.claim_one(1, max_attempts=3, backoff_sec=0))      # 无退避可重试

    def test_ts_detection_tolerates_private_header(self):
        """缺陷②：上游流是「512 字节私有头 + TS」，只查偏移 0 会把合法流判成坏文件。"""
        from ktv.downloader import Downloader
        p = os.path.join(self.tmp, "with-header.ts")
        with open(p, "wb") as fh:
            fh.write(b"\x45" * 512 + (b"\x47" + b"\x00" * 187) * 20)
        ok, offset = Downloader._looks_like_ts(p)
        self.assertTrue(ok)
        self.assertEqual(offset, 512)
        bad = os.path.join(self.tmp, "garbage.ts")
        with open(bad, "wb") as fh:
            fh.write(b"\x11" * 4096)
        self.assertFalse(Downloader._looks_like_ts(bad)[0])


class TestRequirement3MultiSource(StoreCase):
    """需求③：多个歌曲源 + 自动选择。"""

    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.cfg = Config(os.path.join(self.tmp, "c.json"))
        self.store = Store(os.path.join(self.tmp, "t.db"))
        self.cfg.update({"upstream_hosts": ["http://a.example.com", "http://b.example.com",
                                            "http://c.example.com"]})
        self.res = Resolver(self.cfg, self.store)

    def test_configured_order_initially(self):
        self.assertEqual([h.split("//")[1] for h in self.res.hosts()],
                         ["a.example.com", "b.example.com", "c.example.com"])

    def test_success_promotes_host(self):
        self.res.note_ok("http://c.example.com")
        self.assertEqual(self.res.hosts()[0], "http://c.example.com")

    def test_403_cools_down_and_sinks(self):
        self.res.note_fail("http://a.example.com", "code=403 检测到异常行为")
        order = self.res.hosts()
        self.assertEqual(order[-1], "http://a.example.com")
        rep = {x["host"]: x for x in self.res.host_report()}
        self.assertTrue(rep["http://a.example.com"]["cooling"])
        self.assertGreater(rep["http://a.example.com"]["cooldown_left"], 0)

    def test_health_persists_across_instances(self):
        self.res.note_ok("http://b.example.com")
        self.res.note_fail("http://a.example.com", "403")
        again = Resolver(self.cfg, self.store)          # 模拟重启
        self.assertEqual(again.hosts()[0], "http://b.example.com")
        self.assertEqual(again.host_report()[0]["ok"] + again.host_report()[1]["ok"], 1)

    def test_clear_health_restores_configured_order(self):
        self.res.note_ok("http://c.example.com")
        self.res.clear_health()
        self.assertEqual([h.split("//")[1] for h in self.res.hosts()],
                         ["a.example.com", "b.example.com", "c.example.com"])

    def test_admin_upstream_endpoints(self):
        r = admin.build_router(self.cfg, self.store, None)
        import json
        d = json.loads(r.dispatch(req("GET", "/api/upstream/hosts")).body)
        self.assertEqual(len(d["items"]), 3)
        self.assertEqual(d["order"][0], "http://a.example.com")
        d2 = json.loads(r.dispatch(req("POST", "/api/upstream/clear_health")).body)
        self.assertTrue(d2["ok"])


if __name__ == "__main__":
    unittest.main(verbosity=2)

class TestUiAndPacing(unittest.TestCase):
    """2026-10-10 第二批修复的回归测试。"""

    def test_task_tab_fills_after_template(self):
        """缺陷①：loadTasks 必须先渲染模板再调 loader，否则数据被"加载中…"覆盖。"""
        from ktv import ui
        html = ui.PAGE
        fn = html[html.index("async function loadTasks()"):]
        fn = fn[:fn.index("// ── 设置 ──")]
        pos_inner = fn.index("$('#tab-tasks').innerHTML")
        pos_loader = fn.index("loadSyncTasks('syncTasksAll')")
        self.assertLess(pos_inner, pos_loader, "loader 必须晚于 innerHTML 赋值")
        self.assertIn("加载中…", fn)          # 模板里确有占位
        self.assertGreater(pos_loader, fn.rindex("加载中…"), "loader 调用要在占位模板之后")

    def test_upstream_pacing_between_3_and_5s(self):
        """需求②：取链请求之间要有随机间隔（全局串行）。

        这里用很小的间隔快速验证"每两次请求之间确实等了 [min,max] 区间"；
        **默认值 5000/10000 与"间隔 ≥ min"用虚拟时钟精确断言**（见
        `tests/test_autoretry_scan.py::TestPacing`，不真睡 10 秒）。
        """
        from ktv import resolver as R
        import time
        starts = []
        for _ in range(3):
            R._pace(300, 500)                      # 直接测限速（不联网）
            starts.append(time.time())
        deltas = [starts[i + 1] - starts[i] for i in range(len(starts) - 1)]
        for d in deltas:
            self.assertGreater(d, 0.25, f"间隔过小: {d}")
            self.assertLess(d, 0.75, f"间隔过大: {d}")

    def test_interval_config_defaults_and_validation(self):
        cfg = Config(os.path.join(tempfile.mkdtemp(), "c.json"))
        self.assertEqual(cfg.get("upstream_interval_min_ms"), 5000)      # 2026：默认 5~10 秒
        self.assertEqual(cfg.get("upstream_interval_max_ms"), 10000)
        warns = cfg.update({"upstream_interval_min_ms": 8000, "upstream_interval_max_ms": 2000})
        self.assertEqual(cfg.get("upstream_interval_min_ms"), 8000)
        self.assertGreaterEqual(cfg.get("upstream_interval_max_ms"), 8000)   # 最大不得小于最小
        warns2 = cfg.update({"upstream_interval_min_ms": 200, "upstream_interval_max_ms": 300})
        self.assertTrue(any("攻击" in w for w in warns2), warns2)

class TestStorageGuardAndRateLimit(StoreCase):
    """媒体在云盘/FUSE 上时的保护：水位守卫 + 上传速度匹配限速。"""

    def setUp(self):
        from ktv.config import Config
        from ktv.store import Store
        from ktv.downloader import StorageGuard
        self.tmp = tempfile.mkdtemp()
        self.cfg = Config(os.path.join(self.tmp, "c.json"))
        self.cfg.update({"media_dir": os.path.join(self.tmp, "media"),
                         "local_space_path": self.tmp})
        self.store = Store(os.path.join(self.tmp, "t.db"))
        self.guard = StorageGuard(self.cfg, self.store, None)

    def _patch_free(self, local, media):
        import ktv.downloader as D
        real = D.__dict__.get("_real_disk_usage")

        def fake(path, *a, **k):
            class U:
                def __init__(self, free):
                    self.free = free
            return U(local if path == self.cfg.get("local_space_path") else media)
        D.shutil.disk_usage = fake

    def test_ok_when_space_sufficient(self):
        self._patch_free(100 * 1024 ** 3, 100 * 1024 ** 3)
        st = self.guard.sample()
        self.assertTrue(st["ok"])
        self.assertTrue(st["recover"])
        self.assertEqual(st["reason"], "空间充足")

    def test_low_local_space_pauses(self):
        self._patch_free(1 * 1024 ** 3, 100 * 1024 ** 3)          # 本地仅 1GB
        st = self.guard.sample()
        self.assertFalse(st["ok"])
        self.assertIn("本地盘剩余", st["reason"])

    def test_low_media_quota_pauses(self):
        self._patch_free(100 * 1024 ** 3, 1 * 1024 ** 3)          # 云盘配额仅 1GB
        st = self.guard.sample()
        self.assertFalse(st["ok"])
        self.assertIn("云盘剩余", st["reason"])

    def test_hysteresis_on_recovery(self):
        """刚好回到阈值不算恢复（要阈值+余量），避免边界抖动。"""
        min_local = int(self.cfg.get("min_free_local_bytes"))
        self._patch_free(min_local + 1024, 100 * 1024 ** 3)
        st = self.guard.sample()
        self.assertTrue(st["ok"])
        self.assertFalse(st["recover"], "刚好过线不应判定为可恢复")

    def test_space_pause_not_cleared_by_user_resume(self):
        from ktv.downloader import Downloader
        dl = Downloader(self.cfg, self.store)
        dl.pause(by="space")
        self.assertTrue(dl.paused_by_space)
        dl.resume(by="user")                                      # 用户点继续不应解除空间暂停
        self.assertTrue(dl.paused)
        dl.resume(by="space")
        self.assertFalse(dl.paused)

    def test_rate_limiter_throttles(self):
        from ktv.downloader import RateLimiter
        r = RateLimiter(1.0)                                      # 1 MB/s
        # 令牌桶首个额度是"满桶"（不等待），所以按**聚合**验证：
        # 连续取 2MB 应耗时 ≈1 秒（首 1MB 免费 + 之后按速率排队）。
        t0 = time.time()
        for _ in range(8):
            r.acquire(256 * 1024)
        el = time.time() - t0
        self.assertGreater(el, 0.7, f"限速过松: {el}")
        self.assertLess(el, 2.2, f"限速过紧: {el}")
        r.set_rate(0)                                             # 0 = 不限速
        t0 = time.time()
        r.acquire(10 * 1024 * 1024)
        self.assertLess(time.time() - t0, 0.2)

class TestBatchFatal(StoreCase):
    """全库批量：持续补队 + 致命错误自动停止（对应"致命错误时停下"）。"""

    def setUp(self):
        from ktv.batch import BatchManager
        from ktv.config import Config
        from ktv.store import Store
        self.tmp = tempfile.mkdtemp()
        self.cfg = Config(os.path.join(self.tmp, "c.json"))
        self.cfg.update({"data_dir": self.tmp, "media_dir": os.path.join(self.tmp, "m"),
                         "tmp_dir": os.path.join(self.tmp, "t")})
        self.store = Store(os.path.join(self.tmp, "t.db"))
        self.store.import_library(rows((1, "A", "国语", "s", 10), (2, "B", "国语", "s", 9)), "v")
        self.batch = BatchManager(self.cfg, self.store, None, None)

    def test_low_space_streak_triggers_fatal(self):
        stats = self.store.stats()
        self.assertEqual(self.batch.evaluate(stats, low_space=True), "")      # 1 次不算
        self.batch.state["low_space_streak"] = 4
        self.assertIn("存储空间不足", self.batch.evaluate(stats, low_space=True))

    def test_consecutive_failures_trigger_fatal(self):
        stats = {"pending": 5, "failed": 0, "done": 0, "songs": 2}
        self.batch.state["consecutive_failures"] = 19
        self.batch.state["last_progress_at"] = int(time.time())
        self.assertEqual(self.batch.evaluate(stats, False), "")
        self.batch.state["consecutive_failures"] = 20
        self.assertIn("连续失败", self.batch.evaluate(stats, False))

    def test_stale_progress_triggers_fatal(self):
        stats = {"pending": 3, "failed": 0, "done": 0, "songs": 2}
        self.batch.state["consecutive_failures"] = 0
        self.batch.state["last_progress_at"] = int(time.time()) - 21 * 60
        self.assertIn("没有任何新完成", self.batch.evaluate(stats, False))

    def test_healthy_state_no_fatal(self):
        stats = {"pending": 3, "failed": 0, "done": 1, "songs": 2}
        self.batch.state["last_progress_at"] = int(time.time())
        self.assertEqual(self.batch.evaluate(stats, False), "")

    def test_tops_up_queue(self):
        self.store.keep_add([1, 2])                       # 补队只认歌曲表里的歌
        self.batch.start()
        time.sleep(0.2)
        # 手动跑一轮补队逻辑（避免等 30 秒）：候选来自歌曲表，热度降序
        cands = self.store.keep_next_batch(2, "hot", True)
        self.assertEqual([c["tid"] for c in cands], [1, 2])
        self.batch.stop("test")
        self.assertFalse(self.batch.status()["enabled"])

    def test_stop_is_fatal_marked(self):
        self.batch.start()
        self.batch.stop("模拟致命", fatal=True)
        st = self.batch.status()
        self.assertTrue(st["fatal"])
        self.assertIn("模拟致命", st["reason"])