"""
系统后台更新状态机测试。
"""

import errno
import json
import threading
import zipfile
from contextlib import nullcontext
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock, call

import pytest

from app.adapters.system import update as update_module


def _manager(monkeypatch, tmp_path: Path):
    """
    创建使用临时状态目录的更新管理器。
    """
    # 固定升级起点，避免产品版本发布后与测试中的目标版本重合。
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.0")
    monkeypatch.setattr(
        update_module,
        "get_runtime_setting",
        lambda key: tmp_path if key == "TEMP_PATH" else None,
    )
    manager = object.__new__(update_module.SystemUpdateManager)
    manager._lock = threading.RLock()
    manager._download_active = False
    manager._active_target = None
    return manager


def _docker_manager(monkeypatch, tmp_path: Path):
    """
    创建指向隔离 Docker 目录的更新管理器。
    """
    runtime_settings = {
        "TEMP_PATH": tmp_path / "config" / "temp",
        "ROOT_PATH": tmp_path / "app",
        "FRONTEND_PATH": tmp_path / "public",
        "VENV_PATH": tmp_path / "venv",
        "UV_BIN": tmp_path / "uv",
        "MOVIEPILOT_AUTO_UPDATE": False,
        "AUTO_UPDATE_RESOURCE": True,
        "PIP_PROXY": "",
        "PROXY_HOST": "",
    }
    monkeypatch.setattr(
        update_module,
        "get_runtime_setting",
        lambda key: runtime_settings[key],
    )
    monkeypatch.setattr(update_module, "is_docker", lambda: True)
    manager = object.__new__(update_module.SystemUpdateManager)
    manager._lock = threading.RLock()
    manager._download_active = False
    manager._active_target = None
    return manager


def _response(payload, status_code=200):
    """
    构造不访问外网的更新接口响应。
    """
    return SimpleNamespace(status_code=status_code, json=lambda: payload)


def test_status_reads_live_auto_update_setting_without_discarding_cached_update(monkeypatch, tmp_path):
    """
    切换提醒设置立即反映到状态，缓存版本仍供手动升级使用。
    """
    manager = _manager(monkeypatch, tmp_path)
    manager._write_state(state="available", version="v3.1.0", can_update=True)
    for enabled in (True, False, True):
        monkeypatch.setattr(
            update_module, "get_runtime_setting",
            lambda key: tmp_path if key == "TEMP_PATH" else enabled,
        )
        status = manager.get_status()
        assert status.auto_update is enabled
        assert status.state == "available"
        assert status.version == "v3.1.0"
        assert status.can_update is True


def test_sync_docker_dependencies_freezes_lock_for_custom_package_index(
    monkeypatch, tmp_path
):
    """
    自定义镜像源只替换下载地址，不应触发 uv 重新校验或更新锁文件。
    """
    manager = _docker_manager(monkeypatch, tmp_path)
    settings = {
        "TEMP_PATH": tmp_path / "config" / "temp",
        "ROOT_PATH": tmp_path / "app",
        "FRONTEND_PATH": tmp_path / "public",
        "VENV_PATH": tmp_path / "venv",
        "UV_BIN": tmp_path / "uv",
        "PIP_PROXY": "https://mirror.example/simple",
        "PROXY_HOST": "",
    }
    monkeypatch.setattr(update_module, "get_runtime_setting", settings.__getitem__)
    project_dir = tmp_path / "staged"
    project_dir.mkdir()
    calls = []

    def run(command, **kwargs):
        """
        记录 uv 调用并返回成功结果。
        """
        calls.append((command, kwargs))
        return SimpleNamespace(returncode=0)

    monkeypatch.setattr(update_module.subprocess, "run", run)

    assert manager._sync_docker_dependencies(project_dir, force=True) is True

    command = calls[0][0]
    assert "--frozen" in command
    assert "--locked" not in command
    assert command[-2:] == ["--default-index", "https://mirror.example/simple"]


