From 840b26eb124f76e348ca8d9d3b91067b180c4c38 Mon Sep 17 00:00:00 2001 From: refactor Date: Mon, 14 Sep 2026 03:05:38 +0800 Subject: [PATCH] =?UTF-8?q?test(api/services):=20=E4=B8=BAgeneration=5Fcom?= =?UTF-8?q?mon=E5=85=AC=E5=85=B1=E6=9C=8D=E5=8A=A1=E8=A1=A5=E5=85=85?= =?UTF-8?q?=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95=EF=BC=8C=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?CI=E5=A2=9E=E9=87=8F=E8=A6=86=E7=9B=96=E7=8E=87=E9=97=A8?= =?UTF-8?q?=E7=A6=81(20%=E2=86=9287%)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_generation_common.py | 259 +++++++++++++++++++++++++++ 1 file changed, 259 insertions(+) create mode 100644 tests/unit/test_generation_common.py diff --git a/tests/unit/test_generation_common.py b/tests/unit/test_generation_common.py new file mode 100644 index 000000000..baa1b9d38 --- /dev/null +++ b/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)