test(api/services): 为generation_common公共服务补充单元测试，修复CI增量覆盖率门禁(20%→87%)

tests/unit/test_generation_common.py

@@ -0,0 +1,259 @@
+"""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_config" not in plan.config
+        db.commit.assert_called_once()
+
+    def test_merges_title_config_without_title_change(self):
+        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)
+        assert plan.config["title_config"] == {"text": "old"}
+        # 标题未变 → cover 保留
+        assert plan.config.get("cover") == "x"
+
+    def test_title_change_clears_cover(self):
+        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": "new-title"}, db)
+        assert "cover" not in plan.config
+        assert plan.config["title_config"] == {"text": "new-title"}
+
+    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_config"] == {"text": "hi"}
+
+    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)