@pytest.mark.parametrize("auto_update", [False, True])
@pytest.mark.parametrize("auto_update_resource", [False, True])
def test_scheduled_check_respects_independent_switches(
    monkeypatch, tmp_path, auto_update, auto_update_resource
):
    """
    自动检查只访问已开启的目标；手动检查仍可访问两类更新。
    """
    manager = _manager(monkeypatch, tmp_path)
    values = {
        "TEMP_PATH": tmp_path,
        "MOVIEPILOT_AUTO_UPDATE": auto_update,
        "AUTO_UPDATE_RESOURCE": auto_update_resource,
    }
    monkeypatch.setattr(update_module, "get_runtime_setting", values.get)
    checked = []
    monkeypatch.setattr(manager, "_check_application", lambda: checked.append("application"))
    monkeypatch.setattr(manager, "_check_resources", lambda: checked.append("resources"))

    status = manager.check_scheduled()
    assert checked == [
        target for target, enabled in (("application", auto_update), ("resources", auto_update_resource))
        if enabled
    ]
    assert status.auto_update is auto_update
    assert status.auto_update_resource is auto_update_resource

    checked.clear()
    manager.check()
    assert checked == ["application", "resources"]


def test_check_exposes_new_stable_release(monkeypatch, tmp_path):
    """
    主程序更新应跳过最新采集器独立 Release 和后端预发布版本。
    """
    manager = _manager(monkeypatch, tmp_path)
    logs = []
    releases = [
        {"tag_name": "site-adapter-collector-v1.0.1", "prerelease": False, "draft": False},
        {"tag_name": "v3.2.0-beta", "prerelease": True, "draft": False},
        {
            "tag_name": "v3.1.0",
            "name": "MoviePilot v3.1.0",
            "body": "changes",
            "published_at": "2026-08-24T00:00:00Z",
            "prerelease": False,
            "draft": False,
        },
    ]
    monkeypatch.setattr(manager, "_request", lambda: SimpleNamespace(get_res=lambda _url: _response(releases)))
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.0")
    monkeypatch.setattr(update_module.logger, "info", logs.append)

    status = manager.check("application")

    assert status.state == "available"
    assert status.version == "v3.1.0"
    assert status.release_notes == "changes"
    assert status.can_update is True
    assert logs == ["发现 MoviePilot 主程序更新：v3.0.0 -> v3.1.0"]


def test_check_logs_when_application_is_current(monkeypatch, tmp_path):
    """
    手动检查确认已是最新版时记录结果，并保持不可更新的空闲状态。
    """
    manager = _manager(monkeypatch, tmp_path)
    logs = []
    releases = [
        {
            "tag_name": "v3.0.0",
            "name": "MoviePilot v3.0.0",
            "prerelease": False,
            "draft": False,
        }
    ]
    monkeypatch.setattr(manager, "_request", lambda: SimpleNamespace(get_res=lambda _url: _response(releases)))
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.0")
    monkeypatch.setattr(update_module.logger, "info", logs.append)

    status = manager.check("application")

    assert status.state == "idle"
    assert status.can_update is False
    assert logs == ["MoviePilot 主程序已是最新版本：v3.0.0"]


def test_status_refreshes_stale_application_version(monkeypatch, tmp_path):
    """
    旧状态文件中的主程序版本应在读取时更新为当前运行版本。
    """
    manager = _manager(monkeypatch, tmp_path)
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.0")
    manager._write_item("application", state="idle")

    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.3")
    status = manager.get_status()
    application = next(item for item in status.updates if item.type == "application")

    assert status.current_version == "v3.0.3"
    assert application.current_version == "v3.0.3"


def test_public_download_request_omits_github_authorization(monkeypatch, tmp_path):
    """
    公开归档下载不得复用 GitHub API 的认证请求头。
    """
    manager = _manager(monkeypatch, tmp_path)
    settings = {
        "TEMP_PATH": tmp_path,
        "PROXY": {"https": "http://proxy.example:7890"},
        "GITHUB_HEADERS": {"Authorization": "Bearer github-token"},
    }
    monkeypatch.setattr(update_module, "get_runtime_setting", settings.__getitem__)
    request_utils = Mock()
    request_utils.return_value.get_stream.return_value = nullcontext(
        SimpleNamespace(
            status_code=200,
            headers={"content-length": "7"},
            iter_content=lambda **_kwargs: [b"archive"],
        )
    )
    monkeypatch.setattr(update_module, "RequestUtils", request_utils)

    manager._request()
    destination = tmp_path / "backend.zip"
    downloaded, content_length = manager._download_file(
        "https://github.com/jxxghp/MoviePilot/archive/refs/tags/v3.0.2.zip",
        destination,
        0,
        0,
    )

    assert request_utils.call_args_list == [
        call(
            proxies=settings["PROXY"],
            headers=settings["GITHUB_HEADERS"],
            timeout=60,
        ),
        call(proxies=settings["PROXY"], timeout=60),
    ]
    assert downloaded == content_length == 7
    assert destination.read_bytes() == b"archive"


