xiaoxia-saas/tests/unit/test_three_bugs_fix.py

"""Regression tests for 3 bug fixes: flush, append_log, plan_id fallback."""

from __future__ import annotations

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

import pytest

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


# ── Bug 1: db.flush() before pending query ─────────────────────────────────

class TestReplaceAllClipsFlush:
    """replace_all_clips_transactional must flush before querying pending clips."""

    def _make_svc_and_db(self, pending_results):
        """Helper: create service + db mock. pending_results = list returned by pending query."""
        from app.services.edit_plan_service import EditPlanService

        db = MagicMock()
        # Delete query
        delete_query = MagicMock()
        delete_query.filter.return_value.delete.return_value = 0
        # Pending query: single .filter() with multiple conditions
        ready_query = MagicMock()
        ready_query.filter.return_value.all.return_value = pending_results
        db.query.side_effect = [delete_query, ready_query]
        return db, EditPlanService

    def _setup_clip_mocks(self, mock_clip_cls, mock_model_cls, asset_id="asset-1"):
        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_id
        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_model_cls.return_value = MagicMock()
        return mock_entity

    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
    @patch("app.services.edit_plan_service.EditPlanClip")
    def test_flush_called_between_add_and_query(self, mock_clip_cls, mock_model_cls):
        """db.flush() must be called after db.add() and before the pending query."""
        self._setup_clip_mocks(mock_clip_cls, mock_model_cls)
        db, SvcClass = self._make_svc_and_db([])

        clip_repo = MagicMock()
        clip_repo.session = db
        svc = SvcClass.__new__(SvcClass)
        svc._clip_repo = clip_repo

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

        db.flush.assert_called_once()
        # Verify ordering: add → flush → query → commit
        method_names = [c[0] for c in db.method_calls]
        add_idx = method_names.index("add")
        flush_idx = method_names.index("flush")
        commit_idx = method_names.index("commit")
        assert add_idx < flush_idx < commit_idx

    @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel")
    @patch("app.services.edit_plan_service.EditPlanClip")
    def test_flush_marks_new_clips_ready(self, mock_clip_cls, mock_model_cls):
        """After flush, new clips with asset_id are found and marked ready."""
        self._setup_clip_mocks(mock_clip_cls, mock_model_cls)

        # Use a plain object so we can verify attribute mutation
        class FakeClip:
            status = "pending"

        pending_clip = FakeClip()
        db, SvcClass = self._make_svc_and_db([pending_clip])

        clip_repo = MagicMock()
        clip_repo.session = db
        svc = SvcClass.__new__(SvcClass)
        svc._clip_repo = clip_repo

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

        assert pending_clip.status == "ready"


# ── Bug 2: append_log no TypeError ─────────────────────────────────────────

class TestAppendLogNoConflict:
    """append_log must not receive duplicate 'stage' parameter."""

    def _make_task(self):
        from packages.domain.generation_task import GenerationTask
        return GenerationTask.create(
            project_id="proj-1",
            asset_library_id="lib-1",
            strategy_id="one_take",
            template_id="tpl-1",
            asset_ids=["a1"],
            created_by_user_id="user-1",
        )

    def test_append_log_with_stage_as_first_positional(self):
        """append_log(stage, message, ...) works correctly."""
        task = self._make_task()
        task.append_log("render", "some error", level="ERROR", error_type="RuntimeError")

        entries = json.loads(task.logs)
        assert len(entries) == 1
        assert entries[0]["stage"] == "render"
        assert entries[0]["message"] == "some error"
        assert entries[0]["level"] == "ERROR"
        assert entries[0]["error_type"] == "RuntimeError"

    def test_duplicate_stage_raises_type_error(self):
        """Sanity check: passing stage both positionally and as kwarg raises TypeError."""
        task = self._make_task()
        with pytest.raises(TypeError):
            task.append_log(
                "任务失败",  # positional → stage
                "some error",
                level="ERROR",
                stage="render",  # duplicate → TypeError
            )


# ── Bug 3: plan_id fallback in create_generation_task ──────────────────────

class TestPlanIdFallback:
    """Formal generation API should fallback to find plan by template_id + user_id."""

    def test_fallback_code_present_in_source(self):
        """Verify the fallback logic is present in the generation_tasks module."""
        import inspect
        from apps.api.app.api.routes import generation_tasks

        source = inspect.getsource(generation_tasks.create_generation_task)
        assert "SQLAlchemyEditPlanRepository" in source
        assert "list_by_template" in source
        assert "兜底关联编辑计划" in source
        assert "自动关联编辑计划" in source

    def test_fallback_only_runs_when_source_edit_plan_id_empty(self):
        """Verify the condition checks for empty source_edit_plan_id."""
        import inspect
        from apps.api.app.api.routes import generation_tasks

        source = inspect.getsource(generation_tasks.create_generation_task)
        assert "not task.source_edit_plan_id and request.template_id" in source

    def test_list_by_template_method_exists(self):
        """Verify SQLAlchemyEditPlanRepository.list_by_template is callable."""
        from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
            SQLAlchemyEditPlanRepository,
        )

        mock_db = MagicMock()
        mock_session = MagicMock()
        mock_db.query.return_value = mock_session
        mock_session.filter.return_value = mock_session
        mock_session.order_by.return_value = mock_session
        mock_session.offset.return_value = mock_session
        mock_session.limit.return_value.all.return_value = []

        repo = SQLAlchemyEditPlanRepository(mock_db)
        result = repo.list_by_template("tpl-1", limit=20)
        assert result == []