MoviePilot/tests/test_search_sync_async_parity.py

"""
搜索链同步、异步与流式入口的共享业务决策回归测试。
"""

import asyncio
from copy import deepcopy
from types import SimpleNamespace
from typing import Any

import pytest

from app.chain.search import execution as execution_module
from app.chain.search import media as media_module
from app.chain.search import title as title_module
from app.chain.search.facade import SearchChain
from app.domain.context import MediaInfo, TorrentInfo
from app.domain.metainfo import MetaInfo
from app.schemas.types import MediaSource, MediaType


def _media() -> MediaInfo:
    """
    构造具备稳定来源身份和别名的测试媒体。
    """
    return MediaInfo(
        media_source=MediaSource.TMDB,
        media_id="100",
        tmdb_id=100,
        title="测试电影",
        names=["Test Movie"],
        type=MediaType.MOVIE,
        year="2026",
    )


def _torrent(title: str) -> TorrentInfo:
    """
    构造一条无需外部 provider 的测试资源。
    """
    return TorrentInfo(title=title, description="description")


def _chain() -> SearchChain:
    """
    绕过运行时组合根构造可局部注入的搜索链。
    """
    return object.__new__(SearchChain)


def test_id_search_sync_async_share_request_and_missing_plan(monkeypatch):
    """
    ID 搜索双入口应产生完全相同的识别、缓存和缺集参数。
    """
    chain = _chain()
    target = _media()
    sync_calls: list[dict[str, Any]] = []
    async_calls: list[dict[str, Any]] = []
    saved_sync: list[dict[str, Any]] = []
    saved_async: list[dict[str, Any]] = []
    result = [SimpleNamespace(name="context")]

    class FakeMediaChain:
        """
        记录同步与异步识别参数并返回同一媒体快照。
        """

        def recognize_media(self, **kwargs):
            """
            返回同步识别结果。
            """
            sync_calls.append(kwargs)
            return target

        async def async_recognize_media(self, **kwargs):
            """
            返回异步识别结果。
            """
            async_calls.append(kwargs)
            return target

    def process(**kwargs):
        """
        记录同步处理计划。
        """
        sync_calls.append(kwargs)
        return result

    async def async_process(**kwargs):
        """
        记录异步处理计划。
        """
        async_calls.append(kwargs)
        return result

    async def async_save_params(**kwargs):
        """
        记录异步缓存参数。
        """
        saved_async.append(kwargs)

    async def async_save_results(_contexts):
        """
        模拟异步结果缓存。
        """

    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
    chain.cancel_ai_recommend = lambda: None
    chain.save_last_search_params = lambda **kwargs: saved_sync.append(kwargs)
    chain.async_save_last_search_params = async_save_params
    chain.process = process
    chain.async_process = async_process
    chain._save_results = lambda _contexts: None
    chain._async_save_results = async_save_results

    kwargs = {
        "media_source": MediaSource.TMDB,
        "media_id": "100",
        "mtype": MediaType.TV,
        "area": "imdbid",
        "season": 2,
        "sites": [1, 3],
        "cache_local": True,
        "music_type": "recording",
    }
    sync_result = chain.search_by_id(**kwargs)
    async_result = asyncio.run(chain.async_search_by_id(**kwargs))

    assert sync_result is result
    assert async_result is result
    assert sync_calls[0] == async_calls[0] == {
        "media_source": MediaSource.TMDB,
        "media_id": "100",
        "mtype": MediaType.TV,
        "music_type": "recording",
    }
    assert sync_calls[1] == async_calls[1]
    assert sync_calls[1]["sites"] == [1, 3]
    assert sync_calls[1]["area"] == "imdbid"
    missing = sync_calls[1]["no_exists"]
    assert list(missing) == ["tmdb:100"]
    assert list(missing["tmdb:100"]) == [2]
    assert saved_sync == saved_async


