605 lines
20 KiB
Python
605 lines
20 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_rejects_package_version_mismatch() -> None:
|
||
offer = parse_update_offer(
|
||
{
|
||
"enabled": True,
|
||
"has_update": True,
|
||
"force": True,
|
||
"can_install": True,
|
||
"latest_version": "1.3.0",
|
||
"platform": "windows",
|
||
"arch": "x64",
|
||
"package": {
|
||
"type": PACKAGE_TYPE_INNO_SETUP,
|
||
"url": "https://cdn.example.com/DoctorWorkstation-Setup-1.1.0.exe",
|
||
"filename": "DoctorWorkstation-Setup-1.1.0.exe",
|
||
"sha256": "a" * 64,
|
||
"size": 1024,
|
||
},
|
||
},
|
||
current_version="1.2.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
|
||
assert "安装包版本 1.1.0 与发布版本 1.3.0 不一致" in offer.install_unavailable_reason
|
||
|
||
|
||
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,
|
||
ready_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,
|
||
ready_file=ready_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 "$ReadyFile" in script_text
|
||
assert "helper ready" in script_text
|
||
assert "Restart-Application" in script_text
|
||
|
||
|
||
def test_inno_helper_uses_runnable_flags_and_waits_for_ready(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
script = tmp_path / "install_update.ps1"
|
||
script.write_text("", encoding="utf-8")
|
||
installer = tmp_path / "Setup.exe"
|
||
restart_exe = tmp_path / "DoctorWorkstation.exe"
|
||
helper_log = tmp_path / "helper.log"
|
||
installer_log = tmp_path / "inno.log"
|
||
ready_file = tmp_path / "helper.ready"
|
||
captured: dict[str, object] = {}
|
||
|
||
class FakeProcess:
|
||
def poll(self) -> None:
|
||
return None
|
||
|
||
def fake_popen(args: list[str], **kwargs: object) -> FakeProcess:
|
||
captured["args"] = args
|
||
captured.update(kwargs)
|
||
ready_file.write_text("ready", encoding="utf-8")
|
||
return FakeProcess()
|
||
|
||
monkeypatch.setattr(app_update.subprocess, "DETACHED_PROCESS", 8, raising=False)
|
||
monkeypatch.setattr(app_update.subprocess, "CREATE_NEW_PROCESS_GROUP", 512, raising=False)
|
||
monkeypatch.setattr(app_update.subprocess, "CREATE_NO_WINDOW", 134217728, raising=False)
|
||
monkeypatch.setattr(app_update.subprocess, "Popen", fake_popen)
|
||
|
||
app_update._spawn_inno_setup_applier(
|
||
script,
|
||
installer=installer,
|
||
restart_exe=restart_exe,
|
||
helper_log_file=helper_log,
|
||
installer_log_file=installer_log,
|
||
ready_file=ready_file,
|
||
)
|
||
|
||
flags = int(captured["creationflags"])
|
||
detached = int(getattr(app_update.subprocess, "DETACHED_PROCESS", 0))
|
||
assert not detached or flags & detached == 0
|
||
assert flags & int(getattr(app_update.subprocess, "CREATE_NEW_PROCESS_GROUP", 0))
|
||
assert flags & int(getattr(app_update.subprocess, "CREATE_NO_WINDOW", 0))
|
||
assert "-ReadyFile" in captured["args"]
|
||
|
||
|
||
def test_inno_helper_reports_exit_before_ready(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
script = tmp_path / "install_update.ps1"
|
||
script.write_text("", encoding="utf-8")
|
||
|
||
class ExitedProcess:
|
||
def poll(self) -> int:
|
||
return 23
|
||
|
||
monkeypatch.setattr(app_update.subprocess, "Popen", lambda *args, **kwargs: ExitedProcess())
|
||
|
||
with pytest.raises(OSError, match="提前退出（代码 23）"):
|
||
app_update._spawn_inno_setup_applier(
|
||
script,
|
||
installer=tmp_path / "Setup.exe",
|
||
restart_exe=tmp_path / "DoctorWorkstation.exe",
|
||
helper_log_file=tmp_path / "helper.log",
|
||
installer_log_file=tmp_path / "inno.log",
|
||
ready_file=tmp_path / "helper.ready",
|
||
)
|
||
|
||
|
||
@pytest.mark.skipif(app_update.sys.platform != "win32", reason="Windows helper contract")
|
||
def test_inno_helper_executes_bootstrap_with_production_flags(tmp_path: Path) -> None:
|
||
script = tmp_path / "helper probe.ps1"
|
||
script.write_text(
|
||
"\n".join(
|
||
[
|
||
"param(",
|
||
"[int]$TargetPid, [string]$Installer, [string]$RestartExe,",
|
||
"[string]$HelperLogFile, [string]$InstallerLogFile, [string]$ReadyFile",
|
||
")",
|
||
'Set-Content -LiteralPath $ReadyFile -Value "ready" -Encoding UTF8',
|
||
]
|
||
),
|
||
encoding="utf-8-sig",
|
||
)
|
||
ready_file = tmp_path / "helper.ready"
|
||
|
||
app_update._spawn_inno_setup_applier(
|
||
script,
|
||
installer=tmp_path / "Setup.exe",
|
||
restart_exe=tmp_path / "DoctorWorkstation.exe",
|
||
helper_log_file=tmp_path / "helper.log",
|
||
installer_log_file=tmp_path / "inno.log",
|
||
ready_file=ready_file,
|
||
)
|
||
|
||
assert ready_file.read_text(encoding="utf-8-sig").strip() == "ready"
|
||
|
||
|
||
@pytest.mark.skipif(app_update.sys.platform != "win32", reason="Windows helper contract")
|
||
def test_inno_helper_survives_launcher_process_exit(tmp_path: Path) -> None:
|
||
script = tmp_path / "helper parent-exit probe.ps1"
|
||
script.write_text(
|
||
"\n".join(
|
||
[
|
||
"param(",
|
||
"[int]$TargetPid, [string]$Installer, [string]$RestartExe,",
|
||
"[string]$HelperLogFile, [string]$InstallerLogFile, [string]$ReadyFile",
|
||
")",
|
||
'Set-Content -LiteralPath $ReadyFile -Value "ready" -Encoding UTF8',
|
||
"while (Get-Process -Id $TargetPid -ErrorAction SilentlyContinue) {",
|
||
" Start-Sleep -Milliseconds 50",
|
||
"}",
|
||
'Set-Content -LiteralPath $HelperLogFile -Value "parent-exited" -Encoding UTF8',
|
||
]
|
||
),
|
||
encoding="utf-8-sig",
|
||
)
|
||
helper_log = tmp_path / "helper.log"
|
||
ready_file = tmp_path / "helper.ready"
|
||
launcher = (
|
||
"from pathlib import Path; import sys; "
|
||
"from doctor_workstation.services.app_update import _spawn_inno_setup_applier; "
|
||
"root=Path(sys.argv[1]); "
|
||
"_spawn_inno_setup_applier(root/'helper parent-exit probe.ps1', "
|
||
"installer=root/'Setup.exe', restart_exe=root/'DoctorWorkstation.exe', "
|
||
"helper_log_file=root/'helper.log', installer_log_file=root/'inno.log', "
|
||
"ready_file=root/'helper.ready')"
|
||
)
|
||
|
||
launched = app_update.subprocess.run(
|
||
[app_update.sys.executable, "-c", launcher, str(tmp_path)],
|
||
cwd=str(Path.cwd()),
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=10,
|
||
)
|
||
|
||
assert launched.returncode == 0, launched.stderr
|
||
deadline = app_update.time.monotonic() + 5.0
|
||
while not helper_log.is_file() and app_update.time.monotonic() < deadline:
|
||
app_update.time.sleep(0.05)
|
||
assert ready_file.is_file()
|
||
assert helper_log.read_text(encoding="utf-8-sig").strip() == "parent-exited"
|
||
|
||
|
||
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
|