MoviePilot/tests/test_transfer_tmdb_category.py

"""
验证TMDB 分类规则在整理规划中的应用。
"""

import threading
from types import SimpleNamespace
from unittest.mock import Mock

from app.application.transfer.execution import TransferExecutionCheckpoint
from app.application.transfer.workflow import TransferTask

# 此用例覆盖运行时惰性导出，静态检查无法解析该公开入口。
from app.chain.transfer import TransferChain  # pylint: disable=no-name-in-module
from app.domain.context import MediaInfo
from app.domain.metainfo import MetaInfo
from app.schemas.file import FileItem
from app.schemas.system import TransferDirectoryConf
from app.schemas.types import MediaSource, MediaType


def test_transfer_rejects_partial_or_invalid_explicit_identity() -> None:
    """
    整理公共入口不得把半套身份或格式非法来源传入后台任务。
    """
    chain = object.__new__(TransferChain)
    fileitem = FileItem(
        storage="local",
        path="/downloads/Test.Movie.2024.mkv",
        type="file",
    )

    partial_state, partial_message = chain.do_transfer(
        fileitem=fileitem,
        media_source=MediaSource.TMDB,
    )
    invalid_state, invalid_message = chain.do_transfer(
        fileitem=fileitem,
        media_source="plugin source:invalid",
        media_id="1234",
    )

    assert not partial_state
    assert "media_source" in partial_message
    assert not invalid_state
    assert "media_source" in invalid_message


def test_transfer_resolves_complete_identity_before_building_tasks(monkeypatch) -> None:
    """
    整理入口收到完整身份时应精确识别，失败后不得退化为标题识别。
    """
    chain = object.__new__(TransferChain)
    fileitem = FileItem(
        storage="local",
        path="/downloads/Test.Movie.2024.mkv",
        type="file",
    )
    recognize = monkeypatch.setattr
    calls = []

    class FakeMediaChain:
        """
        记录精确识别参数并返回空结果。
        """

        def recognize_media(self, **kwargs):
            """
            模拟显式身份识别失败。
            """
            calls.append(kwargs)
            return None

    recognize("app.chain.transfer.workflow.MediaChain", FakeMediaChain)

    state, message = chain.do_transfer(
        fileitem=fileitem,
        media_source="tmdb",
        media_id="1234",
        mtype=MediaType.MOVIE,
    )

    assert not state
    assert "未识别到媒体信息" in message
    assert calls == [{
        "mtype": MediaType.MOVIE,
        "media_source": MediaSource.TMDB,
        "media_id": "1234",
        "music_type": None,
    }]


def test_transfer_stops_when_automatic_category_has_no_tmdb_result(monkeypatch) -> None:
    """
    启用自动类别目录时，缺少 TMDB 分类必须在文件操作前明确失败。
    """
    chain = object.__new__(TransferChain)
    chain.jobview = SimpleNamespace(try_remove_job=lambda _task: None)
    chain._transfer_admissions = Mock()
    chain._worker_owner_id = "category-owner"
    chain._owned_leases = {
        "task-before-category": ("lease-before-category", float("inf"))
    }
    chain._worker_state_lock = threading.RLock()
    chain.durable_event_writer = Mock()
    chain.runtime_config = SimpleNamespace(
        scrape_follow_tmdb=True,
        ai_agent_enable=True,
        ai_agent_retry_transfer=True,
    )
    chain.queue_failed_transfer_notification = Mock()
    chain._TransferChain__mark_torrent_completed_if_done = Mock()
    record_transfer_failure = Mock()
    add_transfer_fail = Mock()
    monkeypatch.setattr(
        "app.chain.transfer.settlement.record_transfer_failure",
        record_transfer_failure,
    )
    monkeypatch.setattr("app.chain.transfer.settlement.add_transfer_fail", add_transfer_fail)
    chain._transfer_admissions.checkpoint_plan.side_effect = (
        lambda **kwargs: SimpleNamespace(checkpoint=kwargs["checkpoint"])
    )
    step_runner = Mock()
    step_runner.checkpoint.side_effect = lambda transferinfo: (
        TransferExecutionCheckpoint.create(
            payload={
                "outcome": "failed",
                "transferinfo": transferinfo.model_dump(mode="json"),
            },
            operation_ids=("planning-reject",),
        )
    )
    chain._TransferChain__build_durable_step_runner = Mock(
        return_value=step_runner
    )
    chain.transfer_history_repository = SimpleNamespace()
    monkeypatch.setattr(
        "app.chain.transfer.execution.MediaChain",
        lambda: SimpleNamespace(
            supplement_tmdb_info=lambda media, _meta: media,
        ),
    )
    monkeypatch.setattr("app.chain.transfer.filter.MediaChain", lambda: SimpleNamespace(
        supplement_tmdb_info=lambda media, _meta: media,
    ))
    task = TransferTask(
        fileitem=FileItem(
            storage="local",
            path="/downloads/Test.Movie.2024.mkv",
            type="file",
            name="Test.Movie.2024.mkv",
            extension="mkv",
            size=1024,
        ),
        meta=MetaInfo("Test Movie 2024"),
        mediainfo=MediaInfo(
            media_source=MediaSource.AniList,
            media_id="1234",
            anilist_id=1234,
            type=MediaType.MOVIE,
            title="Test Movie",
            year="2024",
        ),
        target_directory=TransferDirectoryConf(
            library_storage="local",
            library_path="/library",
            library_category_folder=True,
        ),
        library_category_folder=True,
        preview=False,
    )
    task.bind_admission_task_id("task-before-category")
    task.bind_execution_lease(
        owner_id="category-owner",
        lease_token="lease-before-category",
    )

    state, message = chain._TransferChain__handle_transfer(task)

    assert not state
    assert message == "媒体识别结果未匹配到媒体分类，无法按媒体类别整理"
    assert task.mediainfo.media_source == MediaSource.AniList
    assert task.mediainfo.media_id == "1234"
    assert task.plan_checkpoint is not None
    assert task.plan_checkpoint.rejection_error == message
    assert task.execution_checkpoint is not None
    chain._transfer_admissions.record_planning_failure.assert_not_called()
    record_transfer_failure.assert_not_called()
    add_transfer_fail.assert_not_called()
    chain.queue_failed_transfer_notification.assert_not_called()
    chain._TransferChain__mark_torrent_completed_if_done.assert_not_called()