From 2bd9102eebed4b047cd9b0d19ad1e24597b74c78 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 24 Aug 2026 12:13:44 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20CI=20=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E7=9A=84=204=20=E4=B8=AA=E5=8D=95=E5=85=83=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_worker_generate_video_task_binding.py: 不依赖 Celery Task 对象的 .name/.run 属性（CI conftest mock 导致 PromiseProxy 不可用）。 改用 _get_original_function() 遍历 __wrapped__.__func__ 链获取原始函数， 通过 inspect.signature 验证 bind=True(self 为首参)、task_id 参数存在、 函数名为 generate_video。 - test_templates_editor_api.py: 删除 TestGenerationRoutes 类（2 个测试）， 对应路由 GET /generation-status 和 GET /generations 已在 PR #1472 中删除。 --- tests/unit/test_templates_editor_api.py | 27 +--------- ...test_worker_generate_video_task_binding.py | 53 ++++++++++++++++--- 2 files changed, 47 insertions(+), 33 deletions(-) diff --git a/tests/unit/test_templates_editor_api.py b/tests/unit/test_templates_editor_api.py index 06044e2a6..783b2a45d 100755 --- a/tests/unit/test_templates_editor_api.py +++ b/tests/unit/test_templates_editor_api.py @@ -1,12 +1,11 @@ """ templates_editor.py 模板编辑器 API 端点单元测试 -覆盖核心端点（25个测试用例）： +覆盖核心端点（23个测试用例）： - 草稿：GET/PUT/发布 - 片段：list/create/get/update/delete/split/merge - BGM：GET/PUT - 时间线：GET -- 生成状态查询 - 预设：BGM预设 """ @@ -395,30 +394,6 @@ class TestTimelineRoute: mock_plan_svc.list_clips.assert_called() -# --------------------------------------------------------------------------- -# 生成端点测试 -# --------------------------------------------------------------------------- - - -class TestGenerationRoutes: - """生成端点测试""" - - def test_generation_status_success(self, client): - c, _, mock_plan_svc = client - resp = c.get(BASE + "/generation-status") - assert resp.status_code == 200 - data = resp.json() - assert "generation_task_id" in data - assert "clips" in data - - def test_generations_list_success(self, client): - c, _, mock_plan_svc = client - resp = c.get(BASE + "/generations") - assert resp.status_code == 200 - data = resp.json() - assert "items" in data or "tasks" in data or isinstance(data, dict) - - # --------------------------------------------------------------------------- # 字幕端点测试 # --------------------------------------------------------------------------- diff --git a/tests/unit/test_worker_generate_video_task_binding.py b/tests/unit/test_worker_generate_video_task_binding.py index 0a51dff5f..adf13957f 100644 --- a/tests/unit/test_worker_generate_video_task_binding.py +++ b/tests/unit/test_worker_generate_video_task_binding.py @@ -6,28 +6,67 @@ _sync_task_config_to_plan directly under the @celery_app.task decorator, so Celery registered the helper as "worker.generate_video". Calling the task with a single task_id raised TypeError and every generation job failed immediately. This test pins the decorator target. + +NOTE: CI conftest may mock Celery so that @celery_app.task does NOT return +a fully functional Task/PromiseProxy object. Tests therefore use multiple +defensive strategies: source-code inspection, __wrapped__.__func__ chain +traversal, and direct attribute checks. """ from __future__ import annotations import inspect +import re -def test_generate_video_task_registered_under_expected_name(): +def _get_original_function(generate_video): + """Walk the __wrapped__ chain to find the original function object.""" + obj = generate_video + seen = set() + while hasattr(obj, "__wrapped__"): + obj_id = id(obj) + if obj_id in seen: + break + seen.add(obj_id) + obj = obj.__wrapped__ + # __wrapped__ may be a bound method — unwrap to the underlying function + if hasattr(obj, "__func__"): + return obj.__func__ + return obj + + +def test_generate_video_task_has_bind_true(): + """The decorator must use bind=True — verified via the original function's + first parameter being 'self' (bind=True convention).""" from worker_app.tasks.generation import generate_video - # Celery task object exposes its registered name - assert generate_video.name == "worker.generate_video" + original = _get_original_function(generate_video) + sig = inspect.signature(original) + params = list(sig.parameters) + assert params[0] == "self", ( + f"bind=True requires 'self' as first param, got {params}" + ) def test_generate_video_task_signature_has_task_id(): + """The original generate_video function must accept task_id as a parameter.""" from worker_app.tasks.generation import generate_video - # For bind=True tasks Celery binds self at call time, so run() signature - # starts directly with task_id (verified on Celery 5.x). - sig = inspect.signature(generate_video.run) + original = _get_original_function(generate_video) + sig = inspect.signature(original) params = list(sig.parameters) - assert params[0] == "task_id", f"expected task_id as first param, got {params}" + assert "task_id" in params, f"expected 'task_id' in params, got {params}" + + +def test_generate_video_preserves_original_function(): + """The original function wrapped by @celery_app.task must be named + 'generate_video' — not '_sync_task_config_to_plan'.""" + from worker_app.tasks.generation import generate_video + + original = _get_original_function(generate_video) + assert original.__name__ == "generate_video", ( + f"expected __name__='generate_video', got '{original.__name__}'" + ) def test_sync_task_config_to_plan_is_plain_function(): -- 2.54.0 From 947bd4dd94e143c84734ce83807b2bb1d9ce2193 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 24 Aug 2026 04:17:34 +0000 Subject: [PATCH 2/2] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_worker_generate_video_task_binding.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/tests/unit/test_worker_generate_video_task_binding.py b/tests/unit/test_worker_generate_video_task_binding.py index adf13957f..17414a0af 100644 --- a/tests/unit/test_worker_generate_video_task_binding.py +++ b/tests/unit/test_worker_generate_video_task_binding.py @@ -43,9 +43,7 @@ def test_generate_video_task_has_bind_true(): original = _get_original_function(generate_video) sig = inspect.signature(original) params = list(sig.parameters) - assert params[0] == "self", ( - f"bind=True requires 'self' as first param, got {params}" - ) + assert params[0] == "self", f"bind=True requires 'self' as first param, got {params}" def test_generate_video_task_signature_has_task_id(): @@ -64,9 +62,7 @@ def test_generate_video_preserves_original_function(): from worker_app.tasks.generation import generate_video original = _get_original_function(generate_video) - assert original.__name__ == "generate_video", ( - f"expected __name__='generate_video', got '{original.__name__}'" - ) + assert original.__name__ == "generate_video", f"expected __name__='generate_video', got '{original.__name__}'" def test_sync_task_config_to_plan_is_plain_function(): -- 2.54.0