def test_imdb_search_falls_back_to_media_title_without_imdb_id():
    """
    IMDb 搜索缺少 IMDb ID 且没有调用方关键字时应请求媒体标题。
    """
    target = _media()

    assert SearchChain._torrent_keyword(None, target, "imdbid") == target.title
    assert SearchChain._torrent_keyword("自定义标题", target, "imdbid") == "自定义标题"


def test_media_process_sync_async_share_keyword_stop_decision(monkeypatch):
    """
    双入口应按相同关键字顺序搜索，并在首个有效结果后同时停止。
    """
    chain = _chain()
    target = _media()
    found = _torrent("测试电影 2026 1080p")
    sync_keywords: list[str] = []
    async_keywords: list[str] = []
    sync_sleeps: list[int] = []
    async_sleeps: list[int] = []
    parsed: list[list[TorrentInfo]] = []

    class FakeMediaChain:
        """
        为同步与异步处理提供等价的附加信息结果。
        """

        def supplement_media_info(self, mediainfo):
            """
            返回同步附加信息。
            """
            return mediainfo

        async def async_supplement_media_info(self, mediainfo):
            """
            返回异步附加信息。
            """
            return mediainfo

    def search_all_sites(**kwargs):
        """
        第二个关键字开始返回资源。
        """
        keyword = kwargs["keyword"]
        sync_keywords.append(keyword)
        return [found] if keyword == "second" else []

    async def async_search_all_sites(**kwargs):
        """
        第二个关键字开始异步返回资源。
        """
        keyword = kwargs["keyword"]
        async_keywords.append(keyword)
        return [found] if keyword == "second" else []

    def parse_result(**kwargs):
        """
        记录进入统一结果处理边界的资源顺序。
        """
        parsed.append(list(kwargs["torrents"]))
        return kwargs["torrents"]

    async def fake_async_sleep(delay):
        """
        记录异步关键字退避但不实际等待。
        """
        async_sleeps.append(delay)

    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
    monkeypatch.setattr(execution_module.random, "randint", lambda _start, _end: 1)
    monkeypatch.setattr(execution_module.time, "sleep", sync_sleeps.append)
    monkeypatch.setattr(execution_module.asyncio, "sleep", fake_async_sleep)
    chain.runtime_config = SimpleNamespace(search_multiple_name=False)
    chain._copy_media_input = deepcopy
    chain._prepare_params = lambda **_kwargs: (None, ["first", "second", "third"])
    chain._SearchChain__search_all_sites = search_all_sites
    chain._SearchChain__async_search_all_sites = async_search_all_sites
    chain._parse_result = parse_result

    sync_result = chain.process(target)
    async_result = asyncio.run(chain.async_process(target))

    assert sync_keywords == async_keywords == ["first", "second"]
    assert sync_sleeps == async_sleeps == [1]
    assert parsed == [[found], [found]]
    assert sync_result == async_result == [found]


def test_keyword_resolution_searches_all_names_when_enabled(monkeypatch):
    """
    开启多名称搜索时同步、异步共用状态机须完整执行并稳定聚合结果。
    """
    chain = _chain()
    target = _media()
    first = _torrent("First Result")
    third = _torrent("Third Result")
    results = {"first": [first], "second": [], "third": [third]}
    sync_keywords = []
    async_keywords = []

    def search(**kwargs):
        """
        记录同步站点调用顺序。
        """
        sync_keywords.append(kwargs["keyword"])
        return results[kwargs["keyword"]]

    async def async_search(**kwargs):
        """
        记录异步站点调用顺序。
        """
        async_keywords.append(kwargs["keyword"])
        return results[kwargs["keyword"]]

    async def supplement(mediainfo):
        """
        异步补充步骤在样例中不改变媒体对象。
        """
        return mediainfo

    async def sleep(_delay):
        """
        隔离关键词间隔，不执行实际等待。
        """

    provider = SimpleNamespace(supplement_media_info=lambda mediainfo: mediainfo,
                               async_supplement_media_info=supplement)
    monkeypatch.setattr(media_module, "MediaChain", lambda: provider)
    monkeypatch.setattr(execution_module.time, "sleep", lambda _delay: None)
    monkeypatch.setattr(execution_module.asyncio, "sleep", sleep)
    chain.runtime_config = SimpleNamespace(search_multiple_name=True)
    chain._prepare_params = lambda **_kwargs: (None, list(results))
    chain._SearchChain__search_all_sites = search
    chain._SearchChain__async_search_all_sites = async_search
    chain._parse_result = lambda **kwargs: list(kwargs["torrents"])

    assert chain.process(target) == [first, third]
    assert asyncio.run(chain.async_process(target)) == [first, third]
    assert sync_keywords == async_keywords == list(results)


