MoviePilot/tests/test_transfer_execution_runner.py

"""验证 TransferChain 步骤 runner 与文件执行器的崩溃恢复边界。"""

from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import Mock

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.application.transfer_execution import (
    TransferExecutionCommand,
    TransferExecutionState,
    TransferOperationObservation,
    TransferOperationObservationState,
    TransferStepIntent,
    TransferStepResult,
)
from app.chain import transfer as transfer_chain_module
from app.db.adapters.transfer_execution import (
    TransactionalTransferExecutionRepository,
)
from app.db.base import Base
from app.db.models.transferexecutionstep import TransferExecutionStep
from app.db.models.transferhistory import TransferHistory
from app.db.models.transferpending import TransferPending
from app.modules.filemanager.transhandler import TransHandler
from app.schemas.workflow import FileItem


@pytest.fixture
def execution_repository():
    """构造带有效 pending 租约的独立执行仓储。"""
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(
        engine,
        tables=[
            TransferPending.__table__,
            TransferHistory.__table__,
            TransferExecutionStep.__table__,
        ],
    )
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    with factory() as session:
        session.add(TransferPending(
            task_id="task-runner",
            storage="local",
            src_path="/source.mkv",
            created_at="2026-08-27 09:00:00",
            state="planned",
            updated_at="2026-08-27 09:00:00",
            input_version=1,
            planning_input={"schema_version": 1},
            input_fingerprint="input",
            checkpoint_version=1,
            checkpoint_payload={"schema_version": 1},
            planned_at="2026-08-27 09:00:00",
            lease_owner="worker",
            lease_token="lease",
            lease_expires_at="2099-01-01 00:00:00.000000",
            heartbeat_at="2026-08-27 01:00:00.000000",
            attempt_count=1,
            execution_state="not_started",
            retry_generation=0,
            retry_count=0,
            settlement_revision=0,
        ))
        session.commit()
    repository = TransactionalTransferExecutionRepository(
        factory,
        local_clock=lambda: datetime(2026, 8, 27, 9, 30, 0),
        lease_clock=lambda: datetime(2026, 8, 27, 1, 30, 0, tzinfo=timezone.utc),
    )
    try:
        yield repository
    finally:
        engine.dispose()


def _runner(repository):
    """构造绑定固定任务、租约与计划身份的 durable runner。"""
    return transfer_chain_module._DurableTransferStepRunner(
        task_id="task-runner",
        lease_token="lease",
        checkpoint_fingerprint="plan",
        repository=repository,
    )


def test_runner_replay_returns_persisted_result_without_repeating_side_effect(
        execution_repository,
):
    """成功步骤重放只能回读结果，不能再次调用外部执行函数。"""
    calls = []
    first = _runner(execution_repository).run(
        phase="transfer",
        kind="copy",
        payload={"source": "/source.mkv", "target": "/target.mkv"},
        execute=lambda: calls.append("executed") or TransferStepResult(
            payload={"item": {"path": "/target.mkv"}}
        ),
        observe=lambda: pytest.fail("新步骤不应执行恢复探测"),
    )
    second = _runner(execution_repository).run(
        phase="transfer",
        kind="copy",
        payload={"source": "/source.mkv", "target": "/target.mkv"},
        execute=lambda: pytest.fail("已成功步骤不得重复执行"),
        observe=lambda: pytest.fail("已成功步骤不得执行恢复探测"),
    )
    assert first == second
    assert calls == ["executed"]


def test_runner_routes_unknown_orphaned_attempt_to_manual_review(
        execution_repository,
):
    """遗留 STARTED 无法严格判断时必须隔离，不能再次执行副作用。"""
    command = TransferExecutionCommand(execution_repository)
    prepared = command.prepare(
        task_id="task-runner",
        lease_token="lease",
        intent=TransferStepIntent.create(
            task_id="task-runner",
            checkpoint_fingerprint="plan",
            ordinal=0,
            phase="provider",
            kind="opaque",
            payload={"provider": "legacy"},
        ),
    )
    command.begin(
        task_id="task-runner",
        lease_token="lease",
        operation_id=prepared.operation_id,
    )

    with pytest.raises(
        transfer_chain_module._TransferManualReviewRequired,
        match="禁止自动重放",
    ):
        _runner(execution_repository).run(
            phase="provider",
            kind="opaque",
            payload={"provider": "legacy"},
            execute=lambda: pytest.fail("未知遗留步骤不得重放"),
            observe=lambda: TransferOperationObservation(
                state=TransferOperationObservationState.UNKNOWN,
                evidence=TransferStepResult(payload={"receipt": None}),
            ),
        )
    snapshot = execution_repository.get_snapshot(task_id="task-runner")
    assert snapshot is not None
    assert snapshot.state is TransferExecutionState.MANUAL_REVIEW


class _ImmediateStepRunner:
    """记录 TransHandler 拆分顺序并立即执行步骤的测试 runner。"""

    def __init__(self) -> None:
        """初始化步骤记录。"""
        self.steps = []

    def run(self, *, phase, kind, payload, execute, observe):
        """记录稳定意图并直接执行，不触发恢复探测。"""
        self.steps.append((phase, kind, payload))
        return execute()


def test_cross_storage_move_materializes_before_independent_source_delete(tmp_path):
    """跨存储 move 必须先复制目标，再用独立步骤删除源。"""
    source_path = tmp_path / "source.mkv"
    source_path.write_bytes(b"movie")
    source_item = FileItem(
        storage="local",
        path=source_path.as_posix(),
        name=source_path.name,
        type="file",
        size=source_path.stat().st_size,
        extension="mkv",
    )
    target_item = FileItem(
        storage="remote",
        path="/library/source.mkv",
        name="source.mkv",
        type="file",
        size=source_item.size,
        extension="mkv",
    )
    source_oper = Mock()
    source_oper.delete.return_value = True
    target_oper = Mock()
    target_oper.get_folder.return_value = FileItem(
        storage="remote", path="/library", name="library", type="dir"
    )
    target_oper.upload.return_value = target_item
    runner = _ImmediateStepRunner()

    result, error = TransHandler._TransHandler__execute_transfer_with_steps(
        step_runner=runner,
        fileitem=source_item,
        target_storage="remote",
        source_oper=source_oper,
        target_oper=target_oper,
        target_file=Path("/library/source.mkv"),
        transfer_type="move",
    )

    assert error == ""
    assert result == target_item
    assert [kind for _phase, kind, _payload in runner.steps] == [
        "materialize_target",
        "delete_move_source",
    ]
    assert runner.steps[0][2]["transfer_type"] == "copy"
    target_oper.upload.assert_called_once()
    source_oper.delete.assert_called_once_with(source_item)