MoviePilot/tests/test_message_queue_sync_async_parity.py

"""
消息队列同步与异步入口的调度决策一致性测试。
"""

import asyncio
import queue
from unittest.mock import Mock, call

import pytest

from app.application.messaging.message import MessageQueueManager


@pytest.fixture
def message_queue(monkeypatch: pytest.MonkeyPatch) -> MessageQueueManager:
    """
    构造不启动监控线程的独立消息队列。
    """
    manager = object.__new__(MessageQueueManager)
    manager.queue = queue.Queue()
    manager.send_callback = Mock()
    monkeypatch.setattr(manager, "_is_in_scheduled_time", lambda _now: False)
    return manager


@pytest.mark.parametrize(
    ("immediately", "scheduled", "expected"),
    [
        (False, False, False),
        (False, True, True),
        (True, False, True),
        (True, True, True),
    ],
)
def test_dispatch_plan_combines_immediate_and_schedule_decisions(
    immediately: bool, scheduled: bool, expected: bool
) -> None:
    """
    共享计划应唯一决定立即发送结论并移除内部控制参数。
    """
    callback = Mock()

    send_now, message = MessageQueueManager._build_message_dispatch(
        send_callback=callback,
        explicit_callback=True,
        args=("payload",),
        kwargs={"immediately": immediately, "token": 1},
        scheduled=scheduled,
    )

    assert send_now is expected
    assert message == {
        "args": ("payload",),
        "kwargs": {"token": 1},
        "send_callback": callback,
    }


def test_default_queue_payload_matches_between_sync_and_async(
    message_queue: MessageQueueManager,
) -> None:
    """
    默认回调双入口在免打扰时段应生成完全相同的队列载荷。
    """
    message_queue.send_message("payload", token=1)
    asyncio.run(message_queue.async_send_message("payload", token=1))

    sync_message = message_queue.queue.get_nowait()
    async_message = message_queue.queue.get_nowait()

    assert sync_message == async_message == {
        "args": ("payload",),
        "kwargs": {"token": 1},
    }
    message_queue.send_callback.assert_not_called()


def test_explicit_queue_payload_matches_between_sync_and_async(
    message_queue: MessageQueueManager,
) -> None:
    """
    显式回调双入口排队时都应保留各消息自己的回调身份。
    """
    callback = Mock()

    message_queue.send_message_for(callback, "payload", token=1)
    asyncio.run(
        message_queue.async_send_message_for(callback, "payload", token=1)
    )

    sync_message = message_queue.queue.get_nowait()
    async_message = message_queue.queue.get_nowait()

    assert sync_message == async_message == {
        "args": ("payload",),
        "kwargs": {"token": 1},
        "send_callback": callback,
    }
    callback.assert_not_called()


def test_immediate_delivery_matches_between_sync_and_async(
    message_queue: MessageQueueManager,
) -> None:
    """
    强制立即发送时双入口都应绕过队列并传递相同业务参数。
    """
    callback = message_queue.send_callback

    message_queue.send_message("payload", token=1, immediately=True)
    asyncio.run(
        message_queue.async_send_message("payload", token=1, immediately=True)
    )

    assert callback.call_args_list == [
        call("payload", token=1),
        call("payload", token=1),
    ]
    assert message_queue.queue.empty()


def test_explicit_immediate_delivery_matches_between_sync_and_async(
    message_queue: MessageQueueManager,
) -> None:
    """
    显式回调强制发送时双入口都应绕过队列并移除控制参数。
    """
    callback = Mock()

    message_queue.send_message_for(
        callback, "payload", token=1, immediately=True
    )
    asyncio.run(
        message_queue.async_send_message_for(
            callback, "payload", token=1, immediately=True
        )
    )

    assert callback.call_args_list == [
        call("payload", token=1),
        call("payload", token=1),
    ]
    assert message_queue.queue.empty()