426 lines
14 KiB
Python
426 lines
14 KiB
Python
"""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 import app_update
|
|
from doctor_workstation.services.api_client import ApiClient
|
|
from doctor_workstation.services.app_update import (
|
|
PACKAGE_TYPE_ARCHIVE,
|
|
PACKAGE_TYPE_INNO_SETUP,
|
|
AppUpdateError,
|
|
UpdatePackage,
|
|
apply_extracted_update,
|
|
apply_inno_setup_update,
|
|
compare_version,
|
|
discover_payload,
|
|
download_package,
|
|
fetch_update_offer,
|
|
normalize_version,
|
|
package_filename,
|
|
parse_update_offer,
|
|
safe_extract_zip,
|
|
validate_installer_download_policy,
|
|
validate_windows_installer,
|
|
)
|
|
|
|
|
|
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_parse_offer_accepts_explicit_inno_setup_type() -> None:
|
|
offer = parse_update_offer(
|
|
{
|
|
"has_update": True,
|
|
"enabled": True,
|
|
"latest_version": "0.2.0",
|
|
"platform": "windows",
|
|
"arch": "x64",
|
|
"package": {
|
|
"url": "https://cdn.example.com/DoctorWorkstation-Setup.exe",
|
|
"sha256": "a" * 64,
|
|
"size": 123,
|
|
"filename": "DoctorWorkstation-Setup.exe",
|
|
"type": "inno_setup",
|
|
},
|
|
"can_install": True,
|
|
},
|
|
current_version="0.1.0",
|
|
platform_name="windows",
|
|
arch="x64",
|
|
)
|
|
assert offer.can_install is True
|
|
assert offer.package is not None
|
|
assert offer.package.type == PACKAGE_TYPE_INNO_SETUP
|
|
|
|
|
|
def test_parse_offer_disables_insecure_inno_setup_transport() -> None:
|
|
offer = parse_update_offer(
|
|
{
|
|
"has_update": True,
|
|
"force": True,
|
|
"enabled": True,
|
|
"latest_version": "0.2.0",
|
|
"platform": "windows",
|
|
"arch": "x64",
|
|
"package": {
|
|
"url": "http://cdn.example.com/DoctorWorkstation-Setup.exe",
|
|
"sha256": "a" * 64,
|
|
"type": "inno_setup",
|
|
},
|
|
"can_install": True,
|
|
},
|
|
current_version="0.1.0",
|
|
platform_name="windows",
|
|
arch="x64",
|
|
)
|
|
assert offer.has_update is True
|
|
assert offer.can_install is False
|
|
assert offer.force is False
|
|
assert offer.package is None
|
|
|
|
|
|
@pytest.mark.parametrize("package_type", ["msi", "script", "unknown"])
|
|
def test_parse_offer_rejects_unknown_package_type(package_type: str) -> None:
|
|
offer = parse_update_offer(
|
|
{
|
|
"has_update": True,
|
|
"latest_version": "0.2.0",
|
|
"platform": "windows",
|
|
"arch": "x64",
|
|
"package": {
|
|
"url": "https://cdn.example.com/update.bin",
|
|
"sha256": "a" * 64,
|
|
"type": package_type,
|
|
},
|
|
"can_install": True,
|
|
},
|
|
current_version="0.1.0",
|
|
platform_name="windows",
|
|
arch="x64",
|
|
)
|
|
assert offer.can_install is False
|
|
assert offer.force is False
|
|
assert offer.package is None
|
|
|
|
|
|
def test_parse_offer_rejects_stale_or_wrong_platform_response() -> None:
|
|
base = {
|
|
"has_update": True,
|
|
"latest_version": "0.1.0",
|
|
"platform": "windows",
|
|
"arch": "x64",
|
|
"can_install": False,
|
|
}
|
|
stale = parse_update_offer(
|
|
base,
|
|
current_version="0.1.0",
|
|
platform_name="windows",
|
|
arch="x64",
|
|
)
|
|
wrong_platform = parse_update_offer(
|
|
{**base, "latest_version": "0.2.0", "platform": "macos"},
|
|
current_version="0.1.0",
|
|
platform_name="windows",
|
|
arch="x64",
|
|
)
|
|
assert stale.has_update is False
|
|
assert wrong_platform.has_update is False
|
|
|
|
|
|
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()
|
|
|
|
|
|
def test_download_package_rejects_declared_size_mismatch(tmp_path: Path) -> None:
|
|
payload = b"short"
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
del request
|
|
return httpx.Response(200, content=payload)
|
|
|
|
destination = tmp_path / "pkg.exe"
|
|
with pytest.raises(AppUpdateError, match="文件大小"):
|
|
download_package(
|
|
"https://cdn.example.com/pkg.exe",
|
|
destination,
|
|
sha256=hashlib.sha256(payload).hexdigest(),
|
|
expected_size=len(payload) + 1,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
assert not destination.exists()
|
|
assert not (tmp_path / "pkg.exe.part").exists()
|
|
|
|
|
|
def test_windows_installer_download_policy_requires_verified_https() -> None:
|
|
with pytest.raises(AppUpdateError, match="HTTPS"):
|
|
validate_installer_download_policy(
|
|
"http://cdn.example.com/setup.exe",
|
|
verify_ssl=True,
|
|
)
|
|
with pytest.raises(AppUpdateError, match="证书校验"):
|
|
validate_installer_download_policy(
|
|
"https://cdn.example.com/setup.exe",
|
|
verify_ssl=False,
|
|
)
|
|
validate_installer_download_policy(
|
|
"http://127.0.0.1/setup.exe",
|
|
verify_ssl=True,
|
|
)
|
|
|
|
|
|
def test_validate_windows_installer_requires_exe_and_pe_header(tmp_path: Path) -> None:
|
|
installer = tmp_path / "Setup.exe"
|
|
installer.write_bytes(b"MZ" + b"\0" * 32)
|
|
assert validate_windows_installer(installer) == installer.resolve()
|
|
|
|
invalid = tmp_path / "invalid.exe"
|
|
invalid.write_bytes(b"PK")
|
|
with pytest.raises(AppUpdateError, match="PE"):
|
|
validate_windows_installer(invalid)
|
|
|
|
|
|
def test_inno_setup_applier_waits_installs_and_restarts_installed_exe(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
install_root = tmp_path / "installed"
|
|
install_root.mkdir()
|
|
installed_exe = install_root / "DoctorWorkstation.exe"
|
|
installed_exe.write_bytes(b"MZ")
|
|
installer = tmp_path / "DoctorWorkstation-Setup.exe"
|
|
installer.write_bytes(b"MZ" + b"\0" * 32)
|
|
spawned: dict[str, Path] = {}
|
|
|
|
monkeypatch.setattr(app_update.sys, "platform", "win32")
|
|
|
|
def capture_spawn(
|
|
script: Path,
|
|
*,
|
|
installer: Path,
|
|
restart_exe: Path,
|
|
helper_log_file: Path,
|
|
installer_log_file: Path,
|
|
) -> None:
|
|
spawned.update(
|
|
script=script,
|
|
installer=installer,
|
|
restart_exe=restart_exe,
|
|
helper_log_file=helper_log_file,
|
|
installer_log_file=installer_log_file,
|
|
)
|
|
|
|
monkeypatch.setattr(app_update, "_spawn_inno_setup_applier", capture_spawn)
|
|
apply_inno_setup_update(installer, install_root=install_root)
|
|
|
|
script_text = spawned["script"].read_text(encoding="utf-8-sig")
|
|
assert spawned["installer"] == installer.resolve()
|
|
assert spawned["restart_exe"] == installed_exe
|
|
assert "/VERYSILENT" in script_text
|
|
assert "/RESTARTEXITCODE=3010" in script_text
|
|
assert "/NOFORCECLOSEAPPLICATIONS" in script_text
|
|
assert "$HelperLogFile" in script_text
|
|
assert "$InstallerLogFile" in script_text
|
|
assert "Restart-Application" in script_text
|
|
|
|
|
|
def test_archive_applier_restarts_from_install_root(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
payload = tmp_path / "payload"
|
|
payload.mkdir()
|
|
(payload / "DoctorWorkstation.exe").write_bytes(b"MZ")
|
|
install_root = tmp_path / "installed"
|
|
install_root.mkdir()
|
|
installed_exe = install_root / "DoctorWorkstation.exe"
|
|
installed_exe.write_bytes(b"MZ")
|
|
captured: dict[str, Path] = {}
|
|
script = tmp_path / "apply.ps1"
|
|
script.write_text("", encoding="utf-8")
|
|
|
|
def capture_script(**kwargs: Path) -> Path:
|
|
captured.update(kwargs)
|
|
return script
|
|
|
|
monkeypatch.setattr(app_update, "_write_apply_script", capture_script)
|
|
monkeypatch.setattr(app_update, "_spawn_applier", lambda *args, **kwargs: None)
|
|
apply_extracted_update(payload, install_root=install_root)
|
|
assert captured["restart_exe"] == installed_exe
|
|
|
|
|
|
def test_inno_setup_applier_reports_helper_start_failure(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
install_root = tmp_path / "installed"
|
|
install_root.mkdir()
|
|
(install_root / "DoctorWorkstation.exe").write_bytes(b"MZ")
|
|
installer = tmp_path / "Setup.exe"
|
|
installer.write_bytes(b"MZ")
|
|
monkeypatch.setattr(app_update.sys, "platform", "win32")
|
|
|
|
def fail_spawn(*args: object, **kwargs: object) -> None:
|
|
del args, kwargs
|
|
raise OSError("blocked")
|
|
|
|
monkeypatch.setattr(app_update, "_spawn_inno_setup_applier", fail_spawn)
|
|
|
|
with pytest.raises(AppUpdateError, match="无法启动 Windows 更新助手"):
|
|
apply_inno_setup_update(installer, install_root=install_root)
|
|
|
|
|
|
def test_package_filename_defaults_match_package_type() -> None:
|
|
archive = UpdatePackage("https://cdn.example.com/", "a" * 64, 0, "")
|
|
installer = UpdatePackage(
|
|
"https://cdn.example.com/",
|
|
"a" * 64,
|
|
0,
|
|
"",
|
|
type=PACKAGE_TYPE_INNO_SETUP,
|
|
)
|
|
assert package_filename(archive, "0.2.0").endswith(".zip")
|
|
assert package_filename(installer, "0.2.0").endswith(".exe")
|
|
assert archive.type == PACKAGE_TYPE_ARCHIVE
|