def test_media_process_stream_shares_keyword_order_and_stop_decision(monkeypatch):
    """
    流入口应复用双入口的关键字顺序、退避次数和首结果短路决策。
    """
    chain = _chain()
    target = _media()
    found = _torrent("测试电影 2026 1080p")
    stream_keywords: list[str] = []
    sleeps: list[int] = []
    parsed: list[list[TorrentInfo]] = []

    class FakeMediaChain:
        """
        为流处理提供无需外部服务的附加信息结果。
        """

        async def async_supplement_media_info(self, mediainfo):
            """
            返回异步附加信息。
            """
            return mediainfo

    async def search_stream(**kwargs):
        """
        第二个关键字输出资源，供共享状态机执行短路。
        """
        keyword = kwargs["keyword"]
        stream_keywords.append(keyword)
        yield {
            "site_id": len(stream_keywords),
            "items": [found] if keyword == "second" else [],
        }

    async def fake_async_sleep(delay):
        """
        记录流关键字退避但不实际等待。
        """
        sleeps.append(delay)

    def parse_result(**kwargs):
        """
        记录流最终过滤前的资源顺序并返回空上下文。
        """
        parsed.append(list(kwargs["torrents"]))
        return []

    async def collect_events():
        """
        完整消费媒体搜索流。
        """
        return [
            event
            async for event in chain.async_process_stream(mediainfo=target)
        ]

    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
    monkeypatch.setattr(execution_module.random, "randint", lambda _start, _end: 1)
    monkeypatch.setattr(execution_module.asyncio, "sleep", fake_async_sleep)
    chain.runtime_config = SimpleNamespace(search_multiple_name=False)
    chain._copy_media_input = deepcopy
    chain._prepare_params = lambda **_kwargs: (None, ["first", "second", "third"])
    chain._SearchChain__async_search_all_sites_stream = search_stream
    chain._parse_result = parse_result

    events = asyncio.run(collect_events())

    assert stream_keywords == ["first", "second", "third"]
    assert sleeps == [1, 1]
    assert parsed == [[found]]
    assert [event["type"] for event in events] == [
        "append",
        "append",
        "progress",
        "append",
        "replace",
        "done",
    ]
    assert events[-1]["candidate_items"] == 1


def test_media_process_empty_keyword_plan_skips_all_provider_io(monkeypatch):
    """
    空关键字计划应由共享状态机直接完成，且三入口均不触发 provider。
    """
    chain = _chain()
    target = _media()
    parsed: list[list[TorrentInfo]] = []

    class FakeMediaChain:
        """
        为三入口提供稳定的媒体附加信息。
        """

        def supplement_media_info(self, mediainfo):
            """
            返回同步附加信息。
            """
            return mediainfo

        async def async_supplement_media_info(self, mediainfo):
            """
            返回异步附加信息。
            """
            return mediainfo

    def unexpected_provider(**_kwargs):
        """
        标记空计划错误触发了同步 provider。
        """
        raise AssertionError("空关键字计划不应触发同步 provider")

    async def unexpected_async_provider(**_kwargs):
        """
        标记空计划错误触发了异步 provider。
        """
        raise AssertionError("空关键字计划不应触发异步 provider")

    async def unexpected_stream_provider(**_kwargs):
        """
        标记空计划错误触发了流式 provider。
        """
        yield {"items": []}
        raise AssertionError("空关键字计划不应触发流式 provider")

    def parse_result(**kwargs):
        """
        记录三入口交付给结果解析边界的空资源。
        """
        parsed.append(list(kwargs["torrents"]))
        return []

    async def collect_events():
        """
        完整消费空关键字媒体搜索流。
        """
        return [
            event
            async for event in chain.async_process_stream(mediainfo=target)
        ]

    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
    chain.runtime_config = SimpleNamespace(search_multiple_name=False)
    chain._copy_media_input = deepcopy
    chain._prepare_params = lambda **_kwargs: (None, [])
    chain._SearchChain__search_all_sites = unexpected_provider
    chain._SearchChain__async_search_all_sites = unexpected_async_provider
    chain._SearchChain__async_search_all_sites_stream = unexpected_stream_provider
    chain._parse_result = parse_result

    assert chain.process(target) == []
    assert asyncio.run(chain.async_process(target)) == []
    events = asyncio.run(collect_events())

    assert parsed == [[], [], []]
    assert [event["type"] for event in events] == [
        "progress",
        "replace",
        "done",
    ]
    assert events[-1]["candidate_items"] == 0


