zyt/app/tests/test_app_update.py

"""Desktop auto-update check, download and payload discovery."""

from __future__ import annotations

import hashlib
import zipfile
from pathlib import Path

import httpx
import pytest

from doctor_workstation.services.api_client import ApiClient
from doctor_workstation.services.app_update import (
    AppUpdateError,
    compare_version,
    discover_payload,
    download_package,
    fetch_update_offer,
    normalize_version,
    parse_update_offer,
    safe_extract_zip,
)


def test_normalize_and_compare_versions() -> None:
    assert normalize_version("0.2") == "0.2.0"
    assert normalize_version("1.2.3.4") == "1.2.3"
    assert normalize_version("nope") == ""
    assert compare_version("0.1.0", "0.2.0") < 0
    assert compare_version("0.2.0", "0.2.0") == 0
    assert compare_version("1.0.0", "0.9.9") > 0


def test_parse_offer_requires_hash_before_install() -> None:
    offer = parse_update_offer(
        {
            "has_update": True,
            "force": True,
            "enabled": True,
            "latest_version": "0.2.0",
            "package": {
                "url": "https://cdn.example.com/app.zip",
                "sha256": "",
                "size": 12,
                "filename": "app.zip",
            },
            "can_install": True,
        },
        current_version="0.1.0",
    )
    assert offer.has_update is True
    assert offer.can_install is False
    assert offer.force is False
    assert offer.package is None


def test_fetch_update_offer_uses_check_endpoint() -> None:
    requests: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        return httpx.Response(
            200,
            json={
                "code": 1,
                "data": {
                    "has_update": True,
                    "force": True,
                    "enabled": True,
                    "latest_version": "0.2.0",
                    "title": "医生工作站 0.2.0",
                    "notes": "修复登录",
                    "package": {
                        "url": "https://cdn.example.com/DoctorWorkstation.zip",
                        "sha256": "a" * 64,
                        "size": 2048,
                        "filename": "DoctorWorkstation.zip",
                    },
                    "can_install": True,
                },
            },
        )

    with ApiClient("https://example.test", transport=httpx.MockTransport(handler)) as client:
        offer = fetch_update_offer(
            client,
            current_version="0.1.0",
            platform_name="windows",
            arch="x64",
        )

    assert offer.has_update is True
    assert offer.force is True
    assert offer.can_install is True
    assert offer.package is not None
    assert "setting.desktop_workstation/check" in str(requests[0].url)
    assert "current_version=0.1.0" in str(requests[0].url)
    assert "platform=windows" in str(requests[0].url)


def test_safe_extract_rejects_zip_slip(tmp_path: Path) -> None:
    archive = tmp_path / "evil.zip"
    with zipfile.ZipFile(archive, "w") as bundle:
        bundle.writestr("../outside.txt", "nope")
    with pytest.raises(AppUpdateError, match="非法路径"):
        safe_extract_zip(archive, tmp_path / "out")


def test_discover_windows_payload_prefers_internal_onedir(tmp_path: Path) -> None:
    wrapped = tmp_path / "DoctorWorkstation"
    wrapped.mkdir()
    (wrapped / "_internal").mkdir()
    (wrapped / "DoctorWorkstation.exe").write_bytes(b"mz")
    (tmp_path / "Start_DoctorWorkstation.bat").write_text("start", encoding="utf-8")
    assert discover_payload(tmp_path, platform_name="windows") == wrapped


def test_discover_macos_payload_finds_app_bundle(tmp_path: Path) -> None:
    app = tmp_path / "DoctorWorkstation.app"
    macos = app / "Contents" / "MacOS"
    macos.mkdir(parents=True)
    (macos / "DoctorWorkstation").write_text("bin", encoding="utf-8")
    assert discover_payload(tmp_path, platform_name="macos") == app


def test_download_package_verifies_sha256_and_reports_progress(tmp_path: Path) -> None:
    payload = b"doctor-workstation-zip"
    digest = hashlib.sha256(payload).hexdigest()
    progress: list[tuple[int, int]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        del request
        return httpx.Response(
            200,
            content=payload,
            headers={"content-length": str(len(payload))},
        )

    destination = tmp_path / "pkg.zip"
    download_package(
        "https://cdn.example.com/pkg.zip",
        destination,
        sha256=digest,
        progress=lambda received, total: progress.append((received, total)),
        transport=httpx.MockTransport(handler),
    )
    assert destination.read_bytes() == payload
    assert progress[-1][0] == len(payload)


def test_download_package_rejects_hash_mismatch(tmp_path: Path) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        del request
        return httpx.Response(200, content=b"tampered")

    destination = tmp_path / "pkg.zip"
    with pytest.raises(AppUpdateError, match="校验失败"):
        download_package(
            "https://cdn.example.com/pkg.zip",
            destination,
            sha256="b" * 64,
            transport=httpx.MockTransport(handler),
        )
    assert not destination.exists()