life-echo/api/tests/test_process_memoir_segments_image_enqueue.py

import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch

from api.tasks.memoir_tasks import MemoirImageSettings, process_memoir_segments


class ProcessMemoirSegmentsImageEnqueueTest(unittest.TestCase):
    @patch("api.tasks.memoir_tasks._update_task_status_sync")
    @patch("api.tasks.memoir_tasks._release_chapter_lock")
    @patch("api.tasks.memoir_tasks._acquire_chapter_lock", return_value=True)
    @patch("api.tasks.memoir_tasks._update_slot_sync")
    @patch("api.tasks.memoir_tasks._classify_chapter_category", return_value="childhood")
    @patch("api.tasks.memoir_tasks._get_or_create_state_sync")
    @patch("api.tasks.memoir_tasks.llm_service.get_llm")
    @patch("api.tasks.memoir_tasks.generate_chapter_images.delay")
    @patch("api.tasks.memoir_tasks.SessionLocal")
    @patch("api.tasks.memoir_tasks.MemoirImageSettings.from_env")
    def test_process_memoir_segments_parses_markdown_wrapped_state_extraction_json(
        self,
        settings_from_env,
        session_local_cls,
        delay_mock,
        get_llm_mock,
        get_state_mock,
        _classify_mock,
        update_slot_mock,
        _acquire_lock_mock,
        _release_lock_mock,
        _update_status_mock,
    ):
        settings_from_env.return_value = MemoirImageSettings(
            enabled=True,
            max_per_chapter=2,
            provider="liblib",
            default_style="watercolor",
            default_size="1024x1024",
            poll_interval_seconds=3,
            max_attempts=20,
            liblib_template_uuid="tpl-uuid",
        )
        get_state_mock.return_value = SimpleNamespace(current_stage="childhood", slots={})
        update_slot_mock.return_value = SimpleNamespace(current_stage="childhood", slots={})
        llm = Mock()
        llm.invoke.side_effect = [
            SimpleNamespace(
                content="""```json
{
  "detected_stage": "childhood",
  "slots": {
    "family_memory": "外婆总在门口等我"
  }
}
```"""
            ),
            SimpleNamespace(content="童年的门前"),
            SimpleNamespace(content="新的章节正文\n\n{{IMAGE:南方小镇的青石板路}}"),
        ]
        get_llm_mock.return_value = llm

        segment = SimpleNamespace(
            id="segment-1",
            transcript_text="那条路我一直记得。\n\n{{{{IMAGE:南方小镇的青石板路}}}}",
            processed=False,
        )

        segments_result = Mock()
        segments_result.scalars.return_value.all.return_value = [segment]

        chapter_result = Mock()
        chapter_result.scalar_one_or_none.return_value = None

        book_result = Mock()
        book_result.scalar_one_or_none.return_value = None

        db = Mock()
        db.execute.side_effect = [segments_result, chapter_result, book_result]
        db.get.return_value = None
        session_local_cls.return_value = db

        events: list[str] = []
        db.commit.side_effect = lambda: events.append("commit")
        delay_mock.side_effect = lambda chapter_id: events.append(f"delay:{chapter_id}")

        task_self = SimpleNamespace(
            request=SimpleNamespace(id="task-1"),
            retry=Mock(side_effect=AssertionError("retry should not be called")),
        )
        process_memoir_segments.run.__func__(task_self, "user-1", ["segment-1"])

        update_slot_mock.assert_called_once_with(
            user_id="user-1",
            stage="childhood",
            slot_name="family_memory",
            snippet="外婆总在门口等我",
            segment_ids=["segment-1"],
            db=db,
        )
        self.assertIn("commit", events)
        delay_events = [event for event in events if event.startswith("delay:")]
        self.assertEqual(len(delay_events), 1)
        self.assertGreater(events.index(delay_events[0]), events.index("commit"))

    @patch("api.tasks.memoir_tasks._update_task_status_sync")
    @patch("api.tasks.memoir_tasks._release_chapter_lock")
    @patch("api.tasks.memoir_tasks._acquire_chapter_lock", return_value=True)
    @patch("api.tasks.memoir_tasks._classify_chapter_category", return_value="childhood")
    @patch("api.tasks.memoir_tasks._get_or_create_state_sync")
    @patch("api.tasks.memoir_tasks.llm_service.get_llm", return_value=None)
    @patch("api.tasks.memoir_tasks.generate_chapter_images.delay")
    @patch("api.tasks.memoir_tasks.SessionLocal")
    @patch("api.tasks.memoir_tasks.MemoirImageSettings.from_env")
    def test_process_memoir_segments_does_not_enqueue_image_jobs_when_feature_disabled(
        self,
        settings_from_env,
        session_local_cls,
        delay_mock,
        _get_llm,
        get_state_mock,
        _classify_mock,
        _acquire_lock_mock,
        _release_lock_mock,
        _update_status_mock,
    ):
        settings_from_env.return_value = MemoirImageSettings(
            enabled=False,
            max_per_chapter=2,
            provider="liblib",
            default_style="watercolor",
            default_size="1024x1024",
            poll_interval_seconds=3,
            max_attempts=20,
            liblib_template_uuid="tpl-uuid",
        )
        get_state_mock.return_value = SimpleNamespace(current_stage="childhood", slots={})

        segment = SimpleNamespace(
            id="segment-1",
            transcript_text="那条路我一直记得。\n\n{{{{IMAGE:南方小镇的青石板路}}}}",
            processed=False,
        )

        segments_result = Mock()
        segments_result.scalars.return_value.all.return_value = [segment]

        chapter_result = Mock()
        chapter_result.scalar_one_or_none.return_value = None

        book_result = Mock()
        book_result.scalar_one_or_none.return_value = None

        db = Mock()
        db.execute.side_effect = [segments_result, chapter_result, book_result]
        db.get.return_value = None
        session_local_cls.return_value = db

        task_self = SimpleNamespace(
            request=SimpleNamespace(id="task-1"),
            retry=Mock(side_effect=AssertionError("retry should not be called")),
        )
        process_memoir_segments.run.__func__(task_self, "user-1", ["segment-1"])

        delay_mock.assert_not_called()