def test_scheduled_check_failure_stays_silent(monkeypatch, tmp_path):
    """
    定时检查失败时保留错误状态信息，但不输出主动通知日志。
    """
    manager = _manager(monkeypatch, tmp_path)
    monkeypatch.setattr(
        manager,
        "_request",
        lambda: SimpleNamespace(get_res=lambda _url: _response({}, status_code=503)),
    )

    status = manager.check("application")

    assert status.state == "idle"
    assert status.error


def test_interrupted_download_becomes_retryable_failure(monkeypatch, tmp_path):
    """
    下载中断后进入可重试的失败状态。
    """
    manager = _manager(monkeypatch, tmp_path)
    manager._write_state(state="downloading", version="v3.1.0")

    status = manager.get_status()

    assert status.state == "failed"
    assert status.can_update is True
    assert "中断" in status.error


def test_installing_application_update_converges_when_runtime_is_newer(
    monkeypatch, tmp_path
):
    """
    当前主程序高于残留目标版本时应清理状态并恢复自动检查。
    """
    manager = _manager(monkeypatch, tmp_path)
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.6")
    manager._write_item(
        "application",
        state="installing",
        version="v3.0.1",
        can_update=False,
        can_install=False,
    )
    checked = []
    monkeypatch.setattr(manager, "_check_application", lambda: checked.append(True))

    status = manager.check("application")

    application = next(item for item in status.updates if item.type == "application")
    assert checked == [True]
    assert application.state == "idle"
    assert application.version is None
    assert status.state == "idle"


def test_ready_application_update_is_discarded_when_runtime_reaches_target(
    monkeypatch, tmp_path
):
    """
    外部升级先于确认安装时应清理过期的主程序待安装包。
    """
    manager = _manager(monkeypatch, tmp_path)
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.1.0")
    manager._merge_prepared_manifest(
        {
            "version": "v3.1.0",
            "backend_archive": "/tmp/backend.zip",
            "frontend_archive": "/tmp/frontend.zip",
        }
    )
    manager._write_item(
        "application",
        state="ready",
        version="v3.1.0",
        can_update=False,
        can_install=True,
    )

    status = manager.get_status()

    application = next(item for item in status.updates if item.type == "application")
    assert application.state == "idle"
    assert application.version is None
    assert application.can_install is False
    assert not (manager._root / "prepared.json").exists()


def test_failed_application_update_clears_after_external_upgrade(monkeypatch, tmp_path):
    """
    旧版本安装失败后改用镜像升级，残留的失败记录和重试包应被清理。
    """
    manager = _manager(monkeypatch, tmp_path)
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.10-1")
    manager._merge_prepared_manifest(
        {
            "version": "v3.0.4",
            "backend_archive": "/tmp/backend.zip",
            "frontend_archive": "/tmp/frontend.zip",
        }
    )
    manager._write_item(
        "application",
        state="failed",
        version="v3.0.4",
        error="Docker 更新包替换失败：[Errno 18] Invalid cross-device link",
        can_update=True,
        can_install=False,
    )

    status = manager.get_status()

    application = next(item for item in status.updates if item.type == "application")
    assert application.state == "idle"
    assert application.version is None
    assert application.error is None
    assert application.can_update is False
    assert status.state == "idle"
    assert status.error is None
    assert not (manager._root / "prepared.json").exists()


def test_available_application_update_clears_after_external_upgrade(monkeypatch, tmp_path):
    """
    关闭自动检查时，外部升级后不应继续提示已低于当前版本的可用更新。
    """
    manager = _manager(monkeypatch, tmp_path)
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.10-1")
    manager._write_item(
        "application",
        state="available",
        version="v3.0.10",
        can_update=True,
    )

    status = manager.get_status()

    application = next(item for item in status.updates if item.type == "application")
    assert application.state == "idle"
    assert application.version is None
    assert status.can_update is False


