MoviePilot/tests/test_subscribe_search_command.py

from types import SimpleNamespace
from unittest.mock import Mock

import pytest

from app.api.dependencies import subscription as subscription_dependencies
from app.application.subscription.delete import SubscribeDeletionCandidate
from app.application.subscription.search import (
    SearchSubscriptionsCommand,
    SubscribeSearchActor,
)
from app.runtime.tasks import TaskRegistry


class _Repository:
    """提供手工订阅搜索测试需要的归属和列表数据。"""

    def __init__(self, candidate=None, subscribe_ids=None):
        """保存预设单条候选和批量编号。"""
        self.candidate = candidate
        self.subscribe_ids = subscribe_ids or []

    async def get_candidate(self, _subscribe_id):
        """返回预设订阅候选。"""
        return self.candidate

    async def list_search_ids(self, username, state):
        """校验普通用户搜索状态并返回预设编号。"""
        assert username == "alice"
        assert state == "R"
        return self.subscribe_ids


def _candidate(username):
    """构造只包含归属信息的订阅候选。"""
    return SubscribeDeletionCandidate(
        subscribe_id=7,
        username=username,
        event_payload={},
    )


@pytest.mark.asyncio
async def test_superuser_search_all_uses_single_global_scheduler_request():
    """管理员搜索全部订阅时保持一次 state=R 的全局调度语义。"""
    scheduled = []
    command = SearchSubscriptionsCommand(
        repository=_Repository(),
        schedule_search=lambda sid, state: scheduled.append((sid, state)),
    )

    assert await command.execute(
        SubscribeSearchActor(username="admin", is_superuser=True)
    ) is True
    assert scheduled == [(None, "R")]


@pytest.mark.asyncio
async def test_regular_user_search_all_schedules_only_owned_subscriptions():
    """普通用户搜索全部时把归属订阅合并为一次后台批次。"""
    scheduled = []
    command = SearchSubscriptionsCommand(
        repository=_Repository(subscribe_ids=[2, 5]),
        schedule_search=lambda sid, state: scheduled.append((sid, state)),
    )

    assert await command.execute(
        SubscribeSearchActor(username="alice", is_superuser=False)
    ) is True
    assert scheduled == [((2, 5), None)]


@pytest.mark.asyncio
async def test_regular_user_search_all_with_no_targets_does_not_schedule():
    """普通用户没有可搜索订阅时不创建空后台任务。"""
    scheduled = []
    command = SearchSubscriptionsCommand(
        repository=_Repository(),
        schedule_search=lambda ids, state: scheduled.append((ids, state)),
    )

    assert await command.execute(
        SubscribeSearchActor(username="alice", is_superuser=False)
    ) is True
    assert scheduled == []


@pytest.mark.asyncio
async def test_targeted_search_rejects_missing_or_other_users_subscription():
    """单条搜索不得泄漏订阅是否属于其他普通用户。"""
    scheduled = []
    command = SearchSubscriptionsCommand(
        repository=_Repository(candidate=_candidate("bob")),
        schedule_search=lambda sid, state: scheduled.append((sid, state)),
    )

    assert await command.execute(
        SubscribeSearchActor(username="alice", is_superuser=False),
        subscribe_id=7,
    ) is False
    assert scheduled == []


@pytest.mark.asyncio
async def test_targeted_search_schedules_accessible_subscription():
    """归属用户搜索单条订阅时提交历史兼容参数。"""
    scheduled = []
    command = SearchSubscriptionsCommand(
        repository=_Repository(candidate=_candidate("alice")),
        schedule_search=lambda sid, state: scheduled.append((sid, state)),
    )

    assert await command.execute(
        SubscribeSearchActor(username="alice", is_superuser=False),
        subscribe_id=7,
    ) is True
    assert scheduled == [((7,), None)]


def test_subscription_search_batch_uses_one_scheduler_generation(monkeypatch):
    """一个后台批次只占用一次调度任务运行权。"""
    calls = []
    monkeypatch.setattr(
        subscription_dependencies,
        "start_scheduler_job",
        lambda job_id, **kwargs: calls.append((job_id, kwargs)),
    )

    subscription_dependencies._start_subscription_search_batch((2, 5), None)

    assert calls == [
        (
            "subscribe_search",
            {"sids": (2, 5), "state": None, "manual": True},
        ),
    ]


@pytest.mark.asyncio
async def test_search_dependency_registers_one_owned_background_batch():
    """请求适配器只登记一个具名后台批次。"""
    registry = TaskRegistry()
    registry.create_sync = Mock()
    repository = _Repository(subscribe_ids=[2, 5])
    runtime = SimpleNamespace(
        subscription=SimpleNamespace(repository=lambda _db: repository),
    )
    command = subscription_dependencies.get_search_subscriptions_command(
        task_registry=registry,
        db=object(),
        runtime=runtime,
    )

    assert await command.execute(
        SubscribeSearchActor(username="alice", is_superuser=False)
    ) is True
    registry.create_sync.assert_called_once_with(
        subscription_dependencies._start_subscription_search_batch,
        (2, 5),
        None,
        owner="api.subscribe.search",
    )