zijunchiang
87aa979abd
版本号：
- 1.5.11 越界（规则 3 规定 patch ≤ 10）→ 修正为 1.6.0（进位到 minor、patch 归 0）
- versionCode 43 不变：43 从未对外发布，本次修正只是把命名拉回约定
合并支线（原在工作区根目录、App 仓库之外，DESIGN.md 自称"支线"）：
- selfhost-source/server/ 自建歌曲源服务器（纯 Python 标准库、零 pip 依赖）
- selfhost-source/deploy/、DESIGN.md、LIBRARY.md、build_inventory.py
- selfhost-source/inventory/ 小清单（playlists.csv、song-stats.md 等）
- ⚠️ inventory 生成物（songs-*.csv / singers-top.csv，共 84MB）不入库，
已写入 .gitignore；需要时用 build_inventory.py 依据 muse.db 重新生成
同时入库的 App 侧支线改动（此前为工作树未提交状态）：
- Whoami.kt（whoami 设备心跳）、DeviceIdentity.kt（设备身份伪装）
- PushDownload.kt / PushDownloadClient.kt（点歌推送）
- AdVideoFilter 带 query/fragment 的地址闸门；KTV_BRIDGE_API 11 → 13
- 修 ktv_api.js 的 global.KtvBridgeApiVersion（12 → 13，与注释及 Kotlin 两处校验一致）
- 对应单测：WhoamiTest / DeviceIdentityTest / PushDownloadTest / AdVideoFilterTest（共 39 例）
文档：
- DESIGN.md 的"在 App 仓库之外"约定改为"已并入仓库统一版本化"
- AGENTS.md 版本号行更新为 1.6.0/43，并新增支线并入说明
验证：assembleRelease + testReleaseUnitTest 全过（39 例）
887 lines
48 KiB
Python
887 lines
48 KiB
Python
#!/usr/bin/env python3
|
||
"""歌曲表（keeps）专项测试：**名单 → 下载 → 删除文件 → 媒体扫描** 的一致性（不启动服务、不联网）。
|
||
|
||
覆盖需求里的每一条：
|
||
1) 增 / 删 / 查 / 分页 / 统计（**曲库没有的编号也允许加入**，标记 matched=false）；
|
||
2) **删除歌曲表条目会删除媒体文件**（临时目录造文件；断言删除与 freed_bytes；并断言
|
||
**不会**删到 media_dir 之外的同名文件，也不会跟软链接逃逸到目录外）；
|
||
3) 导入 merge / replace 与导出往返（列与曲库清单导出一致）；
|
||
**只给「编号」的 CSV 导入后，用曲库自动补全歌名/歌手等字段**（端到端，见 TestKeepSelfFill）；
|
||
4) `keep_next_batch` 只返回歌曲表里未完成、且能从曲库匹配到的歌（不在表里的歌永不下）；
|
||
5) `keep_adopt_media` 四种情形（未知文件**登记不删** / 在表里不是 done → 更新为 done /
|
||
**done 但文件缺失或过小 → 回队** / 目录不可用 → 什么都不改）；
|
||
6) 批量驱动器与下载器只认歌曲表；管理接口 /api/keeps* 全链路。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from ktv import admin
|
||
from ktv.batch import BatchManager
|
||
from ktv.config import Config
|
||
from ktv.downloader import Downloader
|
||
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 jbody(data: dict) -> bytes:
|
||
return json.dumps(data, ensure_ascii=False).encode()
|
||
|
||
|
||
def rows(*specs):
|
||
"""(tid, name, lang, singers, hot) → import_library 需要的一行。"""
|
||
return [(tid, name, lang, "", singers, f"{tid}.ts", hot, "") for tid, name, lang, singers, hot in specs]
|
||
|
||
|
||
class KeepCase(unittest.TestCase):
|
||
"""公共夹具：4 首歌的小曲库 + 一个空的媒体目录（都在临时目录里）。"""
|
||
|
||
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.store = Store(os.path.join(self.tmp, "t.db"))
|
||
self.store.import_library(rows((1, "A歌", "国语", "s1", 10),
|
||
(2, "B歌", "粤语", "s2", 9),
|
||
(3, "C歌", "英语", "s3", 8),
|
||
(4, "D歌", "国语", "s4", 7)), "v1")
|
||
|
||
def tearDown(self):
|
||
try:
|
||
self.store.conn.close() # 显式关连接：避免 ResourceWarning 污染验收输出
|
||
except Exception:
|
||
pass
|
||
|
||
def write_media(self, tid: int, size: int = 2048, base: str | None = None) -> str:
|
||
path = os.path.join(base or self.media, f"{tid}.ts")
|
||
with open(path, "wb") as fh:
|
||
fh.write(b"\x47" * size)
|
||
return path
|
||
|
||
def keeps_state(self, tid: int) -> str:
|
||
row = self.store.q1("SELECT state FROM keeps WHERE tid=?", (tid,))
|
||
return row["state"] if row else ""
|
||
|
||
|
||
class TestKeepCrud(KeepCase):
|
||
"""1) 增 / 删 / 查 / 分页 / 统计（曲库暂缺的编号也允许先加入）。"""
|
||
|
||
def test_add_allows_unknown_tid_and_reports_match(self):
|
||
r = self.store.keep_add([1, 2, 999, "x"])
|
||
# 一律允许加入：999 曲库暂时没有也加进去（matched=false），非法项 "x" 丢弃
|
||
self.assertEqual((r["added"], r["matched"], r["unmatched"]), (3, 2, 1))
|
||
self.assertEqual(r["added"], r["matched"] + r["unmatched"])
|
||
# 重复加入不覆盖状态（已下完的不能被重新拉回待下载）
|
||
self.store.keep_mark_done(1, 111, "/m/1.ts")
|
||
r = self.store.keep_add([1, 3], user="前台", note="客户点的")
|
||
self.assertEqual((r["added"], r["skipped_existing"], r["matched"], r["unmatched"]), (1, 1, 1, 0))
|
||
self.assertEqual(self.keeps_state(1), "done")
|
||
self.assertEqual(self.store.keep_stats()["total"], 4)
|
||
self.assertEqual(self.store.keep_rows(keyword="C歌")["items"][0]["note"], "客户点的")
|
||
|
||
def test_add_by_filter(self):
|
||
r = self.store.keep_add_by_filter(lang="国语")
|
||
self.assertEqual((r["added"], r["matched"], r["unmatched"]), (2, 2, 0)) # A歌、D歌
|
||
self.assertEqual(sorted(x["tid"] for x in self.store.keep_rows()["items"]), [1, 4])
|
||
r2 = self.store.keep_add_by_filter(keyword="B歌", limit=0)
|
||
self.assertEqual(r2["added"], 1)
|
||
self.assertEqual(self.store.keep_stats()["total"], 3)
|
||
self.assertEqual(self.store.keep_add_by_filter(lang="国语")["added"], 0) # 已在表里
|
||
|
||
def test_remove_clears_keeps_and_downloads(self):
|
||
self.store.keep_add([1, 2])
|
||
self.store.enqueue([1, 2])
|
||
res = self.store.keep_remove([1, 999], delete_files=False)
|
||
self.assertEqual(res["removed"], 1)
|
||
self.assertIsNone(self.store.q1("SELECT 1 FROM keeps WHERE tid=1"))
|
||
self.assertIsNone(self.store.q1("SELECT 1 FROM downloads WHERE tid=1")) # 登记一并清掉
|
||
self.assertIsNotNone(self.store.q1("SELECT 1 FROM downloads WHERE tid=2"))
|
||
|
||
def test_rows_filters_paging_and_counts(self):
|
||
self.store.keep_add([1, 2, 3])
|
||
self.store.keep_mark_done(3, 5, "")
|
||
d = self.store.keep_rows(page=1, page_size=2, order="hot", desc=True)
|
||
self.assertEqual(d["total"], 3)
|
||
self.assertEqual([x["tid"] for x in d["items"]], [1, 2]) # 热度降序
|
||
d2 = self.store.keep_rows(page=2, page_size=2, order="hot", desc=True)
|
||
self.assertEqual([x["tid"] for x in d2["items"]], [3])
|
||
self.assertEqual(self.store.keep_rows(state="done")["total"], 1)
|
||
self.assertEqual(self.store.keep_rows(state="want")["total"], 2)
|
||
self.assertEqual(self.store.keep_rows(lang="粤语")["total"], 1)
|
||
self.assertEqual(self.store.keep_rows(keyword="s3")["total"], 1) # 歌手命中
|
||
self.assertEqual(self.store.keep_rows(keyword="2")["total"], 1) # 编号命中
|
||
# counts 是整表统计，不受筛选影响；行内有曲库清单的列 + matched + 状态相关列
|
||
self.assertEqual(self.store.keep_rows(state="want")["counts"]["total"], 3)
|
||
item = self.store.keep_rows(keyword="A歌")["items"][0]
|
||
for k in ("tid", "name", "lang", "album", "singers", "filename", "hot_score", "matched",
|
||
"state", "bytes", "attempts", "last_error", "user", "note",
|
||
"added_at", "updated_at", "finished_at", "download_state"):
|
||
self.assertIn(k, item)
|
||
self.assertIs(item["matched"], True)
|
||
|
||
def test_stats_and_bytes(self):
|
||
self.store.keep_add([1, 2, 3, 4])
|
||
self.store.keep_mark_done(1, 2048, "/m/1.ts")
|
||
self.store.keep_mark_failed(2, "取址失败")
|
||
self.store.keep_mark_downloading(3)
|
||
st = self.store.keep_stats()
|
||
self.assertEqual(st["total"], 4)
|
||
self.assertEqual((st["done"], st["failed"], st["downloading"], st["want"]), (1, 1, 1, 1))
|
||
self.assertEqual((st["matched"], st["unmatched"]), (4, 0))
|
||
self.assertEqual(st["bytes"], 2048)
|
||
self.assertEqual(st["remaining"], 3)
|
||
|
||
|
||
class TestRemoveDeletesFiles(KeepCase):
|
||
"""2) 删除歌曲表条目 = 删除媒体文件（且绝不越界删）。"""
|
||
|
||
def test_remove_deletes_file_and_reports_freed_bytes(self):
|
||
size = 5000
|
||
self.write_media(1, size)
|
||
self.store.keep_add([1])
|
||
self.store.keep_mark_done(1, size, os.path.join(self.media, "1.ts"))
|
||
res = self.store.keep_remove([1], delete_files=True, media_dir=self.media)
|
||
self.assertEqual(res["removed"], 1)
|
||
self.assertEqual(res["deleted_files"], 1)
|
||
self.assertEqual(res["freed_bytes"], size)
|
||
self.assertEqual(res["errors"], [])
|
||
self.assertFalse(os.path.exists(os.path.join(self.media, "1.ts")))
|
||
# 日志里要说明删了哪个文件、多大
|
||
self.assertTrue(any("删除媒体文件" in e["text"] and f"{size} 字节" in e["text"]
|
||
for e in self.store.events(20)))
|
||
|
||
def test_never_deletes_file_outside_media_dir(self):
|
||
outside = os.path.join(self.tmp, "outside")
|
||
os.makedirs(outside, exist_ok=True)
|
||
self.write_media(1, 1000) # media/1.ts
|
||
self.write_media(1, 1000, base=outside) # outside/1.ts（同名的目录外文件）
|
||
self.store.keep_add([1])
|
||
res = self.store.keep_remove([1], delete_files=True, media_dir=self.media)
|
||
self.assertEqual(res["deleted_files"], 1)
|
||
self.assertFalse(os.path.exists(os.path.join(self.media, "1.ts")))
|
||
self.assertTrue(os.path.isfile(os.path.join(outside, "1.ts")), "目录外的同名文件绝不能被删")
|
||
|
||
def test_refuses_symlink_escape(self):
|
||
outside = os.path.join(self.tmp, "outside")
|
||
os.makedirs(outside, exist_ok=True)
|
||
target = self.write_media(2, 777, base=outside)
|
||
os.symlink(target, os.path.join(self.media, "2.ts")) # 软链接指向目录外
|
||
self.store.keep_add([2])
|
||
res = self.store.keep_remove([2], delete_files=True, media_dir=self.media)
|
||
self.assertEqual(res["deleted_files"], 0)
|
||
self.assertTrue(any("不在媒体目录内" in e for e in res["errors"]))
|
||
self.assertTrue(os.path.isfile(target), "软链接目标不能被删")
|
||
|
||
def test_delete_files_false_keeps_file(self):
|
||
self.write_media(1, 100)
|
||
self.store.keep_add([1])
|
||
res = self.store.keep_remove([1], delete_files=False, media_dir=self.media)
|
||
self.assertEqual((res["removed"], res["deleted_files"], res["freed_bytes"]), (1, 0, 0))
|
||
self.assertTrue(os.path.isfile(os.path.join(self.media, "1.ts")))
|
||
|
||
def test_missing_file_is_not_an_error(self):
|
||
self.store.keep_add([1])
|
||
res = self.store.keep_remove([1], delete_files=True, media_dir=self.media)
|
||
self.assertEqual((res["removed"], res["deleted_files"], res["errors"]), (1, 0, []))
|
||
|
||
def test_safe_delete_rejects_bad_input(self):
|
||
self.write_media(1, 10)
|
||
self.assertEqual(Store._safe_delete_media(self.media, "../1")[2], "编号非法")
|
||
self.assertEqual(Store._safe_delete_media(self.media, 1)[0], True)
|
||
self.assertEqual(Store._safe_delete_media("", 1)[2], "媒体目录未配置")
|
||
self.assertEqual(Store._safe_delete_media(self.media, 404)[2], "文件不存在")
|
||
|
||
|
||
class TestKeepCsv(KeepCase):
|
||
"""3) 导入 merge / replace 与导出往返（列与曲库清单导出一致）。"""
|
||
|
||
def _seed(self):
|
||
self.store.keep_add([1, 2, 3])
|
||
self.store.keep_mark_done(2, 2048, os.path.join(self.media, "2.ts"))
|
||
|
||
def test_export_columns_match_library_export(self):
|
||
self._seed()
|
||
head = self.store.keep_csv().splitlines()[0]
|
||
self.assertEqual(head, self.store.library_csv().splitlines()[0])
|
||
self.assertEqual(head, "编号,歌名,语言,专辑,歌手,文件名,热度,下载状态,字节")
|
||
self.assertIn("已下", self.store.keep_csv()) # 状态用中文可读值
|
||
self.assertIn("待下载", self.store.keep_csv())
|
||
|
||
def test_merge_roundtrip_preserves_states(self):
|
||
self._seed()
|
||
self.store.keep_mark_failed(3, "网络抖动")
|
||
text = self.store.keep_csv()
|
||
before = {x["tid"]: x["state"] for x in self.store.keep_rows(page_size=100)["items"]}
|
||
self.store.keep_import_csv("编号\n4\n", mode="replace") # 先整表换成只有 4
|
||
self.assertEqual(self.store.keep_stats()["total"], 1)
|
||
res = self.store.keep_import_csv(text, mode="merge") # 再把导出的内容合并回来
|
||
self.assertEqual(res["errors"], [])
|
||
after = {x["tid"]: x["state"] for x in self.store.keep_rows(page_size=100)["items"]}
|
||
for tid, state in before.items():
|
||
self.assertEqual(after[tid], state)
|
||
|
||
def test_replace_mode_replaces_whole_table(self):
|
||
size = 333
|
||
self.write_media(1, size)
|
||
self.write_media(2, size)
|
||
self._seed()
|
||
self.store.enqueue([1, 2])
|
||
res = self.store.keep_import_csv("编号,下载状态\n3,待下载\n", mode="replace")
|
||
self.assertEqual((res["mode"], res["removed"], res["updated"]), ("replace", 3, 1))
|
||
self.assertEqual([x["tid"] for x in self.store.keep_rows()["items"]], [3])
|
||
self.assertIsNone(self.store.q1("SELECT 1 FROM downloads WHERE tid=1")) # 登记清掉
|
||
# 媒体文件**不在这里删**（删本地文件只发生在「显式移除」，避免误导入清空媒体目录）
|
||
self.assertTrue(os.path.isfile(os.path.join(self.media, "1.ts")))
|
||
|
||
def test_import_aliases_and_bad_rows(self):
|
||
text = ("编号,歌名,下载状态\n"
|
||
"1,A歌,待下载\n"
|
||
"2,B歌,已就绪\n"
|
||
"x,X歌,待下载\n"
|
||
"3,C歌,啥玩意\n"
|
||
"404,不存在,待下载\n")
|
||
res = self.store.keep_import_csv(text, mode="merge")
|
||
# 404 曲库里没有也照样写入（matched=false）；坏编号/坏状态才跳过
|
||
self.assertEqual((res["updated"], res["skipped"], res["matched"], res["unmatched"]),
|
||
(3, 2, 2, 1))
|
||
self.assertEqual(res["updated"], res["matched"] + res["unmatched"])
|
||
self.assertEqual(len(res["errors"]), 2)
|
||
self.assertTrue(any("编号无效" in e for e in res["errors"]))
|
||
self.assertTrue(any("状态无法识别" in e for e in res["errors"]))
|
||
self.assertEqual(self.keeps_state(1), "want")
|
||
self.assertEqual(self.keeps_state(2), "done")
|
||
self.assertEqual(self.keeps_state(404), "want")
|
||
un = self.store.keep_rows(state="unmatched")["items"] # 特殊筛选：曲库匹配不到
|
||
self.assertEqual([x["tid"] for x in un], [404])
|
||
self.assertIs(un[0]["matched"], False)
|
||
|
||
def test_import_only_tid_column_and_empty(self):
|
||
res = self.store.keep_import_csv("编号\n1\n4\n")
|
||
self.assertEqual((res["updated"], res["matched"], res["unmatched"]), (2, 2, 0))
|
||
self.assertEqual(self.keeps_state(1), "want")
|
||
res2 = self.store.keep_import_csv("")
|
||
self.assertEqual(res2["updated"], 0)
|
||
self.assertTrue(res2["errors"])
|
||
self.assertEqual(self.store.keep_import_csv("歌名,歌手\nA,s\n")["errors"],
|
||
["缺少「编号」列（首行应为表头，或第 1 列为编号）"])
|
||
|
||
|
||
def test_import_many_rows_is_chunked(self):
|
||
"""大 CSV：曲库存在性校验/删除登记都要分批（SQLite 的 SQL 变量数有上限，且省内存）。"""
|
||
n = 1200
|
||
self.store.import_library(rows(*[(1000 + i, f"S{i}", "国语", "s", i) for i in range(n)]), "v2")
|
||
max_ph = 0
|
||
orig_q = self.store.q
|
||
|
||
def spy(sql, args=()):
|
||
nonlocal max_ph
|
||
if " IN (" in sql:
|
||
max_ph = max(max_ph, len(tuple(args)))
|
||
return orig_q(sql, args)
|
||
|
||
self.store.q = spy # 盯住所有 IN (...) 查询的参数个数
|
||
text = "编号,下载状态\n" + "".join(f"{1000 + i},待下载\n" for i in range(n))
|
||
res = self.store.keep_import_csv(text, mode="merge")
|
||
self.assertEqual((res["updated"], res["unmatched"]), (n, 0))
|
||
self.assertEqual(self.store.keep_stats()["total"], n)
|
||
res = self.store.keep_import_csv("编号\n1000\n", mode="replace")
|
||
self.assertEqual((res["removed"], res["updated"]), (n, 1))
|
||
self.assertLessEqual(max_ph, 400, f"单条 SQL 参数过多（{max_ph}），没有分批")
|
||
|
||
|
||
class TestKeepSelfFill(KeepCase):
|
||
"""3b) 自行填充：只给「编号」也能用，列表字段实时 LEFT JOIN 曲库补全（端到端，不联网）。"""
|
||
|
||
def test_import_only_tid_csv_then_filled_from_library(self):
|
||
"""端到端：内存 CSV 只含「编号」列 → 导入 → 查询/导出立刻被曲库补全。"""
|
||
res = self.store.keep_import_csv("编号\n1\n3\n777\n", mode="merge")
|
||
self.assertEqual((res["updated"], res["matched"], res["unmatched"]), (3, 2, 1))
|
||
rows = {x["tid"]: x for x in self.store.keep_rows(page_size=100)["items"]}
|
||
# 1 / 3：歌名、歌手、语言、文件名、热度全部由曲库补全
|
||
self.assertIs(rows[1]["matched"], True)
|
||
self.assertEqual((rows[1]["name"], rows[1]["singers"], rows[1]["lang"],
|
||
rows[1]["filename"], rows[1]["hot_score"]),
|
||
("A歌", "s1", "国语", "1.ts", 10))
|
||
self.assertIs(rows[3]["matched"], True)
|
||
self.assertEqual(rows[3]["name"], "C歌")
|
||
# 777：曲库暂时没有也允许加入，matched=false、列表字段为空
|
||
self.assertIs(rows[777]["matched"], False)
|
||
self.assertEqual((rows[777]["name"], rows[777]["singers"], rows[777]["hot_score"]),
|
||
("", "", 0))
|
||
# 导出同样带上补全后的信息，且未匹配的行不会丢
|
||
text = self.store.keep_csv()
|
||
self.assertIn("A歌", text)
|
||
self.assertIn("777", text)
|
||
|
||
def test_unmatched_becomes_matched_after_library_sync(self):
|
||
"""先加曲库没有的编号（matched=false）→ 曲库同步进来 → 自动 matched=true 且字段补全。"""
|
||
self.assertEqual(self.store.keep_add([777])["unmatched"], 1)
|
||
row = self.store.keep_rows(page_size=10)["items"][0]
|
||
self.assertIs(row["matched"], False)
|
||
self.assertEqual((row["name"], row["hot_score"]), ("", 0))
|
||
self.assertEqual(self.store.keep_stats()["unmatched"], 1)
|
||
self.assertEqual([x["tid"] for x in self.store.keep_unmatched()], [777])
|
||
self.assertEqual([x["tid"] for x in self.store.keep_rows(state="unmatched")["items"]], [777])
|
||
self.assertEqual(self.store.keep_remaining(), 0) # 匹配不到 → 不参与下载
|
||
|
||
self.store.import_library(rows((777, "新歌", "国语", "新歌手", 88)), "v2") # 模拟曲库同步
|
||
|
||
row = self.store.keep_rows(page_size=10)["items"][0]
|
||
self.assertIs(row["matched"], True) # JOIN 自动生效，无需额外动作
|
||
self.assertEqual((row["name"], row["singers"], row["lang"], row["hot_score"]),
|
||
("新歌", "新歌手", "国语", 88))
|
||
self.assertEqual(self.store.keep_stats()["unmatched"], 0)
|
||
self.assertEqual(self.store.keep_unmatched(), [])
|
||
self.assertEqual(self.store.keep_rows(state="matched")["total"], 1)
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(10)], [777]) # 现在能下
|
||
|
||
def test_library_metadata_update_is_live(self):
|
||
"""用户表里不存冗余列：曲库改了歌名，歌曲表显示立刻跟着变。"""
|
||
self.store.keep_add([1])
|
||
self.assertEqual(self.store.keep_rows()["items"][0]["name"], "A歌")
|
||
self.store.import_library(rows((1, "A歌改名", "国语", "s1", 10)), "v3")
|
||
self.assertEqual(self.store.keep_rows()["items"][0]["name"], "A歌改名")
|
||
|
||
def test_import_empty_other_columns_is_fine(self):
|
||
"""CSV 带了歌名/歌手/语言等列但留空：一律以曲库为准，不会被空值覆盖。"""
|
||
res = self.store.keep_import_csv("编号,歌名,歌手,语言\n1,,,\n")
|
||
self.assertEqual((res["updated"], res["matched"]), (1, 1))
|
||
row = self.store.keep_rows(page_size=10)["items"][0]
|
||
self.assertEqual((row["name"], row["singers"], row["lang"]), ("A歌", "s1", "国语"))
|
||
|
||
|
||
class TestKeepNextToEnqueue(KeepCase):
|
||
"""4) 补队候选只来自歌曲表里未完成、且能从曲库匹配到的歌。"""
|
||
|
||
def test_only_keeps_and_unfinished(self):
|
||
self.store.keep_add([1, 3])
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(10)], [1, 3]) # 热度降序
|
||
# 曲库里有、但不在歌曲表里的 2/4 永远不出现
|
||
self.assertNotIn(2, [c["tid"] for c in self.store.keep_next_batch(10)])
|
||
|
||
def test_order_limit_and_state_filter(self):
|
||
self.store.keep_add([1, 2, 3, 4])
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(2, "hot", True)], [1, 2])
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(2, "tid", False)], [1, 2])
|
||
self.store.keep_mark_done(1, 1, "")
|
||
self.store.keep_mark_downloading(2)
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(10)], [3, 4])
|
||
self.store.keep_mark_failed(3, "失败也仍会被重试")
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(10)], [3, 4])
|
||
|
||
def test_removed_upstream_not_offered(self):
|
||
self.store.keep_add([1, 2])
|
||
self.store.diff_and_import(rows((2, "B歌", "粤语", "s2", 9)), "v2") # 1 上游已删除
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(10)], [2])
|
||
|
||
def test_unmatched_not_offered_nor_counted(self):
|
||
"""曲库暂无的编号：允许在歌曲表里等着，但不参与补队、也不计入"还没下完"。"""
|
||
self.store.keep_add([1, 777])
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(10)], [1])
|
||
self.assertEqual(self.store.keep_remaining(), 1) # 只算能匹配到的
|
||
self.assertEqual(self.store.keep_stats()["unmatched"], 1)
|
||
|
||
def test_reset_stuck_requeues_downloading_keeps(self):
|
||
"""进程重启：卡在 downloading 的条目必须回到待下载，否则永远补不到队。"""
|
||
self.store.keep_add([1, 2])
|
||
self.store.keep_mark_downloading(1)
|
||
self.store.enqueue([1])
|
||
self.store.claim_one(1)
|
||
self.store.reset_stuck()
|
||
self.assertEqual(self.keeps_state(1), "want")
|
||
self.assertEqual(self.store.q1("SELECT state FROM downloads WHERE tid=1")["state"], "pending")
|
||
self.assertEqual([c["tid"] for c in self.store.keep_next_batch(10)], [1, 2])
|
||
|
||
|
||
class TestKeepAdoptMedia(KeepCase):
|
||
"""5) 媒体目录扫描（`keep_adopt_media`）：认领新文件 / 更新状态 / **缺失回队**（绝不删文件）。"""
|
||
|
||
def test_missing_requeues_and_resets_downloads(self):
|
||
"""done 但文件不存在 → 回到 want，并把 downloads 的完成登记放回 pending。"""
|
||
self.store.keep_add([1])
|
||
self.store.enqueue([1])
|
||
self.store.claim_one(1)
|
||
self.store.finish(1, True, bytes_=5, path=os.path.join(self.media, "1.ts"))
|
||
self.store.keep_mark_done(1, 5, os.path.join(self.media, "1.ts"))
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertTrue(res["ok"], res)
|
||
self.assertEqual(res["requeued"], 1)
|
||
self.assertEqual(self.keeps_state(1), "want")
|
||
self.assertEqual(self.store.q1("SELECT state FROM downloads WHERE tid=1")["state"], "pending")
|
||
|
||
def test_small_file_requeues(self):
|
||
"""文件在、但小于最小字节数 → 同样回队（不能当成"已下好"）。"""
|
||
self.write_media(1, 10)
|
||
self.store.keep_add([1])
|
||
self.store.keep_mark_done(1, 10, os.path.join(self.media, "1.ts"))
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual(res["requeued"], 1)
|
||
self.assertEqual(self.keeps_state(1), "want")
|
||
|
||
def test_done_with_file_present_stays_done(self):
|
||
self.write_media(1, 4096)
|
||
self.store.keep_add([1])
|
||
self.store.keep_mark_done(1, 4096, os.path.join(self.media, "1.ts"))
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual(res["requeued"], 0)
|
||
self.assertEqual(self.keeps_state(1), "done")
|
||
|
||
def test_adopts_existing_file(self):
|
||
"""在表里但不是 done、文件够大 → 认领为 done（避免重复下载）。"""
|
||
self.write_media(1, 4096)
|
||
self.store.keep_add([1])
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual(res["updated"], 1)
|
||
item = self.store.keep_rows(keyword="A歌")["items"][0]
|
||
self.assertEqual((item["state"], item["bytes"]), ("done", 4096))
|
||
|
||
def test_unknown_file_is_adopted_never_deleted(self):
|
||
"""媒体目录里不在歌曲表的文件：**登记进歌曲表**（用户放进来的文件要留住）。"""
|
||
size = 4096
|
||
self.write_media(9, size)
|
||
self.store.keep_add([1])
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual((res["adopted_count"], res["updated"]), (1, 0))
|
||
self.assertTrue(os.path.isfile(os.path.join(self.media, "9.ts"))) # 文件还在
|
||
self.assertEqual(self.keeps_state(9), "done")
|
||
self.assertEqual(self.store.q1("SELECT bytes FROM keeps WHERE tid=9")["bytes"], size)
|
||
self.assertEqual(self.store.keep_rows(state="unmatched")["total"], 1) # 曲库没有 9
|
||
|
||
def test_small_unknown_file_skipped_and_kept(self):
|
||
"""过小的未知文件：跳过（不登记），但**绝不删除**。"""
|
||
self.write_media(9, 10)
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual((res["adopted_count"], res["skipped"]), (0, 1))
|
||
self.assertTrue(os.path.isfile(os.path.join(self.media, "9.ts")))
|
||
self.assertIsNone(self.store.q1("SELECT 1 FROM keeps WHERE tid=9"))
|
||
|
||
def test_media_dir_unavailable_is_skipped(self):
|
||
self.store.keep_add([1])
|
||
self.store.keep_mark_done(1, 5, "")
|
||
res = self.store.keep_adopt_media(os.path.join(self.tmp, "nope"), min_bytes=100, timeout_sec=5)
|
||
self.assertFalse(res["ok"])
|
||
self.assertEqual(res["requeued"], 0)
|
||
self.assertEqual(self.keeps_state(1), "done") # 挂载丢失时绝不误判全部缺失
|
||
|
||
def test_requeue_only_touches_done_rows(self):
|
||
self.store.keep_add([2, 9999999]) # 未下载 + 曲库外
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual(res["requeued"], 0)
|
||
self.assertEqual([self.keeps_state(t) for t in (2, 9999999)], ["want", "want"])
|
||
|
||
|
||
class TestRetiredTablesCleanup(KeepCase):
|
||
"""6) 老库升级：`Store._RETIRED_TABLES` 里列出的历史表会被幂等清除。"""
|
||
|
||
def test_retired_tables_dropped_on_open(self):
|
||
for old in Store._RETIRED_TABLES: # 造出历史遗留表
|
||
self.store.ex(f"CREATE TABLE IF NOT EXISTS {old}(tid INTEGER PRIMARY KEY, v TEXT)")
|
||
self.assertEqual(len(Store._RETIRED_TABLES), 1)
|
||
self.store.conn.close()
|
||
self.store = Store(os.path.join(self.tmp, "t.db")) # 重新打开 = 走一次 _migrate
|
||
names = {r[0] for r in self.store.conn.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='table'")}
|
||
for old in Store._RETIRED_TABLES:
|
||
self.assertNotIn(old, names)
|
||
self.assertIn("keeps", names) # 正常表不受影响
|
||
|
||
|
||
class TestBatchKeeps(KeepCase):
|
||
"""7a) 批量驱动器：只补歌曲表里的歌；全部下完 → 正常收工。"""
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.batch = BatchManager(self.cfg, self.store, None, None)
|
||
self.batch.state["enabled"] = True # 模拟"已在运行"
|
||
|
||
def test_tick_enqueues_only_keeps(self):
|
||
self.store.keep_add([2])
|
||
self.assertTrue(self.batch.tick())
|
||
queued = [r["tid"] for r in self.store.q("SELECT tid FROM downloads ORDER BY tid")]
|
||
self.assertEqual(queued, [2]) # 不在歌曲表里的 1/3/4 不入队
|
||
|
||
def test_status_exposes_keeps_counters(self):
|
||
self.store.keep_add([1, 2])
|
||
self.store.keep_mark_done(1, 100, "/m/1.ts")
|
||
st = self.batch.status()
|
||
self.assertEqual((st["keeps_total"], st["keeps_done"], st["keeps_want"], st["keeps_failed"]),
|
||
(2, 1, 1, 0))
|
||
self.assertEqual(st["keeps_remaining"], 1)
|
||
self.assertEqual(st["keeps_bytes"], 100)
|
||
|
||
def test_finishes_when_all_keeps_done(self):
|
||
self.store.keep_add([1])
|
||
self.assertTrue(self.batch.tick()) # 补队
|
||
self.assertEqual(self.store.stats()["pending"], 1)
|
||
self.store.claim_one(1)
|
||
self.store.finish(1, True, bytes_=1)
|
||
self.store.keep_mark_done(1, 1, os.path.join(self.media, "1.ts"))
|
||
self.assertFalse(self.batch.tick()) # 队列空 + 歌曲表无未完成 → 收工
|
||
st = self.batch.status()
|
||
self.assertFalse(st["enabled"])
|
||
self.assertFalse(st["fatal"]) # **不是致命错误**
|
||
self.assertIn("已按歌曲表下载完毕", st["reason"])
|
||
self.assertTrue(any(e["level"] == "INFO" and "BATCH 已按歌曲表下载完毕" in e["text"]
|
||
for e in self.store.events(20)))
|
||
|
||
def test_untouched_library_is_not_fatal(self):
|
||
self.batch.state["last_progress_at"] = 0 # 极端：从没有过进展
|
||
self.assertFalse(self.batch.tick())
|
||
self.assertFalse(self.batch.status()["fatal"])
|
||
|
||
def test_keeps_remaining_still_running(self):
|
||
self.store.keep_add([1])
|
||
self.assertTrue(self.batch.tick())
|
||
self.assertTrue(self.batch.status()["enabled"])
|
||
|
||
|
||
class TestDownloaderKeeps(KeepCase):
|
||
"""7b) 下载器：成功/失败/抢占都同步歌曲表状态（且能容忍没有该方法的假 store）。"""
|
||
|
||
def test_early_done_path_sets_keeps_done(self):
|
||
self.cfg.update({"min_file_bytes": 1024})
|
||
path = self.write_media(1, 2048)
|
||
self.store.keep_add([1])
|
||
self.store.enqueue([1])
|
||
self.store.claim_one(1)
|
||
self.store.finish(1, True, bytes_=2048, path=path)
|
||
dl = Downloader(self.cfg, self.store)
|
||
dl._download({"tid": 1, "name": "A歌"}) # 走"已就绪"分支，不联网
|
||
item = self.store.keep_rows(keyword="A歌")["items"][0]
|
||
self.assertEqual((item["state"], item["bytes"]), ("done", 2048))
|
||
self.assertEqual(self.store.keep_remaining(), 0)
|
||
|
||
def test_failure_and_downloading_states(self):
|
||
self.store.keep_add([1, 2])
|
||
dl = Downloader(self.cfg, self.store)
|
||
dl._note_downloading(2)
|
||
self.assertEqual(self.keeps_state(2), "downloading")
|
||
self.assertEqual(self.store.q1("SELECT attempts FROM keeps WHERE tid=2")["attempts"], 1)
|
||
dl._fail(1, "文件过小 3B（阈值 6MB）")
|
||
self.assertEqual(self.keeps_state(1), "failed")
|
||
self.assertIn("文件过小", self.store.keep_rows(keyword="A歌")["items"][0]["last_error"])
|
||
|
||
def test_tolerates_store_without_keep_methods(self):
|
||
dl = Downloader(self.cfg, self.store)
|
||
|
||
class Bare:
|
||
def event(self, *a, **k):
|
||
pass
|
||
|
||
dl.store = Bare()
|
||
dl._note_done(1, 1, "p") # 不抛异常即通过
|
||
dl._note_failed(1, "e")
|
||
dl._note_downloading(1)
|
||
|
||
|
||
class TestAdminKeepsEndpoints(KeepCase):
|
||
"""7c) 管理接口 /api/keeps*（以及曲库清单别名）全链路。"""
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.router = admin.build_router(self.cfg, self.store, None)
|
||
|
||
def call(self, method, path, query=None, body=b""):
|
||
return self.router.dispatch(req(method, path, query, body))
|
||
|
||
def test_add_list_stats(self):
|
||
d = json.loads(self.call("POST", "/api/keeps/add", body=jbody({"tids": [1, 2]})).body)
|
||
self.assertTrue(d["ok"])
|
||
self.assertEqual((d["added"], d["stats"]["total"]), (2, 2))
|
||
self.assertEqual((d["matched"], d["unmatched"]), (2, 0))
|
||
d = json.loads(self.call("GET", "/api/keeps", {"state": "want", "page_size": "10"}).body)
|
||
self.assertEqual(d["total"], 2)
|
||
self.assertIn("counts", d)
|
||
d = json.loads(self.call("POST", "/api/keeps/add", body=jbody({"lang": "英语", "limit": 5})).body)
|
||
self.assertEqual(d["added"], 1)
|
||
d = json.loads(self.call("GET", "/api/keeps/stats").body)
|
||
self.assertEqual((d["ok"], d["total"], d["want"]), (True, 3, 3))
|
||
|
||
def test_add_unknown_tid_is_allowed_and_unmatched_endpoint(self):
|
||
"""只给编号（曲库暂时没有）也能加；/api/keeps/unmatched 能列出来核对。"""
|
||
d = json.loads(self.call("POST", "/api/keeps/add", body=jbody({"tids": [1, 777]})).body)
|
||
self.assertTrue(d["ok"])
|
||
self.assertEqual((d["added"], d["matched"], d["unmatched"]), (2, 1, 1))
|
||
d = json.loads(self.call("GET", "/api/keeps", {"state": "unmatched"}).body)
|
||
self.assertEqual([x["tid"] for x in d["items"]], [777])
|
||
self.assertIs(d["items"][0]["matched"], False)
|
||
d = json.loads(self.call("GET", "/api/keeps/unmatched").body)
|
||
self.assertEqual((d["ok"], d["total"], [x["tid"] for x in d["items"]]), (True, 1, [777]))
|
||
# 曲库同步进来后，未匹配列表自动清空，行内字段也补全
|
||
self.store.import_library(rows((777, "新歌", "国语", "新歌手", 88)), "v2")
|
||
d = json.loads(self.call("GET", "/api/keeps/unmatched").body)
|
||
self.assertEqual(d["total"], 0)
|
||
d = json.loads(self.call("GET", "/api/keeps", {"state": "matched", "keyword": "新歌"}).body)
|
||
self.assertEqual([x["tid"] for x in d["items"]], [777])
|
||
self.assertEqual(d["items"][0]["name"], "新歌")
|
||
|
||
def test_add_without_target_is_rejected(self):
|
||
d = json.loads(self.call("POST", "/api/keeps/add", body=jbody({})).body)
|
||
self.assertFalse(d["ok"])
|
||
self.assertTrue(d["error"])
|
||
d = json.loads(self.call("POST", "/api/keeps/add", body=jbody({"limit": -1})).body)
|
||
self.assertFalse(d["ok"])
|
||
|
||
def test_remove_endpoint_returns_impact(self):
|
||
size = 4096
|
||
self.write_media(1, size)
|
||
self.call("POST", "/api/keeps/add", body=jbody({"tids": [1]}))
|
||
self.store.keep_mark_done(1, size, os.path.join(self.media, "1.ts"))
|
||
d = json.loads(self.call("POST", "/api/keeps/remove", body=jbody({"tids": [1]})).body)
|
||
self.assertTrue(d["ok"])
|
||
self.assertEqual((d["removed"], d["deleted_files"], d["freed_bytes"]), (1, 1, size))
|
||
self.assertFalse(os.path.exists(os.path.join(self.media, "1.ts")))
|
||
d2 = json.loads(self.call("POST", "/api/keeps/remove", body=jbody({})).body)
|
||
self.assertFalse(d2["ok"])
|
||
|
||
def test_export_csv_has_bom_and_same_columns(self):
|
||
self.call("POST", "/api/keeps/add", body=jbody({"tids": [1]}))
|
||
resp = self.call("GET", "/api/keeps/export.csv")
|
||
self.assertEqual(resp.status, 200)
|
||
self.assertEqual(resp.content_type, "text/csv; charset=utf-8")
|
||
self.assertEqual(resp.headers["Content-Disposition"], 'attachment; filename="keeps.csv"')
|
||
self.assertTrue(resp.body.startswith(b"\xef\xbb\xbf")) # utf-8-sig BOM
|
||
head = resp.body.decode("utf-8-sig").splitlines()[0]
|
||
self.assertEqual(head, self.store.library_csv().splitlines()[0])
|
||
|
||
def test_import_endpoint_modes(self):
|
||
raw = "编号,下载状态\n1,待下载\n2,已下\n"
|
||
d = json.loads(self.call("POST", "/api/keeps/import", {"mode": "merge"}, raw.encode()).body)
|
||
self.assertTrue(d["ok"])
|
||
self.assertEqual((d["mode"], d["updated"]), ("merge", 2))
|
||
d = json.loads(self.call("POST", "/api/keeps/import",
|
||
body=jbody({"text": "编号\n3\n", "mode": "replace"})).body)
|
||
self.assertEqual((d["mode"], d["removed"], d["updated"]), ("replace", 2, 1))
|
||
d = json.loads(self.call("POST", "/api/keeps/import", {"mode": "oops"}, raw.encode()).body)
|
||
self.assertFalse(d["ok"])
|
||
d = json.loads(self.call("POST", "/api/keeps/import", {"mode": "merge"}, b" ").body)
|
||
self.assertFalse(d["ok"])
|
||
|
||
def test_scan_endpoint_registers_and_requeues(self):
|
||
"""界面上的「扫描媒体目录」：未知文件登记进歌曲表 + done 但文件缺失的条目回队。"""
|
||
self.cfg.update({"media_scan_min_bytes": 100})
|
||
self.write_media(9, 2048)
|
||
self.call("POST", "/api/keeps/add", body=jbody({"tids": [1]}))
|
||
self.store.keep_mark_done(1, 0, "") # done 但文件不存在
|
||
d = json.loads(self.call("POST", "/api/keeps/scan").body)
|
||
self.assertTrue(d["ok"], d)
|
||
self.assertEqual(d["adopted_count"], 1) # 9 被登记
|
||
self.assertEqual(d["requeued"], 1) # 1 被放回待下载
|
||
self.assertTrue(os.path.exists(os.path.join(self.media, "9.ts"))) # 绝不删文件
|
||
self.assertEqual(self.keeps_state(1), "want")
|
||
self.assertEqual(self.keeps_state(9), "done")
|
||
# 最近一次扫描结果（界面「上次媒体扫描」一行）也带上"缺失回队"
|
||
st = json.loads(self.call("GET", "/api/keeps/scan/status").body)["scan"]
|
||
self.assertEqual((st["requeued"], st["adopted"]), (1, 1))
|
||
|
||
def test_removed_reconcile_endpoints_are_gone(self):
|
||
"""两个「对账」接口已删除（旧脚本拿到 404，而不是被静默当成别的接口）。"""
|
||
for path in ("/api/keeps/reconcile", "/api/maintenance/reconcile"):
|
||
resp = self.call("POST", path, body=jbody({}))
|
||
self.assertEqual(resp.status, 404, path)
|
||
|
||
def test_library_endpoints_and_enqueue_explicit_tids(self):
|
||
d = json.loads(self.call("GET", "/api/library/songs", {"page_size": "10"}).body)
|
||
self.assertEqual(d["total"], 4)
|
||
resp = self.call("GET", "/api/library/export.csv")
|
||
self.assertTrue(resp.body.startswith(b"\xef\xbb\xbf"))
|
||
self.assertEqual(resp.body.decode("utf-8-sig").splitlines()[0],
|
||
self.store.keep_csv().splitlines()[0])
|
||
# 自动入队（mode=auto）已删除：不传 tids 一律拒绝
|
||
d = json.loads(self.call("POST", "/api/download/enqueue", body=jbody({"mode": "auto"})).body)
|
||
self.assertFalse(d["ok"])
|
||
# 显式传编号的入队路径保留（下载队列页的"下载选中/本页"还在用）
|
||
self.call("POST", "/api/keeps/add", body=jbody({"tids": [2]}))
|
||
d = json.loads(self.call("POST", "/api/download/enqueue", body=jbody({"tids": [2]})).body)
|
||
self.assertTrue(d["ok"], d)
|
||
self.assertEqual(d["added"], 1)
|
||
|
||
|
||
class TestConfigKeeps(KeepCase):
|
||
"""配置项的取舍：旧「对账」相关键已随按钮一起删除，媒体扫描键保留。"""
|
||
|
||
def test_removed_reconcile_keys_are_gone(self):
|
||
for key in ("keep_prune_orphans", "keep_prune_orphans_on_start",
|
||
"keep_delete_missing_files", "keep_reconcile_interval_sec"):
|
||
self.assertIsNone(self.cfg.get(key))
|
||
|
||
def test_media_scan_defaults(self):
|
||
self.assertTrue(self.cfg.get("media_scan_enabled"))
|
||
self.assertEqual(self.cfg.get("media_scan_interval_sec"), 600)
|
||
self.assertEqual(self.cfg.get("media_scan_timeout_sec"), 30)
|
||
|
||
|
||
class TestKeepAdoptMediaNeverDeletes(KeepCase):
|
||
"""8) ★回归：媒体扫描（认领新文件 + 缺失回队）**永不删除 keeps 行**、也永不删除文件。
|
||
|
||
背景（线上实测）：通过 /api/keeps/add 或 /api/keeps/import 加入 3 条（其中 1 条曲库里没有），
|
||
列表正常显示 3 条；**重启服务或跑一次扫描之后变成 0 条** 是绝不能出现的事。
|
||
这里把语义钉死：曲库匹配不到、文件还没下、文件缺失 —— 一律只改状态，绝不删行。
|
||
"""
|
||
|
||
def test_unmatched_tid_survives_scan(self):
|
||
"""曲库里没有的编号：扫描后仍在，且 matched=false（绝不允许因此删行）。"""
|
||
self.store.keep_add([9999999])
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual(res["requeued"], 0)
|
||
self.assertEqual(self.store.keep_stats()["total"], 1)
|
||
row = self.store.keep_rows()["items"][0]
|
||
self.assertEqual(row["tid"], 9999999)
|
||
self.assertIs(row["matched"], False)
|
||
self.assertEqual(self.store.keep_rows(state="unmatched")["total"], 1)
|
||
|
||
def test_not_downloaded_yet_survives_scan(self):
|
||
"""曲库有、但文件还没下（want）：扫描后仍在且仍是 want。"""
|
||
self.store.keep_add([1, 2, 9999999])
|
||
self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual(self.store.keep_stats()["total"], 3)
|
||
self.assertEqual([self.keeps_state(t) for t in (1, 2)], ["want", "want"])
|
||
|
||
def test_done_but_file_missing_goes_back_to_want_not_deleted(self):
|
||
"""登记 done 但文件不存在：回到 want（回队重下），**不是删掉**。"""
|
||
self.store.keep_add([1])
|
||
self.store.keep_mark_done(1, 12345, os.path.join(self.media, "1.ts"))
|
||
res = self.store.keep_adopt_media(self.media, min_bytes=100, timeout_sec=5)
|
||
self.assertEqual(res["requeued"], 1)
|
||
self.assertEqual(self.keeps_state(1), "want")
|
||
self.assertEqual(self.store.keep_stats()["total"], 1)
|
||
|
||
def test_every_scan_mode_keeps_rows(self):
|
||
"""目录里有过小文件 / 未知文件 / 目录不可用 —— 每种情形都不许**删掉** keeps 行。"""
|
||
self.write_media(9, 10) # 过小：跳过且不删
|
||
self.store.keep_add([1, 2, 9999999])
|
||
self.store.keep_mark_done(1, 1, "") # done 但文件缺失 → 只回队
|
||
for media, kw in ((self.media, {"min_bytes": 100}), (self.media, {"min_bytes": 0}),
|
||
(os.path.join(self.tmp, "nope"), {"min_bytes": 100})):
|
||
self.store.keep_adopt_media(media, timeout_sec=5, **kw)
|
||
tids = [r["tid"] for r in self.store.keep_rows(page_size=50)["items"]]
|
||
for tid in (1, 2, 9999999): # 原有 3 行一行都不能丢
|
||
self.assertIn(tid, tids)
|
||
self.assertLessEqual(self.store.keep_stats()["total"], 4) # 只有 9 可能被认领进来
|
||
self.assertTrue(os.path.isfile(os.path.join(self.media, "9.ts"))) # 目录里的小文件也不删
|
||
|
||
def test_media_dir_unavailable_scan_keeps_rows(self):
|
||
"""媒体目录不可用（FUSE 掉线/很慢）：跳过 + 一行不删 + 返回结构键齐全。"""
|
||
self.store.keep_add([4100081, 4100293, 9999999])
|
||
res = self.store.keep_adopt_media(os.path.join(self.tmp, "nope"), min_bytes=100, timeout_sec=5)
|
||
self.assertFalse(res["ok"])
|
||
self.assertEqual(res["requeued"], 0)
|
||
self.assertEqual(self.store.keep_stats()["total"], 3)
|
||
|
||
def test_reopen_and_scan_keeps_rows_and_content(self):
|
||
"""模拟"重启服务"：重开同一个库 + 跑维护任务同一套扫描 → 行数与内容完全不变。"""
|
||
self.store.keep_add([4100081, 4100293, 9999999], user="前台", note="客户点的")
|
||
before_stats = self.store.keep_stats()
|
||
before_rows = [(r["tid"], r["state"], r["note"], r["matched"])
|
||
for r in self.store.keep_rows(page_size=100)["items"]]
|
||
self.store.conn.close()
|
||
|
||
self.store = Store(os.path.join(self.tmp, "t.db")) # 重开 = 重启
|
||
self.store.reset_stuck() # main.py 启动流程
|
||
self.cfg.update({"media_scan_min_bytes": 100})
|
||
admin.run_media_scan(self.cfg, self.store) # 维护任务的扫描步骤
|
||
|
||
self.assertEqual(self.store.keep_stats(), before_stats)
|
||
self.assertEqual([(r["tid"], r["state"], r["note"], r["matched"])
|
||
for r in self.store.keep_rows(page_size=100)["items"]], before_rows)
|
||
|
||
def test_scan_adopts_unknown_file_and_requeues_missing(self):
|
||
"""扫描的核心作用（旧"对账"的行为）一个都不能丢：未知文件登记、缺失回队，且不删文件。"""
|
||
self.cfg.update({"media_scan_min_bytes": 100})
|
||
self.write_media(9, 4096) # 未知文件
|
||
self.store.keep_add([1, 2])
|
||
self.store.keep_mark_done(1, 4096, os.path.join(self.media, "1.ts")) # done 但文件不存在
|
||
res = admin.run_media_scan(self.cfg, self.store)
|
||
self.assertTrue(res["ok"], res)
|
||
self.assertEqual(res["adopted_count"], 1) # 9 被登记
|
||
self.assertEqual(res["requeued"], 1) # 1 被回队
|
||
self.assertTrue(os.path.isfile(os.path.join(self.media, "9.ts"))) # 绝不删除
|
||
self.assertEqual(self.keeps_state(9), "done")
|
||
self.assertEqual(self.keeps_state(1), "want")
|
||
|
||
|
||
class TestKeepImportReplaceSafety(KeepCase):
|
||
"""9) ★回归：整表替换必须**原子** + 空 CSV **拒执行**（"导入一次就整表没了"的两个入口）。"""
|
||
|
||
def test_replace_with_header_only_is_refused(self):
|
||
self.store.keep_add([1, 2, 3])
|
||
res = self.store.keep_import_csv("编号\n", mode="replace")
|
||
self.assertTrue(res["replace_cancelled"])
|
||
self.assertEqual((res["updated"], res["removed"]), (0, 0))
|
||
self.assertEqual(self.store.keep_stats()["total"], 3) # 一行都没丢
|
||
self.assertEqual(sorted(r["tid"] for r in self.store.keep_rows(page_size=50)["items"]), [1, 2, 3])
|
||
|
||
def test_replace_with_no_valid_row_is_refused(self):
|
||
self.store.keep_add([1, 2, 3])
|
||
for bad in ("编号\nabc\ndef\n", "编号,下载状态\n1,状态乱写\n", "编号\n\n\n"):
|
||
res = self.store.keep_import_csv(bad, mode="replace")
|
||
self.assertTrue(res["replace_cancelled"], bad)
|
||
self.assertTrue(res["errors"])
|
||
self.assertEqual(self.store.keep_stats()["total"], 3) # 每次都被拒，旧行保留
|
||
|
||
def test_replace_still_replaces_when_csv_has_rows(self):
|
||
self.store.keep_add([1, 2, 3])
|
||
res = self.store.keep_import_csv("编号\n2\n4\n", mode="replace")
|
||
self.assertEqual((res["updated"], res["removed"], res["replace_cancelled"]), (2, 3, False))
|
||
self.assertEqual(sorted(r["tid"] for r in self.store.keep_rows(page_size=50)["items"]), [2, 4])
|
||
self.assertEqual(self.store.keep_stats()["total"], 2)
|
||
|
||
def test_replace_rolls_back_when_write_fails(self):
|
||
"""写入阶段抛异常 → 旧行原样保留（旧实现先单独提交 DELETE，会整表清空且无法恢复）。"""
|
||
self.store.keep_add([1, 2, 3])
|
||
|
||
def _boom(conn, payload):
|
||
raise RuntimeError("模拟写入失败（磁盘满 / 库被锁 / FUSE 掉线）")
|
||
|
||
self.store._write_keeps_rows = _boom # 只让写入步骤失败
|
||
with self.assertRaises(RuntimeError):
|
||
self.store.keep_import_csv("编号\n2\n", mode="replace")
|
||
self.assertFalse(self.store.conn.in_transaction) # 事务已整体回滚
|
||
self.assertEqual(sorted(r["tid"] for r in self.store.keep_rows(page_size=50)["items"]), [1, 2, 3])
|
||
self.assertEqual(self.store.keep_stats()["total"], 3)
|
||
|
||
def test_merge_never_deletes(self):
|
||
self.store.keep_add([1, 2, 3])
|
||
res = self.store.keep_import_csv("编号\n4\n", mode="merge")
|
||
self.assertEqual((res["updated"], res["removed"]), (1, 0))
|
||
self.assertEqual(sorted(r["tid"] for r in self.store.keep_rows(page_size=50)["items"]), [1, 2, 3, 4])
|
||
|
||
|
||
class TestKeepDeleteCallSites(KeepCase):
|
||
"""10) 守卫：全项目**只有两处** `DELETE FROM keeps`（源码级检查，防后来人加"清理"）。"""
|
||
|
||
def test_only_two_delete_call_sites(self):
|
||
import re
|
||
base = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "ktv")
|
||
with open(os.path.join(base, "store.py"), encoding="utf-8") as fh:
|
||
src = fh.read()
|
||
hits = [m.start() for m in re.finditer(r"DELETE FROM keeps", src)]
|
||
self.assertEqual(len(hits), 2, "DELETE FROM keeps 只允许 keep_remove() 与 replace_keeps() 两处")
|
||
for pos in hits: # 两处必须落在允许的函数里
|
||
func = re.findall(r"\n def (\w+)\(", src[:pos])[-1]
|
||
self.assertIn(func, ("keep_remove", "replace_keeps"), f"DELETE FROM keeps 出现在 {func}()")
|
||
others = []
|
||
for name in sorted(os.listdir(base)):
|
||
if not name.endswith(".py"):
|
||
continue
|
||
with open(os.path.join(base, name), encoding="utf-8") as fh:
|
||
if "DELETE FROM keeps" in fh.read():
|
||
others.append(name)
|
||
self.assertEqual(others, ["store.py"]) # 别的模块一律不许删 keeps
|
||
self.assertNotIn("keeps", Store._RETIRED_TABLES) # 升级清理也不许 DROP keeps
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|