xiaoxia-saas/tests/unit/test_generation_common.py

"""generation_common 公共服务辅助函数单元测试。

覆盖 query_voice_durations / writeback_edit_plan_config / collect_plan_segments /
resolve_latest_plan_by_template 四个下沉函数的主路径、边界与容错路径。
"""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest

# ═══════════════════════════════════════════════════════════════════════════════
# query_voice_durations
# ═══════════════════════════════════════════════════════════════════════════════


class TestQueryVoiceDurations:
    def _make_db_with_rows(self, rows):
        """构造 MagicMock db，query().filter().all() 返回 rows。"""
        db = MagicMock()
        db.query.return_value.filter.return_value.all.return_value = list(rows)
        return db

    def test_empty_input_returns_empty_list(self):
        from app.services.generation_common import query_voice_durations

        db = MagicMock()
        assert query_voice_durations(db, []) == []
        assert query_voice_durations(db, None) == []
        db.query.assert_not_called()

    def test_all_empty_or_falsy_ids_returns_zero_list(self):
        from app.services.generation_common import query_voice_durations

        db = MagicMock()
        assert query_voice_durations(db, ["", None, ""]) == [0.0, 0.0, 0.0]

    def test_normal_lookup_returns_durations_in_input_order(self):
        from app.services.generation_common import query_voice_durations

        db = self._make_db_with_rows([("v1", 3.5), ("v2", 7.2)])
        result = query_voice_durations(db, ["v1", "v2", "v-missing"])
        assert result == [3.5, 7.2, 0.0]

    def test_duplicate_ids_returns_consistent_durations_preserves_order(self):
        """#1855：同配音 id 多次出现应返回相同时长，保持输入顺序/长度。"""
        from app.services.generation_common import query_voice_durations

        db = self._make_db_with_rows([("v1", 4.0)])
        result = query_voice_durations(db, ["v1", "v1", "v1"])
        assert result == [4.0, 4.0, 4.0]

    def test_non_numeric_duration_coerced_to_zero(self):
        from app.services.generation_common import query_voice_durations

        db = self._make_db_with_rows([("v1", None), ("v2", "not-a-number"), ("v3", 2.0)])
        result = query_voice_durations(db, ["v1", "v2", "v3"])
        assert result == [0.0, 0.0, 2.0]

    def test_db_exception_returns_zeros_and_logs(self, caplog):
        from app.services.generation_common import query_voice_durations

        db = MagicMock()
        db.query.side_effect = RuntimeError("DB boom")
        with caplog.at_level("WARNING"):
            result = query_voice_durations(db, ["v1", "v2"])
        assert result == [0.0, 0.0]
        assert any("配音时长查询失败" in rec.message for rec in caplog.records)


# ═══════════════════════════════════════════════════════════════════════════════
# writeback_edit_plan_config
# ═══════════════════════════════════════════════════════════════════════════════


def _make_plan_model(config=None):
    plan = MagicMock()
    plan.config = config if config is not None else {}
    return plan


