xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 14 Pull Requests 1 Actions 3 Packages Projects Releases 7 Wiki Activity

style: black format test_voice_clone_task.py
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been skipped
Details
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
Details
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
Details
CI/CD Pipeline / Deploy Production (push) Has been skipped
Details
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
Details
CI/CD Pipeline / Frontend Lint (push) Failing after 36h19m6s
Details
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 36h19m6s
Details

Browse Source
This commit is contained in:
灵应
2026-07-08 00:35:21 +08:00
parent ea27360c5f
commit 7ef4b0677a
1 changed files with 8 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_voice_clone_task.py
+8 -24
View File
@@ -63,9 +63,7 @@ class TestProcessVoiceCloneSuccess:
@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) -> None:
"""克隆成功：轮询返回 voice_id，profile 标记为 ready。"""
mock_session = MagicMock()
mock_repo = MagicMock()
@@ -97,9 +95,7 @@ class TestProcessVoiceCloneSuccess:
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()
@@ -134,9 +130,7 @@ class TestProcessVoiceCloneTimeout:
@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) -> None:
"""超时时调用 self.retry() 进行重试，Retry 异常向上传播。"""
mock_session = MagicMock()
mock_repo = MagicMock()
@@ -146,9 +140,7 @@ class TestProcessVoiceCloneTimeout:
mock_repo.get.return_value = profile
mock_repo_cls.return_value = mock_repo
mock_workflow.poll_and_process_clone.side_effect = CosyVoiceTimeoutError(
"任务超时"
)
mock_workflow.poll_and_process_clone.side_effect = CosyVoiceTimeoutError("任务超时")
mock_workflow_cls.return_value = mock_workflow
_mock_db_module.SessionLocal.return_value = mock_session
@@ -175,9 +167,7 @@ class TestProcessVoiceCloneFailure:
@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) -> None:
"""CosyVoice 错误：profile 标记为 failed。"""
mock_session = MagicMock()
mock_repo = MagicMock()
@@ -204,9 +194,7 @@ class TestProcessVoiceCloneFailure:
@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) -> None:
"""意外异常：profile 标记为 failed。"""
mock_session = MagicMock()
mock_repo = MagicMock()
@@ -233,9 +221,7 @@ class TestProcessVoiceCloneFailure:
@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) -> None:
"""metadata 中没有 cosyvoice_task_id 时返回 failed。"""
mock_session = MagicMock()
mock_repo = MagicMock()
@@ -248,9 +234,7 @@ class TestProcessVoiceCloneFailure:
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.poll_and_process_clone.side_effect = CosyVoiceError("missing task_id")
mock_workflow_cls.return_value = mock_workflow
_mock_db_module.SessionLocal.return_value = mock_session