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..17414a0af 100644 --- a/tests/unit/test_worker_generate_video_task_binding.py +++ b/tests/unit/test_worker_generate_video_task_binding.py @@ -6,28 +6,63 @@ _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():