xiaoxia-saas/tests/unit/test_fallback_user_assets.py

"""Unit tests for PR #1338: 确认生成兜底增强 — user_id 查找素材.

覆盖:
- SQLAlchemyAssetRepository.find_ready_videos_by_user
- _auto_fallback_auto_material_mode 策略2 (user_id 兜底)
"""

from __future__ import annotations

import sys
from pathlib import Path
from unittest.mock import MagicMock

import pytest

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

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.adapters.sqlalchemy_impl.models import Base
from packages.domain import Asset, AssetStatus


def _make_repo():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    return SQLAlchemyAssetRepository(session)


class TestFindReadyVideosByUser:
    """SQLAlchemyAssetRepository.find_ready_videos_by_user 测试."""

    def test_returns_ready_videos_for_user(self):
        repo = _make_repo()
        user_id = "user-abc-123"
        v1 = Asset.create(
            project_id="",
            library_id="lib-1",
            name="video1.mp4",
            storage_key="v/v1.mp4",
            mime_type="video/mp4",
            status=AssetStatus.READY,
            uploaded_by_user_id=user_id,
        )
        v2 = Asset.create(
            project_id="",
            library_id="lib-1",
            name="video2.mp4",
            storage_key="v/v2.mp4",
            mime_type="video/mp4",
            status=AssetStatus.READY,
            uploaded_by_user_id=user_id,
        )
        repo.create(v1)
        repo.create(v2)
        result = repo.find_ready_videos_by_user(user_id)
        assert len(result) == 2
        assert {a.id for a in result} == {v1.id, v2.id}

    def test_excludes_non_video_assets(self):
        repo = _make_repo()
        user_id = "user-abc-123"
        video = Asset.create(
            project_id="",
            library_id="lib-1",
            name="clip.mp4",
            storage_key="v/clip.mp4",
            mime_type="video/mp4",
            status=AssetStatus.READY,
            uploaded_by_user_id=user_id,
        )
        image = Asset.create(
            project_id="",
            library_id="lib-1",
            name="photo.jpg",
            storage_key="v/photo.jpg",
            mime_type="image/jpeg",
            status=AssetStatus.READY,
            uploaded_by_user_id=user_id,
        )
        repo.create(video)
        repo.create(image)
        result = repo.find_ready_videos_by_user(user_id)
        assert len(result) == 1
        assert result[0].id == video.id

    def test_excludes_non_ready_assets(self):
        repo = _make_repo()
        user_id = "user-abc-123"
        ready = Asset.create(
            project_id="",
            library_id="lib-1",
            name="ready.mp4",
            storage_key="v/ready.mp4",
            mime_type="video/mp4",
            status=AssetStatus.READY,
            uploaded_by_user_id=user_id,
        )
        uploading = Asset.create(
            project_id="",
            library_id="lib-1",
            name="uploading.mp4",
            storage_key="v/uploading.mp4",
            mime_type="video/mp4",
            status=AssetStatus.UPLOADING,
            uploaded_by_user_id=user_id,
        )
        repo.create(ready)
        repo.create(uploading)
        result = repo.find_ready_videos_by_user(user_id)
        assert len(result) == 1
        assert result[0].id == ready.id

    def test_excludes_other_users_assets(self):
        repo = _make_repo()
        my_video = Asset.create(
            project_id="",
            library_id="lib-1",
            name="mine.mp4",
            storage_key="v/mine.mp4",
            mime_type="video/mp4",
            status=AssetStatus.READY,
            uploaded_by_user_id="user-A",
        )
        other_video = Asset.create(
            project_id="",
            library_id="lib-1",
            name="other.mp4",
            storage_key="v/other.mp4",
            mime_type="video/mp4",
            status=AssetStatus.READY,
            uploaded_by_user_id="user-B",
        )
        repo.create(my_video)
        repo.create(other_video)
        result = repo.find_ready_videos_by_user("user-A")
        assert len(result) == 1
        assert result[0].id == my_video.id

    def test_empty_result_for_unknown_user(self):
        repo = _make_repo()
        result = repo.find_ready_videos_by_user("nonexistent-user")
        assert result == []

    def test_respects_limit(self):
        repo = _make_repo()
        user_id = "user-abc-123"
        for i in range(10):
            asset = Asset.create(
                project_id="",
                library_id="lib-1",
                name=f"video_{i}.mp4",
                storage_key=f"v/v{i}.mp4",
                mime_type="video/mp4",
                status=AssetStatus.READY,
                uploaded_by_user_id=user_id,
            )
            repo.create(asset)
        result = repo.find_ready_videos_by_user(user_id, limit=3)
        assert len(result) == 3


