xiaoxia-saas/tests/unit/test_cover_url_finalize.py

"""Tests for cover_url backfill to GenerationTask.

Verifies _finalize_render_success correctly writes cover_url
from cover_candidates to gen_task.cover_url.
"""

from __future__ import annotations

import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

# Add worker app to sys.path
_WORKER_ROOT = Path(__file__).resolve().parents[2] / "apps" / "worker"
if str(_WORKER_ROOT) not in sys.path:
    sys.path.insert(0, str(_WORKER_ROOT))


class FakeGenTask:
    """Simple stand-in for GenerationTask that tracks attribute assignment."""

    def __init__(self):
        object.__setattr__(self, "_assigned", {})
        self.id = "task-1"
        self.status = MagicMock()
        self.status.value = "running"

    def __setattr__(self, name, value):
        if not name.startswith("_"):
            self._assigned[name] = value
        object.__setattr__(self, name, value)

    def append_log(self, **kwargs):
        pass


def _make_plan():
    plan = MagicMock()
    plan.project_id = "proj-1"
    plan.created_by_user_id = "user-1"
    plan.config = {"batch_id": "batch-1", "mode": "edit_plan", "title": {"text": "test"}}
    plan.mark_completed = MagicMock()
    return plan


def _call_finalize(cover_candidates=None, gen_task=None, plan=None):
    from worker_app.tasks.edit_plan_generation import _finalize_render_success

    plan = plan or _make_plan()
    gen_task = gen_task or FakeGenTask()

    plan_repo = MagicMock()
    clip_repo = MagicMock()
    gen_task_repo = MagicMock()
    gen_task_repo.get.return_value = gen_task
    db = MagicMock()

    with patch("worker_app.tasks.edit_plan_generation.create_video_record_and_dedup"):
        result = _finalize_render_success(
            plan=plan,
            plan_repo=plan_repo,
            clip_repo=clip_repo,
            gen_task_repo=gen_task_repo,
            db=db,
            plan_id="plan-1",
            output_url="https://oss.example.com/output.mp4",
            storage_key="rendered/plan-1/task-1.mp4",
            duration=10.0,
            file_size=1024,
            width=1280,
            height=720,
            rendered_clip_ids=["clip-1"],
            failed_clip_ids=[],
            generation_task_id="task-1",
            output_path=Path("/tmp/output.mp4"),
            engine="unified",
            thumbnail_url="",
            cover_candidates=cover_candidates,
        )

    return result, gen_task, gen_task_repo


class TestFinalizeCoverUrl:

    def test_cover_url_set_from_image_url(self):
        """cover_candidates with image_url should set gen_task.cover_url"""
        candidates = [
            {"image_url": "https://oss.example.com/cover1.jpg", "frame_time": 1.5},
            {"image_url": "https://oss.example.com/cover2.jpg", "frame_time": 3.0},
        ]
        _, gen_task, gen_task_repo = _call_finalize(cover_candidates=candidates)
        assert gen_task.cover_url == "https://oss.example.com/cover1.jpg"
        gen_task_repo.update.assert_called()

    def test_cover_url_fallback_to_url_key(self):
        """Should fallback to 'url' key when 'image_url' is absent"""
        candidates = [{"url": "https://oss.example.com/cover_url_key.jpg"}]
        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
        assert gen_task.cover_url == "https://oss.example.com/cover_url_key.jpg"

    def test_cover_url_not_set_when_empty_list(self):
        """Empty cover_candidates should not set cover_url"""
        _, gen_task, _ = _call_finalize(cover_candidates=[])
        assert "cover_url" not in gen_task._assigned

    def test_cover_url_not_set_when_none(self):
        """None cover_candidates should not set cover_url"""
        _, gen_task, _ = _call_finalize(cover_candidates=None)
        assert "cover_url" not in gen_task._assigned

    def test_cover_url_not_set_when_url_empty(self):
        """Empty URL strings in candidates should not set cover_url"""
        candidates = [{"image_url": "", "url": ""}]
        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
        assert "cover_url" not in gen_task._assigned

    def test_no_generation_task_no_crash(self):
        """Should not crash when gen_task is None"""
        candidates = [{"image_url": "https://oss.example.com/cover.jpg"}]
        gen_task_repo = MagicMock()
        gen_task_repo.get.return_value = None
        result, _, _ = _call_finalize(cover_candidates=candidates)
        assert result["status"] == "completed"

    def test_image_url_priority_over_url(self):
        """image_url should take priority over url key"""
        candidates = [{"image_url": "https://a.jpg", "url": "https://b.jpg"}]
        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
        assert gen_task.cover_url == "https://a.jpg"