xiaoxia-saas/tests/unit/test_default_template_fallback_p0.py

"""P0 #1922 默认模板自动兜底单元测试。

覆盖 get_or_create_default_template_id / resolve_draft_plan_id / list_templates / clips_standalone。
注意：_default_template.py 内部使用函数级 lazy import，测试创建/异常分支时
通过 patch.dict(sys.modules, ...) 临时替换 lazy import 目标为 mock 模块。
"""

from __future__ import annotations

import sys
import types
from datetime import datetime
from unittest.mock import MagicMock, patch

import pytest
from fastapi import BackgroundTasks, HTTPException

# ---------- helpers ----------


def _build_mock_tpl(tid="tpl-exists"):
    tpl = MagicMock()
    tpl.id = tid
    tpl.name = "默认配音模板"
    tpl.mode = "voice_over"
    tpl.category = "default"
    tpl.tags = []
    tpl.is_active = True
    tpl.created_at = datetime(2026, 1, 1)
    tpl.updated_at = datetime(2026, 1, 1)
    seg = MagicMock(id="s1", segment_order=0, duration_min=1.0, duration_max=30.0)
    tpl.segments = [seg]
    return tpl


def _db_chain_existing(tpl):
    q_tpl = MagicMock()
    q_tpl.filter.return_value.order_by.return_value.first.return_value = tpl
    q_seg = MagicMock()
    q_seg.filter.return_value.first.return_value = (1,)
    db = MagicMock()
    db.query.side_effect = [q_tpl, q_seg]
    return db


def _db_chain_empty():
    q = MagicMock()
    q.filter.return_value.order_by.return_value.first.return_value = None
    db = MagicMock()
    db.query.return_value = q
    return db


def _lazy_create_patches(new_tpl_or_exc):
    m_models = types.ModuleType("mock_models")
    m_models.TemplateModel = MagicMock()
    m_models.TemplateClipConfigModel = MagicMock()
    m_repo_mod = types.ModuleType("mock_repo")
    m_repo_mod.SQLAlchemyTemplateRepository = MagicMock()
    m_cmds = types.ModuleType("mock_cmds")
    m_cmds.CreateTemplateCommand = MagicMock()
    m_cmds.SegmentCommand = MagicMock()
    m_uc = types.ModuleType("mock_uc")
    uc_instance = MagicMock()
    if isinstance(new_tpl_or_exc, BaseException):
        uc_instance.execute.side_effect = new_tpl_or_exc
    else:
        uc_instance.execute.return_value = new_tpl_or_exc
    m_uc.CreateTemplateUseCase = MagicMock(return_value=uc_instance)
    patches = patch.dict(
        sys.modules,
        {
            "packages.adapters.sqlalchemy_impl.models": m_models,
            "packages.adapters.sqlalchemy_impl.template_repository": m_repo_mod,
            "packages.application.template.commands": m_cmds,
            "packages.application.template.use_cases": m_uc,
        },
    )
    return patches, uc_instance


# ---------- get_or_create_default_template_id ----------


class TestGetOrCreateDefaultTemplateId:
    def test_existing_valid_template_returned_directly(self):
        from app.api.routes._default_template import get_or_create_default_template_id

        tpl = _build_mock_tpl("tpl-1")
        db = _db_chain_existing(tpl)
        tid = get_or_create_default_template_id(db, "user-1")
        assert tid == "tpl-1"
        db.commit.assert_not_called()

    def test_no_template_creates_default(self):
        from app.api.routes._default_template import get_or_create_default_template_id

        new_tpl = _build_mock_tpl("tpl-new")
        db = _db_chain_empty()
        patches, uc = _lazy_create_patches(new_tpl)
        with patches:
            tid = get_or_create_default_template_id(db, "user-new")
        assert tid == "tpl-new"
        db.commit.assert_called_once()
        uc.execute.assert_called_once()

    def test_create_exception_fallback_finds_existing(self):
        from app.api.routes._default_template import get_or_create_default_template_id

        existing = _build_mock_tpl("tpl-race")
        q_none = MagicMock()
        q_none.filter.return_value.order_by.return_value.first.return_value = None
        q_hit = MagicMock()
        q_hit.filter.return_value.order_by.return_value.first.return_value = existing
        q_seg = MagicMock()
        q_seg.filter.return_value.first.return_value = (1,)
        db = MagicMock()
        db.query.side_effect = [q_none, q_hit, q_seg]
        patches, _ = _lazy_create_patches(Exception("unique violation"))
        with patches:
            tid = get_or_create_default_template_id(db, "user-race")
        assert tid == "tpl-race"
        db.rollback.assert_called_once()

    def test_create_exception_no_fallback_returns_none(self):
        from app.api.routes._default_template import get_or_create_default_template_id

        db = _db_chain_empty()
        patches, _ = _lazy_create_patches(Exception("DB down"))
        with patches:
            tid = get_or_create_default_template_id(db, "user-fail")
        assert tid is None
        db.rollback.assert_called_once()

    def test_existing_template_without_segments_triggers_create(self):
        from app.api.routes._default_template import get_or_create_default_template_id

        tpl_empty = _build_mock_tpl("tpl-empty")
        tpl_empty.segments = []
        q_tpl = MagicMock()
        q_tpl.filter.return_value.order_by.return_value.first.return_value = tpl_empty
        q_seg_none = MagicMock()
        q_seg_none.filter.return_value.first.return_value = None
        new_tpl = _build_mock_tpl("tpl-new")
        db = MagicMock()
        db.query.side_effect = [q_tpl, q_seg_none]
        patches, _ = _lazy_create_patches(new_tpl)
        with patches:
            tid = get_or_create_default_template_id(db, "user-2")
        assert tid == "tpl-new"
        db.commit.assert_called_once()