def test_failed_application_update_is_kept_below_target(monkeypatch, tmp_path):
    """
    当前版本仍低于目标时应保留失败原因和重试入口。
    """
    manager = _manager(monkeypatch, tmp_path)
    monkeypatch.setattr(update_module, "get_app_version", lambda: "v3.0.3")
    manager._write_item(
        "application",
        state="failed",
        version="v3.0.4",
        error="Docker 更新包替换失败",
        can_update=True,
    )

    status = manager.get_status()

    application = next(item for item in status.updates if item.type == "application")
    assert application.state == "failed"
    assert application.version == "v3.0.4"
    assert application.can_update is True


def test_ready_resource_update_clears_after_loaded_version_reaches_target(
    monkeypatch,
    tmp_path,
):
    """
    运行资源已达到目标版本时应清除残留提示，并保留另一类待安装制品。
    """
    versions = ["3.0.1", "3.0.9"]
    monkeypatch.setattr(
        update_module,
        "get_resource_versions",
        lambda: tuple(versions),
    )
    manager = _manager(monkeypatch, tmp_path)
    manager._merge_prepared_manifest(
        {
            "version": "v3.1.0",
            "backend_archive": "/tmp/backend.zip",
            "resource_package_version": "10",
            "resource_files": [{"name": "user.sites.v3.bin"}],
        }
    )
    manager._write_item(
        "resources",
        state="ready",
        version="10",
        current_auth_version="3.0.1",
        current_indexer_version="3.0.9",
        indexer_version="3.0.10",
        can_update=False,
        can_install=True,
    )

    pending = manager.get_status()
    pending_resource = next(item for item in pending.updates if item.type == "resources")
    assert pending_resource.state == "ready"

    versions[1] = "3.0.10"
    current = manager.get_status()
    current_resource = next(item for item in current.updates if item.type == "resources")
    prepared = json.loads((manager._root / "prepared.json").read_text(encoding="utf-8"))

    assert current_resource.state == "idle"
    assert current_resource.current_indexer_version == "3.0.10"
    assert current_resource.indexer_version is None
    assert current_resource.version is None
    assert current_resource.can_install is False
    assert prepared["version"] == "v3.1.0"
    assert prepared["backend_archive"] == "/tmp/backend.zip"
    assert "resource_package_version" not in prepared
    assert "resource_files" not in prepared


def test_download_prepares_matching_backend_and_frontend_archives(monkeypatch, tmp_path):
    """
    更新下载准备相互匹配的后端与前端归档，完成后进入可安装状态。
    """
    manager = _manager(monkeypatch, tmp_path)
    backend_fixture = tmp_path / "source-backend.zip"
    frontend_fixture = tmp_path / "source-frontend.zip"
    with zipfile.ZipFile(backend_fixture, "w") as archive:
        archive.writestr(
            "MoviePilot-v3.1.0/version.py",
            "APP_VERSION = 'v3.1.0'\nFRONTEND_VERSION = 'v3.1.0'\n",
        )
        archive.writestr("MoviePilot-v3.1.0/pyproject.toml", "[project]\n")
        archive.writestr("MoviePilot-v3.1.0/uv.lock", "version = 1\n")
    with zipfile.ZipFile(frontend_fixture, "w") as archive:
        archive.writestr("dist/index.html", "ok")
        archive.writestr("dist/version.txt", "v3.1.0\n")

    fixtures = iter((backend_fixture, frontend_fixture))

    def download(_url, destination, downloaded_before, _total_hint):
        source = next(fixtures)
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_bytes(source.read_bytes())
        size = destination.stat().st_size
        return downloaded_before + size, size

    monkeypatch.setattr(manager, "_download_file", download)
    monkeypatch.setattr(update_module, "is_docker", lambda: True)
    monkeypatch.setattr(
        manager,
        "_fetch_frontend_release",
        lambda _version: {
            "assets": [
                {
                    "name": "dist.zip",
                    "size": frontend_fixture.stat().st_size,
                    "browser_download_url": "https://example.invalid/dist.zip",
                    "digest": f"sha256:{manager._sha256(frontend_fixture)}",
                }
            ]
        },
    )

    manager._write_state(state="downloading", version="v3.1.0")
    manager._active_target = "application"
    manager._download_update("application")

    status = manager.get_status()
    prepared = json.loads((manager._root / "prepared.json").read_text(encoding="utf-8"))
    assert status.state == "ready"
    assert status.progress == 100
    assert status.frontend_version == "v3.1.0"
    assert prepared["backend_sha256"] == manager._sha256(manager._backend_archive)
    assert prepared["frontend_sha256"] == manager._sha256(manager._frontend_archive)


