xiaoxia-saas/tests/unit/test_unify_template_segments.py

"""Tests for unified template segments (read from template_clip_configs)."""

import uuid

<<<<<<< Updated upstream
from unittest.mock import MagicMock

=======
>>>>>>> Stashed changes

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, 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 db_session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(bind=engine)
    session = SessionLocal()
    yield session
    session.close()


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


def _make_template(user_id: str = "user1", name: str = "test") -> Template:
<<<<<<< Updated upstream
    return Template(
        id=uuid.uuid4().hex,
        user_id=user_id,
        name=name,
        mode="one_take",
    )
=======
    return Template(id=uuid.uuid4().hex, user_id=user_id, name=name, mode="one_take")
>>>>>>> Stashed changes


def _make_segment(template_id: str, order: int, material_type: str = "video") -> TemplateSegment:
    return TemplateSegment(
<<<<<<< Updated upstream
        id=uuid.uuid4().hex,
        template_id=template_id,
        segment_order=order,
        duration_min=3.0,
        duration_max=5.0,
        material_type=material_type,
=======
        id=uuid.uuid4().hex, template_id=template_id, segment_order=order,
        duration_min=3.0, duration_max=5.0, material_type=material_type,
>>>>>>> Stashed changes
    )


class TestCreateSegmentsWritesToClipConfigs:
<<<<<<< Updated upstream
    """create_segments 应该写入 template_clip_configs 表"""

    def test_create_segments_writes_clip_configs(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
=======
    def test_create_segments_writes_clip_configs(self, repo, db_session):
>>>>>>> Stashed changes
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, 0, "voiceover")
        repo.create_segments([seg])
<<<<<<< Updated upstream

        # 验证 template_clip_configs 有记录
        clips = db_session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == tpl.id
        ).all()
        assert len(clips) == 1
        assert clips[0].order == 0
        assert clips[0].min_duration == 3.0
        assert clips[0].max_duration == 5.0
        assert clips[0].clip_type == "main"
        assert clips[0].config.get("material_type") == "voiceover"

    def test_create_segments_no_old_table_write(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
        """确保不写入旧 template_segments 表"""
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, 0)
        repo.create_segments([seg])

        old_rows = db_session.query(TemplateSegmentModel).filter(
            TemplateSegmentModel.template_id == tpl.id
        ).all()
=======
        clips = db_session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == tpl.id).all()
        assert len(clips) == 1
        assert clips[0].order == 0
        assert clips[0].config.get("material_type") == "voiceover"

    def test_create_segments_no_old_table_write(self, repo, db_session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0)])
        old_rows = db_session.query(TemplateSegmentModel).filter(
            TemplateSegmentModel.template_id == tpl.id).all()
>>>>>>> Stashed changes
        assert len(old_rows) == 0


class TestReadSegmentsFromClipConfigs:
<<<<<<< Updated upstream
    """list_segments 应该从 template_clip_configs 读取"""

    def test_list_segments_reads_clip_configs(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, 0, "video")
        repo.create_segments([seg])

        segments = repo.list_segments(tpl.id)
        assert len(segments) == 1
        assert segments[0].segment_order == 0
        assert segments[0].duration_min == 3.0
        assert segments[0].duration_max == 5.0
        assert segments[0].material_type == "video"

    def test_list_segments_fallback_to_old_table(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
        """如果 template_clip_configs 为空，回退读旧 template_segments"""
        tpl = _make_template()
        repo.create(tpl)

        # 直接写入旧表（模拟历史数据）
        old_model = TemplateSegmentModel(
            id=uuid.uuid4().hex,
            template_id=tpl.id,
            segment_order=0,
            duration_min=2.0,
            duration_max=4.0,
            material_type="人物",
        )
        db_session.add(old_model)
        db_session.commit()

        segments = repo.list_segments(tpl.id)
        assert len(segments) == 1
        assert segments[0].material_type == "人物"
        assert segments[0].duration_min == 2.0

    def test_list_segments_prefers_clip_configs_over_old(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
        """如果两个表都有数据，优先读 template_clip_configs"""
        tpl = _make_template()
        repo.create(tpl)

        # 写入新表
        seg = _make_segment(tpl.id, 0, "video")
        repo.create_segments([seg])

        # 也写入旧表
        old_model = TemplateSegmentModel(
            id=uuid.uuid4().hex,
            template_id=tpl.id,
            segment_order=99,
            duration_min=1.0,
            duration_max=2.0,
            material_type="stale_data",
        )
        db_session.add(old_model)
        db_session.commit()

        segments = repo.list_segments(tpl.id)
        # 应该只返回新表的数据
        assert len(segments) == 1
        assert segments[0].material_type == "video"
        assert segments[0].segment_order == 0


class TestListByUserSegments:
    """list_by_user 批量加载 segments 应从 template_clip_configs 读取"""

    def test_batch_load_segments_from_clip_configs(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
=======
    def test_list_segments_reads_clip_configs(self, repo, db_session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0, "video")])
        segments = repo.list_segments(tpl.id)
        assert len(segments) == 1
        assert segments[0].material_type == "video"

    def test_list_segments_fallback_to_old_table(self, repo, db_session):
        tpl = _make_template()
        repo.create(tpl)
        old_model = TemplateSegmentModel(
            id=uuid.uuid4().hex, template_id=tpl.id, segment_order=0,
            duration_min=2.0, duration_max=4.0, material_type="人物")
        db_session.add(old_model)
        db_session.commit()
        segments = repo.list_segments(tpl.id)
        assert len(segments) == 1
        assert segments[0].material_type == "人物"

    def test_list_segments_prefers_clip_configs_over_old(self, repo, db_session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0, "video")])
        old_model = TemplateSegmentModel(
            id=uuid.uuid4().hex, template_id=tpl.id, segment_order=99,
            duration_min=1.0, duration_max=2.0, material_type="stale_data")
        db_session.add(old_model)
        db_session.commit()
        segments = repo.list_segments(tpl.id)
        assert len(segments) == 1
        assert segments[0].material_type == "video"