# ---------- resolve_draft_plan_id ----------


class TestResolveDraftPlanId:
    def _services(self, global_tpl=None, draft=None):
        tpl_svc = MagicMock()
        plan_svc = MagicMock()
        tpl_svc.get_template.return_value = global_tpl
        tpl_svc.get_template_draft.return_value = draft
        return (tpl_svc, plan_svc)

    def _user(self, uid="user-1"):
        u = MagicMock()
        u.user.id = uid
        return u

    def test_empty_tid_auto_fallback(self):
        db = MagicMock()
        services = self._services()
        user = self._user()
        old_tpl = _build_mock_tpl("tpl-auto")
        mock_repo = MagicMock()
        mock_repo.get_active.return_value = old_tpl
        mock_plan = MagicMock()
        mock_plan.id = "plan-new"
        mock_gen = MagicMock()
        mock_gen.generate_from_template.return_value = {"plan": mock_plan}
        with (
            patch(
                "app.api.routes.templates_editor.dependencies.SQLAlchemyTemplateRepository",
                return_value=mock_repo,
            ),
            patch(
                "app.api.routes.templates_editor.dependencies.get_or_create_default_template_id",
                return_value="tpl-auto",
            ) as mock_fb,
            patch("app.services.plan_generator_service.PlanGeneratorService", return_value=mock_gen),
        ):
            from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id

            plan_id = resolve_draft_plan_id("", services, user, db, auto_create_default=True)
        assert plan_id == "plan-new"
        mock_fb.assert_called_once_with(db, "user-1")

    def test_empty_tid_no_auto_raises_400(self):
        from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id

        with pytest.raises(HTTPException) as e:
            resolve_draft_plan_id("", self._services(), self._user(), MagicMock(), auto_create_default=False)
        assert e.value.status_code == 400

    def test_missing_template_raises_404(self):
        from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id

        db = MagicMock()
        mock_repo = MagicMock()
        mock_repo.get_active.return_value = None
        with patch(
            "app.api.routes.templates_editor.dependencies.SQLAlchemyTemplateRepository",
            return_value=mock_repo,
        ):
            with pytest.raises(HTTPException) as e:
                resolve_draft_plan_id("tpl-x", self._services(), self._user(), db, auto_create_default=False)
        assert e.value.status_code == 404

    def test_existing_draft_returned_directly(self):
        from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id

        db = MagicMock()
        draft = MagicMock()
        draft.id = "plan-exist"
        services = self._services(draft=draft)
        old_tpl = _build_mock_tpl("tpl-1")
        mock_repo = MagicMock()
        mock_repo.get_active.return_value = old_tpl
        with patch(
            "app.api.routes.templates_editor.dependencies.SQLAlchemyTemplateRepository",
            return_value=mock_repo,
        ):
            plan_id = resolve_draft_plan_id("tpl-1", services, self._user(), db, auto_create_default=False)
        assert plan_id == "plan-exist"

    def test_global_template_creates_draft(self):
        from app.api.routes.templates_editor.dependencies import resolve_draft_plan_id

        db = MagicMock()
        services = self._services(global_tpl=MagicMock(), draft=None)
        new_draft = MagicMock()
        new_draft.id = "plan-g"
        services[0].create_template_draft.return_value = new_draft
        plan_id = resolve_draft_plan_id("tpl-g", services, self._user(), db, auto_create_default=False)
        assert plan_id == "plan-g"
        services[0].create_template_draft.assert_called_once_with("tpl-g", user_id="user-1")


# ---------- list_templates 路由 ----------


