xiaoxia-saas/tests/unit/test_replace_all_clips_transactional.py

"""Tests for EditPlanService.replace_all_clips_transactional."""

from __future__ import annotations

import os
from unittest.mock import MagicMock, PropertyMock, patch

import pytest

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")


class TestReplaceAllClipsTransactional:
    """事务性替换片段方法测试。"""

    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
    @patch("app.services.edit_plan_service.EditPlanClip")
    def test_success_commits_once(self, mock_clip_cls, mock_model_cls):
        """成功时单次 commit，不 rollback。"""
        from app.services.edit_plan_service import EditPlanService

        db = MagicMock()
        # Mock query chain for delete
        query_mock = MagicMock()
        query_mock.filter.return_value.delete.return_value = 3
        db.query.return_value = query_mock

        # Mock query chain for mark_ready (pending_with_asset)
        # After the create loop, query returns empty list (no pending clips with asset)
        ready_query = MagicMock()
        ready_query.filter.return_value.filter.return_value.filter.return_value.all.return_value = []
        db.query.side_effect = [query_mock, ready_query]

        # Mock EditPlanClip.create to return a mock entity
        mock_entity = MagicMock()
        mock_entity.id = "clip-1"
        mock_entity.plan_id = "plan-1"
        mock_entity.clip_type = "main"
        mock_entity.order = 0
        mock_entity.asset_id = "asset-1"
        mock_entity.text_content = ""
        mock_entity.start_time = 0.0
        mock_entity.duration = 3.0
        mock_entity.transition_effect = "cut"
        mock_entity.transition_duration = 0.0
        mock_entity.playback_speed = 1.0
        mock_entity.status.value = "pending"
        mock_entity.config = {}
        mock_clip_cls.create.return_value = mock_entity

        # Mock the model constructor
        mock_model_instance = MagicMock()
        mock_model_cls.return_value = mock_model_instance

        # Mock clip_repo
        clip_repo = MagicMock()
        clip_repo.session = db

        svc = EditPlanService.__new__(EditPlanService)
        svc._clip_repo = clip_repo

        result = svc.replace_all_clips_transactional(
            "plan-1",
            [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}],
        )

        assert result == 1
        db.commit.assert_called_once()
        db.rollback.assert_not_called()
        db.add.assert_called_once_with(mock_model_instance)

    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
    @patch("app.services.edit_plan_service.EditPlanClip")
    def test_failure_rolls_back(self, mock_clip_cls, mock_model_cls):
        """异常时自动 rollback。"""
        from app.services.edit_plan_service import EditPlanService

        db = MagicMock()
        query_mock = MagicMock()
        query_mock.filter.return_value.delete.return_value = 0
        db.query.return_value = query_mock

        # Simulate failure during create
        mock_clip_cls.create.side_effect = ValueError("模拟异常")

        clip_repo = MagicMock()
        clip_repo.session = db

        svc = EditPlanService.__new__(EditPlanService)
        svc._clip_repo = clip_repo

        with pytest.raises(ValueError, match="模拟异常"):
            svc.replace_all_clips_transactional(
                "plan-1",
                [{"asset_id": "bad", "start_time": 0.0, "duration": 1.0, "order": 0}],
            )

        db.rollback.assert_called_once()
        db.commit.assert_not_called()

    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
    @patch("app.services.edit_plan_service.EditPlanClip")
    def test_order_defaults_to_index(self, mock_clip_cls, mock_model_cls):
        """order=0 时使用索引值作为 order。"""
        from app.services.edit_plan_service import EditPlanService

        db = MagicMock()
        query_mock = MagicMock()
        query_mock.filter.return_value.delete.return_value = 0
        db.query.return_value = query_mock

        ready_query = MagicMock()
        ready_query.filter.return_value.filter.return_value.filter.return_value.all.return_value = []
        db.query.side_effect = [query_mock, ready_query]

        mock_entity = MagicMock()
        mock_entity.id = "clip-1"
        mock_entity.plan_id = "plan-1"
        mock_entity.clip_type = "main"
        mock_entity.order = 0  # order=0 → 使用 i=0
        mock_entity.asset_id = "a1"
        mock_entity.text_content = ""
        mock_entity.start_time = 0.0
        mock_entity.duration = 1.0
        mock_entity.transition_effect = "cut"
        mock_entity.transition_duration = 0.0
        mock_entity.playback_speed = 1.0
        mock_entity.status.value = "pending"
        mock_entity.config = {}
        mock_clip_cls.create.return_value = mock_entity

        mock_model_cls.return_value = MagicMock()

        clip_repo = MagicMock()
        clip_repo.session = db

        svc = EditPlanService.__new__(EditPlanService)
        svc._clip_repo = clip_repo

        svc.replace_all_clips_transactional(
            "plan-1",
            [{"asset_id": "a1", "start_time": 0.0, "duration": 1.0, "order": 0}],
        )

        # order=0 → falsy → use index i=0
        create_call = mock_clip_cls.create.call_args
        assert create_call.kwargs["order"] == 0