def test_application_download_keeps_prepared_resource_package(monkeypatch, tmp_path):
    """
    应用包重新下载时不得删除另一类已经准备好的资源包。
    """
    manager = _manager(monkeypatch, tmp_path)
    resource_dir = manager._resource_dir
    resource_dir.mkdir(parents=True)
    resource_file = resource_dir / "user.sites.v3.bin"
    resource_file.write_bytes(b"resource")
    manager._merge_prepared_manifest(
        {
            "resource_package_version": "10",
            "resource_files": [
                {
                    "name": resource_file.name,
                    "type": "indexer",
                    "version": "3.0.8",
                    "path": str(resource_file),
                    "sha256": manager._sha256(resource_file),
                }
            ],
        }
    )
    backend_fixture = tmp_path / "source-backend.zip"
    frontend_fixture = tmp_path / "source-frontend.zip"
    with zipfile.ZipFile(backend_fixture, "w") as archive:
        archive.writestr(
            "MoviePilot-v3.1.0/version.py",
            "APP_VERSION = 'v3.1.0'\nFRONTEND_VERSION = 'v3.1.0'\n",
        )
        archive.writestr("MoviePilot-v3.1.0/pyproject.toml", "[project]\n")
        archive.writestr("MoviePilot-v3.1.0/uv.lock", "version = 1\n")
    with zipfile.ZipFile(frontend_fixture, "w") as archive:
        archive.writestr("dist/index.html", "ok")
        archive.writestr("dist/version.txt", "v3.1.0\n")

    fixtures = iter((backend_fixture, frontend_fixture))

    def download(_url, destination, downloaded_before, _total_hint):
        source = next(fixtures)
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_bytes(source.read_bytes())
        size = destination.stat().st_size
        return downloaded_before + size, size

    monkeypatch.setattr(manager, "_download_file", download)
    monkeypatch.setattr(update_module, "is_docker", lambda: True)
    monkeypatch.setattr(
        manager,
        "_fetch_frontend_release",
        lambda _version: {
            "assets": [
                {
                    "name": "dist.zip",
                    "size": frontend_fixture.stat().st_size,
                    "browser_download_url": "https://example.invalid/dist.zip",
                }
            ]
        },
    )
    manager._write_state(state="downloading", version="v3.1.0")
    manager._active_target = "application"

    manager._download_update("application")

    prepared = json.loads((manager._root / "prepared.json").read_text(encoding="utf-8"))
    assert prepared["resource_package_version"] == "10"
    assert prepared["resource_files"][0]["name"] == "user.sites.v3.bin"


def test_resource_manifest_requires_complete_current_platform_package(monkeypatch, tmp_path):
    """
    资源安装只能接受同时包含索引和当前平台认证文件的完整包。
    """
    manager = _manager(monkeypatch, tmp_path)
    monkeypatch.setattr(
        update_module.ResourceHelper,
        "_get_needed_files",
        classmethod(lambda cls: ["user.sites.v3.bin", "sites.cpython-test.so"]),
    )
    resource_dir = manager._resource_dir
    resource_dir.mkdir(parents=True)
    resource_file = resource_dir / "user.sites.v3.bin"
    resource_file.write_bytes(b"resource")
    prepared = {
        "resource_files": [
            {
                "name": resource_file.name,
                "path": str(resource_file),
                "sha256": manager._sha256(resource_file),
            }
        ]
    }

    with pytest.raises(RuntimeError, match="完整资源包"):
        manager._validate_resource_manifest(prepared)