class TestWritebackEditPlanConfig:
    def test_empty_plan_id_returns_immediately(self):
        from app.services.generation_common import writeback_edit_plan_config

        db = MagicMock()
        writeback_edit_plan_config("", "task1", None, db)
        db.query.assert_not_called()

    def test_plan_not_found_logs_and_returns(self, caplog):
        from app.services.generation_common import writeback_edit_plan_config

        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = None
        with caplog.at_level("WARNING"):
            writeback_edit_plan_config("p999", "task1", None, db)
        db.commit.assert_not_called()
        assert any("plan不存在" in rec.message for rec in caplog.records)

    def test_writes_task_id_preserves_existing_config(self):
        from app.services.generation_common import writeback_edit_plan_config

        plan = _make_plan_model({"other": "keep-me"})
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = plan
        writeback_edit_plan_config("p1", "task-xyz", None, db)
        assert plan.config["generation_task_id"] == "task-xyz"
        assert plan.config["other"] == "keep-me"
        assert "title" not in plan.config or not plan.config.get("title")
        db.commit.assert_called_once()

    def test_merges_title_without_title_change(self):
        """#1901: 写 'title' 字段，未变标题保留 cover。"""
        from app.services.generation_common import writeback_edit_plan_config

        plan = _make_plan_model({"title": {"text": "old"}, "cover": "x"})
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = plan
        writeback_edit_plan_config("p1", "t1", {"text": "old"}, db)
        assert plan.config["title"] == {"text": "old"}
        # 旧 key 不应残留
        assert "title_config" not in plan.config
        # 标题未变 → cover 保留
        assert plan.config.get("cover") == "x"

    def test_merges_title_fallback_to_old_title_config_key(self):
        """#1901: 老数据存在 title_config（无 title）时，也能正确识别旧标题文字。"""
        from app.services.generation_common import writeback_edit_plan_config

        plan = _make_plan_model({"title_config": {"text": "old"}, "cover": "x"})
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = plan
        writeback_edit_plan_config("p1", "t1", {"text": "old"}, db)
        # 写入新 key "title"，旧 key 被清除
        assert plan.config["title"] == {"text": "old"}
        assert "title_config" not in plan.config
        assert plan.config.get("cover") == "x"

    def test_title_change_clears_cover(self):
        """#1901: 标题变化时清 cover，新配置写到 'title'。"""
        from app.services.generation_common import writeback_edit_plan_config

        plan = _make_plan_model({"title": {"text": "old"}, "cover": "x"})
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = plan
        writeback_edit_plan_config("p1", "t1", {"text": "new-title"}, db)
        assert "cover" not in plan.config
        assert plan.config["title"] == {"text": "new-title"}
        assert "title_config" not in plan.config

    def test_title_config_normalizes_legacy_keys(self):
        """#1901: 写入时归一化 font_size/font_preset/font_color → size/font/color，与 worker 对齐。"""
        from app.services.generation_common import writeback_edit_plan_config

        plan = _make_plan_model({})
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = plan
        writeback_edit_plan_config(
            "p1",
            "t1",
            {"text": "hi", "font_size": 32, "font_preset": "楷体", "font_color": "#ff0000", "bold": True},
            db,
        )
        title = plan.config["title"]
        assert title["text"] == "hi"
        assert title["size"] == 32
        assert title["font"] == "楷体"
        assert title["color"] == "#ff0000"
        # 原始 key 保留（方便调用方排查，但归一化后的 key 必须存在）
        assert title["font_size"] == 32

    def test_config_not_dict_treated_as_empty(self):
        from app.services.generation_common import writeback_edit_plan_config

        plan = _make_plan_model(config=None)
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = plan
        writeback_edit_plan_config("p1", "t1", {"text": "hi"}, db)
        assert plan.config["generation_task_id"] == "t1"
        assert plan.config["title"] == {"text": "hi"}
        assert "title_config" not in plan.config

    def test_exception_triggers_rollback_and_logs(self, caplog):
        from app.services.generation_common import writeback_edit_plan_config

        db = MagicMock()
        db.query.return_value.filter.return_value.first.side_effect = RuntimeError("fail")
        with caplog.at_level("WARNING"):
            writeback_edit_plan_config("p1", "t1", None, db)
        db.rollback.assert_called_once()
        assert any("回写plan.config异常" in rec.message for rec in caplog.records)

    def test_exception_with_rollback_also_failing_is_safe(self, caplog):
        """外层异常后，db.rollback() 自己也抛异常时也不应中断（pass 兜底）。"""
        from app.services.generation_common import writeback_edit_plan_config

        db = MagicMock()
        db.query.return_value.filter.return_value.first.side_effect = RuntimeError("fail")
        db.rollback.side_effect = RuntimeError("rollback boom")
        with caplog.at_level("WARNING"):
            # 不应抛出异常
            writeback_edit_plan_config("p1", "t1", None, db)
        assert any("回写plan.config异常" in rec.message for rec in caplog.records)


# ═══════════════════════════════════════════════════════════════════════════════
# collect_plan_segments
# ═══════════════════════════════════════════════════════════════════════════════


def _make_clip(asset_id, start, duration):
    c = MagicMock()
    c.asset_id = asset_id
    c.start_time = start
    c.duration = duration
    return c


class TestCollectPlanSegments:
    def test_empty_plan_returns_empty(self):
        from app.services.generation_common import collect_plan_segments

        repo = MagicMock()
        repo.list_by_plan.return_value = []
        assert collect_plan_segments("p1", repo) == {}

    def test_single_page_collects_segments(self):
        from app.services.generation_common import collect_plan_segments

        repo = MagicMock()
        repo.list_by_plan.side_effect = [
            [_make_clip("a1", 0.0, 5.0), _make_clip("a1", 10.0, 3.0), _make_clip("a2", 2.0, 4.0)],
            [],
        ]
        segs = collect_plan_segments("p1", repo, page_size=500)
        assert segs["a1"] == [(0.0, 5.0), (10.0, 13.0)]
        assert segs["a2"] == [(2.0, 6.0)]

    def test_pagination_walks_all_batches(self):
        from app.services.generation_common import collect_plan_segments

        repo = MagicMock()
        page1 = [_make_clip("a1", 0.0, 1.0)] * 2
        page2 = [_make_clip("a2", 0.0, 2.0)] * 2
        page3 = [_make_clip("a3", 0.0, 1.0)]  # short final batch → stop
        repo.list_by_plan.side_effect = [page1, page2, page3]
        segs = collect_plan_segments("p1", repo, page_size=2)
        assert set(segs.keys()) == {"a1", "a2", "a3"}
        assert repo.list_by_plan.call_count == 3

    def test_skips_zero_or_negative_duration_clips(self):
        from app.services.generation_common import collect_plan_segments

        repo = MagicMock()
        repo.list_by_plan.side_effect = [
            [_make_clip(None, 0.0, 5.0), _make_clip("a1", 0.0, 0.0), _make_clip("a1", 1.0, -1.0)],
            [],
        ]
        assert collect_plan_segments("p1", repo) == {}


