fix: 修复 CI 失败的 4 个单元测试
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 49s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 21s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 32s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m34s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m48s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m51s
AI Code Review / AI Code Review (pull_request) Successful in 2m9s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m15s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 555h11m10s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 555h11m58s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 555h11m12s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 555h11m59s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 555h12m3s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 555h12m5s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 555h12m5s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 555h45m3s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 555h45m56s

- 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 中删除。
This commit is contained in:
xiaoxia
2026-08-24 12:13:44 +08:00
parent 706b9057d6
commit 2bd9102eeb
2 changed files with 47 additions and 33 deletions
+1 -26
View File
@@ -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)
# ---------------------------------------------------------------------------
# 字幕端点测试
# ---------------------------------------------------------------------------
@@ -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():