class TestAutoFallbackAutoMaterialModeUserId:
    """_auto_fallback_auto_material_mode user_id 兜底策略测试."""

    def _make_plan_check(self, project_id="", template_id="tmpl-1"):
        plan = MagicMock()
        plan.project_id = project_id
        plan.template_id = template_id
        plan.config = {}
        return plan

    def _make_clip(self, clip_id="clip-1"):
        clip = MagicMock()
        clip.id = clip_id
        clip.asset_id = ""
        return clip

    def test_skips_when_no_clips_without_asset(self):
        from app.api.routes.templates_editor._fallback import (
            _auto_fallback_auto_material_mode,
        )

        svc = MagicMock()
        plan_check = self._make_plan_check()
        _auto_fallback_auto_material_mode(
            svc,
            "plan-1",
            plan_check,
            [],
            MagicMock(),
            MagicMock(),
            user_id="user-1",
        )
        svc.assign_asset.assert_not_called()

    def test_strategy2_user_id_fallback(self):
        from app.api.routes.templates_editor._fallback import (
            _auto_fallback_auto_material_mode,
        )

        svc = MagicMock()
        plan_check = self._make_plan_check(project_id="")
        clip = self._make_clip("clip-1")
        mock_asset = MagicMock()
        mock_asset.id = "asset-from-user"
        mock_asset.status = AssetStatus.READY
        mock_asset.mime_type = "video/mp4"
        asset_repo = MagicMock()
        asset_repo.find_ready_videos_by_user.return_value = [mock_asset]
        asset_library_repo = MagicMock()
        _auto_fallback_auto_material_mode(
            svc,
            "plan-1",
            plan_check,
            [clip],
            asset_library_repo,
            asset_repo,
            user_id="user-123",
        )
        asset_repo.find_ready_videos_by_user.assert_called_once_with("user-123")
        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-user")

    def test_strategy1_takes_priority_over_strategy2(self):
        from app.api.routes.templates_editor._fallback import (
            _auto_fallback_auto_material_mode,
        )

        svc = MagicMock()
        plan_check = self._make_plan_check(project_id="proj-1")
        clip = self._make_clip("clip-1")
        mock_lib = MagicMock()
        mock_lib.id = "lib-video"
        mock_lib.kind = MagicMock()
        mock_lib.kind.value = "video"
        mock_asset = MagicMock()
        mock_asset.id = "asset-from-project"
        mock_asset.status = "ready"
        mock_asset.mime_type = "video/mp4"
        asset_library_repo = MagicMock()
        asset_library_repo.find_by_project.return_value = [mock_lib]
        asset_repo = MagicMock()
        asset_repo.find_by_library.return_value = [mock_asset]
        _auto_fallback_auto_material_mode(
            svc,
            "plan-1",
            plan_check,
            [clip],
            asset_library_repo,
            asset_repo,
            user_id="user-123",
        )
        asset_library_repo.find_by_project.assert_called_once_with("proj-1")
        asset_repo.find_ready_videos_by_user.assert_not_called()
        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-project")

    def test_falls_back_when_project_has_no_videos(self):
        from app.api.routes.templates_editor._fallback import (
            _auto_fallback_auto_material_mode,
        )

        svc = MagicMock()
        plan_check = self._make_plan_check(project_id="proj-1")
        clip = self._make_clip("clip-1")
        asset_library_repo = MagicMock()
        asset_library_repo.find_by_project.return_value = []
        mock_asset = MagicMock()
        mock_asset.id = "asset-from-user"
        mock_asset.status = AssetStatus.READY
        mock_asset.mime_type = "video/mp4"
        asset_repo = MagicMock()
        asset_repo.find_ready_videos_by_user.return_value = [mock_asset]
        _auto_fallback_auto_material_mode(
            svc,
            "plan-1",
            plan_check,
            [clip],
            asset_library_repo,
            asset_repo,
            user_id="user-123",
        )
        asset_repo.find_ready_videos_by_user.assert_called_once_with("user-123")
        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-user")

    def test_no_assets_found_does_nothing(self):
        from app.api.routes.templates_editor._fallback import (
            _auto_fallback_auto_material_mode,
        )

        svc = MagicMock()
        plan_check = self._make_plan_check(project_id="")
        clip = self._make_clip("clip-1")
        asset_repo = MagicMock()
        asset_repo.find_ready_videos_by_user.return_value = []
        asset_library_repo = MagicMock()
        _auto_fallback_auto_material_mode(
            svc,
            "plan-1",
            plan_check,
            [clip],
            asset_library_repo,
            asset_repo,
            user_id="user-123",
        )
        svc.assign_asset.assert_not_called()

    def test_no_user_id_skips_strategy2(self):
        from app.api.routes.templates_editor._fallback import (
            _auto_fallback_auto_material_mode,
        )

        svc = MagicMock()
        plan_check = self._make_plan_check(project_id="")
        clip = self._make_clip("clip-1")
        asset_repo = MagicMock()
        asset_library_repo = MagicMock()
        _auto_fallback_auto_material_mode(
            svc,
            "plan-1",
            plan_check,
            [clip],
            asset_library_repo,
            asset_repo,
            user_id="",
        )
        asset_repo.find_ready_videos_by_user.assert_not_called()
        svc.assign_asset.assert_not_called()