xiaoxia-saas/tests/unit/test_worker_generate_video_task_binding.py

"""Regression test: ensure worker.generate_video Celery task is bound to the
real generate_video function, not a helper introduced above it.

Context (P0 incident 2026-08-23): a refactor inserted helper function
_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 _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

    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

    original = _get_original_function(generate_video)
    sig = inspect.signature(original)
    params = list(sig.parameters)
    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():
    """Helper must NOT be registered as a Celery task."""
    from worker_app.tasks.generation import _sync_task_config_to_plan

    assert not hasattr(
        _sync_task_config_to_plan, "run"
    ), "_sync_task_config_to_plan must be a plain function, not a Celery task"