def test_id_search_stream_failure_preserves_order_and_skips_result_cache(
    monkeypatch,
):
    """
    识别失败时流入口应先保存请求状态，只输出错误且不保存空结果。
    """
    chain = _chain()
    order: list[str] = []

    class FakeMediaChain:
        """
        模拟异步识别失败。
        """

        async def async_recognize_media(self, **_kwargs):
            """
            记录识别顺序并返回失败。
            """
            order.append("recognize")
            return None

    async def save_params(**_kwargs):
        """
        记录请求状态保存顺序。
        """
        order.append("params")

    async def save_results(_contexts):
        """
        若失败路径错误保存结果则留下可断言记录。
        """
        order.append("results")

    async def collect_events():
        """
        完整消费识别失败的搜索流。
        """
        return [
            event
            async for event in chain.async_search_by_id_stream(
                media_source=MediaSource.TMDB,
                media_id="404",
                cache_local=True,
            )
        ]

    monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain)
    chain.cancel_ai_recommend = lambda: order.append("cancel")
    chain.async_save_last_search_params = save_params
    chain._async_save_results = save_results

    events = asyncio.run(collect_events())

    assert order == ["cancel", "params", "recognize"]
    assert events == [
        {"type": "error", "success": False, "message": "媒体信息识别失败"}
    ]


def test_title_search_sync_async_share_plan_filter_and_projection():
    """
    标题搜索双入口应共享 provider 参数、过滤决策和可替换元数据投影。
    """
    chain = _chain()
    keep = _torrent("Movie 2026 1080p")
    drop = _torrent("Movie 2026 2160p")
    provider_calls: list[dict[str, Any]] = []
    filter_calls: list[list[TorrentInfo]] = []
    meta_calls: list[str] = []

    def search_all_sites(**kwargs):
        """
        记录同步 provider 参数。
        """
        provider_calls.append(kwargs)
        return [keep, drop]

    async def async_search_all_sites(**kwargs):
        """
        记录异步 provider 参数。
        """
        provider_calls.append(kwargs)
        return [keep, drop]

    def filter_torrents(torrents, rule_groups=None):
        """
        记录共享过滤输入并保留首条资源。
        """
        assert rule_groups == ["quality"]
        filter_calls.append(list(torrents))
        return [keep]

    def build_meta(torrent, _mtype):
        """
        模拟插件或测试替换的稳定元数据构造点。
        """
        meta_calls.append(torrent.title)
        return MetaInfo(title=f"mapped:{torrent.title}")

    chain._SearchChain__search_all_sites = search_all_sites
    chain._SearchChain__async_search_all_sites = async_search_all_sites
    chain._filter_title_search_torrents = filter_torrents
    chain._build_title_search_meta = build_meta

    sync_result = chain.search_by_title(
        title="Movie", page=2, sites=[1], mtype=MediaType.MOVIE, rule_groups=["quality"]
    )
    async_result = asyncio.run(
        chain.async_search_by_title(
            title="Movie",
            page=2,
            sites=[1],
            mtype=MediaType.MOVIE,
            rule_groups=["quality"],
        )
    )

    assert provider_calls == [
        {"keyword": "Movie", "sites": [1], "page": 2, "mtype": MediaType.MOVIE},
        {"keyword": "Movie", "sites": [1], "page": 2, "mtype": MediaType.MOVIE},
    ]
    assert filter_calls == [[keep, drop], [keep, drop]]
    assert meta_calls == [keep.title, keep.title]
    assert [context.meta_info.title for context in sync_result] == [
        context.meta_info.title for context in async_result
    ] == [f"mapped:{keep.title}"]