def test_request_install_rejects_modified_prepared_package(monkeypatch, tmp_path):
    """
    已准备的更新包被修改后，安装请求必须拒绝校验失败的内容。
    """
    manager = _manager(monkeypatch, tmp_path)
    manager._root.mkdir(parents=True)
    manager._backend_archive.write_bytes(b"backend")
    manager._frontend_archive.write_bytes(b"frontend")
    (manager._root / "prepared.json").write_text(
        json.dumps(
            {
                "version": "v3.1.0",
                "frontend_version": "v3.1.0",
                "backend_archive": str(manager._backend_archive),
                "frontend_archive": str(manager._frontend_archive),
                "backend_sha256": "invalid",
                "frontend_sha256": manager._sha256(manager._frontend_archive),
            }
        ),
        encoding="utf-8",
    )
    manager._write_state(state="ready", version="v3.1.0", can_install=True)

    success, message = manager.request_install()

    assert success is False
    assert "后端更新包校验失败" in message
    assert not manager._install_file.exists()
    assert manager.get_status().state == "failed"


def test_cancel_install_returns_prepared_update_to_ready(monkeypatch, tmp_path):
    """
    取消安装等待后，已准备的更新恢复到可安装状态。
    """
    manager = _manager(monkeypatch, tmp_path)
    manager._root.mkdir(parents=True)
    manager._install_file.write_text("{}", encoding="utf-8")
    manager._write_state(state="installing", version="v3.1.0")

    manager.cancel_install("restart failed")

    status = manager.get_status()
    assert status.state == "ready"
    assert status.can_install is True
    assert status.error == "restart failed"
    assert not manager._install_file.exists()


