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 ./build_linux.sh
``` ```
Результат: `dist-linux/plate_config_app-linux-x86_64` и папки конфигураций Результат: бинарник `dist-linux/plate_config_app-linux-x86_64`, папки
рядом с ним. Именно бинарник с этим именем публикуется как release-артефакт для конфигураций рядом с ним и архив
автообновления. Для работы с USB-COM пользователь Debian должен состоять в `dist-linux/plate_config_app-linux-x86_64.zip`. Для автообновления публикуется
группе `dialout`. ZIP-архив. Для работы с USB-COM пользователь Debian должен состоять в группе
`dialout`.
## Предварительные требования ## Предварительные требования
@ -30,7 +31,9 @@ chmod +x build_linux.sh
.\build_exe.ps1 -InstallBuildDeps .\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` - запускает PyInstaller по `plate_config_app.spec`
- переименовывает бинарник в `plate_config_app-windows-x86_64.exe` - переименовывает бинарник в `plate_config_app-windows-x86_64.exe`
- копирует папки `light_modes/`, `input_mappings/`, `commands/` в `dist/` - копирует папки `light_modes/`, `input_mappings/`, `commands/` в `dist/`
- выводит путь и размер готового EXE - создаёт `plate_config_app-windows-x86_64.zip` с бинарником и ресурсами
- выводит пути и размеры готовых EXE и ZIP
### 3. Только очистка (без сборки) ### 3. Только очистка (без сборки)
@ -94,4 +98,5 @@ Hidden imports: `qtawesome`, `Crypto.Cipher._raw_aes`, `Crypto.Util._raw_api`
| `commands/` | Наборы команд (.csv) | | `commands/` | Наборы команд (.csv) |
Эти файлы можно редактировать без пересборки — достаточно изменить их Эти файлы можно редактировать без пересборки — достаточно изменить их
рядом с EXE и перезапустить приложение. рядом с EXE и перезапустить приложение. При установке OTA-обновления каталоги,
которые входят в новый ZIP-пакет, полностью заменяются их новыми версиями.

View File

@ -201,5 +201,6 @@ health-check) вынесены в `QThread`-подклассы. Связь с UI
.\build_exe.ps1 -InstallBuildDeps .\build_exe.ps1 -InstallBuildDeps
``` ```
Результат: `dist\plate_config_app-windows-x86_64.exe` + папки Результат: `dist\plate_config_app-windows-x86_64.exe`, ZIP-архив
`dist\light_modes\`, `dist\input_mappings\`, `dist\commands\`. `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` численно под настройками только тогда, когда удалённый тег формата `vX.Y.Z` численно
новее: сначала сравнивается major, затем minor и patch. новее: сначала сравнивается major, затем minor и patch.
Приложение скачивает исполняемый release-артефакт текущей платформы, проверяет Приложение скачивает ZIP-пакет текущей платформы, проверяет его целостность и
его формат, завершает старый процесс, заменяет бинарник и запускается заново. безопасность, завершает старый процесс, заменяет бинарник и все каталоги из
Файл `settings.json` и пользовательские папки ресурсов не заменяются. Для пакета, а затем запускается заново. Перед заменой создаётся резервная копия;
работы проверки релизы репозитория должны быть доступны без авторизации. она удаляется только после подтверждения запуска новой версии. Если запуск
не удался, предыдущая версия восстанавливается, а сообщение записывается
в `update-error.log`. Файл `settings.json` и другие файлы,
которых нет в архиве, сохраняются. Для работы проверки релизы репозитория должны
быть доступны без авторизации.
Имена release-артефактов: Имена release-артефактов:
- Windows x86_64: `plate_config_app-windows-x86_64.exe`; - Windows x86_64: `plate_config_app-windows-x86_64.zip`;
- Linux x86_64: `plate_config_app-linux-x86_64`. - 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.""" """Plate Config App package."""
__version__ = "2.0.0" __version__ = "2.0.2"

View File

@ -2,11 +2,13 @@ from __future__ import annotations
import sys import sys
from PySide6.QtCore import QTimer
from PySide6.QtGui import QFont from PySide6.QtGui import QFont
from PySide6.QtWidgets import QApplication from PySide6.QtWidgets import QApplication
from .main_window import MainWindow from .main_window import MainWindow
from .services import settings_service from .services import settings_service
from .services.update_service import signal_update_ready
from .ui.icons import ICON_COLOR_PRIMARY, icon from .ui.icons import ICON_COLOR_PRIMARY, icon
from .ui.theme import build_stylesheet, get_palette, create_qpalette from .ui.theme import build_stylesheet, get_palette, create_qpalette
@ -27,4 +29,5 @@ def main() -> None:
if not app_icon.isNull(): if not app_icon.isNull():
window.setWindowIcon(app_icon) window.setWindowIcon(app_icon)
window.show() window.show()
QTimer.singleShot(0, signal_update_ready)
sys.exit(app.exec()) sys.exit(app.exec())

View File

@ -6,12 +6,14 @@ import os
import platform import platform
import re import re
import shutil import shutil
import stat
import subprocess import subprocess
import sys import sys
import uuid import uuid
import zipfile
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path, PurePosixPath
from typing import BinaryIO from typing import BinaryIO
from urllib.parse import quote, unquote, urlparse from urllib.parse import quote, unquote, urlparse
from urllib.request import Request, urlopen from urllib.request import Request, urlopen
@ -27,9 +29,13 @@ HTTP_TIMEOUT_SECONDS = 10.0
DOWNLOAD_TIMEOUT_SECONDS = 30.0 DOWNLOAD_TIMEOUT_SECONDS = 30.0
DOWNLOAD_CHUNK_SIZE = 1024 * 1024 DOWNLOAD_CHUNK_SIZE = 1024 * 1024
MAX_DOWNLOAD_BYTES = 300 * 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( VERSION_PATTERN = re.compile(
r"[vV]?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)" 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): 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()) return tuple(int(component) for component in match.groups())
def platform_asset_name( def platform_executable_name(
*, *,
system: str | None = None, system: str | None = None,
machine: str | None = None, machine: str | None = None,
@ -81,6 +87,30 @@ def platform_asset_name(
raise UpdateError(f"Автообновление не поддерживает платформу {system_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: def extract_release_tag(final_url: str) -> str:
parsed = urlparse(final_url) parsed = urlparse(final_url)
if parsed.scheme != "https" or parsed.hostname != UPDATE_HOST: 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: def _validate_executable_signature(signature: bytes, executable_name: str) -> None:
with path.open("rb") as stream: if executable_name.casefold().endswith(".exe"):
signature = stream.read(4)
if asset_name.casefold().endswith(".exe"):
valid = signature.startswith(b"MZ") valid = signature.startswith(b"MZ")
else: else:
valid = signature == b"\x7fELF" valid = signature == b"\x7fELF"
@ -153,6 +181,103 @@ def _validate_download(path: Path, asset_name: str) -> None:
raise UpdateError("Сервер вернул файл, не похожий на исполняемый артефакт") 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( def download_update(
info: UpdateInfo, info: UpdateInfo,
target_directory: Path, target_directory: Path,
@ -162,7 +287,7 @@ def download_update(
progress: Callable[[int], None] | None = None, progress: Callable[[int], None] | None = None,
cancelled: Callable[[], bool] | None = None, cancelled: Callable[[], bool] | None = None,
) -> Path: ) -> Path:
"""Скачать и проверить новый бинарник рядом с текущим приложением.""" """Скачать и проверить ZIP-пакет обновления рядом с приложением."""
target_directory = target_directory.resolve() target_directory = target_directory.resolve()
if not target_directory.is_dir(): if not target_directory.is_dir():
raise UpdateError("Каталог приложения не найден") raise UpdateError("Каталог приложения не найден")
@ -213,9 +338,8 @@ def download_update(
raise UpdateError("Сервер вернул пустой файл обновления") raise UpdateError("Сервер вернул пустой файл обновления")
if total > 0 and received != total: if total > 0 and received != total:
raise UpdateError("Файл обновления скачан не полностью") raise UpdateError("Файл обновления скачан не полностью")
_validate_download(temporary_path, info.asset_name) executable_name = archive_executable_name(info.asset_name)
if not info.asset_name.casefold().endswith(".exe"): _validate_update_archive(temporary_path, executable_name)
temporary_path.chmod(0o755)
if progress is not None: if progress is not None:
progress(100) progress(100)
return temporary_path return temporary_path
@ -230,30 +354,112 @@ def current_executable() -> Path:
return Path(sys.executable).resolve() 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: def _write_windows_updater(script_path: Path) -> None:
script = r'''param( script = r'''param(
[int]$ApplicationProcessId, [int]$ApplicationProcessId,
[string]$Source, [string]$Staging,
[string]$Target [string]$Target,
[string]$Backup,
[string]$ReadyFile
) )
$ErrorActionPreference = "Stop" $ErrorActionPreference = "Stop"
$backup = "$Target.old" $applicationDirectory = Split-Path -Parent $Target
$errorLog = Join-Path (Split-Path -Parent $Target) "update-error.log" $errorLog = Join-Path $applicationDirectory "update-error.log"
$installed = [System.Collections.Generic.List[string]]::new()
$backedUp = [System.Collections.Generic.List[string]]::new()
$restarted = $null
try { try {
Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $backup -Force -ErrorAction SilentlyContinue Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
if (Test-Path -LiteralPath $Target) { New-Item -ItemType Directory -Path $Backup | Out-Null
Move-Item -LiteralPath $Target -Destination $backup -Force $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 if (-not (Test-Path -LiteralPath $Target -PathType Leaf)) {
Start-Process -FilePath $Target -WorkingDirectory (Split-Path -Parent $Target) throw "В пакете обновления отсутствует исполняемый файл"
Remove-Item -LiteralPath $backup -Force -ErrorAction SilentlyContinue }
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 Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force
} catch { } catch {
if (-not (Test-Path -LiteralPath $Target) -and (Test-Path -LiteralPath $backup)) { $failure = $_
Move-Item -LiteralPath $backup -Destination $Target -Force 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 exit 1
} }
''' '''
@ -264,28 +470,89 @@ def _write_linux_updater(script_path: Path) -> None:
script = r'''#!/bin/sh script = r'''#!/bin/sh
set -eu set -eu
application_pid="$1" application_pid="$1"
source_path="$2" staging_path="$2"
target_path="$3" target_path="$3"
backup_path="${target_path}.old" backup_path="$4"
error_log="$(dirname -- "$target_path")/update-error.log" 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 while kill -0 "$application_pid" 2>/dev/null; do
sleep 0.2 sleep 0.2
done done
rm -f -- "$backup_path" rm -rf -- "$backup_path"
if [ -e "$target_path" ]; then mkdir -p -- "$backup_path"
mv -f -- "$target_path" "$backup_path" : >"$installed_file"
fi : >"$backed_up_file"
if mv -f -- "$source_path" "$target_path"; then for source_path in "$staging_path"/*; do
chmod 755 -- "$target_path" name="$(basename -- "$source_path")"
"$target_path" >/dev/null 2>&1 & destination="$application_dir/$name"
rm -f -- "$backup_path" "$0" if [ -e "$destination" ]; then
else mv -f -- "$destination" "$backup_path/$name"
if [ ! -e "$target_path" ] && [ -e "$backup_path" ]; then printf '%s\n' "$name" >>"$backed_up_file"
mv -f -- "$backup_path" "$target_path"
fi 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 exit 1
fi 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.write_text(script, encoding="utf-8", newline="\n")
script_path.chmod(0o700) script_path.chmod(0o700)
@ -298,61 +565,84 @@ def launch_update(
process_id: int | None = None, process_id: int | None = None,
system: str | None = None, system: str | None = None,
) -> None: ) -> None:
"""Запустить внешний helper, который заменит бинарник после выхода.""" """Распаковать обновление и запустить helper для замены пакета."""
source = downloaded_path.resolve() archive_path = downloaded_path.resolve()
target = (executable_path or current_executable()).resolve() target = (executable_path or current_executable()).resolve()
if not source.is_file(): if not archive_path.is_file():
raise UpdateError("Скачанный файл обновления не найден") raise UpdateError("Скачанный архив обновления не найден")
if source.parent != target.parent:
staged = target.parent / source.name
shutil.move(str(source), str(staged))
source = staged.resolve()
if not os.access(target.parent, os.W_OK): if not os.access(target.parent, os.W_OK):
raise UpdateError("Нет прав на замену приложения") raise UpdateError("Нет прав на замену приложения")
updater_id = uuid.uuid4().hex updater_id = uuid.uuid4().hex
application_pid = process_id or os.getpid() application_pid = process_id or os.getpid()
system_name = (system or platform.system()).casefold() 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": if system_name == "windows":
script_path = target.parent / f".plate_config_updater-{updater_id}.ps1" script_path = target.parent / f".plate_config_updater-{updater_id}.ps1"
_write_windows_updater(script_path) try:
creation_flags = ( _write_windows_updater(script_path)
getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) creation_flags = (
| getattr(subprocess, "CREATE_NO_WINDOW", 0) getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
) | getattr(subprocess, "CREATE_NO_WINDOW", 0)
subprocess.Popen( )
[ subprocess.Popen(
"powershell.exe", [
"-NoProfile", "powershell.exe",
"-ExecutionPolicy", "-NoProfile",
"Bypass", "-ExecutionPolicy",
"-File", "Bypass",
str(script_path), "-File",
str(application_pid), str(script_path),
str(source), str(application_pid),
str(target), str(staging_path),
], str(target),
close_fds=True, str(backup_path),
creationflags=creation_flags, 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 return
if system_name == "linux": if system_name == "linux":
script_path = target.parent / f".plate_config_updater-{updater_id}.sh" script_path = target.parent / f".plate_config_updater-{updater_id}.sh"
_write_linux_updater(script_path) try:
subprocess.Popen( _write_linux_updater(script_path)
[ subprocess.Popen(
"/bin/sh", [
str(script_path), "/bin/sh",
str(application_pid), str(script_path),
str(source), str(application_pid),
str(target), str(staging_path),
], str(target),
cwd=str(target.parent), str(backup_path),
close_fds=True, str(ready_path),
start_new_session=True, ],
stdin=subprocess.DEVNULL, cwd=str(target.parent),
stdout=subprocess.DEVNULL, close_fds=True,
stderr=subprocess.DEVNULL, 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 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.QtCore import QThread, QTimer, Qt, Signal
from PySide6.QtGui import QColor from PySide6.QtGui import QColor
from PySide6.QtWidgets import ( from PySide6.QtWidgets import (
QApplication,
QFrame, QFrame,
QGraphicsDropShadowEffect, QGraphicsDropShadowEffect,
QHBoxLayout, QHBoxLayout,
@ -230,9 +229,7 @@ class StartTab(QWidget):
path.unlink(missing_ok=True) path.unlink(missing_ok=True)
self._on_update_failed(str(exc)) self._on_update_failed(str(exc))
return return
app = QApplication.instance() self.window().close()
if app is not None:
app.quit()
def _on_update_failed(self, detail: str) -> None: def _on_update_failed(self, detail: str) -> None:
self._update_btn.setText("Скачать обновление") self._update_btn.setText("Скачать обновление")

View File

@ -54,10 +54,28 @@ if ($InstallBuildDeps) {
# ── Сборка ───────────────────────────────────────────────────────── # ── Сборка ─────────────────────────────────────────────────────────
Write-Host "" Write-Host ""
Write-Host "=== Сборка EXE ===" 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) { if ($buildExitCode -ne 0) {
throw "PyInstaller завершился с ошибкой (код $LASTEXITCODE)." throw "PyInstaller завершился с ошибкой (код $buildExitCode)."
} }
$builtExe = Join-Path $projectRoot "dist\plate_config_app.exe" $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) $size = [math]::Round((Get-Item $exe).Length / 1MB, 1)
$archiveSize = [math]::Round((Get-Item $archive).Length / 1MB, 1)
Write-Host "" Write-Host ""
Write-Host "Сборка завершена: $exe ($size MB)" 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" build_dir="$project_root/build-linux"
dist_dir="$project_root/dist-linux" dist_dir="$project_root/dist-linux"
artifact_name="plate_config_app-linux-x86_64" artifact_name="plate_config_app-linux-x86_64"
archive_name="plate_config_app-linux-x86_64.zip"
cd "$project_root" cd "$project_root"
@ -41,5 +42,18 @@ Plate Config App для Debian 13 x86_64
sudo usermod -aG dialout "$USER" sudo usermod -aG dialout "$USER"
EOF 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" printf 'Готово: %s\n' "$dist_dir/$artifact_name"
sha256sum "$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 tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import Mock, patch from unittest.mock import patch
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
@ -34,7 +34,7 @@ class StartTabUpdateTests(unittest.TestCase):
info = UpdateInfo( info = UpdateInfo(
tag="v2.1.0", tag="v2.1.0",
version="2.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", download_url="https://git.robotail.pro/download/test",
) )
@ -98,20 +98,18 @@ class StartTabUpdateTests(unittest.TestCase):
warning.assert_called_once() warning.assert_called_once()
tab.shutdown() 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) tab = StartTab(lambda: None)
downloaded = Path("downloaded-update") downloaded = Path("downloaded-update")
application = Mock()
with ( with (
patch("app.widgets.start_tab.launch_update") as launcher, 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) tab._on_update_downloaded(downloaded)
launcher.assert_called_once_with(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()) self.assertEqual("Перезапуск…", tab._update_btn.text())
tab.shutdown() tab.shutdown()
@ -153,7 +151,7 @@ class StartTabUpdateTests(unittest.TestCase):
return UpdateInfo( return UpdateInfo(
tag="v2.1.0", tag="v2.1.0",
version="2.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", download_url="https://git.robotail.pro/download/test",
) )

View File

@ -10,6 +10,7 @@ import sys
import tempfile import tempfile
import time import time
import unittest import unittest
import zipfile
from pathlib import Path from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
@ -18,6 +19,7 @@ from app.services.update_service import (
UpdateCancelled, UpdateCancelled,
UpdateError, UpdateError,
UpdateInfo, UpdateInfo,
archive_executable_name,
build_download_url, build_download_url,
check_for_update, check_for_update,
download_update, download_update,
@ -26,6 +28,8 @@ from app.services.update_service import (
normalize_version, normalize_version,
parse_version, parse_version,
platform_asset_name, platform_asset_name,
platform_executable_name,
signal_update_ready,
) )
@ -56,6 +60,20 @@ class FakeResponse:
return self._stream.read(size) 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): class UpdateServiceTests(unittest.TestCase):
def test_normalizes_version_prefix_and_whitespace(self) -> None: def test_normalizes_version_prefix_and_whitespace(self) -> None:
self.assertEqual("2.0.0", normalize_version(" v2.0.0 ")) self.assertEqual("2.0.0", normalize_version(" v2.0.0 "))
@ -84,12 +102,9 @@ class UpdateServiceTests(unittest.TestCase):
self.assertEqual("v2.1.0", tag) self.assertEqual("v2.1.0", tag)
def test_rejects_login_redirect(self) -> None: def test_rejects_invalid_release_redirects(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:
invalid_urls = ( invalid_urls = (
"https://git.robotail.pro/user/login",
"http://git.robotail.pro/Robotail/plate-config-app/releases/tag/v2.1.0", "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", "https://example.com/Robotail/plate-config-app/releases/tag/v2.1.0",
) )
@ -98,49 +113,28 @@ class UpdateServiceTests(unittest.TestCase):
extract_release_tag(url) extract_release_tag(url)
def test_build_download_url_escapes_tag_and_asset(self) -> None: 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 test_equal_or_older_tag_does_not_offer_update(self) -> None:
def opener(_request, timeout: float): cases = (
self.assertGreater(timeout, 0) ("2.0.0", "v2.0.0"),
return FakeResponse( ("2.0.0", "v1.99.99"),
"https://git.robotail.pro/Robotail/plate-config-app/" ("2.5.0", "v2.4.99"),
"releases/tag/v2.0.0" ("2.5.4", "v2.5.3"),
)
result = check_for_update(
"2.0.0",
opener=opener,
system="Windows",
machine="AMD64",
) )
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_newer_major_minor_and_patch_versions_offer_zip(self) -> None:
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:
cases = ( cases = (
("1.9.9", "v2.0.0"), ("1.9.9", "v2.0.0"),
("2.4.9", "v2.5.0"), ("2.4.9", "v2.5.0"),
@ -151,33 +145,16 @@ class UpdateServiceTests(unittest.TestCase):
result = check_for_update( result = check_for_update(
current, current,
opener=self._release_opener(remote), opener=self._release_opener(remote),
system="Linux", system="Windows",
machine="x86_64", machine="x86_64",
) )
self.assertIsNotNone(result) self.assertIsNotNone(result)
assert result is not None
def test_older_major_minor_and_patch_versions_do_not_offer_update(self) -> None: self.assertEqual("plate_config_app-windows-x86_64.zip", result.asset_name)
cases = ( self.assertTrue(result.download_url.endswith("/" + result.asset_name))
("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)
def test_rejects_invalid_current_or_remote_version(self) -> None: def test_rejects_invalid_current_or_remote_version(self) -> None:
cases = ( for current, remote in (("development", "v2.1.0"), ("2.0.0", "v2.1")):
("development", "v2.1.0"),
("2.0.0", "v2.1"),
)
for current, remote in cases:
with self.subTest(current=current, remote=remote), self.assertRaises( with self.subTest(current=current, remote=remote), self.assertRaises(
UpdateError UpdateError
): ):
@ -197,32 +174,49 @@ class UpdateServiceTests(unittest.TestCase):
self.assertIsInstance(raised.exception.__cause__, OSError) self.assertIsInstance(raised.exception.__cause__, OSError)
def test_platform_asset_names(self) -> None: def test_platform_names(self) -> None:
self.assertEqual( self.assertEqual(
"plate_config_app-windows-x86_64.exe", "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"), platform_asset_name(system="Windows", machine="AMD64"),
) )
self.assertEqual( self.assertEqual(
"plate_config_app-linux-x86_64", "plate_config_app-linux-x86_64.zip",
platform_asset_name(system="Linux", machine="x86_64"), 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, "архитектуру"): with self.assertRaisesRegex(UpdateError, "архитектуру"):
platform_asset_name(system="Linux", machine="arm64") platform_asset_name(system="Linux", machine="arm64")
with self.assertRaisesRegex(UpdateError, "платформу"): with self.assertRaisesRegex(UpdateError, "платформу"):
platform_asset_name(system="Darwin", machine="x86_64") 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: def test_signals_restart_only_when_ready_file_is_requested(self) -> None:
data = b"MZ" + b"test-update" with tempfile.TemporaryDirectory() as directory:
info = UpdateInfo( marker = Path(directory) / "ready"
tag="v2.1.0", with patch.dict(os.environ, {update_service.UPDATE_READY_ENV: str(marker)}):
version="2.1.0", signal_update_ready()
asset_name="plate_config_app-windows-x86_64.exe", self.assertEqual(str(os.getpid()), marker.read_text(encoding="ascii"))
download_url=( self.assertNotIn(update_service.UPDATE_READY_ENV, os.environ)
"https://git.robotail.pro/Robotail/plate-config-app/releases/" marker.unlink()
"download/v2.1.0/plate_config_app-windows-x86_64.exe" 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] = [] progress_values: list[int] = []
@ -241,271 +235,323 @@ class UpdateServiceTests(unittest.TestCase):
self.assertEqual(data, result.read_bytes()) self.assertEqual(data, result.read_bytes())
self.assertEqual(100, progress_values[-1]) self.assertEqual(100, progress_values[-1])
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux") def test_downloads_and_validates_linux_zip(self) -> None:
def test_downloads_linux_binary_and_makes_it_executable(self) -> None:
data = b"\x7fELF" + b"test-update"
info = self._update_info(system="linux") info = self._update_info(system="linux")
data = make_update_archive("plate_config_app-linux-x86_64", b"\x7fELFnew")
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
result = download_update( result = download_update(
info, info,
Path(directory), Path(directory),
opener=lambda _request, timeout: FakeResponse( opener=lambda _request, timeout: FakeResponse(info.download_url, data),
info.download_url,
data,
content_length=len(data),
),
) )
self.assertEqual(data, result.read_bytes()) with zipfile.ZipFile(result) as archive:
self.assertTrue(result.stat().st_mode & stat.S_IXUSR) self.assertEqual(
["plate_config_app-linux-x86_64"],
def test_rejects_html_instead_of_binary(self) -> None: archive.namelist(),
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,
),
) )
def test_rejects_login_or_foreign_download_redirect(self) -> None: def test_rejects_invalid_or_incomplete_download(self) -> None:
info = self._update_info() info = self._update_info()
redirects = ( valid_data = make_update_archive(
"https://git.robotail.pro/user/login", "plate_config_app-windows-x86_64.exe",
"https://example.com/update.exe", b"MZnew",
) )
cases = (
for redirect in redirects: (b"<html>login</html>", None),
with self.subTest(redirect=redirect), tempfile.TemporaryDirectory() as directory: (valid_data, len(valid_data) + 10),
path = Path(directory) (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): with self.assertRaises(UpdateError):
download_update( download_update(
info, info,
path, target,
opener=lambda _request, timeout, url=redirect: FakeResponse( opener=lambda _request, timeout, payload=data, length=content_length: (
url, FakeResponse(
b"MZupdate", 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() 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( with tempfile.TemporaryDirectory() as directory, patch.object(
update_service, update_service,
"MAX_DOWNLOAD_BYTES", "MAX_DOWNLOAD_BYTES",
4, 4,
): ):
path = Path(directory) for content_length in (8, None):
with self.assertRaisesRegex(UpdateError, "превышает"): with self.subTest(content_length=content_length), self.assertRaisesRegex(
download_update( UpdateError,
info, "превышает",
path, ):
opener=lambda _request, timeout: FakeResponse( download_update(
info.download_url, info,
b"MZupdate", Path(directory),
content_length=8, opener=lambda _request, timeout, length=content_length: FakeResponse(
), info.download_url,
) b"12345678",
self.assertEqual([], list(path.iterdir())) content_length=length,
),
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()))
def test_cancels_download_and_removes_partial_file(self) -> None: def test_cancels_download_and_removes_partial_file(self) -> None:
info = self._update_info() info = self._update_info()
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
path = Path(directory) target = Path(directory)
with self.assertRaises(UpdateCancelled): with self.assertRaises(UpdateCancelled):
download_update( download_update(
info, info,
path, target,
opener=lambda _request, timeout: FakeResponse( opener=lambda _request, timeout: FakeResponse(
info.download_url, info.download_url,
b"MZupdate", b"partial",
), ),
cancelled=lambda: True, cancelled=lambda: True,
) )
self.assertEqual([], list(path.iterdir())) self.assertEqual([], list(target.iterdir()))
def test_rejects_missing_or_read_only_target_directory(self) -> None: def test_rejects_missing_or_read_only_target_directory(self) -> None:
info = self._update_info() info = self._update_info()
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
missing = Path(directory) / "missing"
with self.assertRaisesRegex(UpdateError, "не найден"): 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 patch("app.services.update_service.os.access", return_value=False):
with self.assertRaisesRegex(UpdateError, "Нет прав"): with self.assertRaisesRegex(UpdateError, "Нет прав"):
download_update(info, Path(directory)) 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: with tempfile.TemporaryDirectory() as directory:
root = Path(directory) root = Path(directory)
source = root / ".downloaded-update"
target = root / "plate_config_app-windows-x86_64.exe" target = root / "plate_config_app-windows-x86_64.exe"
source.write_bytes(b"MZnew") archive = root / ".downloaded-update.zip"
target.write_bytes(b"MZold") 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: with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update( launch_update(
source, archive,
executable_path=target, executable_path=target,
process_id=123, process_id=123,
system="Windows", system="Windows",
) )
scripts = list(root.glob(".plate_config_updater-*.ps1")) scripts = list(root.glob(".plate_config_updater-*.ps1"))
staging = list(root.glob(".plate_config_update-*"))
self.assertEqual(1, len(scripts)) 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") script = scripts[0].read_text(encoding="utf-8-sig")
self.assertIn("Wait-Process", script) self.assertIn("$Staging", script)
self.assertIn("Move-Item", 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() 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: with tempfile.TemporaryDirectory() as directory:
root = Path(directory) root = Path(directory)
source = root / ".downloaded-update"
target = root / "plate_config_app-linux-x86_64" target = root / "plate_config_app-linux-x86_64"
source.write_bytes(b"\x7fELFnew") archive = root / ".downloaded-update.zip"
target.write_bytes(b"\x7fELFold") 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: with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update( launch_update(
source, archive,
executable_path=target, executable_path=target,
process_id=123, process_id=123,
system="Linux", system="Linux",
) )
scripts = list(root.glob(".plate_config_updater-*.sh")) scripts = list(root.glob(".plate_config_updater-*.sh"))
staging = list(root.glob(".plate_config_update-*"))
self.assertEqual(1, len(scripts)) 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") script = scripts[0].read_text(encoding="utf-8")
self.assertIn("kill -0", script) self.assertIn("staging_path", script)
self.assertIn("mv -f", script) self.assertIn("rollback", script)
self.assertIn("PLATE_CONFIG_UPDATE_READY_FILE", script)
self.assertIn("PYINSTALLER_RESET_ENVIRONMENT=1", script)
popen.assert_called_once() 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: with tempfile.TemporaryDirectory() as directory:
root = Path(directory) root = Path(directory)
download_directory = root / "downloads" downloads = root / "downloads"
application_directory = root / "application" application = root / "application"
download_directory.mkdir() downloads.mkdir()
application_directory.mkdir() application.mkdir()
source = download_directory / ".downloaded-update" target = application / "plate_config_app-linux-x86_64"
target = application_directory / "plate_config_app-linux-x86_64" archive = downloads / "update.zip"
source.write_bytes(b"\x7fELFnew")
target.write_bytes(b"\x7fELFold") 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: with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update( launch_update(
source, archive,
executable_path=target, executable_path=target,
process_id=123, process_id=123,
system="Linux", system="Linux",
) )
staged = application_directory / source.name staging = list(application.glob(".plate_config_update-*"))
self.assertFalse(source.exists()) self.assertEqual(1, len(staging))
self.assertTrue(staged.exists()) self.assertFalse(archive.exists())
self.assertIn(str(staged.resolve()), popen.call_args.args[0]) 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: with tempfile.TemporaryDirectory() as directory:
root = Path(directory) root = Path(directory)
target = root / "plate_config_app" target = root / "plate_config_app"
source = root / ".downloaded-update" archive = root / "update.zip"
target.write_bytes(b"old") target.write_bytes(b"old")
with self.assertRaisesRegex(UpdateError, "не найден"): 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, "платформу"): 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 @staticmethod
def _update_info(*, system: str = "windows") -> UpdateInfo: def _update_info(*, system: str = "windows") -> UpdateInfo:
extension = ".exe" if system == "windows" else "" asset_name = f"plate_config_app-{system}-x86_64.zip"
asset_name = f"plate_config_app-{system}-x86_64{extension}"
return UpdateInfo( return UpdateInfo(
tag="v2.1.0", tag="v2.1.0",
version="2.1.0", version="2.1.0",
@ -533,90 +579,137 @@ class ReplacementIntegrationTests(unittest.TestCase):
return hashlib.sha256(path.read_bytes()).hexdigest() return hashlib.sha256(path.read_bytes()).hexdigest()
@staticmethod @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 deadline = time.monotonic() + 20
while time.monotonic() < deadline: while time.monotonic() < deadline:
if target.is_file() and ReplacementIntegrationTests._digest(target) == expected_digest: if (
if not list(target.parent.glob(".plate_config_updater-*")): target.is_file()
return 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) time.sleep(0.1)
raise AssertionError("Внешний helper не заменил приложение за 20 секунд") raise AssertionError("Внешний helper не откатил пакет за 20 секунд")
@unittest.skipUnless(sys.platform == "win32", "Тест предназначен для Windows") @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")) system_root = Path(os.environ.get("SystemRoot", r"C:\Windows"))
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
root = Path(directory) root = Path(directory)
source = root / ".downloaded-update.exe"
target = root / "plate_config_app-windows-x86_64.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) shutil.copy2(system_root / "System32" / "where.exe", target)
expected_digest = self._digest(source) self._prepare_old_files(root)
archive.write_bytes(
processes: list[subprocess.Popen] = [] make_update_archive(
real_popen = subprocess.Popen target.name,
new_executable,
def start_process(*args, **kwargs): extra_files={"commands/version.txt": b"new"},
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",
) )
)
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) processes[0].wait(timeout=5)
self.assertFalse(target.with_suffix(target.suffix + ".old").exists()) self._assert_package_preserved(root)
deadline = time.monotonic() + 10 self._wait_for_windows_unlock(target)
while True:
try:
target.unlink()
break
except PermissionError:
if time.monotonic() >= deadline:
raise
time.sleep(0.1)
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux") @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: with tempfile.TemporaryDirectory() as directory:
root = Path(directory) root = Path(directory)
source = root / ".downloaded-update"
target = root / "plate_config_app-linux-x86_64" 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) shutil.copy2("/bin/false", target)
expected_digest = self._digest(source) self._prepare_old_files(root)
archive.write_bytes(
processes: list[subprocess.Popen] = [] make_update_archive(
real_popen = subprocess.Popen target.name,
new_executable,
def start_process(*args, **kwargs): extra_files={"commands/version.txt": b"new"},
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",
) )
)
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) 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__": if __name__ == "__main__":