def test_title_search_sync_async_share_empty_result_decisions(monkeypatch):
    """
    标题双入口应共享无候选与过滤为空的警告，并跳过结果缓存。
    """
    chain = _chain()
    candidate = _torrent("Movie 2026 1080p")
    provider_result: list[TorrentInfo] = []
    warnings: list[str] = []
    filter_calls: list[list[TorrentInfo]] = []
    saved_results: list[list[Any]] = []

    def search_all_sites(**_kwargs):
        """
        返回当前场景的同步候选快照。
        """
        return list(provider_result)

    async def async_search_all_sites(**_kwargs):
        """
        返回当前场景的异步候选快照。
        """
        return list(provider_result)

    def filter_torrents(torrents, rule_groups=None):
        """
        记录过滤输入并模拟候选全部被规则排除。
        """
        assert rule_groups == ["quality"]
        filter_calls.append(list(torrents))
        return []

    async def save_params(**_kwargs):
        """
        模拟异步请求参数缓存。
        """

    async def save_results(contexts):
        """
        记录不应发生的异步结果缓存。
        """
        saved_results.append(list(contexts))

    monkeypatch.setattr(title_module.logger, "warning", warnings.append)
    chain._SearchChain__search_all_sites = search_all_sites
    chain._SearchChain__async_search_all_sites = async_search_all_sites
    chain._filter_title_search_torrents = filter_torrents
    chain._build_title_search_meta = lambda torrent, _mtype: MetaInfo(
        title=torrent.title
    )
    chain.cancel_ai_recommend = lambda: None
    chain.save_last_search_params = lambda **_kwargs: None
    chain.async_save_last_search_params = save_params
    chain._save_results = lambda contexts: saved_results.append(list(contexts))
    chain._async_save_results = save_results

    for torrents, warning in (
        ([], "Movie 未搜索到资源"),
        ([candidate], "Movie 没有符合过滤规则的资源"),
    ):
        provider_result[:] = torrents
        warning_count = len(warnings)
        assert chain.search_by_title(
            title="Movie",
            cache_local=True,
            rule_groups=["quality"],
        ) == []
        assert asyncio.run(
            chain.async_search_by_title(
                title="Movie",
                cache_local=True,
                rule_groups=["quality"],
            )
        ) == []
        assert warnings[warning_count:] == [warning, warning]

    assert filter_calls == [[candidate], [candidate]]
    assert saved_results == []


def test_title_stream_preserves_provider_and_result_order():
    """
    标题流应按 provider 批次顺序追加，并以相同顺序发布累计完成结果。
    """
    chain = _chain()
    first = _torrent("Movie 2026 1080p")
    second = _torrent("Movie 2026 2160p")
    provider_params: list[dict[str, Any]] = []

    async def search_stream(**kwargs):
        """
        按固定顺序输出两个 provider 批次。
        """
        provider_params.append(kwargs)
        yield {"type": "append", "site_id": 1, "items": [first]}
        yield {"type": "append", "site_id": 2, "items": [second]}

    async def collect_events():
        """
        完整消费标题搜索流。
        """
        return [
            event
            async for event in chain.async_search_by_title_stream(
                title="Movie", sites=[1, 2], rule_groups=[]
            )
        ]

    chain._SearchChain__async_search_all_sites_stream = search_stream
    chain._filter_title_search_torrents = lambda torrents, rule_groups=None: torrents
    chain._build_title_search_meta = (
        lambda torrent, _mtype: MetaInfo(title=torrent.title)
    )

    events = asyncio.run(collect_events())

    assert provider_params == [
        {"keyword": "Movie", "sites": [1, 2], "page": 0, "mtype": None}
    ]
    assert [event["type"] for event in events] == ["append", "append", "done"]
    assert [event["total_items"] for event in events] == [1, 2, 2]
    assert [
        item["torrent_info"]["title"] for item in events[-1]["items"]
    ] == [first.title, second.title]
    assert events[-1]["candidate_items"] == 2


