mirror of
https://github.com/JefferyHcool/BiliNote.git
synced 2026-06-28 03:01:28 +08:00
4a87c5b93b
issue #402 衍生问题：whisper 模型后台下载失败时，/transcriber_models_status 只回传 downloading/downloaded 两个布尔，failed 态被直接丢弃，于是前端表现为 「点了下载没反应、状态一直未下载、且无任何错误提示」。 后端：新增轻量模块 model_download_state 统一维护下载状态（downloading/done/ failed）与失败原因，config.py 的下载触发与状态查询共享同一份内存态；状态接口 新增 failed 字段，失败时附带 error（仓库 404、网络中断、本地路径缺 model.bin 等）。 前端：模型管理列表新增「下载失败」红色徽标 + 错误详情，按钮在失败后变为「重试」； 自定义模型项同样展示失败图标与原因；并对「本次新出现的失败」弹一次 toast 主动提示。 测试：新增 test_model_download_state 覆盖状态流转（downloading/done/failed、 失败原因透传、downloaded 覆盖 failed、重下清错、mlx key 隔离）。 已用 docker compose 启动整套栈验证：触发本地路径缺失与 HF 仓库 404 两种失败， /transcriber_models_status 均正确回传 failed:true + error。 Refs #402 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
114 lines
4.2 KiB
Python
114 lines
4.2 KiB
Python
"""Unit tests for app.transcriber.model_download_state（模型下载状态 + 失败原因跟踪）。
|
||
|
||
与 test_whisper_models 一样按文件路径隔离加载，并桩掉 app.utils.logger，
|
||
避免触发 app/__init__.py（会 import faster_whisper 等重依赖）。
|
||
"""
|
||
import importlib.util
|
||
import logging
|
||
import pathlib
|
||
import sys
|
||
import types
|
||
import unittest
|
||
|
||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||
MODULE_PATH = ROOT / "app" / "transcriber" / "model_download_state.py"
|
||
|
||
|
||
def _load_module():
|
||
if "app" not in sys.modules:
|
||
app_pkg = types.ModuleType("app")
|
||
app_pkg.__path__ = []
|
||
sys.modules["app"] = app_pkg
|
||
if "app.utils" not in sys.modules:
|
||
utils_pkg = types.ModuleType("app.utils")
|
||
utils_pkg.__path__ = []
|
||
sys.modules["app.utils"] = utils_pkg
|
||
if "app.utils.logger" not in sys.modules:
|
||
logger_mod = types.ModuleType("app.utils.logger")
|
||
logger_mod.get_logger = lambda name=None: logging.getLogger(name or "test")
|
||
sys.modules["app.utils.logger"] = logger_mod
|
||
spec = importlib.util.spec_from_file_location("model_download_state_under_test", MODULE_PATH)
|
||
assert spec and spec.loader
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
|
||
|
||
ds = _load_module()
|
||
|
||
|
||
class TestDownloadState(unittest.TestCase):
|
||
def setUp(self):
|
||
# 模块级单例，测试间互相隔离
|
||
ds._status.clear()
|
||
ds._errors.clear()
|
||
|
||
def test_unknown_key_defaults(self):
|
||
row = ds.status_row("tiny", downloaded=False)
|
||
self.assertEqual(
|
||
row,
|
||
{"model_size": "tiny", "downloaded": False, "downloading": False, "failed": False},
|
||
)
|
||
self.assertNotIn("error", row)
|
||
self.assertFalse(ds.is_downloading("tiny"))
|
||
|
||
def test_downloading(self):
|
||
ds.mark_downloading("tiny")
|
||
self.assertTrue(ds.is_downloading("tiny"))
|
||
row = ds.status_row("tiny", downloaded=False)
|
||
self.assertTrue(row["downloading"])
|
||
self.assertFalse(row["failed"])
|
||
|
||
def test_failed_surfaces_error(self):
|
||
ds.mark_failed("tiny", "401 Repository Not Found")
|
||
row = ds.status_row("tiny", downloaded=False)
|
||
self.assertTrue(row["failed"])
|
||
self.assertFalse(row["downloading"])
|
||
self.assertEqual(row["error"], "401 Repository Not Found")
|
||
self.assertEqual(ds.get_error("tiny"), "401 Repository Not Found")
|
||
|
||
def test_failed_without_message_has_no_error_field(self):
|
||
ds.mark_failed("tiny")
|
||
row = ds.status_row("tiny", downloaded=False)
|
||
self.assertTrue(row["failed"])
|
||
self.assertNotIn("error", row)
|
||
|
||
def test_downloaded_overrides_failed(self):
|
||
# 先失败后又下好：downloaded=True 时不应再回传 failed/error
|
||
ds.mark_failed("tiny", "boom")
|
||
row = ds.status_row("tiny", downloaded=True)
|
||
self.assertFalse(row["failed"])
|
||
self.assertTrue(row["downloaded"])
|
||
self.assertNotIn("error", row)
|
||
|
||
def test_mark_done_clears_error(self):
|
||
ds.mark_failed("tiny", "boom")
|
||
ds.mark_done("tiny")
|
||
self.assertIsNone(ds.get_error("tiny"))
|
||
row = ds.status_row("tiny", downloaded=True)
|
||
self.assertFalse(row["failed"])
|
||
|
||
def test_redownload_clears_previous_error(self):
|
||
ds.mark_failed("tiny", "boom")
|
||
ds.mark_downloading("tiny") # 重新开始下载
|
||
self.assertIsNone(ds.get_error("tiny"))
|
||
row = ds.status_row("tiny", downloaded=False)
|
||
self.assertTrue(row["downloading"])
|
||
self.assertFalse(row["failed"])
|
||
self.assertNotIn("error", row)
|
||
|
||
def test_mlx_key_is_independent(self):
|
||
# mlx 用 "mlx-{size}" 前缀，与 fast-whisper 的同名档位互不影响
|
||
ds.mark_failed("mlx-tiny", "mlx boom")
|
||
ds.mark_downloading("tiny")
|
||
whisper_row = ds.status_row("tiny", downloaded=False)
|
||
mlx_row = ds.status_row("tiny", downloaded=False, key="mlx-tiny")
|
||
self.assertTrue(whisper_row["downloading"])
|
||
self.assertFalse(whisper_row["failed"])
|
||
self.assertTrue(mlx_row["failed"])
|
||
self.assertEqual(mlx_row["error"], "mlx boom")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|