diff --git a/BUILD_EXE.md b/BUILD_EXE.md index 1d3e9ad..aac8a76 100644 --- a/BUILD_EXE.md +++ b/BUILD_EXE.md @@ -13,10 +13,11 @@ chmod +x build_linux.sh ./build_linux.sh ``` -Результат: `dist-linux/plate_config_app-linux-x86_64` и папки конфигураций -рядом с ним. Именно бинарник с этим именем публикуется как release-артефакт для -автообновления. Для работы с USB-COM пользователь Debian должен состоять в -группе `dialout`. +Результат: бинарник `dist-linux/plate_config_app-linux-x86_64`, папки +конфигураций рядом с ним и архив +`dist-linux/plate_config_app-linux-x86_64.zip`. Для автообновления публикуется +ZIP-архив. Для работы с USB-COM пользователь Debian должен состоять в группе +`dialout`. ## Предварительные требования @@ -30,7 +31,9 @@ chmod +x build_linux.sh .\build_exe.ps1 -InstallBuildDeps ``` -Готовый файл: `dist\plate_config_app-windows-x86_64.exe` +Готовые файлы: `dist\plate_config_app-windows-x86_64.exe` и +`dist\plate_config_app-windows-x86_64.zip`. Для автообновления публикуется +ZIP-архив. ## Команды по отдельности @@ -52,7 +55,8 @@ python -m pip install -r requirements-build.txt - запускает PyInstaller по `plate_config_app.spec` - переименовывает бинарник в `plate_config_app-windows-x86_64.exe` - копирует папки `light_modes/`, `input_mappings/`, `commands/` в `dist/` -- выводит путь и размер готового EXE +- создаёт `plate_config_app-windows-x86_64.zip` с бинарником и ресурсами +- выводит пути и размеры готовых EXE и ZIP ### 3. Только очистка (без сборки) @@ -94,4 +98,5 @@ Hidden imports: `qtawesome`, `Crypto.Cipher._raw_aes`, `Crypto.Util._raw_api` | `commands/` | Наборы команд (.csv) | Эти файлы можно редактировать без пересборки — достаточно изменить их -рядом с EXE и перезапустить приложение. +рядом с EXE и перезапустить приложение. При установке OTA-обновления каталоги, +которые входят в новый ZIP-пакет, полностью заменяются их новыми версиями. diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 3b77d3c..cd71783 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -201,5 +201,6 @@ health-check) вынесены в `QThread`-подклассы. Связь с UI .\build_exe.ps1 -InstallBuildDeps ``` -Результат: `dist\plate_config_app-windows-x86_64.exe` + папки -`dist\light_modes\`, `dist\input_mappings\`, `dist\commands\`. +Результат: `dist\plate_config_app-windows-x86_64.exe`, ZIP-архив +`dist\plate_config_app-windows-x86_64.zip` и папки `dist\light_modes\`, +`dist\input_mappings\`, `dist\commands\`. diff --git a/README.md b/README.md index 49eae4d..f16eb25 100644 --- a/README.md +++ b/README.md @@ -59,15 +59,17 @@ под настройками только тогда, когда удалённый тег формата `vX.Y.Z` численно новее: сначала сравнивается major, затем minor и patch. -Приложение скачивает исполняемый release-артефакт текущей платформы, проверяет -его формат, завершает старый процесс, заменяет бинарник и запускается заново. -Файл `settings.json` и пользовательские папки ресурсов не заменяются. Для -работы проверки релизы репозитория должны быть доступны без авторизации. +Приложение скачивает ZIP-пакет текущей платформы, проверяет его целостность и +безопасность, завершает старый процесс, заменяет бинарник и все каталоги из +пакета, а затем запускается заново. Перед заменой создаётся резервная копия для +автоматического восстановления при ошибке. Файл `settings.json` и другие файлы, +которых нет в архиве, сохраняются. Для работы проверки релизы репозитория должны +быть доступны без авторизации. Имена release-артефактов: -- Windows x86_64: `plate_config_app-windows-x86_64.exe`; -- Linux x86_64: `plate_config_app-linux-x86_64`. +- Windows x86_64: `plate_config_app-windows-x86_64.zip`; +- Linux x86_64: `plate_config_app-linux-x86_64.zip`. --- diff --git a/app/__init__.py b/app/__init__.py index 054714f..29f49db 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -1,3 +1,3 @@ """Plate Config App package.""" -__version__ = "2.0.0" +__version__ = "2.0.1" diff --git a/app/services/update_service.py b/app/services/update_service.py index f3ddc60..a4b8e01 100644 --- a/app/services/update_service.py +++ b/app/services/update_service.py @@ -6,12 +6,14 @@ import os import platform import re import shutil +import stat import subprocess import sys import uuid +import zipfile from collections.abc import Callable from dataclasses import dataclass -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import BinaryIO from urllib.parse import quote, unquote, urlparse from urllib.request import Request, urlopen @@ -27,9 +29,12 @@ HTTP_TIMEOUT_SECONDS = 10.0 DOWNLOAD_TIMEOUT_SECONDS = 30.0 DOWNLOAD_CHUNK_SIZE = 1024 * 1024 MAX_DOWNLOAD_BYTES = 300 * 1024 * 1024 +MAX_UNPACKED_BYTES = 600 * 1024 * 1024 +MAX_ARCHIVE_MEMBERS = 10_000 VERSION_PATTERN = re.compile( r"[vV]?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)" ) +TOP_LEVEL_NAME_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*") class UpdateError(RuntimeError): @@ -65,7 +70,7 @@ def parse_version(value: str) -> tuple[int, int, int]: return tuple(int(component) for component in match.groups()) -def platform_asset_name( +def platform_executable_name( *, system: str | None = None, machine: str | None = None, @@ -81,6 +86,30 @@ def platform_asset_name( raise UpdateError(f"Автообновление не поддерживает платформу {system_name}") +def platform_asset_name( + *, + system: str | None = None, + machine: str | None = None, +) -> str: + executable_name = platform_executable_name(system=system, machine=machine) + if executable_name.casefold().endswith(".exe"): + executable_name = executable_name[:-4] + return f"{executable_name}.zip" + + +def archive_executable_name(asset_name: str) -> str: + names = { + "plate_config_app-windows-x86_64.zip": ( + "plate_config_app-windows-x86_64.exe" + ), + "plate_config_app-linux-x86_64.zip": "plate_config_app-linux-x86_64", + } + try: + return names[asset_name] + except KeyError as exc: + raise UpdateError("Неизвестное имя архива обновления") from exc + + def extract_release_tag(final_url: str) -> str: parsed = urlparse(final_url) if parsed.scheme != "https" or parsed.hostname != UPDATE_HOST: @@ -142,10 +171,8 @@ def check_for_update( ) -def _validate_download(path: Path, asset_name: str) -> None: - with path.open("rb") as stream: - signature = stream.read(4) - if asset_name.casefold().endswith(".exe"): +def _validate_executable_signature(signature: bytes, executable_name: str) -> None: + if executable_name.casefold().endswith(".exe"): valid = signature.startswith(b"MZ") else: valid = signature == b"\x7fELF" @@ -153,6 +180,103 @@ def _validate_download(path: Path, asset_name: str) -> None: raise UpdateError("Сервер вернул файл, не похожий на исполняемый артефакт") +def _validated_archive_members( + archive: zipfile.ZipFile, + executable_name: str, +) -> list[zipfile.ZipInfo]: + members = archive.infolist() + if len(members) > MAX_ARCHIVE_MEMBERS: + raise UpdateError("В архиве обновления слишком много файлов") + + total_size = 0 + normalized_names: set[str] = set() + executable_member: zipfile.ZipInfo | None = None + for member in members: + raw_name = member.filename + if not raw_name or "\\" in raw_name or "\x00" in raw_name: + raise UpdateError("Архив обновления содержит небезопасный путь") + relative_path = PurePosixPath(raw_name.rstrip("/")) + if ( + relative_path.is_absolute() + or not relative_path.parts + or any(part in {"", ".", ".."} for part in relative_path.parts) + ): + raise UpdateError("Архив обновления содержит небезопасный путь") + if not TOP_LEVEL_NAME_PATTERN.fullmatch(relative_path.parts[0]): + raise UpdateError("Архив обновления содержит недопустимое имя") + + normalized_name = relative_path.as_posix().casefold() + if normalized_name in normalized_names: + raise UpdateError("Архив обновления содержит повторяющиеся пути") + normalized_names.add(normalized_name) + + unix_mode = member.external_attr >> 16 + file_type = stat.S_IFMT(unix_mode) + if file_type not in {0, stat.S_IFREG, stat.S_IFDIR}: + raise UpdateError("Архив обновления содержит ссылку или специальный файл") + if member.flag_bits & 0x1: + raise UpdateError("Зашифрованные архивы обновления не поддерживаются") + + if not member.is_dir(): + total_size += member.file_size + if total_size > MAX_UNPACKED_BYTES: + raise UpdateError("Распакованный размер обновления превышает предел") + if relative_path.as_posix() == executable_name: + if member.is_dir(): + raise UpdateError("Исполняемый файл обновления имеет неверный тип") + executable_member = member + + if executable_member is None: + raise UpdateError("В архиве обновления не найден исполняемый файл") + with archive.open(executable_member, "r") as executable: + _validate_executable_signature(executable.read(4), executable_name) + return members + + +def _validate_update_archive(path: Path, executable_name: str) -> None: + try: + with zipfile.ZipFile(path, "r") as archive: + _validated_archive_members(archive, executable_name) + bad_member = archive.testzip() + except (OSError, zipfile.BadZipFile, RuntimeError) as exc: + raise UpdateError(f"Не удалось проверить ZIP-архив обновления: {exc}") from exc + if bad_member is not None: + raise UpdateError(f"Повреждён файл в архиве обновления: {bad_member}") + + +def _extract_update_archive( + archive_path: Path, + staging_directory: Path, + executable_name: str, +) -> Path: + try: + staging_directory.mkdir() + with zipfile.ZipFile(archive_path, "r") as archive: + members = _validated_archive_members(archive, executable_name) + for member in members: + relative_path = PurePosixPath(member.filename.rstrip("/")) + destination = staging_directory.joinpath(*relative_path.parts) + if member.is_dir(): + destination.mkdir(parents=True, exist_ok=True) + continue + destination.parent.mkdir(parents=True, exist_ok=True) + with ( + archive.open(member, "r") as source, + destination.open("xb") as output, + ): + shutil.copyfileobj(source, output) + + executable_path = staging_directory / executable_name + with executable_path.open("rb") as executable: + _validate_executable_signature(executable.read(4), executable_name) + if not executable_name.casefold().endswith(".exe"): + executable_path.chmod(0o755) + return executable_path + except Exception: + shutil.rmtree(staging_directory, ignore_errors=True) + raise + + def download_update( info: UpdateInfo, target_directory: Path, @@ -162,7 +286,7 @@ def download_update( progress: Callable[[int], None] | None = None, cancelled: Callable[[], bool] | None = None, ) -> Path: - """Скачать и проверить новый бинарник рядом с текущим приложением.""" + """Скачать и проверить ZIP-пакет обновления рядом с приложением.""" target_directory = target_directory.resolve() if not target_directory.is_dir(): raise UpdateError("Каталог приложения не найден") @@ -213,9 +337,8 @@ def download_update( raise UpdateError("Сервер вернул пустой файл обновления") if total > 0 and received != total: raise UpdateError("Файл обновления скачан не полностью") - _validate_download(temporary_path, info.asset_name) - if not info.asset_name.casefold().endswith(".exe"): - temporary_path.chmod(0o755) + executable_name = archive_executable_name(info.asset_name) + _validate_update_archive(temporary_path, executable_name) if progress is not None: progress(100) return temporary_path @@ -233,26 +356,51 @@ def current_executable() -> Path: def _write_windows_updater(script_path: Path) -> None: script = r'''param( [int]$ApplicationProcessId, - [string]$Source, - [string]$Target + [string]$Staging, + [string]$Target, + [string]$Backup ) $ErrorActionPreference = "Stop" -$backup = "$Target.old" -$errorLog = Join-Path (Split-Path -Parent $Target) "update-error.log" +$applicationDirectory = Split-Path -Parent $Target +$errorLog = Join-Path $applicationDirectory "update-error.log" +$installed = [System.Collections.Generic.List[string]]::new() +$backedUp = [System.Collections.Generic.List[string]]::new() try { Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue - Remove-Item -LiteralPath $backup -Force -ErrorAction SilentlyContinue - if (Test-Path -LiteralPath $Target) { - Move-Item -LiteralPath $Target -Destination $backup -Force + Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue + New-Item -ItemType Directory -Path $Backup | Out-Null + $entries = @(Get-ChildItem -LiteralPath $Staging -Force) + foreach ($entry in $entries) { + $destination = Join-Path $applicationDirectory $entry.Name + if (Test-Path -LiteralPath $destination) { + Move-Item -LiteralPath $destination -Destination $Backup -Force + $backedUp.Add($entry.Name) + } + Move-Item -LiteralPath $entry.FullName -Destination $destination -Force + $installed.Add($entry.Name) } - Move-Item -LiteralPath $Source -Destination $Target -Force - Start-Process -FilePath $Target -WorkingDirectory (Split-Path -Parent $Target) - Remove-Item -LiteralPath $backup -Force -ErrorAction SilentlyContinue + if (-not (Test-Path -LiteralPath $Target -PathType Leaf)) { + throw "В пакете обновления отсутствует исполняемый файл" + } + Start-Process -FilePath $Target -WorkingDirectory $applicationDirectory + Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue + Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force } catch { - if (-not (Test-Path -LiteralPath $Target) -and (Test-Path -LiteralPath $backup)) { - Move-Item -LiteralPath $backup -Destination $Target -Force + foreach ($name in $installed) { + $destination = Join-Path $applicationDirectory $name + Remove-Item -LiteralPath $destination -Recurse -Force -ErrorAction SilentlyContinue } + foreach ($name in $backedUp) { + $backupItem = Join-Path $Backup $name + $destination = Join-Path $applicationDirectory $name + Remove-Item -LiteralPath $destination -Recurse -Force -ErrorAction SilentlyContinue + if (Test-Path -LiteralPath $backupItem) { + Move-Item -LiteralPath $backupItem -Destination $destination -Force + } + } + Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue + Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue $_ | Out-File -LiteralPath $errorLog -Encoding utf8 exit 1 } @@ -264,28 +412,60 @@ def _write_linux_updater(script_path: Path) -> None: script = r'''#!/bin/sh set -eu application_pid="$1" -source_path="$2" +staging_path="$2" target_path="$3" -backup_path="${target_path}.old" -error_log="$(dirname -- "$target_path")/update-error.log" +backup_path="$4" +application_dir="$(dirname -- "$target_path")" +error_log="$application_dir/update-error.log" +installed_file="$backup_path/.installed" +backed_up_file="$backup_path/.backed_up" +rollback() { + exit_code="$?" + trap - EXIT HUP INT TERM + if [ -f "$installed_file" ]; then + while IFS= read -r name; do + rm -rf -- "$application_dir/$name" + done < "$installed_file" + fi + if [ -f "$backed_up_file" ]; then + while IFS= read -r name; do + rm -rf -- "$application_dir/$name" + if [ -e "$backup_path/$name" ]; then + mv -f -- "$backup_path/$name" "$application_dir/$name" + fi + done < "$backed_up_file" + fi + rm -rf -- "$staging_path" "$backup_path" + printf '%s\n' "Не удалось заменить приложение" >"$error_log" + if [ "$exit_code" -eq 0 ]; then + exit_code=1 + fi + exit "$exit_code" +} +trap rollback EXIT HUP INT TERM while kill -0 "$application_pid" 2>/dev/null; do sleep 0.2 done -rm -f -- "$backup_path" -if [ -e "$target_path" ]; then - mv -f -- "$target_path" "$backup_path" -fi -if mv -f -- "$source_path" "$target_path"; then - chmod 755 -- "$target_path" - "$target_path" >/dev/null 2>&1 & - rm -f -- "$backup_path" "$0" -else - if [ ! -e "$target_path" ] && [ -e "$backup_path" ]; then - mv -f -- "$backup_path" "$target_path" +rm -rf -- "$backup_path" +mkdir -p -- "$backup_path" +: >"$installed_file" +: >"$backed_up_file" +for source_path in "$staging_path"/*; do + name="$(basename -- "$source_path")" + destination="$application_dir/$name" + if [ -e "$destination" ]; then + mv -f -- "$destination" "$backup_path/$name" + printf '%s\n' "$name" >>"$backed_up_file" fi - printf '%s\n' "Не удалось заменить приложение" >"$error_log" - exit 1 -fi + mv -f -- "$source_path" "$destination" + printf '%s\n' "$name" >>"$installed_file" +done +test -f "$target_path" +chmod 755 -- "$target_path" +"$target_path" >/dev/null 2>&1 & +trap - EXIT HUP INT TERM +rm -rf -- "$backup_path" "$staging_path" +rm -f -- "$0" ''' script_path.write_text(script, encoding="utf-8", newline="\n") script_path.chmod(0o700) @@ -298,61 +478,81 @@ def launch_update( process_id: int | None = None, system: str | None = None, ) -> None: - """Запустить внешний helper, который заменит бинарник после выхода.""" - source = downloaded_path.resolve() + """Распаковать обновление и запустить helper для замены пакета.""" + archive_path = downloaded_path.resolve() target = (executable_path or current_executable()).resolve() - if not source.is_file(): - raise UpdateError("Скачанный файл обновления не найден") - if source.parent != target.parent: - staged = target.parent / source.name - shutil.move(str(source), str(staged)) - source = staged.resolve() + if not archive_path.is_file(): + raise UpdateError("Скачанный архив обновления не найден") if not os.access(target.parent, os.W_OK): raise UpdateError("Нет прав на замену приложения") updater_id = uuid.uuid4().hex application_pid = process_id or os.getpid() system_name = (system or platform.system()).casefold() + if system_name not in {"windows", "linux"}: + raise UpdateError(f"Автообновление не поддерживает платформу {system_name}") + + staging_path = target.parent / f".plate_config_update-{updater_id}" + backup_path = target.parent / f".plate_config_backup-{updater_id}" + script_path: Path | None = None + try: + _extract_update_archive(archive_path, staging_path, target.name) + archive_path.unlink() + except Exception: + shutil.rmtree(staging_path, ignore_errors=True) + raise + if system_name == "windows": script_path = target.parent / f".plate_config_updater-{updater_id}.ps1" - _write_windows_updater(script_path) - creation_flags = ( - getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) - | getattr(subprocess, "CREATE_NO_WINDOW", 0) - ) - subprocess.Popen( - [ - "powershell.exe", - "-NoProfile", - "-ExecutionPolicy", - "Bypass", - "-File", - str(script_path), - str(application_pid), - str(source), - str(target), - ], - close_fds=True, - creationflags=creation_flags, - ) + try: + _write_windows_updater(script_path) + creation_flags = ( + getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + | getattr(subprocess, "CREATE_NO_WINDOW", 0) + ) + subprocess.Popen( + [ + "powershell.exe", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + str(script_path), + str(application_pid), + str(staging_path), + str(target), + str(backup_path), + ], + close_fds=True, + creationflags=creation_flags, + ) + except Exception: + script_path.unlink(missing_ok=True) + shutil.rmtree(staging_path, ignore_errors=True) + raise return if system_name == "linux": script_path = target.parent / f".plate_config_updater-{updater_id}.sh" - _write_linux_updater(script_path) - subprocess.Popen( - [ - "/bin/sh", - str(script_path), - str(application_pid), - str(source), - str(target), - ], - cwd=str(target.parent), - close_fds=True, - start_new_session=True, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) + try: + _write_linux_updater(script_path) + subprocess.Popen( + [ + "/bin/sh", + str(script_path), + str(application_pid), + str(staging_path), + str(target), + str(backup_path), + ], + cwd=str(target.parent), + close_fds=True, + start_new_session=True, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except Exception: + script_path.unlink(missing_ok=True) + shutil.rmtree(staging_path, ignore_errors=True) + raise return - raise UpdateError(f"Автообновление не поддерживает платформу {system_name}") diff --git a/build_exe.ps1 b/build_exe.ps1 index 920ab1f..4ba840d 100644 --- a/build_exe.ps1 +++ b/build_exe.ps1 @@ -82,6 +82,18 @@ foreach ($folder in $resourceFolders) { } } +$archive = Join-Path $projectRoot "dist\plate_config_app-windows-x86_64.zip" +$archiveEntries = @($exe) +foreach ($folder in $resourceFolders) { + $resourcePath = Join-Path $projectRoot "dist\$folder" + if (Test-Path $resourcePath) { + $archiveEntries += $resourcePath + } +} +Compress-Archive -LiteralPath $archiveEntries -DestinationPath $archive -Force + $size = [math]::Round((Get-Item $exe).Length / 1MB, 1) +$archiveSize = [math]::Round((Get-Item $archive).Length / 1MB, 1) Write-Host "" Write-Host "Сборка завершена: $exe ($size MB)" +Write-Host "ZIP-архив: $archive ($archiveSize MB)" diff --git a/build_linux.sh b/build_linux.sh index 732d8c9..5bfd15d 100644 --- a/build_linux.sh +++ b/build_linux.sh @@ -7,6 +7,7 @@ venv_dir="$project_root/.venv-linux" build_dir="$project_root/build-linux" dist_dir="$project_root/dist-linux" artifact_name="plate_config_app-linux-x86_64" +archive_name="plate_config_app-linux-x86_64.zip" cd "$project_root" @@ -41,5 +42,18 @@ Plate Config App для Debian 13 x86_64 sudo usermod -aG dialout "$USER" EOF +( + cd "$dist_dir" + "$venv_dir/bin/python" -m zipfile -c \ + "$archive_name" \ + "$artifact_name" \ + light_modes \ + input_mappings \ + commands \ + README.txt +) + printf 'Готово: %s\n' "$dist_dir/$artifact_name" sha256sum "$dist_dir/$artifact_name" +printf 'ZIP-архив: %s\n' "$dist_dir/$archive_name" +sha256sum "$dist_dir/$archive_name" diff --git a/tests/test_start_tab.py b/tests/test_start_tab.py index 874856c..d1779ce 100644 --- a/tests/test_start_tab.py +++ b/tests/test_start_tab.py @@ -34,7 +34,7 @@ class StartTabUpdateTests(unittest.TestCase): info = UpdateInfo( tag="v2.1.0", version="2.1.0", - asset_name="plate_config_app-windows-x86_64.exe", + asset_name="plate_config_app-windows-x86_64.zip", download_url="https://git.robotail.pro/download/test", ) @@ -153,7 +153,7 @@ class StartTabUpdateTests(unittest.TestCase): return UpdateInfo( tag="v2.1.0", version="2.1.0", - asset_name="plate_config_app-windows-x86_64.exe", + asset_name="plate_config_app-windows-x86_64.zip", download_url="https://git.robotail.pro/download/test", ) diff --git a/tests/test_update_service.py b/tests/test_update_service.py index 6718c45..5303718 100644 --- a/tests/test_update_service.py +++ b/tests/test_update_service.py @@ -10,6 +10,7 @@ import sys import tempfile import time import unittest +import zipfile from pathlib import Path from unittest.mock import patch @@ -18,6 +19,7 @@ from app.services.update_service import ( UpdateCancelled, UpdateError, UpdateInfo, + archive_executable_name, build_download_url, check_for_update, download_update, @@ -26,6 +28,7 @@ from app.services.update_service import ( normalize_version, parse_version, platform_asset_name, + platform_executable_name, ) @@ -56,6 +59,20 @@ class FakeResponse: return self._stream.read(size) +def make_update_archive( + executable_name: str, + executable_data: bytes, + *, + extra_files: dict[str, bytes] | None = None, +) -> bytes: + stream = io.BytesIO() + with zipfile.ZipFile(stream, "w", zipfile.ZIP_DEFLATED) as archive: + archive.writestr(executable_name, executable_data) + for name, data in (extra_files or {}).items(): + archive.writestr(name, data) + return stream.getvalue() + + class UpdateServiceTests(unittest.TestCase): def test_normalizes_version_prefix_and_whitespace(self) -> None: self.assertEqual("2.0.0", normalize_version(" v2.0.0 ")) @@ -84,12 +101,9 @@ class UpdateServiceTests(unittest.TestCase): self.assertEqual("v2.1.0", tag) - def test_rejects_login_redirect(self) -> None: - with self.assertRaises(UpdateError): - extract_release_tag("https://git.robotail.pro/user/login") - - def test_rejects_insecure_or_foreign_release_redirect(self) -> None: + def test_rejects_invalid_release_redirects(self) -> None: invalid_urls = ( + "https://git.robotail.pro/user/login", "http://git.robotail.pro/Robotail/plate-config-app/releases/tag/v2.1.0", "https://example.com/Robotail/plate-config-app/releases/tag/v2.1.0", ) @@ -98,49 +112,28 @@ class UpdateServiceTests(unittest.TestCase): extract_release_tag(url) def test_build_download_url_escapes_tag_and_asset(self) -> None: - url = build_download_url("v2.1.0 rc/1", "plate config.exe") + url = build_download_url("v2.1.0 rc/1", "plate config.zip") - self.assertTrue(url.endswith("/v2.1.0%20rc%2F1/plate%20config.exe")) + self.assertTrue(url.endswith("/v2.1.0%20rc%2F1/plate%20config.zip")) - def test_equal_tag_does_not_offer_update(self) -> None: - def opener(_request, timeout: float): - self.assertGreater(timeout, 0) - return FakeResponse( - "https://git.robotail.pro/Robotail/plate-config-app/" - "releases/tag/v2.0.0" - ) - - result = check_for_update( - "2.0.0", - opener=opener, - system="Windows", - machine="AMD64", + def test_equal_or_older_tag_does_not_offer_update(self) -> None: + cases = ( + ("2.0.0", "v2.0.0"), + ("2.0.0", "v1.99.99"), + ("2.5.0", "v2.4.99"), + ("2.5.4", "v2.5.3"), ) + for current, remote in cases: + with self.subTest(current=current, remote=remote): + result = check_for_update( + current, + opener=self._release_opener(remote), + system="Windows", + machine="AMD64", + ) + self.assertIsNone(result) - self.assertIsNone(result) - - def test_different_tag_selects_windows_asset(self) -> None: - def opener(_request, timeout: float): - self.assertGreater(timeout, 0) - return FakeResponse( - "https://git.robotail.pro/Robotail/plate-config-app/" - "releases/tag/v2.1.0" - ) - - result = check_for_update( - "2.0.0", - opener=opener, - system="Windows", - machine="x86_64", - ) - - self.assertIsNotNone(result) - assert result is not None - self.assertEqual("v2.1.0", result.tag) - self.assertEqual("plate_config_app-windows-x86_64.exe", result.asset_name) - self.assertTrue(result.download_url.endswith("/v2.1.0/" + result.asset_name)) - - def test_newer_major_minor_and_patch_versions_offer_update(self) -> None: + def test_newer_major_minor_and_patch_versions_offer_zip(self) -> None: cases = ( ("1.9.9", "v2.0.0"), ("2.4.9", "v2.5.0"), @@ -151,33 +144,16 @@ class UpdateServiceTests(unittest.TestCase): result = check_for_update( current, opener=self._release_opener(remote), - system="Linux", + system="Windows", machine="x86_64", ) self.assertIsNotNone(result) - - def test_older_major_minor_and_patch_versions_do_not_offer_update(self) -> None: - cases = ( - ("2.0.0", "v1.99.99"), - ("2.5.0", "v2.4.99"), - ("2.5.4", "v2.5.3"), - ) - for current, remote in cases: - with self.subTest(current=current, remote=remote): - result = check_for_update( - current, - opener=self._release_opener(remote), - system="Linux", - machine="x86_64", - ) - self.assertIsNone(result) + assert result is not None + self.assertEqual("plate_config_app-windows-x86_64.zip", result.asset_name) + self.assertTrue(result.download_url.endswith("/" + result.asset_name)) def test_rejects_invalid_current_or_remote_version(self) -> None: - cases = ( - ("development", "v2.1.0"), - ("2.0.0", "v2.1"), - ) - for current, remote in cases: + for current, remote in (("development", "v2.1.0"), ("2.0.0", "v2.1")): with self.subTest(current=current, remote=remote), self.assertRaises( UpdateError ): @@ -197,32 +173,38 @@ class UpdateServiceTests(unittest.TestCase): self.assertIsInstance(raised.exception.__cause__, OSError) - def test_platform_asset_names(self) -> None: + def test_platform_names(self) -> None: self.assertEqual( "plate_config_app-windows-x86_64.exe", + platform_executable_name(system="Windows", machine="AMD64"), + ) + self.assertEqual( + "plate_config_app-windows-x86_64.zip", platform_asset_name(system="Windows", machine="AMD64"), ) self.assertEqual( - "plate_config_app-linux-x86_64", + "plate_config_app-linux-x86_64.zip", platform_asset_name(system="Linux", machine="x86_64"), ) + self.assertEqual( + "plate_config_app-linux-x86_64", + archive_executable_name("plate_config_app-linux-x86_64.zip"), + ) - def test_rejects_unsupported_platform_or_architecture(self) -> None: + def test_rejects_unsupported_platform_architecture_or_asset(self) -> None: with self.assertRaisesRegex(UpdateError, "архитектуру"): platform_asset_name(system="Linux", machine="arm64") with self.assertRaisesRegex(UpdateError, "платформу"): platform_asset_name(system="Darwin", machine="x86_64") + with self.assertRaisesRegex(UpdateError, "Неизвестное"): + archive_executable_name("update.zip") - def test_downloads_and_validates_windows_binary(self) -> None: - data = b"MZ" + b"test-update" - info = UpdateInfo( - tag="v2.1.0", - version="2.1.0", - asset_name="plate_config_app-windows-x86_64.exe", - download_url=( - "https://git.robotail.pro/Robotail/plate-config-app/releases/" - "download/v2.1.0/plate_config_app-windows-x86_64.exe" - ), + def test_downloads_and_validates_windows_zip(self) -> None: + info = self._update_info() + data = make_update_archive( + "plate_config_app-windows-x86_64.exe", + b"MZnew", + extra_files={"commands/default.csv": b"04;Update"}, ) progress_values: list[int] = [] @@ -241,271 +223,317 @@ class UpdateServiceTests(unittest.TestCase): self.assertEqual(data, result.read_bytes()) self.assertEqual(100, progress_values[-1]) - @unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux") - def test_downloads_linux_binary_and_makes_it_executable(self) -> None: - data = b"\x7fELF" + b"test-update" + def test_downloads_and_validates_linux_zip(self) -> None: info = self._update_info(system="linux") + data = make_update_archive("plate_config_app-linux-x86_64", b"\x7fELFnew") with tempfile.TemporaryDirectory() as directory: result = download_update( info, Path(directory), - opener=lambda _request, timeout: FakeResponse( - info.download_url, - data, - content_length=len(data), - ), + opener=lambda _request, timeout: FakeResponse(info.download_url, data), ) - self.assertEqual(data, result.read_bytes()) - self.assertTrue(result.stat().st_mode & stat.S_IXUSR) - - def test_rejects_html_instead_of_binary(self) -> None: - info = UpdateInfo( - tag="v2.1.0", - version="2.1.0", - asset_name="plate_config_app-linux-x86_64", - download_url=( - "https://git.robotail.pro/Robotail/plate-config-app/releases/" - "download/v2.1.0/plate_config_app-linux-x86_64" - ), - ) - - with tempfile.TemporaryDirectory() as directory: - path = Path(directory) - with self.assertRaises(UpdateError): - download_update( - info, - path, - opener=lambda _request, timeout: FakeResponse( - info.download_url, - b"login", - ), - ) - self.assertEqual([], list(path.iterdir())) - - def test_rejects_incomplete_download(self) -> None: - data = b"MZpartial" - info = UpdateInfo( - tag="v2.1.0", - version="2.1.0", - asset_name="plate_config_app-windows-x86_64.exe", - download_url=( - "https://git.robotail.pro/Robotail/plate-config-app/releases/" - "download/v2.1.0/plate_config_app-windows-x86_64.exe" - ), - ) - - with tempfile.TemporaryDirectory() as directory: - with self.assertRaises(UpdateError): - download_update( - info, - Path(directory), - opener=lambda _request, timeout: FakeResponse( - info.download_url, - data, - content_length=len(data) + 10, - ), + with zipfile.ZipFile(result) as archive: + self.assertEqual( + ["plate_config_app-linux-x86_64"], + archive.namelist(), ) - def test_rejects_login_or_foreign_download_redirect(self) -> None: + def test_rejects_invalid_or_incomplete_download(self) -> None: info = self._update_info() - redirects = ( - "https://git.robotail.pro/user/login", - "https://example.com/update.exe", + valid_data = make_update_archive( + "plate_config_app-windows-x86_64.exe", + b"MZnew", ) - - for redirect in redirects: - with self.subTest(redirect=redirect), tempfile.TemporaryDirectory() as directory: - path = Path(directory) + cases = ( + (b"login", None), + (valid_data, len(valid_data) + 10), + (b"", None), + ) + for data, content_length in cases: + with self.subTest(size=len(data)), tempfile.TemporaryDirectory() as directory: + target = Path(directory) with self.assertRaises(UpdateError): download_update( info, - path, - opener=lambda _request, timeout, url=redirect: FakeResponse( - url, - b"MZupdate", + target, + opener=lambda _request, timeout, payload=data, length=content_length: ( + FakeResponse( + info.download_url, + payload, + content_length=length, + ) ), ) - self.assertEqual([], list(path.iterdir())) + self.assertEqual([], list(target.iterdir())) - def test_rejects_oversized_declared_download(self) -> None: + def test_rejects_login_or_foreign_download_redirect(self) -> None: info = self._update_info() + data = make_update_archive( + "plate_config_app-windows-x86_64.exe", + b"MZnew", + ) + redirects = ( + "https://git.robotail.pro/user/login", + "https://example.com/update.zip", + ) + for redirect in redirects: + with self.subTest(redirect=redirect), tempfile.TemporaryDirectory() as directory: + with self.assertRaises(UpdateError): + download_update( + info, + Path(directory), + opener=lambda _request, timeout, url=redirect: FakeResponse( + url, + data, + ), + ) + def test_rejects_oversized_downloads(self) -> None: + info = self._update_info() with tempfile.TemporaryDirectory() as directory, patch.object( update_service, "MAX_DOWNLOAD_BYTES", 4, ): - path = Path(directory) - with self.assertRaisesRegex(UpdateError, "превышает"): - download_update( - info, - path, - opener=lambda _request, timeout: FakeResponse( - info.download_url, - b"MZupdate", - content_length=8, - ), - ) - self.assertEqual([], list(path.iterdir())) - - def test_rejects_oversized_stream_without_content_length(self) -> None: - info = self._update_info() - - with tempfile.TemporaryDirectory() as directory, patch.object( - update_service, - "MAX_DOWNLOAD_BYTES", - 4, - ): - path = Path(directory) - with self.assertRaisesRegex(UpdateError, "превышает"): - download_update( - info, - path, - opener=lambda _request, timeout: FakeResponse( - info.download_url, - b"MZupdate", - ), - ) - self.assertEqual([], list(path.iterdir())) - - def test_rejects_empty_download(self) -> None: - info = self._update_info() - - with tempfile.TemporaryDirectory() as directory: - path = Path(directory) - with self.assertRaisesRegex(UpdateError, "пустой"): - download_update( - info, - path, - opener=lambda _request, timeout: FakeResponse( - info.download_url, - ), - ) - self.assertEqual([], list(path.iterdir())) + for content_length in (8, None): + with self.subTest(content_length=content_length), self.assertRaisesRegex( + UpdateError, + "превышает", + ): + download_update( + info, + Path(directory), + opener=lambda _request, timeout, length=content_length: FakeResponse( + info.download_url, + b"12345678", + content_length=length, + ), + ) def test_cancels_download_and_removes_partial_file(self) -> None: info = self._update_info() - with tempfile.TemporaryDirectory() as directory: - path = Path(directory) + target = Path(directory) with self.assertRaises(UpdateCancelled): download_update( info, - path, + target, opener=lambda _request, timeout: FakeResponse( info.download_url, - b"MZupdate", + b"partial", ), cancelled=lambda: True, ) - self.assertEqual([], list(path.iterdir())) + self.assertEqual([], list(target.iterdir())) def test_rejects_missing_or_read_only_target_directory(self) -> None: info = self._update_info() with tempfile.TemporaryDirectory() as directory: - missing = Path(directory) / "missing" with self.assertRaisesRegex(UpdateError, "не найден"): - download_update(info, missing) - + download_update(info, Path(directory) / "missing") with patch("app.services.update_service.os.access", return_value=False): with self.assertRaisesRegex(UpdateError, "Нет прав"): download_update(info, Path(directory)) - def test_prepares_windows_replacement_helper(self) -> None: + def test_rejects_archive_without_executable(self) -> None: + data = self._custom_archive({"commands/default.csv": b"04;Update"}) + + self._assert_invalid_archive(data, "исполняемый файл") + + def test_rejects_archive_path_traversal(self) -> None: + data = self._custom_archive( + { + "plate_config_app-windows-x86_64.exe": b"MZnew", + "../outside.txt": b"unsafe", + } + ) + + self._assert_invalid_archive(data, "небезопасный путь") + + def test_rejects_archive_symlink(self) -> None: + stream = io.BytesIO() + with zipfile.ZipFile(stream, "w") as archive: + archive.writestr("plate_config_app-windows-x86_64.exe", b"MZnew") + link = zipfile.ZipInfo("commands/link") + link.create_system = 3 + link.external_attr = (stat.S_IFLNK | 0o777) << 16 + archive.writestr(link, "../../outside") + + self._assert_invalid_archive(stream.getvalue(), "ссылку") + + def test_rejects_duplicate_archive_paths(self) -> None: + stream = io.BytesIO() + with zipfile.ZipFile(stream, "w") as archive: + archive.writestr("plate_config_app-windows-x86_64.exe", b"MZnew") + archive.writestr("commands/Test.csv", b"one") + archive.writestr("commands/test.csv", b"two") + + self._assert_invalid_archive(stream.getvalue(), "повторяющиеся") + + def test_rejects_excessive_unpacked_size(self) -> None: + data = make_update_archive( + "plate_config_app-windows-x86_64.exe", + b"MZnew", + extra_files={"commands/large.csv": b"12345678"}, + ) + with patch.object(update_service, "MAX_UNPACKED_BYTES", 8): + self._assert_invalid_archive(data, "Распакованный размер") + + def test_prepares_windows_package_replacement_helper(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) - source = root / ".downloaded-update" target = root / "plate_config_app-windows-x86_64.exe" - source.write_bytes(b"MZnew") + archive = root / ".downloaded-update.zip" target.write_bytes(b"MZold") + archive.write_bytes( + make_update_archive( + target.name, + b"MZnew", + extra_files={"commands/default.csv": b"new"}, + ) + ) with patch("app.services.update_service.subprocess.Popen") as popen: launch_update( - source, + archive, executable_path=target, process_id=123, system="Windows", ) scripts = list(root.glob(".plate_config_updater-*.ps1")) + staging = list(root.glob(".plate_config_update-*")) self.assertEqual(1, len(scripts)) + self.assertEqual(1, len(staging)) + self.assertFalse(archive.exists()) + self.assertEqual(b"new", (staging[0] / "commands/default.csv").read_bytes()) script = scripts[0].read_text(encoding="utf-8-sig") - self.assertIn("Wait-Process", script) - self.assertIn("Move-Item", script) + self.assertIn("$Staging", script) + self.assertIn("$backedUp", script) popen.assert_called_once() - creation_flags = popen.call_args.kwargs["creationflags"] - expected_flags = getattr(update_service.subprocess, "CREATE_NEW_PROCESS_GROUP", 0) - expected_flags |= getattr(update_service.subprocess, "CREATE_NO_WINDOW", 0) - self.assertEqual(expected_flags, creation_flags) - def test_prepares_linux_replacement_helper(self) -> None: + @unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux") + def test_prepares_linux_package_replacement_helper(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) - source = root / ".downloaded-update" target = root / "plate_config_app-linux-x86_64" - source.write_bytes(b"\x7fELFnew") + archive = root / ".downloaded-update.zip" target.write_bytes(b"\x7fELFold") + archive.write_bytes( + make_update_archive( + target.name, + b"\x7fELFnew", + extra_files={"commands/default.csv": b"new"}, + ) + ) with patch("app.services.update_service.subprocess.Popen") as popen: launch_update( - source, + archive, executable_path=target, process_id=123, system="Linux", ) scripts = list(root.glob(".plate_config_updater-*.sh")) + staging = list(root.glob(".plate_config_update-*")) self.assertEqual(1, len(scripts)) + self.assertEqual(1, len(staging)) + self.assertTrue((staging[0] / target.name).stat().st_mode & stat.S_IXUSR) script = scripts[0].read_text(encoding="utf-8") - self.assertIn("kill -0", script) - self.assertIn("mv -f", script) + self.assertIn("staging_path", script) + self.assertIn("rollback", script) popen.assert_called_once() - def test_stages_download_next_to_executable(self) -> None: + def test_extracts_external_archive_next_to_executable(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) - download_directory = root / "downloads" - application_directory = root / "application" - download_directory.mkdir() - application_directory.mkdir() - source = download_directory / ".downloaded-update" - target = application_directory / "plate_config_app-linux-x86_64" - source.write_bytes(b"\x7fELFnew") + downloads = root / "downloads" + application = root / "application" + downloads.mkdir() + application.mkdir() + target = application / "plate_config_app-linux-x86_64" + archive = downloads / "update.zip" target.write_bytes(b"\x7fELFold") + archive.write_bytes(make_update_archive(target.name, b"\x7fELFnew")) with patch("app.services.update_service.subprocess.Popen") as popen: launch_update( - source, + archive, executable_path=target, process_id=123, system="Linux", ) - staged = application_directory / source.name - self.assertFalse(source.exists()) - self.assertTrue(staged.exists()) - self.assertIn(str(staged.resolve()), popen.call_args.args[0]) + staging = list(application.glob(".plate_config_update-*")) + self.assertEqual(1, len(staging)) + self.assertFalse(archive.exists()) + self.assertIn(str(staging[0]), popen.call_args.args[0]) - def test_rejects_missing_download_or_unsupported_platform(self) -> None: + def test_failed_helper_launch_removes_staging_and_script(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + target = root / "plate_config_app-windows-x86_64.exe" + archive = root / "update.zip" + target.write_bytes(b"MZold") + archive.write_bytes(make_update_archive(target.name, b"MZnew")) + + with patch( + "app.services.update_service.subprocess.Popen", + side_effect=OSError("helper failed"), + ), self.assertRaisesRegex(OSError, "helper failed"): + launch_update( + archive, + executable_path=target, + process_id=123, + system="Windows", + ) + + self.assertEqual([], list(root.glob(".plate_config_update-*"))) + self.assertEqual([], list(root.glob(".plate_config_updater-*"))) + + def test_rejects_missing_archive_or_unsupported_platform(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) target = root / "plate_config_app" - source = root / ".downloaded-update" + archive = root / "update.zip" target.write_bytes(b"old") with self.assertRaisesRegex(UpdateError, "не найден"): - launch_update(source, executable_path=target, system="Linux") + launch_update(archive, executable_path=target, system="Linux") - source.write_bytes(b"new") + archive.write_bytes(b"not-a-zip") with self.assertRaisesRegex(UpdateError, "платформу"): - launch_update(source, executable_path=target, system="Darwin") + launch_update(archive, executable_path=target, system="Darwin") + + def _assert_invalid_archive(self, data: bytes, message: str) -> None: + info = self._update_info() + with tempfile.TemporaryDirectory() as directory: + target = Path(directory) + with self.assertRaisesRegex(UpdateError, message): + download_update( + info, + target, + opener=lambda _request, timeout: FakeResponse( + info.download_url, + data, + ), + ) + self.assertEqual([], list(target.iterdir())) + + @staticmethod + def _custom_archive(files: dict[str, bytes]) -> bytes: + stream = io.BytesIO() + with zipfile.ZipFile(stream, "w", zipfile.ZIP_DEFLATED) as archive: + for name, data in files.items(): + archive.writestr(name, data) + return stream.getvalue() @staticmethod def _update_info(*, system: str = "windows") -> UpdateInfo: - extension = ".exe" if system == "windows" else "" - asset_name = f"plate_config_app-{system}-x86_64{extension}" + asset_name = f"plate_config_app-{system}-x86_64.zip" return UpdateInfo( tag="v2.1.0", version="2.1.0", @@ -533,90 +561,135 @@ class ReplacementIntegrationTests(unittest.TestCase): return hashlib.sha256(path.read_bytes()).hexdigest() @staticmethod - def _wait_for_replacement(target: Path, expected_digest: str) -> None: + def _wait_for_replacement( + target: Path, + expected_digest: str, + replaced_resource: Path, + ) -> None: deadline = time.monotonic() + 20 while time.monotonic() < deadline: - if target.is_file() and ReplacementIntegrationTests._digest(target) == expected_digest: - if not list(target.parent.glob(".plate_config_updater-*")): - return + if ( + target.is_file() + and ReplacementIntegrationTests._digest(target) == expected_digest + and replaced_resource.read_text(encoding="utf-8") == "new" + and not list(target.parent.glob(".plate_config_updater-*")) + and not list(target.parent.glob(".plate_config_update-*")) + and not list(target.parent.glob(".plate_config_backup-*")) + ): + return time.sleep(0.1) - raise AssertionError("Внешний helper не заменил приложение за 20 секунд") + raise AssertionError("Внешний helper не заменил пакет за 20 секунд") @unittest.skipUnless(sys.platform == "win32", "Тест предназначен для Windows") - def test_windows_helper_replaces_binary(self) -> None: + def test_windows_helper_replaces_full_package(self) -> None: system_root = Path(os.environ.get("SystemRoot", r"C:\Windows")) with tempfile.TemporaryDirectory() as directory: root = Path(directory) - source = root / ".downloaded-update.exe" target = root / "plate_config_app-windows-x86_64.exe" - shutil.copy2(system_root / "System32" / "whoami.exe", source) + archive = root / "update.zip" + new_executable = (system_root / "System32" / "whoami.exe").read_bytes() shutil.copy2(system_root / "System32" / "where.exe", target) - expected_digest = self._digest(source) - - processes: list[subprocess.Popen] = [] - real_popen = subprocess.Popen - - def start_process(*args, **kwargs): - process = real_popen(*args, **kwargs) - processes.append(process) - return process - - with patch( - "app.services.update_service.subprocess.Popen", - side_effect=start_process, - ): - launch_update( - source, - executable_path=target, - process_id=2_147_483_647, - system="Windows", + self._prepare_old_files(root) + archive.write_bytes( + make_update_archive( + target.name, + new_executable, + extra_files={"commands/version.txt": b"new"}, ) + ) + expected_digest = hashlib.sha256(new_executable).hexdigest() + processes = self._launch_and_capture(archive, target, "Windows") - self._wait_for_replacement(target, expected_digest) + self._wait_for_replacement( + target, + expected_digest, + root / "commands/version.txt", + ) processes[0].wait(timeout=5) - self.assertFalse(target.with_suffix(target.suffix + ".old").exists()) - deadline = time.monotonic() + 10 - while True: - try: - target.unlink() - break - except PermissionError: - if time.monotonic() >= deadline: - raise - time.sleep(0.1) + self._assert_package_replaced(root) + self._wait_for_windows_unlock(target) @unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux") - def test_linux_helper_replaces_binary(self) -> None: + def test_linux_helper_replaces_full_package(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) - source = root / ".downloaded-update" target = root / "plate_config_app-linux-x86_64" - shutil.copy2("/bin/true", source) + archive = root / "update.zip" + new_executable = Path("/bin/true").read_bytes() shutil.copy2("/bin/false", target) - expected_digest = self._digest(source) - - processes: list[subprocess.Popen] = [] - real_popen = subprocess.Popen - - def start_process(*args, **kwargs): - process = real_popen(*args, **kwargs) - processes.append(process) - return process - - with patch( - "app.services.update_service.subprocess.Popen", - side_effect=start_process, - ): - launch_update( - source, - executable_path=target, - process_id=2_147_483_647, - system="Linux", + self._prepare_old_files(root) + archive.write_bytes( + make_update_archive( + target.name, + new_executable, + extra_files={"commands/version.txt": b"new"}, ) + ) + expected_digest = hashlib.sha256(new_executable).hexdigest() + processes = self._launch_and_capture(archive, target, "Linux") - self._wait_for_replacement(target, expected_digest) + self._wait_for_replacement( + target, + expected_digest, + root / "commands/version.txt", + ) processes[0].wait(timeout=5) - self.assertFalse(target.with_name(target.name + ".old").exists()) + self._assert_package_replaced(root) + self.assertTrue(target.stat().st_mode & stat.S_IXUSR) + + @staticmethod + def _prepare_old_files(root: Path) -> None: + commands = root / "commands" + commands.mkdir() + (commands / "version.txt").write_text("old", encoding="utf-8") + (commands / "obsolete.txt").write_text("remove", encoding="utf-8") + (root / "settings.json").write_text("preserve", encoding="utf-8") + + @staticmethod + def _assert_package_replaced(root: Path) -> None: + self_settings = (root / "settings.json").read_text(encoding="utf-8") + if self_settings != "preserve": + raise AssertionError("settings.json не сохранён") + if (root / "commands/obsolete.txt").exists(): + raise AssertionError("Старое содержимое каталога не удалено") + + @staticmethod + def _launch_and_capture( + archive: Path, + target: Path, + system: str, + ) -> list[subprocess.Popen]: + processes: list[subprocess.Popen] = [] + real_popen = subprocess.Popen + + def start_process(*args, **kwargs): + process = real_popen(*args, **kwargs) + processes.append(process) + return process + + with patch( + "app.services.update_service.subprocess.Popen", + side_effect=start_process, + ): + launch_update( + archive, + executable_path=target, + process_id=2_147_483_647, + system=system, + ) + return processes + + @staticmethod + def _wait_for_windows_unlock(target: Path) -> None: + deadline = time.monotonic() + 10 + while True: + try: + target.unlink() + return + except PermissionError: + if time.monotonic() >= deadline: + raise + time.sleep(0.1) if __name__ == "__main__":