fix(transcriber): 下载失败时透传错误到前端并提示

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>
This commit is contained in:
huangjianwu
2026-06-23 10:19:38 +08:00
parent 3841719d5a
commit 4a87c5b93b
5 changed files with 281 additions and 54 deletions

View File

@@ -0,0 +1,113 @@
"""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()