class TestListByUserSegments:
    def test_batch_load_segments_from_clip_configs(self, repo, db_session):
>>>>>>> Stashed changes
        tpl1 = _make_template(name="tpl1")
        tpl2 = _make_template(name="tpl2")
        repo.create(tpl1)
        repo.create(tpl2)
<<<<<<< Updated upstream

        repo.create_segments([_make_segment(tpl1.id, 0, "video")])
        repo.create_segments([_make_segment(tpl2.id, 0, "voiceover"), _make_segment(tpl2.id, 1, "video")])

        templates = repo.list_by_user("user1")
        assert len(templates) == 2

        tpl1_result = next(t for t in templates if t.id == tpl1.id)
        tpl2_result = next(t for t in templates if t.id == tpl2.id)

        assert len(tpl1_result.segments) == 1
        assert tpl1_result.segments[0].material_type == "video"

        assert len(tpl2_result.segments) == 2
        assert tpl2_result.segments[0].material_type == "voiceover"
        assert tpl2_result.segments[1].material_type == "video"


class TestDeleteSegments:
    """delete_segments_by_template 应清理 template_clip_configs"""

    def test_delete_segments_clears_clip_configs(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0)])

        count = repo.delete_segments_by_template(tpl.id)
        assert count == 1

        clips = db_session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == tpl.id
        ).all()
        assert len(clips) == 0

    def test_delete_template_clears_both_tables(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0)])

        # 也写入旧表
        old_model = TemplateSegmentModel(
            id=uuid.uuid4().hex,
            template_id=tpl.id,
            segment_order=0,
            duration_min=1.0,
            duration_max=2.0,
        )
        db_session.add(old_model)
        db_session.commit()

        repo.delete(tpl.id, "user1")

        clips = db_session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == tpl.id
        ).all()
        assert len(clips) == 0

        old_rows = db_session.query(TemplateSegmentModel).filter(
            TemplateSegmentModel.template_id == tpl.id
        ).all()
=======
        repo.create_segments([_make_segment(tpl1.id, 0, "video")])
        repo.create_segments([_make_segment(tpl2.id, 0, "voiceover"), _make_segment(tpl2.id, 1, "video")])
        templates = repo.list_by_user("user1")
        assert len(templates) == 2
        tpl1_r = next(t for t in templates if t.id == tpl1.id)
        tpl2_r = next(t for t in templates if t.id == tpl2.id)
        assert len(tpl1_r.segments) == 1
        assert len(tpl2_r.segments) == 2


class TestDeleteSegments:
    def test_delete_segments_clears_clip_configs(self, repo, db_session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0)])
        count = repo.delete_segments_by_template(tpl.id)
        assert count >= 1
        clips = db_session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == tpl.id).all()
        assert len(clips) == 0

    def test_delete_template_clears_both_tables(self, repo, db_session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0)])
        old_model = TemplateSegmentModel(
            id=uuid.uuid4().hex, template_id=tpl.id, segment_order=0,
            duration_min=1.0, duration_max=2.0)
        db_session.add(old_model)
        db_session.commit()
        repo.delete(tpl.id, "user1")
        clips = db_session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == tpl.id).all()
        assert len(clips) == 0
        old_rows = db_session.query(TemplateSegmentModel).filter(
            TemplateSegmentModel.template_id == tpl.id).all()
>>>>>>> Stashed changes
        assert len(old_rows) == 0


class TestCopyTemplateSegments:
<<<<<<< Updated upstream
    """copy_template 应从 template_clip_configs 复制 segments"""

    def test_copy_preserves_segments(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([
            _make_segment(tpl.id, 0, "video"),
            _make_segment(tpl.id, 1, "voiceover"),
        ])

        copied = repo.copy_template(tpl.id, "user1", "copy")
        assert len(copied.segments) == 2
        assert copied.segments[0].material_type == "video"
        assert copied.segments[1].material_type == "voiceover"

        # 验证新模板的 clip_configs 有数据
        clips = db_session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == copied.id
        ).all()
=======
    def test_copy_preserves_segments(self, repo, db_session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0, "video"), _make_segment(tpl.id, 1, "voiceover")])
        copied = repo.copy_template(tpl.id, "user1", "copy")
        assert len(copied.segments) == 2
        clips = db_session.query(TemplateClipConfigModel).filter(
            TemplateClipConfigModel.template_id == copied.id).all()
>>>>>>> Stashed changes
        assert len(clips) == 2


class TestNullMaterialType:
<<<<<<< Updated upstream
    """material_type 为 None 时不应报错"""

    def test_null_material_type(self, repo: SQLAlchemyTemplateRepository, db_session: Session):
        tpl = _make_template()
        repo.create(tpl)
        seg = _make_segment(tpl.id, 0, None)
        repo.create_segments([seg])

=======
    def test_null_material_type(self, repo, db_session):
        tpl = _make_template()
        repo.create(tpl)
        repo.create_segments([_make_segment(tpl.id, 0, None)])
>>>>>>> Stashed changes
        segments = repo.list_segments(tpl.id)
        assert len(segments) == 1
        assert segments[0].material_type is None