@pytest.mark.parametrize(
    ("exdev_target", "failure"),
    [
        (None, None),
        ("app", None),
        ("public", None),
        ("app", "backup"),
        ("public", "backup"),
        (None, "stage"),
    ],
)
def test_apply_prepared_application_replaces_docker_payload_and_preserves_plugins(
    monkeypatch, tmp_path, exdev_target, failure
):
    """
    Docker 更新应兼容 OverlayFS，并在备份或切换失败时保留旧载荷。
    """
    manager = _docker_manager(monkeypatch, tmp_path)
    app_dir = manager._docker_app_dir
    public_dir = manager._docker_public_dir
    plugin_dir = app_dir / "app" / "plugins"
    resource_dir = app_dir / "app" / "application" / "site"
    plugin_dir.mkdir(parents=True)
    resource_dir.mkdir(parents=True)
    public_dir.mkdir()
    (plugin_dir / "__init__.py").write_text("# compatibility\n", encoding="utf-8")
    (plugin_dir / "local_plugin.py").write_text("local\n", encoding="utf-8")
    (resource_dir / "user.sites.v3.bin").write_text("old-resource\n", encoding="utf-8")
    (resource_dir / "legacy.py").write_text("old-source\n", encoding="utf-8")
    (app_dir / "old.py").write_text("old\n", encoding="utf-8")
    (app_dir / "pyproject.toml").write_text("old-project\n", encoding="utf-8")
    (app_dir / "uv.lock").write_text("old-lock\n", encoding="utf-8")
    (public_dir / "index.html").write_text("old-front\n", encoding="utf-8")
    manager._root.mkdir(parents=True, exist_ok=True)

    with zipfile.ZipFile(manager._backend_archive, "w") as archive:
        archive.writestr(
            "MoviePilot-v3.1.0/version.py",
            "APP_VERSION = 'v3.1.0'\nFRONTEND_VERSION = 'v3.1.0'\n",
        )
        archive.writestr("MoviePilot-v3.1.0/pyproject.toml", "[project]\n")
        archive.writestr("MoviePilot-v3.1.0/uv.lock", "version = 1\n")
        archive.writestr("MoviePilot-v3.1.0/new.py", "new\n")
        archive.writestr(
            "MoviePilot-v3.1.0/app/plugins/__init__.py",
            "new compatibility\n",
        )
        archive.writestr(
            "MoviePilot-v3.1.0/app/application/site/__init__.py",
            "",
        )
        archive.writestr(
            "MoviePilot-v3.1.0/app/application/site/auth.py",
            "new-auth\n",
        )
    with zipfile.ZipFile(manager._frontend_archive, "w") as archive:
        archive.writestr("dist/index.html", "new-front\n")
        archive.writestr("dist/version.txt", "v3.1.0\n")

    prepared = {
        "targets": ["application"],
        "version": "v3.1.0",
        "frontend_version": "v3.1.0",
        "backend_archive": str(manager._backend_archive),
        "frontend_archive": str(manager._frontend_archive),
        "backend_sha256": manager._sha256(manager._backend_archive),
        "frontend_sha256": manager._sha256(manager._frontend_archive),
    }
    (manager._root / "prepared.json").write_text(
        json.dumps(prepared), encoding="utf-8"
    )
    manager._install_file.write_text(json.dumps(prepared), encoding="utf-8")
    sync_calls = []
    monkeypatch.setattr(
        manager,
        "_sync_docker_dependencies",
        lambda project_dir, **kwargs: sync_calls.append((project_dir, kwargs)),
    )

    original_replace = Path.replace
    original_copytree = update_module.shutil.copytree

    def replace(source, target):
        """
        模拟 OverlayFS 目录重命名限制及新载荷切换失败。
        """
        if exdev_target == "app" and source == app_dir:
            raise OSError(errno.EXDEV, "Invalid cross-device link")
        if exdev_target == "public" and source == public_dir:
            raise OSError(errno.EXDEV, "Invalid cross-device link")
        if failure == "stage" and source.name == "App" and target == app_dir:
            raise OSError(errno.EIO, "stage install failed")
        return original_replace(source, target)

    def copytree(source, target, *args, **kwargs):
        """
        模拟 OverlayFS 备份复制失败，确保旧目录未被删除。
        """
        if failure == "backup" and Path(target).name.endswith(".__update_previous__"):
            raise OSError(errno.ENOSPC, "backup failed")
        return original_copytree(source, target, *args, **kwargs)

    monkeypatch.setattr(Path, "replace", replace)
    monkeypatch.setattr(update_module.shutil, "copytree", copytree)

    success, message = manager.apply_prepared_update()

    if failure:
        assert success is False
        assert (app_dir / "old.py").read_text(encoding="utf-8") == "old\n"
        assert (public_dir / "index.html").read_text(encoding="utf-8") == "old-front\n"
        assert not (app_dir / "new.py").exists()
        assert (manager._root / "prepared.json").exists()
        assert not manager._docker_previous_app_dir.exists()
        assert not manager._docker_previous_public_dir.exists()
        return

    assert success is True
    assert message == "已下载的更新已替换到 Docker 程序目录"
    assert len(sync_calls) == 1
    assert sync_calls[0][0].name == "App"
    assert sync_calls[0][1] == {}
    assert (app_dir / "new.py").read_text(encoding="utf-8") == "new\n"
    assert not (app_dir / "old.py").exists()
    assert (app_dir / "app" / "plugins" / "__init__.py").read_text(
        encoding="utf-8"
    ) == "new compatibility\n"
    assert (app_dir / "app" / "plugins" / "local_plugin.py").exists()
    assert (resource_dir / "user.sites.v3.bin").read_text(encoding="utf-8") == "old-resource\n"
    assert (resource_dir / "auth.py").read_text(encoding="utf-8") == "new-auth\n"
    assert not (resource_dir / "legacy.py").exists()
    assert (public_dir / "index.html").read_text(encoding="utf-8") == "new-front\n"
    assert not manager._install_file.exists()
    assert not (manager._root / "prepared.json").exists()
    assert not manager._docker_pending_file.exists()
    assert not manager._docker_previous_app_dir.exists()
    assert not manager._docker_previous_public_dir.exists()


