From ce1bd8e7c705e0285e83b88ef46e7bea81fe80d3 Mon Sep 17 00:00:00 2001 From: saas-backend Date: Sat, 26 Sep 2026 14:30:39 +0800 Subject: [PATCH] =?UTF-8?q?test(voice-clone):=20=E4=BF=AE=E5=A4=8D=20=5Fre?= =?UTF-8?q?solve=5Ftask=20=E8=B7=A8=20Celery=20=E7=89=88=E6=9C=AC=E9=B2=81?= =?UTF-8?q?=E6=A3=92=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _get_current_object() 加 try/except 兜底，无 active app context 时不抛 AttributeError - _resolve_task 统一返回 (callable, mock_self, real_task) 三元组 - timeout 重试用例复用 real_task patch retry，不再二次调用无保护的 _get_current_object() - mock_self 判断改为 is not None（语义更清晰，兼容 MagicMock 真值） --- tests/unit/test_voice_clone_task.py | 94 ++++++++++++++++------------- 1 file changed, 53 insertions(+), 41 deletions(-) diff --git a/tests/unit/test_voice_clone_task.py b/tests/unit/test_voice_clone_task.py index e2f850594..ecbcc8d0b 100644 --- a/tests/unit/test_voice_clone_task.py +++ b/tests/unit/test_voice_clone_task.py @@ -8,13 +8,14 @@ Celery bind=True 任务的底层函数签名为 (self, profile_id)， CosyVoiceService 在 voice_clone.py 中被实例化传入 workflow，必须 mock 防止真实初始化。 -跨环境兼容： - Python 3.13 + Celery 5.4.0 → import 返回 Celery Proxy - → _get_current_object() 返回 Task 实例 → .run 是 bound method（self 已绑定） - → 调用方式：task.run(profile_id)，retry mock 在 task.run.retry - Python 3.10 + Celery 5.4.0 → import 返回原始函数（装饰器未生效） - → 签名 (self, profile_id)，需手动传 mock_self - → 调用方式：func(mock_self, profile_id)，retry mock 在 mock_self.retry +跨环境兼容（_resolve_task）： + 不同 Celery 版本 / Python 版本 / 是否有 active Celery app，task 对象形态不同： + 1) Celery Proxy（LocalProxy/LazyProxy）：import 结果是代理对象，调用 + _get_current_object() 可能抛 RuntimeError（无 active context），必须 try 保护。 + 成功取到真实 Task 实例后，使用 bound method .run。 + 2) Celery Task 实例（bind=True 时 @task 返回的典型形态）：直接有 .run/.retry。 + 3) 原始函数（某些环境装饰器未生效或 patch 时序问题）：需手动传 mock_self。 + 统一返回 (callable, mock_self, real_task)，调用方不需要重复解析。 """ from __future__ import annotations @@ -58,24 +59,33 @@ def _make_mock_profile( def _resolve_task(task_obj): - """解析 Celery 任务对象，返回 (callable, mock_self_or_none)。 + """解析 Celery 任务对象，兼容 Proxy / Task 实例 / 原始函数三种形态。 - 跨环境兼容 Celery Proxy / Task 实例 / 原始函数三种情况。 + 所有分支均做异常保护，避免因 Celery Proxy 在无 app context 时抛错导致测试挂掉。 Returns: - tuple: (callable, mock_self) - - Proxy/Task: callable 是 bound method task.run，mock_self=None - - 原始函数: callable 是原始函数，mock_self 需由调用方提供 + tuple: (callable, mock_self, real_task) + - callable: 最终执行用的可调用对象 + - mock_self: 仅原始函数分支需要手动传入 mock self；其他分支为 None + - real_task: 真实 Task 实例（Proxy 分支为 _get_current_object() 结果； + Task 分支为 task_obj 本身；原始函数分支为 None）。用于 patch .retry。 """ - # Case 1: Celery Proxy → 提取 Task 实例的 .run（bound method） + # Case 1: Celery Proxy → 安全尝试 _get_current_object() if hasattr(task_obj, "_get_current_object"): - real_task = task_obj._get_current_object() - return real_task.run, None + try: + real_task = task_obj._get_current_object() + if real_task is not None and hasattr(real_task, "run"): + return real_task.run, None, real_task + except Exception: + # 无 active app context 或 Proxy 未绑定，退化为其他分支处理 + pass + # Case 2: Celery Task 实例（非 Proxy） if hasattr(task_obj, "run") and hasattr(task_obj, "retry"): - return task_obj.run, None - # Case 3: 原始函数（CI 环境中装饰器未生效） - return task_obj, MagicMock() + return task_obj.run, None, task_obj + + # Case 3: 原始函数（装饰器未生效） + return task_obj, MagicMock(), None # ── 成功场景 ────────────────────────────────────────────── @@ -110,8 +120,8 @@ class TestProcessVoiceCloneSuccess: from worker_app.tasks.voice_clone import process_voice_clone - func, mock_self = _resolve_task(process_voice_clone) - args = (mock_self, "profile-123") if mock_self else ("profile-123",) + func, mock_self, _ = _resolve_task(process_voice_clone) + args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",) result = func(*args) assert result["ok"] is True @@ -142,14 +152,16 @@ class TestProcessVoiceCloneSuccess: mock_repo_cls.return_value = mock_repo mock_workflow_cls.return_value = mock_workflow - mock_workflow.poll_and_process_clone.side_effect = VoiceCloneNotFoundError("Voice clone nonexistent not found") + mock_workflow.poll_and_process_clone.side_effect = VoiceCloneNotFoundError( + "Voice clone nonexistent not found" + ) mock_session_local.return_value = mock_session from worker_app.tasks.voice_clone import process_voice_clone - func, mock_self = _resolve_task(process_voice_clone) - args = (mock_self, "nonexistent") if mock_self else ("nonexistent",) + func, mock_self, _ = _resolve_task(process_voice_clone) + args = (mock_self, "nonexistent") if mock_self is not None else ("nonexistent",) result = func(*args) assert result["ok"] is False @@ -189,24 +201,24 @@ class TestProcessVoiceCloneTimeout: from worker_app.tasks.voice_clone import process_voice_clone - func, mock_self = _resolve_task(process_voice_clone) + func, mock_self, real_task = _resolve_task(process_voice_clone) - # 设置 retry mock：根据环境不同，retry 在不同对象上 - if mock_self is None: - # Proxy/Task 环境：retry 在 Task 实例上（func 是 bound method task.run） - real_task = process_voice_clone._get_current_object() - mock_retry = MagicMock() - mock_retry.side_effect = Retry("retrying") - with patch.object(real_task, "retry", mock_retry): - with pytest.raises(Retry): - func("profile-123") - mock_retry.assert_called_once() - else: + if mock_self is not None: # 原始函数环境：retry 在 mock_self 上 mock_self.retry.side_effect = Retry("retrying") with pytest.raises(Retry): func(mock_self, "profile-123") mock_self.retry.assert_called_once() + else: + # Proxy/Task 环境：retry 在 Task 实例上。用 _resolve_task 返回的 real_task， + # 避免再次 _get_current_object() 在无 context 时抛 AttributeError。 + retry_target = real_task if real_task is not None else process_voice_clone + mock_retry = MagicMock() + mock_retry.side_effect = Retry("retrying") + with patch.object(retry_target, "retry", mock_retry): + with pytest.raises(Retry): + func("profile-123") + mock_retry.assert_called_once() mock_session.rollback.assert_called_once() mock_session.close.assert_called_once() @@ -243,8 +255,8 @@ class TestProcessVoiceCloneFailure: from worker_app.tasks.voice_clone import process_voice_clone - func, mock_self = _resolve_task(process_voice_clone) - args = (mock_self, "profile-123") if mock_self else ("profile-123",) + func, mock_self, _ = _resolve_task(process_voice_clone) + args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",) result = func(*args) assert result["ok"] is False @@ -277,8 +289,8 @@ class TestProcessVoiceCloneFailure: from worker_app.tasks.voice_clone import process_voice_clone - func, mock_self = _resolve_task(process_voice_clone) - args = (mock_self, "profile-123") if mock_self else ("profile-123",) + func, mock_self, _ = _resolve_task(process_voice_clone) + args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",) result = func(*args) assert result["ok"] is False @@ -311,8 +323,8 @@ class TestProcessVoiceCloneFailure: from worker_app.tasks.voice_clone import process_voice_clone - func, mock_self = _resolve_task(process_voice_clone) - args = (mock_self, "profile-123") if mock_self else ("profile-123",) + func, mock_self, _ = _resolve_task(process_voice_clone) + args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",) result = func(*args) assert result["ok"] is False