Compare commits

...

2 Commits

Author SHA1 Message Date
35c9693740 fix: stabilize OTA updates in v2.0.2
- Исправлен перезапуск приложения после замены файлов и добавлено подтверждение запуска новой версии.
- Обеспечены откат при ошибке и корректное завершение фоновых задач.
- Исключены посторонние DLL из Windows-сборки, обновлены тесты и инструкция ручного перехода с v2.0.1.
2026-09-17 10:30:18 +03:00
3a4491f0c4 feat: add ZIP package updates
- Сборки Windows и Linux теперь дополнительно формируют ZIP-архивы с бинарником и ресурсами.

- OTA-проверка выбирает ZIP-артефакт текущей платформы вместо отдельного исполняемого файла.

- Добавлена безопасная проверка архивов, защита от traversal, ссылок, повреждений и превышения лимитов.

- Обновление транзакционно заменяет все верхнеуровневые файлы и каталоги пакета с резервным копированием и откатом.

- Пользовательские файлы, отсутствующие в ZIP, включая settings.json, сохраняются.

- Версия приложения повышена до 2.0.1, документация и кроссплатформенные тесты обновлены.
2026-09-16 14:07:01 +03:00
11 changed files with 857 additions and 417 deletions

View File

@ -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-пакет, полностью заменяются их новыми версиями.

View File

@ -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\`.

View File

@ -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 последующие обновления можно устанавливать кнопкой в приложении.
---

View File

@ -1,3 +1,3 @@
"""Plate Config App package."""
__version__ = "2.0.0"
__version__ = "2.0.2"

View File

@ -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())

View File

@ -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}")

View File

@ -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("Скачать обновление")

View File

@ -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)"

View File

@ -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"

View File

@ -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",
)

View File

@ -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__":