Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 35c9693740 | |||
| 3a4491f0c4 |
19
BUILD_EXE.md
19
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-пакет, полностью заменяются их новыми версиями.
|
||||
|
||||
@ -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\`.
|
||||
|
||||
21
README.md
21
README.md
@ -59,15 +59,24 @@
|
||||
под настройками только тогда, когда удалённый тег формата `vX.Y.Z` численно
|
||||
новее: сначала сравнивается major, затем minor и patch.
|
||||
|
||||
Приложение скачивает исполняемый release-артефакт текущей платформы, проверяет
|
||||
его формат, завершает старый процесс, заменяет бинарник и запускается заново.
|
||||
Файл `settings.json` и пользовательские папки ресурсов не заменяются. Для
|
||||
работы проверки релизы репозитория должны быть доступны без авторизации.
|
||||
Приложение скачивает ZIP-пакет текущей платформы, проверяет его целостность и
|
||||
безопасность, завершает старый процесс, заменяет бинарник и все каталоги из
|
||||
пакета, а затем запускается заново. Перед заменой создаётся резервная копия;
|
||||
она удаляется только после подтверждения запуска новой версии. Если запуск
|
||||
не удался, предыдущая версия восстанавливается, а сообщение записывается
|
||||
в `update-error.log`. Файл `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`.
|
||||
|
||||
Для перехода с уже установленной v2.0.1 на v2.0.2 обновите приложение вручную:
|
||||
закройте его, скачайте ZIP-пакет своей платформы и распакуйте содержимое в
|
||||
каталог приложения с заменой прежних файлов. `settings.json` удалять не нужно.
|
||||
Начиная с v2.0.2 последующие обновления можно устанавливать кнопкой в приложении.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -1,3 +1,3 @@
|
||||
"""Plate Config App package."""
|
||||
|
||||
__version__ = "2.0.0"
|
||||
__version__ = "2.0.2"
|
||||
|
||||
@ -2,11 +2,13 @@ from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
from PySide6.QtCore import QTimer
|
||||
from PySide6.QtGui import QFont
|
||||
from PySide6.QtWidgets import QApplication
|
||||
|
||||
from .main_window import MainWindow
|
||||
from .services import settings_service
|
||||
from .services.update_service import signal_update_ready
|
||||
from .ui.icons import ICON_COLOR_PRIMARY, icon
|
||||
from .ui.theme import build_stylesheet, get_palette, create_qpalette
|
||||
|
||||
@ -27,4 +29,5 @@ def main() -> None:
|
||||
if not app_icon.isNull():
|
||||
window.setWindowIcon(app_icon)
|
||||
window.show()
|
||||
QTimer.singleShot(0, signal_update_ready)
|
||||
sys.exit(app.exec())
|
||||
|
||||
@ -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,13 @@ 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
|
||||
UPDATE_READY_ENV = "PLATE_CONFIG_UPDATE_READY_FILE"
|
||||
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 +71,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 +87,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 +172,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 +181,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 +287,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 +338,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
|
||||
@ -230,30 +354,112 @@ def current_executable() -> Path:
|
||||
return Path(sys.executable).resolve()
|
||||
|
||||
|
||||
def signal_update_ready() -> None:
|
||||
ready_file = os.environ.pop(UPDATE_READY_ENV, None)
|
||||
if ready_file:
|
||||
try:
|
||||
Path(ready_file).write_text(str(os.getpid()), encoding="ascii")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _write_windows_updater(script_path: Path) -> None:
|
||||
script = r'''param(
|
||||
[int]$ApplicationProcessId,
|
||||
[string]$Source,
|
||||
[string]$Target
|
||||
[string]$Staging,
|
||||
[string]$Target,
|
||||
[string]$Backup,
|
||||
[string]$ReadyFile
|
||||
)
|
||||
$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()
|
||||
$restarted = $null
|
||||
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) {
|
||||
$deadline = (Get-Date).AddSeconds(30)
|
||||
while ($true) {
|
||||
try {
|
||||
Move-Item -LiteralPath $destination -Destination $Backup -Force -ErrorAction Stop
|
||||
break
|
||||
} catch {
|
||||
if ((Get-Date) -ge $deadline) { throw }
|
||||
Start-Sleep -Milliseconds 250
|
||||
}
|
||||
}
|
||||
$backedUp.Add($entry.Name)
|
||||
}
|
||||
Move-Item -LiteralPath $entry.FullName -Destination $destination -Force -ErrorAction Stop
|
||||
$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 "В пакете обновления отсутствует исполняемый файл"
|
||||
}
|
||||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||||
$previousReadyFile = $env:PLATE_CONFIG_UPDATE_READY_FILE
|
||||
$previousResetEnvironment = $env:PYINSTALLER_RESET_ENVIRONMENT
|
||||
try {
|
||||
$env:PLATE_CONFIG_UPDATE_READY_FILE = $ReadyFile
|
||||
$env:PYINSTALLER_RESET_ENVIRONMENT = "1"
|
||||
$restarted = Start-Process -FilePath $Target -WorkingDirectory $applicationDirectory -PassThru
|
||||
} finally {
|
||||
$env:PLATE_CONFIG_UPDATE_READY_FILE = $previousReadyFile
|
||||
$env:PYINSTALLER_RESET_ENVIRONMENT = $previousResetEnvironment
|
||||
}
|
||||
$deadline = (Get-Date).AddSeconds(60)
|
||||
while (-not (Test-Path -LiteralPath $ReadyFile -PathType Leaf)) {
|
||||
if ($restarted.HasExited) { throw "Новая версия завершилась до запуска интерфейса" }
|
||||
if ((Get-Date) -ge $deadline) { throw "Новая версия не подтвердила запуск" }
|
||||
Start-Sleep -Milliseconds 250
|
||||
}
|
||||
if ($restarted.HasExited) { throw "Новая версия завершилась сразу после запуска" }
|
||||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||||
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
|
||||
$failure = $_
|
||||
if ($restarted -and -not $restarted.HasExited) {
|
||||
& taskkill.exe /PID $restarted.Id /T /F *> $null
|
||||
Start-Sleep -Milliseconds 500
|
||||
}
|
||||
$_ | Out-File -LiteralPath $errorLog -Encoding utf8
|
||||
$rollbackErrors = [System.Collections.Generic.List[string]]::new()
|
||||
foreach ($name in $installed) {
|
||||
$destination = Join-Path $applicationDirectory $name
|
||||
if (Test-Path -LiteralPath $destination) {
|
||||
try {
|
||||
Remove-Item -LiteralPath $destination -Recurse -Force -ErrorAction Stop
|
||||
} catch {
|
||||
$rollbackErrors.Add($_.ToString())
|
||||
}
|
||||
}
|
||||
}
|
||||
foreach ($name in $backedUp) {
|
||||
$backupItem = Join-Path $Backup $name
|
||||
$destination = Join-Path $applicationDirectory $name
|
||||
if (Test-Path -LiteralPath $backupItem) {
|
||||
try {
|
||||
Move-Item -LiteralPath $backupItem -Destination $destination -Force -ErrorAction Stop
|
||||
} catch {
|
||||
$rollbackErrors.Add($_.ToString())
|
||||
}
|
||||
}
|
||||
}
|
||||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||||
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
|
||||
if ($rollbackErrors.Count -eq 0) {
|
||||
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
|
||||
}
|
||||
@($failure.ToString()) + @($rollbackErrors) | Out-File -LiteralPath $errorLog -Encoding utf8
|
||||
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force -ErrorAction SilentlyContinue
|
||||
exit 1
|
||||
}
|
||||
'''
|
||||
@ -264,28 +470,89 @@ 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"
|
||||
ready_path="$5"
|
||||
application_dir="$(dirname -- "$target_path")"
|
||||
error_log="$application_dir/update-error.log"
|
||||
installed_file="$backup_path/.installed"
|
||||
backed_up_file="$backup_path/.backed_up"
|
||||
new_pid=""
|
||||
rollback() {
|
||||
exit_code="$?"
|
||||
trap - EXIT HUP INT TERM
|
||||
set +e
|
||||
if [ -n "$new_pid" ]; then
|
||||
kill "$new_pid" 2>/dev/null
|
||||
wait "$new_pid" 2>/dev/null
|
||||
fi
|
||||
rollback_failed=0
|
||||
if [ -f "$installed_file" ]; then
|
||||
while IFS= read -r name; do
|
||||
rm -rf -- "$application_dir/$name" || rollback_failed=1
|
||||
done < "$installed_file"
|
||||
fi
|
||||
if [ -f "$backed_up_file" ]; then
|
||||
while IFS= read -r name; do
|
||||
if [ -e "$backup_path/$name" ]; then
|
||||
mv -f -- "$backup_path/$name" "$application_dir/$name" || rollback_failed=1
|
||||
fi
|
||||
done < "$backed_up_file"
|
||||
fi
|
||||
rm -f -- "$ready_path"
|
||||
rm -rf -- "$staging_path"
|
||||
if [ "$rollback_failed" -eq 0 ]; then
|
||||
rm -rf -- "$backup_path"
|
||||
fi
|
||||
printf '%s\n' "Не удалось заменить приложение" >"$error_log"
|
||||
rm -f -- "$0"
|
||||
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"
|
||||
mv -f -- "$source_path" "$destination"
|
||||
printf '%s\n' "$name" >>"$installed_file"
|
||||
done
|
||||
test -f "$target_path"
|
||||
chmod 755 -- "$target_path"
|
||||
rm -f -- "$ready_path"
|
||||
PLATE_CONFIG_UPDATE_READY_FILE="$ready_path" PYINSTALLER_RESET_ENVIRONMENT=1 "$target_path" >/dev/null 2>&1 &
|
||||
new_pid=$!
|
||||
attempts=0
|
||||
while [ ! -f "$ready_path" ]; do
|
||||
if ! kill -0 "$new_pid" 2>/dev/null; then
|
||||
exit 1
|
||||
fi
|
||||
attempts=$((attempts + 1))
|
||||
if [ "$attempts" -ge 240 ]; then
|
||||
exit 1
|
||||
fi
|
||||
sleep 0.25
|
||||
done
|
||||
if ! kill -0 "$new_pid" 2>/dev/null; then
|
||||
exit 1
|
||||
fi
|
||||
trap - EXIT HUP INT TERM
|
||||
rm -f -- "$ready_path"
|
||||
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 +565,84 @@ 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}"
|
||||
ready_path = target.parent / f".plate_config_ready-{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),
|
||||
str(ready_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),
|
||||
str(ready_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}")
|
||||
|
||||
@ -5,7 +5,6 @@ from pathlib import Path
|
||||
from PySide6.QtCore import QThread, QTimer, Qt, Signal
|
||||
from PySide6.QtGui import QColor
|
||||
from PySide6.QtWidgets import (
|
||||
QApplication,
|
||||
QFrame,
|
||||
QGraphicsDropShadowEffect,
|
||||
QHBoxLayout,
|
||||
@ -230,9 +229,7 @@ class StartTab(QWidget):
|
||||
path.unlink(missing_ok=True)
|
||||
self._on_update_failed(str(exc))
|
||||
return
|
||||
app = QApplication.instance()
|
||||
if app is not None:
|
||||
app.quit()
|
||||
self.window().close()
|
||||
|
||||
def _on_update_failed(self, detail: str) -> None:
|
||||
self._update_btn.setText("Скачать обновление")
|
||||
|
||||
@ -54,10 +54,28 @@ if ($InstallBuildDeps) {
|
||||
# ── Сборка ─────────────────────────────────────────────────────────
|
||||
Write-Host ""
|
||||
Write-Host "=== Сборка EXE ==="
|
||||
& $venvPython -m PyInstaller --noconfirm --clean plate_config_app.spec
|
||||
$pythonHome = (& $venvPython -c "import sys; print(sys.base_prefix)").Trim()
|
||||
if ($LASTEXITCODE -ne 0 -or -not (Test-Path -LiteralPath $pythonHome -PathType Container)) {
|
||||
throw "Не удалось определить каталог Python для сборки."
|
||||
}
|
||||
$buildPath = @(
|
||||
(Split-Path -Parent $venvPython),
|
||||
$pythonHome,
|
||||
(Join-Path $pythonHome "DLLs"),
|
||||
(Join-Path $env:SystemRoot "System32"),
|
||||
$env:SystemRoot
|
||||
) -join ";"
|
||||
$originalPath = $env:PATH
|
||||
try {
|
||||
$env:PATH = $buildPath
|
||||
& $venvPython -m PyInstaller --noconfirm --clean plate_config_app.spec
|
||||
$buildExitCode = $LASTEXITCODE
|
||||
} finally {
|
||||
$env:PATH = $originalPath
|
||||
}
|
||||
|
||||
if ($LASTEXITCODE -ne 0) {
|
||||
throw "PyInstaller завершился с ошибкой (код $LASTEXITCODE)."
|
||||
if ($buildExitCode -ne 0) {
|
||||
throw "PyInstaller завершился с ошибкой (код $buildExitCode)."
|
||||
}
|
||||
|
||||
$builtExe = Join-Path $projectRoot "dist\plate_config_app.exe"
|
||||
@ -82,6 +100,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)"
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -4,7 +4,7 @@ import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
from unittest.mock import patch
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
@ -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",
|
||||
)
|
||||
|
||||
@ -98,20 +98,18 @@ class StartTabUpdateTests(unittest.TestCase):
|
||||
warning.assert_called_once()
|
||||
tab.shutdown()
|
||||
|
||||
def test_successful_download_launches_update_and_quits(self) -> None:
|
||||
def test_successful_download_launches_update_and_closes_window(self) -> None:
|
||||
tab = StartTab(lambda: None)
|
||||
downloaded = Path("downloaded-update")
|
||||
application = Mock()
|
||||
|
||||
with (
|
||||
patch("app.widgets.start_tab.launch_update") as launcher,
|
||||
patch("app.widgets.start_tab.QApplication") as application_class,
|
||||
patch.object(tab, "close") as close_window,
|
||||
):
|
||||
application_class.instance.return_value = application
|
||||
tab._on_update_downloaded(downloaded)
|
||||
|
||||
launcher.assert_called_once_with(downloaded)
|
||||
application.quit.assert_called_once_with()
|
||||
close_window.assert_called_once_with()
|
||||
self.assertEqual("Перезапуск…", tab._update_btn.text())
|
||||
tab.shutdown()
|
||||
|
||||
@ -153,7 +151,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",
|
||||
)
|
||||
|
||||
|
||||
@ -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,8 @@ from app.services.update_service import (
|
||||
normalize_version,
|
||||
parse_version,
|
||||
platform_asset_name,
|
||||
platform_executable_name,
|
||||
signal_update_ready,
|
||||
)
|
||||
|
||||
|
||||
@ -56,6 +60,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 +102,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 +113,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 +145,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 +174,49 @@ 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_signals_restart_only_when_ready_file_is_requested(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
marker = Path(directory) / "ready"
|
||||
with patch.dict(os.environ, {update_service.UPDATE_READY_ENV: str(marker)}):
|
||||
signal_update_ready()
|
||||
self.assertEqual(str(os.getpid()), marker.read_text(encoding="ascii"))
|
||||
self.assertNotIn(update_service.UPDATE_READY_ENV, os.environ)
|
||||
marker.unlink()
|
||||
signal_update_ready()
|
||||
self.assertFalse(marker.exists())
|
||||
|
||||
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 +235,323 @@ 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"<html>login</html>",
|
||||
),
|
||||
)
|
||||
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"<html>login</html>", 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)
|
||||
self.assertIn("AddSeconds(30)", script)
|
||||
self.assertIn("$restarted.HasExited", script)
|
||||
self.assertIn("$ReadyFile", script)
|
||||
self.assertIn('$env:PYINSTALLER_RESET_ENVIRONMENT = "1"', 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)
|
||||
self.assertIn("PLATE_CONFIG_UPDATE_READY_FILE", script)
|
||||
self.assertIn("PYINSTALLER_RESET_ENVIRONMENT=1", 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 +579,137 @@ 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_rollback(
|
||||
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.is_file()
|
||||
and replaced_resource.read_text(encoding="utf-8") == "old"
|
||||
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-*"))
|
||||
and (target.parent / "update-error.log").is_file()
|
||||
):
|
||||
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_rolls_back_when_new_app_exits_early(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 = self._digest(target)
|
||||
processes = self._launch_and_capture(archive, target, "Windows")
|
||||
|
||||
self._wait_for_replacement(target, expected_digest)
|
||||
self._wait_for_rollback(
|
||||
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_preserved(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_rolls_back_when_new_app_exits_early(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 = self._digest(target)
|
||||
processes = self._launch_and_capture(archive, target, "Linux")
|
||||
|
||||
self._wait_for_replacement(target, expected_digest)
|
||||
self._wait_for_rollback(
|
||||
target,
|
||||
expected_digest,
|
||||
root / "commands/version.txt",
|
||||
)
|
||||
processes[0].wait(timeout=5)
|
||||
self.assertFalse(target.with_name(target.name + ".old").exists())
|
||||
self._assert_package_preserved(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_preserved(root: Path) -> None:
|
||||
self_settings = (root / "settings.json").read_text(encoding="utf-8")
|
||||
if self_settings != "preserve":
|
||||
raise AssertionError("settings.json не сохранён")
|
||||
if not (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__":
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user