# ═══════════════════════════════════════════════════════════════════════════════
# resolve_latest_plan_by_template
# ═══════════════════════════════════════════════════════════════════════════════


class TestResolveLatestPlanByTemplate:
    @pytest.mark.parametrize("tid", ["", None, "   "])
    def test_empty_template_returns_none(self, tid):
        from app.services.generation_common import resolve_latest_plan_by_template

        db = MagicMock()
        assert resolve_latest_plan_by_template(db, template_id=tid, user_id="u1") is None
        db.query.assert_not_called()

    def test_returns_latest_plan_id(self):
        from app.services.generation_common import resolve_latest_plan_by_template

        db = MagicMock()
        latest = MagicMock(id="plan-xyz")
        db.query.return_value.filter.return_value.order_by.return_value.first.return_value = latest
        assert resolve_latest_plan_by_template(db, template_id=" tpl1 ", user_id="u1") == "plan-xyz"

    def test_no_plan_returns_none(self):
        from app.services.generation_common import resolve_latest_plan_by_template

        db = MagicMock()
        db.query.return_value.filter.return_value.order_by.return_value.first.return_value = None
        assert resolve_latest_plan_by_template(db, template_id="tpl", user_id="u") is None

    def test_db_exception_returns_none_and_logs(self, caplog):
        from app.services.generation_common import resolve_latest_plan_by_template

        db = MagicMock()
        db.query.side_effect = RuntimeError("boom")
        with caplog.at_level("WARNING"):
            assert resolve_latest_plan_by_template(db, template_id="tpl", user_id="u") is None
        assert any("查找最新plan失败" in rec.message for rec in caplog.records)


# ═══════════════════════════════════════════════════════════════════════════════
# collect_plan_atom_clip_ids (#1970)
# ═══════════════════════════════════════════════════════════════════════════════


def _make_atom_clip(atom_clip_id):
    c = MagicMock()
    c.atom_clip_id = atom_clip_id
    return c


class TestCollectPlanAtomClipIds:
    def test_empty_plan_returns_empty(self):
        from app.services.generation_common import collect_plan_atom_clip_ids

        repo = MagicMock()
        repo.list_by_plan.return_value = []
        assert collect_plan_atom_clip_ids("p1", repo) == []

    def test_collects_non_empty_ids_and_ignores_blank(self):
        from app.services.generation_common import collect_plan_atom_clip_ids

        repo = MagicMock()
        repo.list_by_plan.side_effect = [
            [
                _make_atom_clip("atom-1"),
                _make_atom_clip(""),
                _make_atom_clip("atom-2"),
            ],
            [],
        ]
        assert collect_plan_atom_clip_ids("p1", repo) == ["atom-1", "atom-2"]

    def test_missing_attribute_treated_as_blank(self):
        from app.services.generation_common import collect_plan_atom_clip_ids

        legacy = MagicMock()
        del legacy.atom_clip_id  # 旧对象无该属性
        repo = MagicMock()
        repo.list_by_plan.side_effect = [[legacy, _make_atom_clip("atom-9")], []]
        assert collect_plan_atom_clip_ids("p1", repo) == ["atom-9"]


# ═══════════════════════════════════════════════════════════════════════════════
# writeback_edit_plan_config：#1970 dedup_enabled / video_index
# ═══════════════════════════════════════════════════════════════════════════════


class TestWritebackDedupAndVideoIndex:
    def test_writes_dedup_enabled_and_video_index(self):
        from app.services.generation_common import writeback_edit_plan_config

        plan = _make_plan_model({})
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = plan
        writeback_edit_plan_config("p1", "t1", None, db, dedup_enabled=False, video_index=3)
        assert plan.config["dedup_enabled"] is False
        assert plan.config["video_index"] == 3
        assert plan.config["generation_task_id"] == "t1"

    def test_none_dedup_does_not_touch_flag(self):
        from app.services.generation_common import writeback_edit_plan_config

        plan = _make_plan_model({"dedup_enabled": True})
        db = MagicMock()
        db.query.return_value.filter.return_value.first.return_value = plan
        writeback_edit_plan_config("p1", "t1", None, db)
        assert plan.config["dedup_enabled"] is True
        assert "video_index" not in plan.config