From 08a52c7bba279a5dfadfa0af19323133704f4a61 Mon Sep 17 00:00:00 2001 From: saas-backend Date: Sat, 26 Sep 2026 15:13:06 +0800 Subject: [PATCH] =?UTF-8?q?test(voice-clone):=20=E4=B8=BA=20cleanup=5Fstal?= =?UTF-8?q?e=5Fprocessing=20=E8=A1=A5=E5=8D=95=E6=B5=8B=EF=BC=8C=E6=BB=A1?= =?UTF-8?q?=E8=B6=B3=E5=A2=9E=E9=87=8F=E8=A6=86=E7=9B=96=E7=8E=87=E9=97=A8?= =?UTF-8?q?=E6=A7=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 5 个用例覆盖： - 无卡死记录不 commit - 单条超时记录标 failed + 错误信息含分钟数 - 自定义 timeout 反映在错误信息中 - 多条记录批量清理且只 commit 一次 - query 链路包含 status/updated_at 双过滤条件 使用 sys.modules 预注入假模型类，避免 SQLAlchemy/DB 依赖。 --- tests/unit/test_voice_clone_cleanup.py | 153 +++++++++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100755 tests/unit/test_voice_clone_cleanup.py diff --git a/tests/unit/test_voice_clone_cleanup.py b/tests/unit/test_voice_clone_cleanup.py new file mode 100755 index 000000000..6e608f90d --- /dev/null +++ b/tests/unit/test_voice_clone_cleanup.py @@ -0,0 +1,153 @@ +"""SQLAlchemyVoiceCloneProfileRepository.cleanup_stale_processing 单元测试。 + +通过 monkeypatch sys.modules['packages.adapters.sqlalchemy_impl.models']， +注入一个具备 SQLAlchemy 列比较语义（== / < 返回可链式 .all() 的 mock）的假模型类， +不依赖真实 DB，也不会触发 SQLAlchemy 映射。 +""" + +from __future__ import annotations + +import sys +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + + +class _Col: + """模拟 SQLAlchemy Column：比较运算返回 MagicMock，可被 filter 链式调用。""" + + def __init__(self, name: str): + self._name = name + + def __eq__(self, other): # type: ignore[override] + return MagicMock(name=f"{self._name}=={other!r}") + + def __ne__(self, other): # type: ignore[override] + return MagicMock(name=f"{self._name}!={other!r}") + + def __lt__(self, other): + return MagicMock(name=f"{self._name}<{other!r}") + + def __gt__(self, other): + return MagicMock(name=f"{self._name}>{other!r}") + + def __le__(self, other): + return MagicMock(name=f"{self._name}<={other!r}") + + def __ge__(self, other): + return MagicMock(name=f"{self._name}>={other!r}") + + def __hash__(self): + return id(self) + + +class _FakeVoiceCloneProfileModel: + """假模型：类属性是 _Col；实例上可读写 status/error_message/updated_at。""" + + status = _Col("status") + updated_at = _Col("updated_at") + id = _Col("id") + error_message = _Col("error_message") + + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + +# ── 预注入 mock 模型模块，避免真实 import 拉起 DB / SQLAlchemy 映射 ── +_fake_models = SimpleNamespace(VoiceCloneProfileModel=_FakeVoiceCloneProfileModel) +sys.modules.setdefault("packages.adapters.sqlalchemy_impl.models", _fake_models) +if "packages.adapters.sqlalchemy_impl.voice_clone_profile_repository" in sys.modules: + mod = sys.modules["packages.adapters.sqlalchemy_impl.voice_clone_profile_repository"] + mod.VoiceCloneProfileModel = _FakeVoiceCloneProfileModel # type: ignore[attr-defined] + +from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import ( + SQLAlchemyVoiceCloneProfileRepository, +) + + +def _make_fake_row( + *, + status: str = "processing", + updated_at: datetime | None = None, + error_message: str = "", +) -> _FakeVoiceCloneProfileModel: + return _FakeVoiceCloneProfileModel( + status=status, + error_message=error_message, + updated_at=updated_at or datetime.now(UTC), + ) + + +def _make_repo(fake_rows: list[_FakeVoiceCloneProfileModel]): + """构造 repo + mock session。 + + 生产代码使用 .query(Model).filter(A, B).all()（一次 filter，两个表达式参数）。 + """ + session = MagicMock() + filtered = MagicMock() + filtered.all.return_value = list(fake_rows) + session.query.return_value.filter.return_value = filtered + + repo = SQLAlchemyVoiceCloneProfileRepository.__new__(SQLAlchemyVoiceCloneProfileRepository) + repo.session = session + return repo, session + + +class TestCleanupStaleProcessing: + """cleanup_stale_processing 行为测试。""" + + def test_no_stale_records_returns_zero_and_no_commit(self): + """无卡死记录时返回 0，不调用 commit。""" + repo, session = _make_repo([]) + assert repo.cleanup_stale_processing() == 0 + session.commit.assert_not_called() + + def test_stale_record_marked_failed_with_timeout_message(self): + """超时 processing 记录被标记为 failed，错误信息包含超时分钟数。""" + old = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=15)) + repo, session = _make_repo([old]) + + count = repo.cleanup_stale_processing(timeout_minutes=10) + + assert count == 1 + assert old.status == "failed" + assert "超时" in old.error_message + assert "10" in old.error_message + session.commit.assert_called_once() + + def test_error_message_reflects_custom_timeout(self): + """自定义 timeout_minutes 会反映在错误信息里。""" + old = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(hours=1)) + repo, _session = _make_repo([old]) + + repo.cleanup_stale_processing(timeout_minutes=5) + + assert old.status == "failed" + assert "5" in old.error_message + + def test_multiple_stale_records_all_cleaned_in_single_commit(self): + """多条卡死记录都被清理，返回正确计数并只 commit 一次。""" + m1 = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=20)) + m2 = _make_fake_row(updated_at=datetime.now(UTC) - timedelta(minutes=11)) + repo, session = _make_repo([m1, m2]) + + assert repo.cleanup_stale_processing(timeout_minutes=10) == 2 + assert m1.status == "failed" + assert m2.status == "failed" + session.commit.assert_called_once() + + def test_queries_model_with_status_and_updated_at_filters(self): + """query 以模型类为参数，filter 同时传入 status=='processing' 与 updated_at