#!/usr/bin/env python3
"""App 点歌推送（`POST /push_download`）+「固定 URL 下载」回归自测（不联网、用临时库）。

覆盖：
  ① 鉴权：无/错 token → 401；header `X-Push-Token` 与 body `token` 两种口径都认；
  ② 校验与闸门：裸 IP / wb66 / demo / ac16.vip / r2.dev / 文件名不符 → 400 且**绝不入库**；
     `tid` 非正整数 / `url` 非 http(s) → 400；
  ③ 去重三态：already_done（done + 有文件）/ already_queued（歌曲表或队列里已在）/
     recently_pushed（同 tid 短时间再推）；
  ④ 接受路径：歌曲表出现该 tid（`note='App 点歌推送'`、`user='app'`、曲库里没有也允许）+
     队列里有该项且带推送 URL（`source='app-push'`）；
  ⑤ 限流：超过 `push_rate_limit_per_min` → 429；`GET /push_download/status` 计数正确；
  ⑥ 下载器：固定 URL **不调用** `get_song_url`（打桩计数）；固定 URL 失败 → 回退取址且事件有记录；
  ⑦ 队列：`next_pending()` 能取到"曲库里没有、但带固定 URL"的推送任务；老库迁移会补列。
"""
import json
import os
import sqlite3
import sys
import tempfile
import unittest
import urllib.error
import urllib.parse
from unittest import mock

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

from ktv import pushdl
from ktv import resolver as R
from ktv import service
from ktv.config import Config
from ktv.downloader import Downloader
from ktv.store import Store
from ktv.web import Request

TID = 7200088
# 合法推送地址：域名 + 路径以 /<编号>.ts 结尾 + 带 query（真源形态）
PUSH_URL = f"http://cdn.example.com/480p/{TID}.ts?token=abc&t=1700000000"
# "下载一定失败"的固定地址（下载器打桩按 host 判定）
BAD_PUSH_URL = f"http://bad.example.com/480p/{TID}.ts?token=expired&t=1"
RESOLVED_URL = f"http://txog.ktvsky.com/480p/{TID}.ts?token=xyz&t=2"

# 能被 `Downloader._looks_like_ts` 认出来的最小 TS 流（3 个 188 字节包同步点）
TS_BYTES = (b"\x47" + b"\x00" * 187) * 3 + b"\x00" * 512


def req(method, path, query=None, body=b"", headers=None, client_ip="10.0.0.1"):
    return Request(method, path, query or {}, headers or {}, body, client_ip=client_ip)


def form(**kv) -> bytes:
    return urllib.parse.urlencode(kv).encode()


