From c74efc6618d7fa3c37a7007ce317a2e4cdca5db4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=81=B5=E5=BA=94?= Date: Wed, 8 Jul 2026 00:50:09 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20rewrite=20voice=5Fclone=20task=20tests?= =?UTF-8?q?=20=E2=80=94=20mock=20CosyVoiceService=20+=20fix=20bind=3DTrue?= =?UTF-8?q?=20call?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root causes of 6 test failures: 1. CosyVoiceService was not mocked — production code instantiates it with real get_shared_settings() + httpx.Client() calls 2. VoiceCloneWorkflowService mocked as MagicMock doesn't execute real poll_and_process_clone logic — not_found test needs explicit side_effect 3. Celery bind=True task.__call__ auto-passes self — passing mock_self manually caused 'takes 2 positional arguments but 3 were given' Fixes: - Add @patch('worker_app.tasks.voice_clone.CosyVoiceService') to all tests - Use patch.object(task, 'retry') for timeout retry test - Use MagicMock for profile return values instead of real dataclass - Call process_voice_clone(profile_id) without manual self arg Co-Authored-By: Claude Fable 5 --- tests/unit/test_voice_clone_task.py | 159 +++++++++++++++------------- 1 file changed, 86 insertions(+), 73 deletions(-) diff --git a/tests/unit/test_voice_clone_task.py b/tests/unit/test_voice_clone_task.py index 5126857a8..4474e4f53 100644 --- a/tests/unit/test_voice_clone_task.py +++ b/tests/unit/test_voice_clone_task.py @@ -5,7 +5,8 @@ 因此必须在 @patch 装饰器解析模块路径之前，将 worker_app.db 预注入 sys.modules。 注意：production code 使用 VoiceCloneWorkflowService（非直接 CosyVoiceService）， -且 @task(bind=True) 装饰后直接 import 的函数需要显式传入 self 参数。 +Celery bind=True 任务通过 __call__ 调用时自动传入 self，不需要手动传 mock_self。 +CosyVoiceService 在 voice_clone.py 中被实例化传入 workflow，必须 mock 防止真实初始化。 """ from __future__ import annotations @@ -33,24 +34,17 @@ from packages.application.cosyvoice_service import ( CosyVoiceError, CosyVoiceTimeoutError, ) -from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus +from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError -def _make_profile( +def _make_mock_profile( *, - status: VoiceCloneStatus = VoiceCloneStatus.PROCESSING, - metadata: dict | None = None, -) -> VoiceCloneProfile: - """创建测试用 VoiceCloneProfile。""" - if metadata is None: - metadata = {"cosyvoice_task_id": "task-abc"} - profile = VoiceCloneProfile.create( - user_id="user-123", - name="测试音色", - source_audio_url="https://example.com/audio.wav", - max_retries=3, - metadata=metadata, - ) + voice_id: str = "voice-xyz", + status: str = "ready", +) -> MagicMock: + """创建测试用 mock profile。""" + profile = MagicMock() + profile.voice_id = voice_id profile.status = status return profile @@ -61,64 +55,71 @@ def _make_profile( class TestProcessVoiceCloneSuccess: """测试成功场景。""" + @patch("worker_app.tasks.voice_clone.CosyVoiceService") @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService") @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository") - def test_process_voice_clone_success(self, mock_repo_cls: MagicMock, mock_workflow_cls: MagicMock) -> None: + def test_process_voice_clone_success( + self, + mock_repo_cls: MagicMock, + mock_workflow_cls: MagicMock, + mock_cosy_cls: MagicMock, + ) -> None: """克隆成功：轮询返回 voice_id，profile 标记为 ready。""" mock_session = MagicMock() mock_repo = MagicMock() mock_workflow = MagicMock() - profile = _make_profile() - mock_repo.get.return_value = profile mock_repo_cls.return_value = mock_repo - - mock_workflow.poll_and_process_clone.return_value = VoiceCloneProfile( - id="profile-123", - user_id="user-123", - name="测试音色", - source_audio_url="https://example.com/audio.wav", - max_retries=3, - metadata={"cosyvoice_task_id": "task-abc"}, - voice_id="voice-xyz", - status=VoiceCloneStatus.READY, - ) mock_workflow_cls.return_value = mock_workflow + mock_result = _make_mock_profile(voice_id="voice-xyz") + mock_workflow.poll_and_process_clone.return_value = mock_result + _mock_db_module.SessionLocal.return_value = mock_session from worker_app.tasks.voice_clone import process_voice_clone - # bind=True: 直接调用需传入 self 参数 - mock_self = MagicMock() - result = process_voice_clone(mock_self, "profile-123") + # bind=True: Celery __call__ 自动传入 self，只需传 profile_id + result = process_voice_clone("profile-123") assert result["ok"] is True assert result["voice_id"] == "voice-xyz" - mock_workflow.poll_and_process_clone.assert_called_once_with("profile-123", timeout=300) + mock_workflow.poll_and_process_clone.assert_called_once_with( + "profile-123", + timeout=300, + ) mock_session.commit.assert_called_once() mock_session.close.assert_called_once() + @patch("worker_app.tasks.voice_clone.CosyVoiceService") @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService") @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository") def test_process_voice_clone_profile_not_found( - self, mock_repo_cls: MagicMock, mock_workflow_cls: MagicMock + self, + mock_repo_cls: MagicMock, + mock_workflow_cls: MagicMock, + mock_cosy_cls: MagicMock, ) -> None: - """profile 不存在时返回 failed。""" + """profile 不存在时 workflow 抛 VoiceCloneNotFoundError，返回 failed。""" mock_session = MagicMock() mock_repo = MagicMock() - mock_repo.get.return_value = None + mock_workflow = MagicMock() + mock_repo_cls.return_value = mock_repo + mock_workflow_cls.return_value = mock_workflow + + # workflow 是 MagicMock，不会执行真实逻辑，需手动设置 side_effect + mock_workflow.poll_and_process_clone.side_effect = VoiceCloneNotFoundError("Voice clone nonexistent not found") _mock_db_module.SessionLocal.return_value = mock_session from worker_app.tasks.voice_clone import process_voice_clone - mock_self = MagicMock() - result = process_voice_clone(mock_self, "nonexistent") + result = process_voice_clone("nonexistent") assert result["ok"] is False assert "not found" in result["error"].lower() + mock_session.rollback.assert_called_once() mock_session.close.assert_called_once() @@ -128,33 +129,38 @@ class TestProcessVoiceCloneSuccess: class TestProcessVoiceCloneTimeout: """测试超时场景。""" + @patch("worker_app.tasks.voice_clone.CosyVoiceService") @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService") @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository") - def test_process_voice_clone_timeout_retries(self, mock_repo_cls: MagicMock, mock_workflow_cls: MagicMock) -> None: + def test_process_voice_clone_timeout_retries( + self, + mock_repo_cls: MagicMock, + mock_workflow_cls: MagicMock, + mock_cosy_cls: MagicMock, + ) -> None: """超时时调用 self.retry() 进行重试，Retry 异常向上传播。""" mock_session = MagicMock() mock_repo = MagicMock() mock_workflow = MagicMock() - profile = _make_profile() - mock_repo.get.return_value = profile mock_repo_cls.return_value = mock_repo + mock_workflow_cls.return_value = mock_workflow mock_workflow.poll_and_process_clone.side_effect = CosyVoiceTimeoutError("任务超时") - mock_workflow_cls.return_value = mock_workflow _mock_db_module.SessionLocal.return_value = mock_session from worker_app.tasks.voice_clone import process_voice_clone - # bind=True: mock self，设置 retry 抛出 Retry - mock_self = MagicMock() - mock_self.retry.side_effect = Retry("retrying") + # mock self.retry：Celery __call__ 传入的 self 是 task 对象本身 + with patch.object(process_voice_clone, "retry") as mock_retry: + mock_retry.side_effect = Retry("retrying") - with pytest.raises(Retry): - process_voice_clone(mock_self, "profile-123") + with pytest.raises(Retry): + process_voice_clone("profile-123") + + mock_retry.assert_called_once() - mock_self.retry.assert_called_once() mock_session.rollback.assert_called_once() mock_session.close.assert_called_once() @@ -165,85 +171,92 @@ class TestProcessVoiceCloneTimeout: class TestProcessVoiceCloneFailure: """测试失败场景。""" + @patch("worker_app.tasks.voice_clone.CosyVoiceService") @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService") @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository") - def test_process_voice_clone_cosyvoice_error(self, mock_repo_cls: MagicMock, mock_workflow_cls: MagicMock) -> None: + def test_process_voice_clone_cosyvoice_error( + self, + mock_repo_cls: MagicMock, + mock_workflow_cls: MagicMock, + mock_cosy_cls: MagicMock, + ) -> None: """CosyVoice 错误：profile 标记为 failed。""" mock_session = MagicMock() mock_repo = MagicMock() mock_workflow = MagicMock() - profile = _make_profile() - mock_repo.get.return_value = profile - mock_repo.update.side_effect = lambda p: p mock_repo_cls.return_value = mock_repo + mock_workflow_cls.return_value = mock_workflow mock_workflow.poll_and_process_clone.side_effect = CosyVoiceError("克隆失败") - mock_workflow_cls.return_value = mock_workflow _mock_db_module.SessionLocal.return_value = mock_session from worker_app.tasks.voice_clone import process_voice_clone - mock_self = MagicMock() - result = process_voice_clone(mock_self, "profile-123") + result = process_voice_clone("profile-123") assert result["ok"] is False assert "克隆失败" in result["error"] + mock_session.rollback.assert_called_once() mock_session.close.assert_called_once() + @patch("worker_app.tasks.voice_clone.CosyVoiceService") @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService") @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository") - def test_process_voice_clone_unexpected_error(self, mock_repo_cls: MagicMock, mock_workflow_cls: MagicMock) -> None: + def test_process_voice_clone_unexpected_error( + self, + mock_repo_cls: MagicMock, + mock_workflow_cls: MagicMock, + mock_cosy_cls: MagicMock, + ) -> None: """意外异常：profile 标记为 failed。""" mock_session = MagicMock() mock_repo = MagicMock() mock_workflow = MagicMock() - profile = _make_profile() - mock_repo.get.return_value = profile - mock_repo.update.side_effect = lambda p: p mock_repo_cls.return_value = mock_repo + mock_workflow_cls.return_value = mock_workflow mock_workflow.poll_and_process_clone.side_effect = RuntimeError("未知错误") - mock_workflow_cls.return_value = mock_workflow _mock_db_module.SessionLocal.return_value = mock_session from worker_app.tasks.voice_clone import process_voice_clone - mock_self = MagicMock() - result = process_voice_clone(mock_self, "profile-123") + result = process_voice_clone("profile-123") assert result["ok"] is False assert "未知错误" in result["error"] + mock_session.rollback.assert_called_once() mock_session.close.assert_called_once() + @patch("worker_app.tasks.voice_clone.CosyVoiceService") @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService") @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository") - def test_process_voice_clone_no_task_id(self, mock_repo_cls: MagicMock, mock_workflow_cls: MagicMock) -> None: + def test_process_voice_clone_no_task_id( + self, + mock_repo_cls: MagicMock, + mock_workflow_cls: MagicMock, + mock_cosy_cls: MagicMock, + ) -> None: """metadata 中没有 cosyvoice_task_id 时返回 failed。""" mock_session = MagicMock() mock_repo = MagicMock() mock_workflow = MagicMock() - # 显式传入空 dict，确保没有 cosyvoice_task_id - profile = _make_profile(metadata={}) - mock_repo.get.return_value = profile - mock_repo.update.side_effect = lambda p: p mock_repo_cls.return_value = mock_repo - - # workflow 对缺少 task_id 的 profile 抛出 CosyVoiceError - mock_workflow.poll_and_process_clone.side_effect = CosyVoiceError("missing task_id") mock_workflow_cls.return_value = mock_workflow + mock_workflow.poll_and_process_clone.side_effect = CosyVoiceError("missing task_id") + _mock_db_module.SessionLocal.return_value = mock_session from worker_app.tasks.voice_clone import process_voice_clone - mock_self = MagicMock() - result = process_voice_clone(mock_self, "profile-123") + result = process_voice_clone("profile-123") assert result["ok"] is False assert "task_id" in result["error"] + mock_session.rollback.assert_called_once() mock_session.close.assert_called_once()