def test_id_search_resolution_freezes_success_and_failure_effect_order():
    """
    ID 搜索状态机应唯一决定成功保存顺序与识别失败短路。
    """
    recognition_params = {"media_source": MediaSource.TMDB, "media_id": "100"}
    cache_params = {**recognition_params, "sites": [1]}
    contexts = [SimpleNamespace(name="context")]

    success = media_module._id_search_resolution(
        recognition_params=recognition_params,
        cache_params=cache_params,
        season=2,
        sites=[1],
        area="title",
        cache_local=True,
        failure_keyword="tmdb:100",
    )
    assert isinstance(next(success), media_module._IdSearchCacheRequest)
    assert isinstance(success.send(None), media_module._IdSearchRecognizeRequest)
    process_request = success.send(_media())
    assert isinstance(process_request, media_module._IdSearchProcessRequest)
    assert list(process_request.params["no_exists"]["tmdb:100"]) == [2]
    save_request = success.send(contexts)
    assert isinstance(save_request, media_module._IdSearchSaveRequest)
    assert save_request.contexts is contexts
    with pytest.raises(StopIteration) as success_completed:
        success.send(None)
    assert success_completed.value.value.contexts is contexts

    failure = media_module._id_search_resolution(
        recognition_params=recognition_params,
        cache_params=cache_params,
        season=None,
        sites=None,
        area="title",
        cache_local=False,
        failure_keyword="tmdb:100",
    )
    assert isinstance(next(failure), media_module._IdSearchRecognizeRequest)
    with pytest.raises(StopIteration) as failure_completed:
        failure.send(None)
    assert failure_completed.value.value.contexts == []
    assert failure_completed.value.value.warning == "tmdb:100 媒体信息识别失败！"


def test_title_search_resolution_owns_filter_short_circuit_and_save_order():
    """
    标题状态机应在过滤失败时短路，并仅保存成功投影。
    """
    search_params = {"keyword": "Movie", "sites": [1], "page": 0}
    cache_params = {"keyword": "Movie", "area": "title"}
    torrent = _torrent("Movie 2026")
    contexts = [SimpleNamespace(name="context")]

    success = title_module._title_search_resolution(
        title="Movie",
        search_params=search_params,
        cache_params=cache_params,
        cache_local=True,
        mtype=MediaType.MOVIE,
        rule_groups=["quality"],
    )
    assert isinstance(next(success), title_module._TitleSearchCacheRequest)
    assert isinstance(success.send(None), title_module._TitleSearchProviderRequest)
    resolve_request = success.send([torrent])
    assert isinstance(resolve_request, title_module._TitleSearchResolveRequest)
    save_request = success.send(
        title_module._TitleSearchResult(contexts=contexts)
    )
    assert isinstance(save_request, title_module._TitleSearchSaveRequest)
    assert save_request.contexts is contexts
    with pytest.raises(StopIteration) as success_completed:
        success.send(None)
    assert success_completed.value.value.contexts is contexts

    failure = title_module._title_search_resolution(
        title="Movie",
        search_params=search_params,
        cache_params=cache_params,
        cache_local=True,
        mtype=MediaType.MOVIE,
        rule_groups=["quality"],
    )
    next(failure)
    failure.send(None)
    failure.send([torrent])
    with pytest.raises(StopIteration) as completed:
        failure.send(
            title_module._TitleSearchResult(
                contexts=[],
                warning="Movie 没有符合过滤规则的资源",
            )
        )
    assert completed.value.value.contexts == []