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


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