@pytest.mark.parametrize("failure", [None, "backup", "install"])
def test_apply_prepared_resources_replaces_complete_docker_resource_package(
    monkeypatch, tmp_path, failure
):
    """
    镜像层目录禁止重命名时仍能更新，备份或安装失败则保留旧资源。
    """
    manager = _docker_manager(monkeypatch, tmp_path)
    monkeypatch.setattr(
        update_module.ResourceHelper,
        "_get_needed_files",
        classmethod(lambda cls: ["user.sites.v3.bin", "sites.cpython-test.so"]),
    )
    monkeypatch.setattr(update_module, "get_resource_versions", lambda: ("1", "1"))
    resource_dir = manager._docker_app_dir / "app" / "application" / "site"
    resource_dir.mkdir(parents=True)
    (manager._docker_app_dir / "keep.py").parent.mkdir(parents=True, exist_ok=True)
    (manager._docker_app_dir / "keep.py").write_text("keep\n", encoding="utf-8")
    (resource_dir / "user.sites.v3.bin").write_bytes(b"old-index")
    (resource_dir / "sites.cpython-old.so").write_bytes(b"old-native")
    prepared_files = []
    for name, content in (
        ("user.sites.v3.bin", b"new-index"),
        ("sites.cpython-test.so", b"new-native"),
    ):
        path = manager._resource_dir / name
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(content)
        prepared_files.append(
            {"name": name, "path": str(path), "sha256": manager._sha256(path)}
        )
    prepared = {
        "targets": ["resources"],
        "resource_package_version": "10",
        "resource_files": prepared_files,
    }
    manager._root.mkdir(parents=True, exist_ok=True)
    (manager._root / "prepared.json").write_text(
        json.dumps(prepared), encoding="utf-8"
    )
    manager._install_file.write_text(json.dumps(prepared), encoding="utf-8")

    original_replace = Path.replace
    original_copytree = update_module.shutil.copytree

    def replace(source, target):
        """
        模拟 OverlayFS 镜像层重命名限制及新资源提交失败。
        """
        if source == resource_dir:
            raise OSError(errno.EXDEV, "Invalid cross-device link")
        if failure == "install" and source.parent.name.startswith(".moviepilot-resource-update-"):
            raise OSError(errno.EIO, "install failed")
        return original_replace(source, target)

    def copytree(source, target, *args, **kwargs):
        """
        模拟备份复制失败，确保尚未触碰运行目录。
        """
        if failure == "backup" and Path(target).name.endswith(".__prepared_previous__"):
            raise OSError(errno.ENOSPC, "backup failed")
        return original_copytree(source, target, *args, **kwargs)

    monkeypatch.setattr(Path, "replace", replace)
    monkeypatch.setattr(update_module.shutil, "copytree", copytree)
    success, _message = manager.apply_prepared_update()

    if failure:
        assert success is False
        assert (resource_dir / "user.sites.v3.bin").read_bytes() == b"old-index"
        assert (resource_dir / "sites.cpython-old.so").read_bytes() == b"old-native"
        assert not (resource_dir / "sites.cpython-test.so").exists()
        assert (manager._root / "prepared.json").exists()
        return
    assert success is True
    assert (manager._docker_app_dir / "keep.py").read_text(encoding="utf-8") == "keep\n"
    assert (resource_dir / "user.sites.v3.bin").read_bytes() == b"new-index"
    assert (resource_dir / "sites.cpython-test.so").read_bytes() == b"new-native"
    assert not (resource_dir / "sites.cpython-old.so").exists()
    assert not manager._install_file.exists()
    assert not (manager._root / "prepared.json").exists()


@pytest.mark.parametrize(
    ("current", "expected_state"),
    [("v3.0.10", "available"), ("v3.0.10-1", "idle")],
)
def test_check_accepts_hotfix_release_suffix(monkeypatch, tmp_path, current, expected_state):
    """
    -N 后缀的临时修复版本视为稳定版，并按数字大于同号正式版。
    """
    manager = _manager(monkeypatch, tmp_path)
    releases = [
        {"tag_name": "v3.0.11-beta", "prerelease": True, "draft": False},
        {"tag_name": "v3.0.10-1", "name": "v3.0.10-1", "prerelease": False, "draft": False},
        {"tag_name": "v3.0.10", "name": "v3.0.10", "prerelease": False, "draft": False},
    ]
    monkeypatch.setattr(manager, "_request", lambda: SimpleNamespace(get_res=lambda _url: _response(releases)))
    monkeypatch.setattr(update_module, "get_app_version", lambda: current)

    status = manager.check("application")

    assert status.state == expected_state
    if expected_state == "available":
        assert status.version == "v3.0.10-1"