class TestTemplatesListRoute:
    def _user(self):
        u = MagicMock()
        u.user.id = "user-1"
        return u

    def test_list_templates_calls_fallback_before_query(self):
        from app.api.routes.templates import list_templates

        db = MagicMock()
        repo = MagicMock()
        tpl = _build_mock_tpl("tpl-1")
        fake_list_uc = MagicMock()
        fake_list_uc.execute.return_value = [tpl]
        fake_count_uc = MagicMock()
        fake_count_uc.execute.return_value = 1
        fake_resp = MagicMock(items=[MagicMock(id="tpl-1")], total=1)
        with (
            patch("app.api.routes.templates.get_or_create_default_template_id") as mock_fb,
            patch("app.api.routes.templates.ListTemplatesUseCase", return_value=fake_list_uc),
            patch("app.api.routes.templates.CountTemplatesUseCase", return_value=fake_count_uc),
            patch(
                "app.api.routes.templates.TemplateResponse.model_validate",
                return_value=MagicMock(id="tpl-1"),
            ),
            patch("app.api.routes.templates.ListTemplatesResponse", return_value=fake_resp),
        ):
            resp = list_templates(
                mode=None,
                category=None,
                tag=None,
                page=1,
                page_size=20,
                current_user=self._user(),
                repo=repo,
                db=db,
            )
        mock_fb.assert_called_once_with(db, "user-1")
        assert resp.total == 1
        fake_list_uc.execute.assert_called_once()


# ---------- clips_standalone 端点 ----------


class TestClipsStandaloneEndpoint:
    def _user(self):
        u = MagicMock()
        u.user.id = "user-1"
        return u

    def test_empty_template_id_uses_fallback(self):
        import app.api.routes.clips_standalone as cs_mod

        body = cs_mod.StandaloneClipsRequest(asset_ids=["a1"])
        mock_core_ret = MagicMock(created_count=3)
        with (
            patch.object(cs_mod, "get_or_create_default_template_id", return_value="tpl-auto") as mock_fb,
            patch.object(cs_mod, "resolve_draft_plan_id", return_value="plan-1") as mock_resolve,
            patch.object(cs_mod, "create_clips_from_assets_editor", return_value=mock_core_ret) as mock_core,
            patch.object(cs_mod, "EditTemplateService"),
            patch.object(cs_mod, "EditPlanService"),
        ):
            resp = cs_mod.create_clips_from_assets(
                body=body,
                background_tasks=BackgroundTasks(),
                db=MagicMock(),
                current_user=self._user(),
                asset_repo=MagicMock(),
            )
        mock_fb.assert_called_once()
        assert resp.created_count == 3
        assert mock_core.call_args.kwargs["template_id"] == "tpl-auto"
        assert mock_core.call_args.kwargs["plan_id"] == "plan-1"
        assert mock_resolve.call_args.kwargs["auto_create_default"] is False

    def test_provided_template_id_skips_fallback(self):
        import app.api.routes.clips_standalone as cs_mod

        body = cs_mod.StandaloneClipsRequest(asset_ids=["a1"], template_id="tpl-custom")
        mock_core_ret = MagicMock(created_count=2)
        with (
            patch.object(cs_mod, "get_or_create_default_template_id") as mock_fb,
            patch.object(cs_mod, "resolve_draft_plan_id", return_value="plan-c"),
            patch.object(cs_mod, "create_clips_from_assets_editor", return_value=mock_core_ret),
            patch.object(cs_mod, "EditTemplateService"),
            patch.object(cs_mod, "EditPlanService"),
        ):
            resp = cs_mod.create_clips_from_assets(
                body=body,
                background_tasks=BackgroundTasks(),
                db=MagicMock(),
                current_user=self._user(),
                asset_repo=MagicMock(),
            )
        mock_fb.assert_not_called()
        assert resp.created_count == 2

    def test_fallback_none_raises_500(self):
        import app.api.routes.clips_standalone as cs_mod

        body = cs_mod.StandaloneClipsRequest(asset_ids=["a1"])
        with (
            patch.object(cs_mod, "get_or_create_default_template_id", return_value=None),
            patch.object(cs_mod, "EditTemplateService"),
            patch.object(cs_mod, "EditPlanService"),
        ):
            with pytest.raises(HTTPException) as e:
                cs_mod.create_clips_from_assets(
                    body=body,
                    background_tasks=BackgroundTasks(),
                    db=MagicMock(),
                    current_user=self._user(),
                    asset_repo=MagicMock(),
                )
        assert e.value.status_code == 500


# ---------- router 注册 ----------


class TestRouterRegistration:
    def test_clips_standalone_router_has_endpoint(self):
        import app.api.routes.clips_standalone as cs_mod

        paths = [r.path for r in cs_mod.router.routes]
        assert "/clips/from-assets" in paths
        assert cs_mod.router.tags and "Clips" in cs_mod.router.tags

    def test_api_router_imports_clips_standalone(self):
        import app.api.router as main_router

        assert main_router.clips_standalone_router is not None
        assert main_router.templates_router is not None