xiaoxia-saas/tests/unit/test_1970_atom_plan_generation.py

"""#1970 PlanGeneratorService 原子片段选片端到端单元测试.

用 SQLite 内存库 + 真实仓储验证：注入 atom_clip_repo 后，正式生成（非预览）
从原子片段选片，EditPlanClip.atom_clip_id 落库；预览模式保持旧路径。
"""

from __future__ import annotations

import os
import sys
from pathlib import Path

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

import pytest
from app.services.plan_generator_service import PlanGeneratorService
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
    SQLAlchemyAssetAtomClipRepository,
)
from packages.adapters.sqlalchemy_impl.models import Base
from packages.domain.asset_atom_clip import AssetAtomClip
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
from packages.domain.editing_mode import EditingMode
from packages.domain.template_clip_config import ClipType, TemplateClipConfig


class _FakeAsset:
    def __init__(self, aid, duration):
        self.id = aid
        self.duration = duration
        self.quality_score = 60.0
        self.metadata = {}
        self.created_at = None


class FakeAssetRepo:
    def __init__(self, durations):
        self._durations = durations

    def get(self, aid):
        return _FakeAsset(aid, self._durations[aid]) if aid in self._durations else None


@pytest.fixture()
def db_session():
    engine = create_engine("sqlite://")
    # 只建相关表，避免全模型依赖
    Base.metadata.create_all(
        engine,
        tables=[
            Base.metadata.tables["edit_plans"],
            Base.metadata.tables["edit_plan_clips"],
            Base.metadata.tables["asset_atom_clips"],
        ],
    )
    connection = engine.connect()
    Session = sessionmaker(bind=connection)
    session = Session()
    yield session
    session.close()
    connection.close()


def _template(mode=EditingMode.ONE_TAKE.value):
    return EditTemplate(
        id="tpl-1",
        name="测试模板",
        editing_mode=mode,
        status=EditTemplateStatus.ACTIVE,
    )


def _clip_configs(n=3):
    return [
        TemplateClipConfig(
            id=f"cfg-{i}",
            template_id="tpl-1",
            clip_type=ClipType.MAIN,
            order=i,
            min_duration=3.0,
            max_duration=6.0,
        )
        for i in range(n)
    ]


class TestAtomClipPlanGeneration:
    def test_generation_uses_atom_clips(self, db_session):
        atom_repo = SQLAlchemyAssetAtomClipRepository(db_session)
        # 两个素材各 30s，各切若干片段
        clips_a = [AssetAtomClip.create("asset-a", i * 5.0, i * 5.0 + 5.0, i) for i in range(6)]
        clips_b = [AssetAtomClip.create("asset-b", i * 5.0, i * 5.0 + 5.0, i) for i in range(6)]
        atom_repo.batch_create(clips_a + clips_b)
        db_session.commit()

        svc = PlanGeneratorService(
            db_session,
            asset_repo=FakeAssetRepo({"asset-a": 30.0, "asset-b": 30.0}),
            atom_clip_repo=atom_repo,
        )
        result = svc.generate_from_template(
            template=_template(),
            clip_configs=_clip_configs(3),
            asset_ids=["asset-a", "asset-b"],
            created_by_user_id="user-1",
        )
        clips = result["clips"]
        assert len(clips) == 3
        # 每个 clip 都绑定了原子片段
        atom_ids = [c.atom_clip_id for c in clips]
        assert all(atom_ids)
        # 同一原子片段一个视频只用一次
        assert len(set(atom_ids)) == 3
        # start_time/duration 与选中片段一致
        for c in clips:
            assert c.start_time >= 0
            assert 0 < c.duration <= 6.0 + 0.01
        # asset_id 与 atom_clip 归属一致
        for c in clips:
            assert c.asset_id.startswith("asset-")

    def test_fallback_when_atom_clips_not_ready(self, db_session):
        """素材没有 atom_clips 时内存兜底切片，仍能选出片段。"""
        atom_repo = SQLAlchemyAssetAtomClipRepository(db_session)
        svc = PlanGeneratorService(
            db_session,
            asset_repo=FakeAssetRepo({"old-asset": 20.0}),
            atom_clip_repo=atom_repo,
        )
        result = svc.generate_from_template(
            template=_template(),
            clip_configs=_clip_configs(3),
            asset_ids=["old-asset"],
            created_by_user_id="user-1",
        )
        clips = result["clips"]
        # 兜底片段不落库、无持久 ID，clip 不绑定 atom_clip_id（回退旧路径）或绑定运行时 ID
        # 关键：必须成功选出素材，不报错
        assert all(c.asset_id == "old-asset" for c in clips)

    def test_preview_mode_keeps_legacy_path(self, db_session):
        """随机预览模式走旧路径，不要求 atom clips。"""
        atom_repo = SQLAlchemyAssetAtomClipRepository(db_session)
        svc = PlanGeneratorService(
            db_session,
            asset_repo=FakeAssetRepo({"asset-a": 30.0, "asset-b": 30.0, "asset-c": 30.0}),
            atom_clip_repo=atom_repo,
        )
        result = svc.generate_from_template(
            template=_template(),
            clip_configs=_clip_configs(3),
            asset_ids=["asset-a", "asset-b", "asset-c"],
            created_by_user_id="user-1",
            random_preview=True,
        )
        clips = result["clips"]
        assert len(clips) == 3
        assert {c.asset_id for c in clips} == {"asset-a", "asset-b", "asset-c"}
        # 预览路径不绑定 atom_clip_id
        assert all(not c.atom_clip_id for c in clips)

    def test_no_atom_repo_uses_legacy_path(self, db_session):
        """未注入 atom_clip_repo（旧调用方）时行为不变。"""
        svc = PlanGeneratorService(
            db_session,
            asset_repo=FakeAssetRepo({"asset-a": 30.0, "asset-b": 30.0, "asset-c": 30.0}),
        )
        result = svc.generate_from_template(
            template=_template(),
            clip_configs=_clip_configs(3),
            asset_ids=["asset-a", "asset-b", "asset-c"],
            created_by_user_id="user-1",
        )
        clips = result["clips"]
        assert len(clips) == 3
        assert {c.asset_id for c in clips} == {"asset-a", "asset-b", "asset-c"}