class PushCase(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")
        os.makedirs(self.media, exist_ok=True)
        self.cfg.update({"media_dir": self.media, "data_dir": self.tmp,
                         "push_token": "local", "push_rate_limit_per_min": 60, "push_dedupe_sec": 300})
        self.store = Store(os.path.join(self.tmp, "t.db"))
        self.router = service.build_router(self.cfg, self.store, None)

    def tearDown(self):
        # 显式关连接：否则 ResourceWarning（unclosed database）会污染验收输出
        try:
            self.store.conn.close()
        except Exception:
            pass

    def call(self, r):
        return self.router.dispatch(r)

    def push(self, **kv):
        return self.call(req("POST", "/push_download", body=form(**kv)))

    def status(self) -> dict:
        return json.loads(self.call(req("GET", "/push_download/status")).body)

    def body_of(self, resp) -> dict:
        return json.loads(resp.body)

    # ── 断言辅助 ──
    def assert_not_persisted(self, tid=TID):
        """闸门/校验不通过时**绝不入库**：歌曲表与下载队列都不能出现该编号。"""
        self.assertIsNone(self.store.q1("SELECT 1 FROM keeps WHERE tid=?", (tid,)))
        self.assertIsNone(self.store.q1("SELECT 1 FROM downloads WHERE tid=?", (tid,)))


# ══════════════════════ ① 鉴权 ══════════════════════

class TestAuth(PushCase):
    def test_missing_token_401(self):
        resp = self.push(tid=TID, url=PUSH_URL)
        self.assertEqual(resp.status, 401)
        self.assertEqual(self.body_of(resp), {"ok": False, "error": "鉴权失败"})
        self.assert_not_persisted()

    def test_wrong_token_401(self):
        resp = self.push(tid=TID, url=PUSH_URL, token="nope")
        self.assertEqual(resp.status, 401)
        resp2 = self.call(req("POST", "/push_download", body=form(tid=TID, url=PUSH_URL),
                              headers={"X-Push-Token": "nope"}))
        self.assertEqual(resp2.status, 401)
        self.assert_not_persisted()

    def test_token_via_header_or_body(self):
        a = self.push(tid=TID, url=PUSH_URL, token="local")
        self.assertEqual(a.status, 200, a.body)
        # 换一首、改用 header 鉴权
        b = self.call(req("POST", "/push_download", body=form(tid=TID + 1, url=f"http://cdn.example.com/{TID + 1}.ts"),
                          headers={"X-Push-Token": "local"}))
        self.assertEqual(b.status, 200, b.body)

    def test_empty_config_token_falls_back_to_local(self):
        """`push_token` 配成空串 = 回落到 "local"（绝不能变成"谁都能推"）。"""
        self.cfg.update({"push_token": ""})
        self.assertEqual(self.push(tid=TID, url=PUSH_URL).status, 401)
        self.assertEqual(self.push(tid=TID, url=PUSH_URL, token="local").status, 200)


# ══════════════════════ ② 校验 / 闸门 ══════════════════════

class TestValidateAndGate(PushCase):
    def test_bad_tid_400(self):
        for bad in ("", "abc", "0", "-3", "1.5", "  "):
            resp = self.push(tid=bad, url=PUSH_URL, token="local")
            self.assertEqual(resp.status, 400, f"tid={bad!r} → {resp.body!r}")
            self.assertFalse(self.body_of(resp)["ok"])
        self.assert_not_persisted()

    def test_bad_url_400(self):
        for bad in ("", "ftp://cdn.example.com/{}.ts".format(TID), "notaurl", "//cdn.example.com/1.ts"):
            resp = self.push(tid=TID, url=bad, token="local")
            self.assertEqual(resp.status, 400, f"url={bad!r} → {resp.body!r}")
        self.assert_not_persisted()

    def test_gate_rejects_untrusted_and_not_persisted(self):
        """闸门拒收（裸 IP / wb66 / demo / ac16.vip / r2.dev / 文件名不符）→ 400 且未入库。"""
        cases = [
            ("http://1.2.3.4/480p/%d.ts" % TID, "裸 IP"),
            ("http://wb66.cn/f/d/7b7c9/%d.ts" % TID, "wb66"),
            ("http://demo.example.com/480p/%d.ts" % TID, "demo"),
            ("http://gz.ac16.vip/480p/%d.ts" % TID, "ac16.vip"),
            ("http://pub-abc123.r2.dev/ss/%d.ts" % TID, "r2.dev"),
            ("http://cdn.example.com/480p/9999999.ts", "文件名不符"),
        ]
        for url, why in cases:
            with self.subTest(why=why):
                resp = self.push(tid=TID, url=url, token="local")
                self.assertEqual(resp.status, 400, f"{why} → {resp.body!r}")
                self.assertIn("地址未通过闸门", self.body_of(resp)["error"])
                self.assert_not_persisted()

    def test_rejected_counts(self):
        self.push(tid=TID, url="http://1.2.3.4/%d.ts" % TID, token="local")
        st = self.status()
        self.assertEqual(st["rejected"], 1)
        self.assertEqual(st["accepted"], 0)
        self.assertEqual(st["duplicate"], 0)


# ══════════════════════ ③ 去重三态 ══════════════════════

class TestDedupe(PushCase):
    def test_already_done(self):
        """a) 歌曲表 done 且媒体文件在 → already_done。"""
        self.store.keep_add([TID])
        self.store.keep_mark_done(TID, 999, os.path.join(self.media, f"{TID}.ts"))
        with open(os.path.join(self.media, f"{TID}.ts"), "wb") as fh:
            fh.write(TS_BYTES)
        d = self.body_of(self.push(tid=TID, url=PUSH_URL, token="local"))
        self.assertEqual(d, {"ok": True, "tid": TID, "duplicate": True,
                             "reason": "already_done", "queued": False})
        self.assertIsNone(self.store.q1("SELECT 1 FROM downloads WHERE tid=?", (TID,)))
        self.assertEqual(self.status()["duplicate"], 1)

    def test_already_done_without_file_is_accepted(self):
        """done 但文件不在 = 不算已下完 → 接受入队（表里的状态不动，与 keep_add 语义一致）。"""
        self.store.keep_add([TID])
        self.store.keep_mark_done(TID, 999, os.path.join(self.media, f"{TID}.ts"))   # 文件并不存在
        d = self.body_of(self.push(tid=TID, url=PUSH_URL, token="local"))
        self.assertFalse(d["duplicate"])
        self.assertTrue(d["queued"])
        self.assertEqual(self.store.q1("SELECT state FROM keeps WHERE tid=?", (TID,))["state"], "done")
        self.assertEqual(self.store.q1("SELECT state FROM downloads WHERE tid=?", (TID,))["state"], "pending")

    def test_already_queued_by_keeps_want(self):
        """b) 歌曲表 want → already_queued。"""
        self.store.keep_add([TID])
        d = self.body_of(self.push(tid=TID, url=PUSH_URL, token="local"))
        self.assertEqual(d["reason"], "already_queued")
        self.assertFalse(d["queued"])

    def test_already_queued_by_downloads_pending(self):
        """b) 队列里 pending（哪怕歌曲表没有该编号）→ already_queued。"""
        self.store.ex("INSERT INTO downloads(tid,state) VALUES(?,?)", (TID, "pending"))
        d = self.body_of(self.push(tid=TID, url=PUSH_URL, token="local"))
        self.assertEqual(d, {"ok": True, "tid": TID, "duplicate": True,
                             "reason": "already_queued", "queued": False})

    def test_recently_pushed(self):
        """c) 同 tid 短时间再推（队列项已被移除，例如用户先从歌曲表移除）→ recently_pushed。"""
        first = self.body_of(self.push(tid=TID, url=PUSH_URL, token="local"))
        self.assertTrue(first["queued"])
        # 模拟"队列/歌曲表已不存在"，但服务器还记得刚刚推过这个编号
        self.store.ex("DELETE FROM downloads WHERE tid=?", (TID,))
        self.store.ex("DELETE FROM keeps WHERE tid=?", (TID,))
        second = self.body_of(self.push(tid=TID, url=PUSH_URL, token="local"))
        self.assertEqual(second, {"ok": True, "tid": TID, "duplicate": True,
                                  "reason": "recently_pushed", "queued": False})
        self.assertEqual(self.status()["duplicate"], 1)

    def test_dedupe_sec_zero_disables_recent_check(self):
        self.cfg.update({"push_dedupe_sec": 0})
        self.push(tid=TID, url=PUSH_URL, token="local")
        self.store.ex("DELETE FROM downloads WHERE tid=?", (TID,))
        self.store.ex("DELETE FROM keeps WHERE tid=?", (TID,))
        d = self.body_of(self.push(tid=TID, url=PUSH_URL, token="local"))
        self.assertFalse(d["duplicate"])          # 关闭去重窗口 → 直接接受


# ══════════════════════ ④ 接受路径 ══════════════════════

class TestAccept(PushCase):
    def test_accept_registers_keeps_and_queue(self):
        """曲库里**没有**该编号也允许登记（matched=false），并带推送 URL 入队。"""
        self.assertIsNone(self.store.q1("SELECT 1 FROM songs WHERE tid=?", (TID,)))
        d = self.body_of(self.push(tid=TID, url=PUSH_URL, name="测试歌", token="local"))
        self.assertEqual(d, {"ok": True, "tid": TID, "duplicate": False, "queued": True})

        k = self.store.q1("SELECT * FROM keeps WHERE tid=?", (TID,))
        self.assertIsNotNone(k)
        self.assertEqual(k["state"], "want")
        self.assertEqual(k["note"], "App 点歌推送")
        self.assertEqual(k["user"], "app")

        row = self.store.q1("SELECT * FROM downloads WHERE tid=?", (TID,))
        self.assertIsNotNone(row)
        self.assertEqual(row["state"], "pending")
        self.assertEqual(row["push_url"], PUSH_URL)
        self.assertEqual(row["source"], "app-push")

        st = self.status()
        self.assertEqual((st["accepted"], st["duplicate"], st["rejected"]), (1, 0, 0))
        self.assertGreater(st["last_at"], 0)
        # 事件里应有一条"接受"（每 tid 一条，便于追溯）
        self.assertTrue(any("App 点歌推送：接受" in e["text"] for e in self.store.events(20)))

    def test_accept_json_body(self):
        """也支持 `application/json`（App 侧两种 body 都能用）。"""
        body = json.dumps({"tid": TID, "url": PUSH_URL, "name": "测试歌", "token": "local"}).encode()
        resp = self.call(req("POST", "/push_download", body=body,
                             headers={"Content-Type": "application/json"}))
        self.assertEqual(resp.status, 200, resp.body)
        self.assertTrue(self.body_of(resp)["queued"])

    def test_accept_when_already_in_library_without_keeps(self):
        """曲库里有、但歌曲表还没有的编号：接受并登记（与"曲库里没有"同一条路径）。"""
        self.store.import_library([(TID, "一帘好梦", "国语", "", "胡云鹏", f"{TID}.ts", 10, "")], "v")
        d = self.body_of(self.push(tid=TID, url=PUSH_URL, token="local"))
        self.assertTrue(d["queued"])
        self.assertEqual(self.store.q1("SELECT note FROM keeps WHERE tid=?", (TID,))["note"], "App 点歌推送")


# ══════════════════════ ⑤ 限流与计数接口 ══════════════════════

class TestRateLimit(PushCase):
    def test_rate_limit_429_and_counters(self):
        self.cfg.update({"push_rate_limit_per_min": 2})
        urls = [f"http://cdn.example.com/480p/72000{i:02d}.ts" for i in range(4)]
        codes = [self.push(tid=7200000 + i, url=urls[i], token="local").status for i in range(4)]
        self.assertEqual(codes, [200, 200, 429, 429])

        resp = self.push(tid=7200000, url=urls[0], token="local")
        self.assertEqual(resp.status, 429)
        self.assertIn("推送过于频繁", self.body_of(resp)["error"])
        self.assertIn("Retry-After", resp.headers)

        st = self.status()
        self.assertEqual(st["accepted"], 2)
        self.assertEqual(st["rate_limited"], 3)
        self.assertGreater(st["last_at"], 0)
        self.assertEqual(st["rate_limit_per_min"], 2)

    def test_rate_limit_per_ip(self):
        """限流按来源 IP 分开计数：换个 IP 不受影响。"""
        self.cfg.update({"push_rate_limit_per_min": 1})
        a = self.call(req("POST", "/push_download", body=form(tid=7200001, url="http://cdn.example.com/7200001.ts",
                                                              token="local"), client_ip="10.0.0.1"))
        b = self.call(req("POST", "/push_download", body=form(tid=7200001, url="http://cdn.example.com/7200001.ts",
                                                              token="local"), client_ip="10.0.0.1"))
        c = self.call(req("POST", "/push_download", body=form(tid=7200002, url="http://cdn.example.com/7200002.ts",
                                                              token="local"), client_ip="10.0.0.2"))
        self.assertEqual((a.status, b.status, c.status), (200, 429, 200))

    def test_zero_rate_limit_disables(self):
        self.cfg.update({"push_rate_limit_per_min": 0})
        for i in range(5):
            self.assertEqual(self.push(tid=7200100 + i, url=f"http://cdn.example.com/{7200100 + i}.ts",
                                       token="local").status, 200)
        self.assertEqual(self.status()["rate_limited"], 0)

    def test_status_shape(self):
        st = self.status()
        for key in ("ok", "accepted", "duplicate", "rejected", "rate_limited", "last_at"):
            self.assertIn(key, st)
        self.assertEqual(st["ok"], True)
        self.assertEqual((st["accepted"], st["duplicate"], st["rejected"], st["rate_limited"]),
                         (0, 0, 0, 0))


# ══════════════════════ ⑥ 下载器：固定 URL 与回退 ══════════════════════

class FixedUrlCase(unittest.TestCase):
    MIN_BYTES = 200

    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({"media_dir": self.media, "data_dir": self.tmp, "tmp_dir": os.path.join(self.tmp, "tmp"),
                         "min_file_bytes": self.MIN_BYTES, "upstream_hosts": ["http://up.example.com"],
                         "push_rate_limit_per_min": 60})
        self.store = Store(os.path.join(self.tmp, "t.db"))
        self.dl = Downloader(self.cfg, self.store)
        self.fetched: list[str] = []
        self.url_calls = {"get_song_url": 0, "get_token": 0}

        def fake_token(host, mac, sn, timeout=20, ua="", min_interval_ms=300, max_interval_ms=None):
            self.url_calls["get_token"] += 1
            return "TK"

        def fake_url(host, tid, token, mac, sn, resolution="720", h265=False, timeout=20,
                     ua="", min_interval_ms=300, max_interval_ms=None):
            self.url_calls["get_song_url"] += 1
            return RESOLVED_URL, "ok"

        self._patches = [mock.patch.object(R, "get_token", fake_token),
                         mock.patch.object(R, "get_song_url", fake_url)]
        for p in self._patches:
            p.start()
        # 下载本身的网络细节全部打桩（_head_size / _fetch 是这条链路唯一碰网络的地方）
        self.dl._head_size = lambda url, timeout: 0
        self.dl._fetch = self._fake_fetch

    def tearDown(self):
        for p in self._patches:
            p.stop()
        try:
            self.store.conn.close()
        except Exception:
            pass

    def _fake_fetch(self, url, tmp, timeout, tid, expected):
        """打桩下载：坏 host 一律 403（模拟推送地址过期），其余写一段合法 TS。"""
        self.fetched.append(url)
        if "bad." in url:
            # ⚠️ HTTPError 继承自 tempfile 包装器：抛之前必须 close()，否则 GC 时会有
            # ResourceWarning 污染验收输出（unittest 默认把警告打到 stderr）。
            err = urllib.error.HTTPError(url, 403, "Forbidden", {}, None)
            try:
                raise err
            finally:
                err.close()
        os.makedirs(os.path.dirname(tmp), exist_ok=True)
        with open(tmp, "wb") as fh:
            fh.write(TS_BYTES)
        return len(TS_BYTES)

    def events_text(self) -> str:
        return "\n".join(e["text"] for e in self.store.events(50))

    def test_fixed_url_skips_resolve(self):
        """固定 URL 可下载 → **完全不调用取址**（get_song_url 计数为 0），且计入同一套校验。"""
        self.store.push_enqueue(TID, PUSH_URL)
        self.dl._download({"tid": TID, "push_url": PUSH_URL, "name": "", "source": "app-push"})
        self.assertEqual(self.url_calls["get_song_url"], 0)
        self.assertEqual(self.url_calls["get_token"], 0)
        self.assertEqual(self.fetched, [PUSH_URL])
        row = self.store.q1("SELECT * FROM downloads WHERE tid=?", (TID,))
        self.assertEqual(row["state"], "done")
        self.assertEqual(row["source_host"], "app-push")
        self.assertTrue(os.path.isfile(os.path.join(self.media, f"{TID}.ts")))
        self.assertEqual(self.store.q1("SELECT state FROM keeps WHERE tid=?", (TID,))["state"], "done")

    def test_fixed_url_failure_falls_back_to_resolve(self):
        """固定 URL 失败（403/过期）→ 回退正常取址重试一次，事件里有记录，固定地址被清掉。"""
        self.store.push_enqueue(TID, BAD_PUSH_URL)
        self.dl._download({"tid": TID, "push_url": BAD_PUSH_URL, "name": "", "source": "app-push"})
        self.assertEqual(self.fetched, [BAD_PUSH_URL, RESOLVED_URL])   # 先推的地址，再取址回退
        self.assertEqual(self.url_calls["get_song_url"], 1)
        row = self.store.q1("SELECT * FROM downloads WHERE tid=?", (TID,))
        self.assertEqual(row["state"], "done")
        self.assertEqual(row["push_url"], "")                          # 死地址不再留着坑以后的重试
        self.assertEqual(row["url"], RESOLVED_URL)
        self.assertEqual(row["source_host"], "http://up.example.com")   # 取址命中源（配置里带协议）
        text = self.events_text()
        self.assertIn("App 推送地址下载失败", text)
        self.assertIn("回退正常取址", text)

    def test_untrusted_push_url_goes_straight_to_resolve(self):
        """库里若是不可信地址（被手工改过）→ 闸门直接拒收 + 清掉，走正常取址（同样一次闸门校验）。"""
        bad = f"http://1.2.3.4/480p/{TID}.ts"
        self.store.push_enqueue(TID, bad)
        self.dl._download({"tid": TID, "push_url": bad, "name": "", "source": "app-push"})
        self.assertEqual(self.fetched, [RESOLVED_URL])
        self.assertEqual(self.url_calls["get_song_url"], 1)
        self.assertIn("未过闸门", self.events_text())
        self.assertEqual(self.store.q1("SELECT push_url FROM downloads WHERE tid=?", (TID,))["push_url"], "")

    def test_normal_task_still_goes_through_resolve(self):
        """没有固定地址的常规任务：照旧走取址（回归保护）。"""
        self.store.ex("INSERT INTO downloads(tid,state) VALUES(?,?)", (TID, "pending"))
        self.dl._download({"tid": TID, "push_url": "", "name": "", "source": ""})
        self.assertEqual(self.fetched, [RESOLVED_URL])
        self.assertEqual(self.url_calls["get_song_url"], 1)

    def test_fixed_url_with_bad_ts_falls_back(self):
        """固定 URL 下到"非 TS"也算失败 → 回退取址（同一套 TS 校验，绝不绕过）。"""
        def junk_fetch(url, tmp, timeout, tid, expected):
            self.fetched.append(url)
            os.makedirs(os.path.dirname(tmp), exist_ok=True)
            if url == PUSH_URL:
                with open(tmp, "wb") as fh:
                    fh.write(b"\x00" * (self.MIN_BYTES + 10))      # 够大但不是 TS
                return self.MIN_BYTES + 10
            with open(tmp, "wb") as fh:
                fh.write(TS_BYTES)
            return len(TS_BYTES)

        self.dl._fetch = junk_fetch
        self.store.push_enqueue(TID, PUSH_URL)
        self.dl._download({"tid": TID, "push_url": PUSH_URL, "name": "", "source": "app-push"})
        self.assertEqual(self.fetched, [PUSH_URL, RESOLVED_URL])
        self.assertIn("不是 TS 流", self.events_text())
        self.assertEqual(self.store.q1("SELECT state FROM downloads WHERE tid=?", (TID,))["state"], "done")


# ══════════════════════ ⑦ 事件日志口径（聚合，不刷屏）══════════════════════

class TestLogAggregation(PushCase):
    def test_repeats_are_aggregated_into_one_line(self):
        """**重复**不逐条写：同一分钟内的多条重复只在分钟翻页时汇总成一条（日志量是硬要求）。"""
        clock = {"t": 1_700_000_000.0}

        class FakeTime:
            @staticmethod
            def time():
                return clock["t"]

        with mock.patch.object(pushdl, "time", FakeTime):
            self.push(tid=TID, url=PUSH_URL, token="local")          # 接受
            self.push(tid=TID, url=PUSH_URL, token="local")          # 重复 ×3
            self.push(tid=TID, url=PUSH_URL, token="local")
            self.push(tid=TID, url=PUSH_URL, token="local")
            texts = [e["text"] for e in self.store.events(20)]
            self.assertEqual(len([t for t in texts if "接受" in t]), 1)   # 接受：每 tid 一条
            self.assertEqual([t for t in texts if "重复跳过" in t], [])   # 当前分钟还没翻页，先不写

            clock["t"] += 61                                        # 进入下一分钟 → 触发汇总
            self.push(tid=TID + 5, url=f"http://cdn.example.com/{TID + 5}.ts", token="local")
            dup = [t["text"] for t in self.store.events(20) if "重复跳过" in t["text"]]
            self.assertEqual(len(dup), 1)                            # 3 条重复 → 恰好一条
            self.assertIn("重复跳过 3 首", dup[0])
            self.assertIn("already_queued=3", dup[0])

    def test_rejected_aggregated_with_sample(self):
        clock = {"t": 1_700_000_000.0}

        class FakeTime:
            @staticmethod
            def time():
                return clock["t"]

        with mock.patch.object(pushdl, "time", FakeTime):
            for _ in range(3):
                self.push(tid=TID, url="http://1.2.3.4/%d.ts" % TID, token="local")   # 三连拒
            clock["t"] += 61
            self.push(tid=TID + 6, url=f"http://cdn.example.com/{TID + 6}.ts", token="local")
            rej = [t["text"] for t in self.store.events(20) if "拒绝" in t["text"]]
            self.assertEqual(len(rej), 1)
            self.assertIn("拒绝 3 条", rej[0])
            self.assertIn("示例：", rej[0])                            # 保留一条明细便于定位
            self.assertIn("地址未通过闸门", rej[0])


# ══════════════════════ ⑧ 队列 / 老库迁移 ══════════════════════

class TestQueueAndMigration(unittest.TestCase):
    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"))

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

    def test_next_pending_includes_pushed_tid_without_library(self):
        """曲库里没有、但有固定 URL 的推送任务，必须能被下载池取到。"""
        self.store.push_enqueue(TID, PUSH_URL)
        items = self.store.next_pending(10, "hot", True, 3, 20)
        ours = [i for i in items if i["tid"] == TID]
        self.assertEqual(len(ours), 1)
        self.assertEqual(ours[0]["push_url"], PUSH_URL)
        self.assertEqual(ours[0]["source"], "app-push")

    def test_next_pending_still_requires_library_for_normal_rows(self):
        """没有固定 URL 的常规任务仍要求编号在曲库里（不能因为改了 JOIN 就放开）。"""
        self.store.ex("INSERT INTO downloads(tid,state) VALUES(?,?)", (999999, "pending"))
        self.assertEqual([i for i in self.store.next_pending(10, "hot", True, 3, 20) if i["tid"] == 999999], [])

    def test_migrate_old_db_adds_push_columns(self):
        """老库（downloads 没有 push_url/source 两列）打开后会自动补齐，不丢原有数据。"""
        path = os.path.join(self.tmp, "old.db")
        conn = sqlite3.connect(path)
        conn.executescript("""
        CREATE TABLE downloads (
            tid            INTEGER PRIMARY KEY,
            state          TEXT NOT NULL DEFAULT 'pending',
            bytes          INTEGER NOT NULL DEFAULT 0,
            expected_bytes INTEGER NOT NULL DEFAULT 0,
            md5            TEXT NOT NULL DEFAULT '',
            path           TEXT NOT NULL DEFAULT '',
            url            TEXT NOT NULL DEFAULT '',
            source_host    TEXT NOT NULL DEFAULT '',
            attempts       INTEGER NOT NULL DEFAULT 0,
            last_error     TEXT NOT NULL DEFAULT '',
            enqueued_at    INTEGER NOT NULL DEFAULT 0,
            started_at     INTEGER NOT NULL DEFAULT 0,
            finished_at    INTEGER NOT NULL DEFAULT 0,
            updated_at     INTEGER NOT NULL DEFAULT 0
        );
        INSERT INTO downloads(tid,state,bytes) VALUES(7009126,'done',123);
        """)
        conn.commit()
        conn.close()

        old = Store(path)
        cols = {r[1] for r in old.q("PRAGMA table_info(downloads)")}
        self.assertIn("push_url", cols)
        self.assertIn("source", cols)
        row = old.q1("SELECT * FROM downloads WHERE tid=?", (7009126,))
        self.assertEqual((row["state"], row["bytes"], row["push_url"], row["source"]),
                         ("done", 123, "", ""))
        # 补齐后可正常写入固定地址
        old.push_enqueue(7009126, PUSH_URL)
        self.assertEqual(old.q1("SELECT push_url, source FROM downloads WHERE tid=?",
                                (7009126,))["push_url"], PUSH_URL)
        old.conn.close()


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