MoviePilot/tests/test_search_recommend_execution.py

"""
AI 搜索推荐执行与代际栅栏回归测试。
"""

import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock

from app.application.configuration import ChainRuntimeConfig
from app.chain.search import SearchChain
from app.chain.search import recommend as recommend_module
from app.chain.search.recommend import recommend_coordinator
from app.runtime.tasks import TaskRegistry


def _result(title: str):
    """
    构造一条具有稳定推荐身份的候选资源。
    """
    return SimpleNamespace(
        torrent_info=SimpleNamespace(
            site=1,
            torrent_id=title,
            title=title,
            description="",
            size=1024,
            seeders=1,
            category="电影",
        )
    )


def _chain() -> SearchChain:
    """
    构造只包含推荐执行所需依赖的搜索链。
    """
    chain = object.__new__(SearchChain)
    chain.runtime_config = ChainRuntimeConfig(
        media_extensions=(),
        ai_agent_enable=True,
        ai_recommend_enabled=True,
        ai_recommend_max_items=50,
    )
    chain.remove_cache = lambda _filename: None
    chain.async_save_cache = AsyncMock()
    return chain


def test_request_hash_distinguishes_same_count_candidates():
    """
    候选数量相同但资源身份不同时不得复用同一推荐请求。
    """
    first = SearchChain._calculate_recommend_request_hash(None, 1, [_result("资源一")])
    second = SearchChain._calculate_recommend_request_hash(None, 1, [_result("资源二")])

    assert first != second


def test_execution_helpers_are_not_exported_on_search_facade():
    """
    推荐执行步骤应留在 owner 内部，不扩大插件可见的 SearchChain 门面。
    """
    assert "_generate_recommend_indices" not in SearchChain.__dict__
    assert "_run_recommend" not in SearchChain.__dict__
    assert "_parse_recommend_response" not in SearchChain.__dict__
    assert "_recommend_prompt" not in SearchChain.__dict__


def test_empty_candidates_publish_error_and_finish(monkeypatch):
    """
    没有有效候选时应发布错误并结束任务，不调用模型或写缓存。
    """

    async def scenario():
        recommend_coordinator.cancel()
        registry = TaskRegistry()
        monkeypatch.setattr(recommend_module, "get_task_registry", lambda: registry)
        chain = _chain()
        chain._invoke_recommend_llm = AsyncMock(side_effect=AssertionError("空候选不应调用模型"))
        try:
            chain.start_recommend_task(
                filtered_indices=None,
                search_results_count=1,
                results=[SimpleNamespace(torrent_info=None)],
            )
            task = recommend_coordinator.snapshot().task
            assert task is not None
            await task

            state = recommend_coordinator.snapshot()
            assert state.error == "没有可用于AI推荐的资源"
            assert not state.running
            assert state.task is None
            chain._invoke_recommend_llm.assert_not_awaited()
            chain.async_save_cache.assert_not_awaited()
        finally:
            recommend_coordinator.cancel()

    asyncio.run(scenario())


def test_stale_generation_cannot_publish_or_cache(monkeypatch):
    """
    旧任务即使忽略取消并返回结果，也不得覆盖新代际状态和缓存。
    """

    async def scenario():
        recommend_coordinator.cancel()
        registry = TaskRegistry()
        monkeypatch.setattr(recommend_module, "get_task_registry", lambda: registry)
        chain = _chain()
        first_started = asyncio.Event()
        release_first = asyncio.Event()
        saved = []
        call_count = 0

        async def invoke(_text: str) -> str:
            """
            让第一代任务跨越第二代完成点后再返回。
            """
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                first_started.set()
                try:
                    await release_first.wait()
                except asyncio.CancelledError:
                    await release_first.wait()
            return "[0]"

        async def save_cache(payload, filename):
            """
            记录真正越过代际栅栏的缓存写入。
            """
            saved.append((filename, payload))

        chain._invoke_recommend_llm = invoke
        chain.async_save_cache = save_cache
        try:
            chain.start_recommend_task(None, 1, [_result("旧资源")])
            first_task = recommend_coordinator.snapshot().task
            assert first_task is not None
            await asyncio.wait_for(first_started.wait(), timeout=1)

            chain.start_recommend_task(None, 1, [_result("新资源")])
            second_state = recommend_coordinator.snapshot()
            second_task = second_state.task
            assert second_task is not None
            await second_task

            release_first.set()
            await first_task

            state = recommend_coordinator.snapshot()
            assert state.request_hash == second_state.request_hash
            assert state.result == [0]
            assert not state.running
            assert state.task is None
            assert saved == [
                (
                    "__ai_recommend_indices__",
                    {"request_hash": second_state.request_hash, "results": [0]},
                )
            ]
        finally:
            release_first.set()
            task = recommend_coordinator.cancel()
            if task and not task.done():
                task.cancel()
                await task

    asyncio.run(scenario())