xiaoxia-saas/tests/unit/test_generation_worker_fixes.py

"""Tests for generation.py worker-side fixes: segment durations + preview resolution."""

from __future__ import annotations

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

# Add worker app to path
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")


def _patch_session_local(mock_session):
    """Patch worker_app.db.SessionLocal robustly even when other tests
    have pre-registered a MagicMock for worker_app.db in sys.modules.
    Uses patch.dict to inject a clean module so that
    'from worker_app.db import SessionLocal' resolves correctly."""
    from types import ModuleType
    _fresh_db = ModuleType("worker_app.db")
    _fresh_db.SessionLocal = lambda *a, **kw: mock_session
    return patch.dict(sys.modules, {"worker_app.db": _fresh_db})


class TestLoadTemplateSegmentDurations:
    """_load_template_segment_durations 单元测试 (covers lines 198-226)."""

    def test_empty_template_id(self):
        """空 template_id 直接返回空列表。"""
        from worker_app.tasks.generation import _load_template_segment_durations

        result = _load_template_segment_durations("")
        assert result == []

    def test_loads_durations_ordered(self):
        """按 segment_order 排序返回 duration_max 列表。"""
        from worker_app.tasks.generation import _load_template_segment_durations

        mock_seg1 = MagicMock(duration_max=5.0)
        mock_seg2 = MagicMock(duration_max=8.0)
        mock_seg3 = MagicMock(duration_max=3.0)

        mock_query = MagicMock()
        mock_query.filter.return_value.order_by.return_value.all.return_value = [
            mock_seg1,
            mock_seg2,
            mock_seg3,
        ]
        mock_session = MagicMock()
        mock_session.query.return_value = mock_query

        with _patch_session_local(mock_session):
            result = _load_template_segment_durations("tpl_123")

        assert result == [5.0, 8.0, 3.0]

    def test_filters_zero_and_negative(self):
        """duration_max <= 0 的 segment 被过滤。"""
        from worker_app.tasks.generation import _load_template_segment_durations

        mock_seg_valid = MagicMock(duration_max=5.0)
        mock_seg_zero = MagicMock(duration_max=0.0)
        mock_seg_none = MagicMock(duration_max=None)

        mock_query = MagicMock()
        mock_query.filter.return_value.order_by.return_value.all.return_value = [
            mock_seg_valid,
            mock_seg_zero,
            mock_seg_none,
        ]
        mock_session = MagicMock()
        mock_session.query.return_value = mock_query

        with _patch_session_local(mock_session):
            result = _load_template_segment_durations("tpl_456")

        assert result == [5.0]

    def test_db_error_returns_empty(self):
        """数据库异常返回空列表，不抛出。"""
        from worker_app.tasks.generation import _load_template_segment_durations

        from types import ModuleType
        _err_db = ModuleType("worker_app.db")
        def _raise(*a, **kw):
            raise Exception("DB down")
        _err_db.SessionLocal = _raise
        with patch.dict(sys.modules, {"worker_app.db": _err_db}):
            result = _load_template_segment_durations("tpl_789")

        assert result == []

    def test_empty_segments_returns_empty(self):
        """没有 segment 时返回空列表。"""
        from worker_app.tasks.generation import _load_template_segment_durations

        mock_query = MagicMock()
        mock_query.filter.return_value.order_by.return_value.all.return_value = []
        mock_session = MagicMock()
        mock_session.query.return_value = mock_query

        with _patch_session_local(mock_session):
            result = _load_template_segment_durations("tpl_empty")

        assert result == []


class TestDurationCappingInBuildPlan:
    """_build_plan_and_clips_from_task 时长约束测试 (covers lines 322-333)."""

    def _make_temp_video(self, tmpdir: Path, name: str = "v.mp4") -> Path:
        p = tmpdir / name
        p.write_bytes(b"\x00" * 100)
        return p

    def test_clips_capped_by_segment_max(self):
        """clip 时长超过 segment duration_max 时截断。"""
        import tempfile

        from worker_app.tasks.generation import _build_plan_and_clips_from_task

        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = Path(tmpdir)
            paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)]

            with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
                with patch(
                    "worker_app.tasks.generation._load_template_segment_durations",
                    return_value=[5.0, 4.0, 3.0],
                ):
                    with patch(
                        "worker_app.tasks.generation._load_template_clip_configs",
                        return_value=[],
                    ):
                        _, clips, _ = _build_plan_and_clips_from_task(
                            task_id="test_cap",
                            downloaded_paths=paths,
                            mode="one_take",
                            template_id="tpl_test",
                        )

            assert clips[0].duration == 5.0
            assert clips[1].duration == 4.0
            assert clips[2].duration == 3.0

    def test_clips_not_capped_when_under_max(self):
        """clip 时长小于 segment duration_max 时不截断。"""
        import tempfile

        from worker_app.tasks.generation import _build_plan_and_clips_from_task

        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = Path(tmpdir)
            paths = [self._make_temp_video(tmpdir)]

            with patch("worker_app.tasks.generation.probe_duration", return_value=3.0):
                with patch(
                    "worker_app.tasks.generation._load_template_segment_durations",
                    return_value=[5.0],
                ):
                    with patch(
                        "worker_app.tasks.generation._load_template_clip_configs",
                        return_value=[],
                    ):
                        _, clips, _ = _build_plan_and_clips_from_task(
                            task_id="test_no_cap",
                            downloaded_paths=paths,
                            mode="one_take",
                            template_id="tpl_test",
                        )

            assert clips[0].duration == 3.0

    def test_no_capping_without_template(self):
        """无 template_id 时不截断。"""
        import tempfile

        from worker_app.tasks.generation import _build_plan_and_clips_from_task

        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = Path(tmpdir)
            paths = [self._make_temp_video(tmpdir)]

            with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
                _, clips, _ = _build_plan_and_clips_from_task(
                    task_id="test_no_tpl",
                    downloaded_paths=paths,
                    mode="one_take",
                    template_id="",
                )

            assert clips[0].duration == 30.0

    def test_partial_segments_only_caps_matching(self):
        """segment 数量少于 clip 时，只截断有对应 segment 的 clip。"""
        import tempfile

        from worker_app.tasks.generation import _build_plan_and_clips_from_task

        with tempfile.TemporaryDirectory() as tmpdir:
            tmpdir = Path(tmpdir)
            paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)]

            with patch("worker_app.tasks.generation.probe_duration", return_value=20.0):
                with patch(
                    "worker_app.tasks.generation._load_template_segment_durations",
                    return_value=[5.0],  # only 1 segment for 3 clips
                ):
                    with patch(
                        "worker_app.tasks.generation._load_template_clip_configs",
                        return_value=[],
                    ):
                        _, clips, _ = _build_plan_and_clips_from_task(
                            task_id="test_partial",
                            downloaded_paths=paths,
                            mode="one_take",
                            template_id="tpl_test",
                        )

            assert clips[0].duration == 5.0  # capped
            assert clips[1].duration == 20.0  # not capped (no matching segment)
            assert clips[2].duration == 20.0  # not capped