xiaoxia-saas/tests/unit/test_unify_template_segments.py

"""统一模板 segments 数据源单元测试。

验证 template_repository 从 template_clip_configs 读取 segments，
写入走 template_clip_configs，回退兼容 template_segments。
"""

from __future__ import annotations

import sys
import uuid
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from packages.adapters.sqlalchemy_impl.models import (
    Base,
    TemplateClipConfigModel,
    TemplateModel,
    TemplateSegmentModel,
)
from packages.adapters.sqlalchemy_impl.template_repository import (
    SQLAlchemyTemplateRepository,
)
from packages.domain.template import Template, TemplateSegment


@pytest.fixture()
def session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    s = Session()
    try:
        yield s
    finally:
        s.close()


@pytest.fixture()
def repo(session):
    return SQLAlchemyTemplateRepository(session)


def _make_template(template_id=None, user_id="u1", name="测试模板", mode="one_take"):
    tid = template_id or str(uuid.uuid4())
    return Template(
        id=tid,
        user_id=user_id,
        name=name,
        mode=mode,
        category="",
        tags=[],
        estimated_duration=30.0,
        is_active=True,
        segments=[],
    )


def _make_segment(template_id, order=1, material_type=None):
    return TemplateSegment(
        id=str(uuid.uuid4()),
        template_id=template_id,
        segment_order=order,
        duration_min=5.0,
        duration_max=10.0,
        material_type=material_type,
    )


class TestCreateSegments:
    def test_writes_to_clip_configs(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, order=1)
        repo.create_segments([seg])
        clips = session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == tpl.id).all()
        assert len(clips) == 1
        assert clips[0].clip_type == "main"
        assert clips[0].order == 1
        assert clips[0].min_duration == 5.0

    def test_material_type_stored_in_config(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, order=1, material_type="voiceover")
        repo.create_segments([seg])
        clip = session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == tpl.id).first()
        assert clip.config["material_type"] == "voiceover"


class TestListSegments:
    def test_reads_from_clip_configs(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, order=1, material_type="voiceover")
        repo.create_segments([seg])
        result = repo.list_segments(tpl.id)
        assert len(result) == 1
        assert result[0].material_type == "voiceover"

    def test_fallback_to_old_table(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        old = TemplateSegmentModel(
            id=str(uuid.uuid4()),
            template_id=tpl.id,
            segment_order=1,
            duration_min=3.0,
            duration_max=8.0,
            material_type="场景",
        )
        session.add(old)
        session.commit()
        result = repo.list_segments(tpl.id)
        assert len(result) == 1
        assert result[0].material_type == "场景"

    def test_clip_configs_takes_priority(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, order=1)
        repo.create_segments([seg])
        old = TemplateSegmentModel(
            id=str(uuid.uuid4()), template_id=tpl.id, segment_order=1, duration_min=1.0, duration_max=2.0
        )
        session.add(old)
        session.commit()
        result = repo.list_segments(tpl.id)
        assert len(result) == 1
        assert result[0].duration_min == 5.0


class TestListByUser:
    def test_batch_loads_from_clip_configs(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, order=1, material_type="人物")
        repo.create_segments([seg])
        result = repo.list_by_user("u1")
        assert len(result) == 1
        assert len(result[0].segments) == 1
        assert result[0].segments[0].material_type == "人物"

    def test_fallback_for_old_data(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        old = TemplateSegmentModel(
            id=str(uuid.uuid4()), template_id=tpl.id, segment_order=1, duration_min=2.0, duration_max=6.0
        )
        session.add(old)
        session.commit()
        result = repo.list_by_user("u1")
        assert len(result) == 1
        assert len(result[0].segments) == 1
        assert result[0].segments[0].duration_min == 2.0

    def test_valid_only_filters_templates_without_segments(self, repo, session):
        """#1769: valid_only=True 时排除两张片段表都没有记录的无效模板."""
        # 有效模板：有 clip_configs
        valid_clip = _make_template(name="有效模板-clip_configs")
        repo.create(valid_clip)
        repo.create_segments([_make_segment(valid_clip.id, order=1)])
        # 有效模板：仅有旧表 template_segments 记录
        valid_old = _make_template(name="有效模板-old_segments")
        repo.create(valid_old)
        old = TemplateSegmentModel(
            id=str(uuid.uuid4()),
            template_id=valid_old.id,
            segment_order=1,
            duration_min=2.0,
            duration_max=6.0,
        )
        session.add(old)
        session.commit()
        # 无效模板：两张表都没有记录
        invalid = _make_template(name="无效模板-无片段")
        repo.create(invalid)

        # 默认不过滤：编辑器视角能看到全部 3 个模板
        all_templates = repo.list_by_user("u1")
        assert len(all_templates) == 3
        assert repo.count_by_user("u1") == 3

        # valid_only=True：剪辑页视角只返回 2 个有效模板
        valid_templates = repo.list_by_user("u1", valid_only=True)
        assert {t.name for t in valid_templates} == {"有效模板-clip_configs", "有效模板-old_segments"}
        assert all(len(t.segments) > 0 for t in valid_templates)
        assert repo.count_by_user("u1", valid_only=True) == 2

    def test_valid_only_with_filters_and_pagination(self, repo, session):
        """valid_only 与其他过滤/分页条件组合使用."""
        tpl = _make_template(name="口播模板", mode="voice_over")
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, order=1, material_type="人物")])
        _invalid = _make_template(name="口播无效模板", mode="voice_over")
        repo.create(_invalid)

        result = repo.list_by_user("u1", mode="voice_over", valid_only=True)
        assert len(result) == 1
        assert result[0].name == "口播模板"
        assert repo.count_by_user("u1", mode="voice_over", valid_only=True) == 1


class TestCopyTemplate:
    def test_copy_writes_to_clip_configs(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, order=1, material_type="voiceover")
        repo.create_segments([seg])
        copied = repo.copy_template(tpl.id, "u1", "副本模板")
        assert copied.id != tpl.id
        clips = session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == copied.id).all()
        assert len(clips) == 1
        assert clips[0].config["material_type"] == "voiceover"

    def test_copy_empty_segments(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        copied = repo.copy_template(tpl.id, "u1", "空副本")
        assert len(copied.segments) == 0


class TestDelete:
    def test_delete_cleans_both_tables(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, order=1)
        repo.create_segments([seg])
        old = TemplateSegmentModel(
            id=str(uuid.uuid4()), template_id=tpl.id, segment_order=1, duration_min=1.0, duration_max=2.0
        )
        session.add(old)
        session.commit()
        repo.delete(tpl.id, "u1")
        c1 = session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == tpl.id).count()
        c2 = session.query(TemplateSegmentModel).filter(TemplateSegmentModel.template_id == tpl.id).count()
        assert c1 == 0
        assert c2 == 0

    def test_delete_segments_by_template(self, repo, session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, order=1)
        repo.create_segments([seg])
        old = TemplateSegmentModel(
            id=str(uuid.uuid4()), template_id=tpl.id, segment_order=2, duration_min=1.0, duration_max=2.0
        )
        session.add(old)
        session.commit()
        count = repo.delete_segments_by_template(tpl.id)
        assert count == 2