Compare commits

..

No commits in common. "master" and "v2.0.0" have entirely different histories.

11 changed files with 410 additions and 850 deletions

View File

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

View File

@ -201,6 +201,5 @@ health-check) вынесены в `QThread`-подклассы. Связь с UI
.\build_exe.ps1 -InstallBuildDeps
```
Результат: `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\`.
Результат: `dist\plate_config_app-windows-x86_64.exe` + папки
`dist\light_modes\`, `dist\input_mappings\`, `dist\commands\`.

View File

@ -59,24 +59,15 @@
под настройками только тогда, когда удалённый тег формата `vX.Y.Z` численно
новее: сначала сравнивается major, затем minor и patch.
Приложение скачивает ZIP-пакет текущей платформы, проверяет его целостность и
безопасность, завершает старый процесс, заменяет бинарник и все каталоги из
пакета, а затем запускается заново. Перед заменой создаётся резервная копия;
она удаляется только после подтверждения запуска новой версии. Если запуск
не удался, предыдущая версия восстанавливается, а сообщение записывается
в `update-error.log`. Файл `settings.json` и другие файлы,
которых нет в архиве, сохраняются. Для работы проверки релизы репозитория должны
быть доступны без авторизации.
Приложение скачивает исполняемый release-артефакт текущей платформы, проверяет
его формат, завершает старый процесс, заменяет бинарник и запускается заново.
Файл `settings.json` и пользовательские папки ресурсов не заменяются. Для
работы проверки релизы репозитория должны быть доступны без авторизации.
Имена release-артефактов:
- 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 последующие обновления можно устанавливать кнопкой в приложении.
- Windows x86_64: `plate_config_app-windows-x86_64.exe`;
- Linux x86_64: `plate_config_app-linux-x86_64`.
---

View File

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

View File

@ -2,13 +2,11 @@ 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
@ -29,5 +27,4 @@ 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,14 +6,12 @@ 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, PurePosixPath
from pathlib import Path
from typing import BinaryIO
from urllib.parse import quote, unquote, urlparse
from urllib.request import Request, urlopen
@ -29,13 +27,9 @@ 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):
@ -71,7 +65,7 @@ def parse_version(value: str) -> tuple[int, int, int]:
return tuple(int(component) for component in match.groups())
def platform_executable_name(
def platform_asset_name(
*,
system: str | None = None,
machine: str | None = None,
@ -87,30 +81,6 @@ def platform_executable_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:
@ -172,8 +142,10 @@ def check_for_update(
)
def _validate_executable_signature(signature: bytes, executable_name: str) -> None:
if executable_name.casefold().endswith(".exe"):
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"):
valid = signature.startswith(b"MZ")
else:
valid = signature == b"\x7fELF"
@ -181,103 +153,6 @@ def _validate_executable_signature(signature: bytes, executable_name: str) -> No
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,
@ -287,7 +162,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("Каталог приложения не найден")
@ -338,8 +213,9 @@ def download_update(
raise UpdateError("Сервер вернул пустой файл обновления")
if total > 0 and received != total:
raise UpdateError("Файл обновления скачан не полностью")
executable_name = archive_executable_name(info.asset_name)
_validate_update_archive(temporary_path, executable_name)
_validate_download(temporary_path, info.asset_name)
if not info.asset_name.casefold().endswith(".exe"):
temporary_path.chmod(0o755)
if progress is not None:
progress(100)
return temporary_path
@ -354,112 +230,30 @@ 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]$Staging,
[string]$Target,
[string]$Backup,
[string]$ReadyFile
[string]$Source,
[string]$Target
)
$ErrorActionPreference = "Stop"
$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
$backup = "$Target.old"
$errorLog = Join-Path (Split-Path -Parent $Target) "update-error.log"
try {
Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue
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)
Remove-Item -LiteralPath $backup -Force -ErrorAction SilentlyContinue
if (Test-Path -LiteralPath $Target) {
Move-Item -LiteralPath $Target -Destination $backup -Force
}
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
Move-Item -LiteralPath $Source -Destination $Target -Force
Start-Process -FilePath $Target -WorkingDirectory (Split-Path -Parent $Target)
Remove-Item -LiteralPath $backup -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force
} catch {
$failure = $_
if ($restarted -and -not $restarted.HasExited) {
& taskkill.exe /PID $restarted.Id /T /F *> $null
Start-Sleep -Milliseconds 500
if (-not (Test-Path -LiteralPath $Target) -and (Test-Path -LiteralPath $backup)) {
Move-Item -LiteralPath $backup -Destination $Target -Force
}
$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
$_ | Out-File -LiteralPath $errorLog -Encoding utf8
exit 1
}
'''
@ -470,89 +264,28 @@ def _write_linux_updater(script_path: Path) -> None:
script = r'''#!/bin/sh
set -eu
application_pid="$1"
staging_path="$2"
source_path="$2"
target_path="$3"
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
backup_path="${target_path}.old"
error_log="$(dirname -- "$target_path")/update-error.log"
while kill -0 "$application_pid" 2>/dev/null; do
sleep 0.2
done
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"
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"
fi
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
printf '%s\n' "Не удалось заменить приложение" >"$error_log"
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)
@ -565,84 +298,61 @@ def launch_update(
process_id: int | None = None,
system: str | None = None,
) -> None:
"""Распаковать обновление и запустить helper для замены пакета."""
archive_path = downloaded_path.resolve()
"""Запустить внешний helper, который заменит бинарник после выхода."""
source = downloaded_path.resolve()
target = (executable_path or current_executable()).resolve()
if not archive_path.is_file():
raise UpdateError("Скачанный архив обновления не найден")
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 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"
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
_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,
)
return
if system_name == "linux":
script_path = target.parent / f".plate_config_updater-{updater_id}.sh"
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
_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,
)
return
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")

View File

@ -5,6 +5,7 @@ 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,
@ -229,7 +230,9 @@ class StartTab(QWidget):
path.unlink(missing_ok=True)
self._on_update_failed(str(exc))
return
self.window().close()
app = QApplication.instance()
if app is not None:
app.quit()
def _on_update_failed(self, detail: str) -> None:
self._update_btn.setText("Скачать обновление")

View File

@ -54,28 +54,10 @@ if ($InstallBuildDeps) {
# ── Сборка ─────────────────────────────────────────────────────────
Write-Host ""
Write-Host "=== Сборка EXE ==="
$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
}
& $venvPython -m PyInstaller --noconfirm --clean plate_config_app.spec
if ($buildExitCode -ne 0) {
throw "PyInstaller завершился с ошибкой (код $buildExitCode)."
if ($LASTEXITCODE -ne 0) {
throw "PyInstaller завершился с ошибкой (код $LASTEXITCODE)."
}
$builtExe = Join-Path $projectRoot "dist\plate_config_app.exe"
@ -100,18 +82,6 @@ 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,7 +7,6 @@ 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"
@ -42,18 +41,5 @@ 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 patch
from unittest.mock import Mock, 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.zip",
asset_name="plate_config_app-windows-x86_64.exe",
download_url="https://git.robotail.pro/download/test",
)
@ -98,18 +98,20 @@ class StartTabUpdateTests(unittest.TestCase):
warning.assert_called_once()
tab.shutdown()
def test_successful_download_launches_update_and_closes_window(self) -> None:
def test_successful_download_launches_update_and_quits(self) -> None:
tab = StartTab(lambda: None)
downloaded = Path("downloaded-update")
application = Mock()
with (
patch("app.widgets.start_tab.launch_update") as launcher,
patch.object(tab, "close") as close_window,
patch("app.widgets.start_tab.QApplication") as application_class,
):
application_class.instance.return_value = application
tab._on_update_downloaded(downloaded)
launcher.assert_called_once_with(downloaded)
close_window.assert_called_once_with()
application.quit.assert_called_once_with()
self.assertEqual("Перезапуск…", tab._update_btn.text())
tab.shutdown()
@ -151,7 +153,7 @@ class StartTabUpdateTests(unittest.TestCase):
return UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name="plate_config_app-windows-x86_64.zip",
asset_name="plate_config_app-windows-x86_64.exe",
download_url="https://git.robotail.pro/download/test",
)

View File

@ -10,7 +10,6 @@ import sys
import tempfile
import time
import unittest
import zipfile
from pathlib import Path
from unittest.mock import patch
@ -19,7 +18,6 @@ from app.services.update_service import (
UpdateCancelled,
UpdateError,
UpdateInfo,
archive_executable_name,
build_download_url,
check_for_update,
download_update,
@ -28,8 +26,6 @@ from app.services.update_service import (
normalize_version,
parse_version,
platform_asset_name,
platform_executable_name,
signal_update_ready,
)
@ -60,20 +56,6 @@ 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 "))
@ -102,9 +84,12 @@ class UpdateServiceTests(unittest.TestCase):
self.assertEqual("v2.1.0", tag)
def test_rejects_invalid_release_redirects(self) -> None:
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:
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",
)
@ -113,28 +98,49 @@ 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.zip")
url = build_download_url("v2.1.0 rc/1", "plate config.exe")
self.assertTrue(url.endswith("/v2.1.0%20rc%2F1/plate%20config.zip"))
self.assertTrue(url.endswith("/v2.1.0%20rc%2F1/plate%20config.exe"))
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"),
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",
)
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)
def test_newer_major_minor_and_patch_versions_offer_zip(self) -> None:
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:
cases = (
("1.9.9", "v2.0.0"),
("2.4.9", "v2.5.0"),
@ -145,16 +151,33 @@ class UpdateServiceTests(unittest.TestCase):
result = check_for_update(
current,
opener=self._release_opener(remote),
system="Windows",
system="Linux",
machine="x86_64",
)
self.assertIsNotNone(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_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)
def test_rejects_invalid_current_or_remote_version(self) -> None:
for current, remote in (("development", "v2.1.0"), ("2.0.0", "v2.1")):
cases = (
("development", "v2.1.0"),
("2.0.0", "v2.1"),
)
for current, remote in cases:
with self.subTest(current=current, remote=remote), self.assertRaises(
UpdateError
):
@ -174,49 +197,32 @@ class UpdateServiceTests(unittest.TestCase):
self.assertIsInstance(raised.exception.__cause__, OSError)
def test_platform_names(self) -> None:
def test_platform_asset_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.zip",
"plate_config_app-linux-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_architecture_or_asset(self) -> None:
def test_rejects_unsupported_platform_or_architecture(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_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"},
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"
),
)
progress_values: list[int] = []
@ -235,323 +241,271 @@ class UpdateServiceTests(unittest.TestCase):
self.assertEqual(data, result.read_bytes())
self.assertEqual(100, progress_values[-1])
def test_downloads_and_validates_linux_zip(self) -> None:
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
def test_downloads_linux_binary_and_makes_it_executable(self) -> None:
data = b"\x7fELF" + b"test-update"
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),
opener=lambda _request, timeout: FakeResponse(
info.download_url,
data,
content_length=len(data),
),
)
with zipfile.ZipFile(result) as archive:
self.assertEqual(
["plate_config_app-linux-x86_64"],
archive.namelist(),
)
self.assertEqual(data, result.read_bytes())
self.assertTrue(result.stat().st_mode & stat.S_IXUSR)
def test_rejects_invalid_or_incomplete_download(self) -> None:
info = self._update_info()
valid_data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
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"
),
)
cases = (
(b"<html>login</html>", None),
(valid_data, len(valid_data) + 10),
(b"", None),
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"
),
)
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,
target,
opener=lambda _request, timeout, payload=data, length=content_length: (
FakeResponse(
info.download_url,
payload,
content_length=length,
)
),
)
self.assertEqual([], list(target.iterdir()))
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:
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",
"https://example.com/update.exe",
)
for redirect in redirects:
with self.subTest(redirect=redirect), tempfile.TemporaryDirectory() as directory:
path = Path(directory)
with self.assertRaises(UpdateError):
download_update(
info,
Path(directory),
path,
opener=lambda _request, timeout, url=redirect: FakeResponse(
url,
data,
b"MZupdate",
),
)
self.assertEqual([], list(path.iterdir()))
def test_rejects_oversized_downloads(self) -> None:
def test_rejects_oversized_declared_download(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory, patch.object(
update_service,
"MAX_DOWNLOAD_BYTES",
4,
):
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,
),
)
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()))
def test_cancels_download_and_removes_partial_file(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory:
target = Path(directory)
path = Path(directory)
with self.assertRaises(UpdateCancelled):
download_update(
info,
target,
path,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
b"partial",
b"MZupdate",
),
cancelled=lambda: True,
)
self.assertEqual([], list(target.iterdir()))
self.assertEqual([], list(path.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, Path(directory) / "missing")
download_update(info, missing)
with patch("app.services.update_service.os.access", return_value=False):
with self.assertRaisesRegex(UpdateError, "Нет прав"):
download_update(info, Path(directory))
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:
def test_prepares_windows_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"
archive = root / ".downloaded-update.zip"
source.write_bytes(b"MZnew")
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(
archive,
source,
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("$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)
self.assertIn("Wait-Process", script)
self.assertIn("Move-Item", 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)
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
def test_prepares_linux_package_replacement_helper(self) -> None:
def test_prepares_linux_replacement_helper(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / ".downloaded-update"
target = root / "plate_config_app-linux-x86_64"
archive = root / ".downloaded-update.zip"
source.write_bytes(b"\x7fELFnew")
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(
archive,
source,
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("staging_path", script)
self.assertIn("rollback", script)
self.assertIn("PLATE_CONFIG_UPDATE_READY_FILE", script)
self.assertIn("PYINSTALLER_RESET_ENVIRONMENT=1", script)
self.assertIn("kill -0", script)
self.assertIn("mv -f", script)
popen.assert_called_once()
def test_extracts_external_archive_next_to_executable(self) -> None:
def test_stages_download_next_to_executable(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
downloads = root / "downloads"
application = root / "application"
downloads.mkdir()
application.mkdir()
target = application / "plate_config_app-linux-x86_64"
archive = downloads / "update.zip"
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")
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(
archive,
source,
executable_path=target,
process_id=123,
system="Linux",
)
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])
staged = application_directory / source.name
self.assertFalse(source.exists())
self.assertTrue(staged.exists())
self.assertIn(str(staged.resolve()), popen.call_args.args[0])
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:
def test_rejects_missing_download_or_unsupported_platform(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app"
archive = root / "update.zip"
source = root / ".downloaded-update"
target.write_bytes(b"old")
with self.assertRaisesRegex(UpdateError, "не найден"):
launch_update(archive, executable_path=target, system="Linux")
launch_update(source, executable_path=target, system="Linux")
archive.write_bytes(b"not-a-zip")
source.write_bytes(b"new")
with self.assertRaisesRegex(UpdateError, "платформу"):
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()
launch_update(source, executable_path=target, system="Darwin")
@staticmethod
def _update_info(*, system: str = "windows") -> UpdateInfo:
asset_name = f"plate_config_app-{system}-x86_64.zip"
extension = ".exe" if system == "windows" else ""
asset_name = f"plate_config_app-{system}-x86_64{extension}"
return UpdateInfo(
tag="v2.1.0",
version="2.1.0",
@ -579,137 +533,90 @@ class ReplacementIntegrationTests(unittest.TestCase):
return hashlib.sha256(path.read_bytes()).hexdigest()
@staticmethod
def _wait_for_rollback(
target: Path,
expected_digest: str,
replaced_resource: Path,
) -> None:
def _wait_for_replacement(target: Path, expected_digest: str) -> None:
deadline = time.monotonic() + 20
while time.monotonic() < deadline:
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
if target.is_file() and ReplacementIntegrationTests._digest(target) == expected_digest:
if not list(target.parent.glob(".plate_config_updater-*")):
return
time.sleep(0.1)
raise AssertionError("Внешний helper не откатил пакет за 20 секунд")
raise AssertionError("Внешний helper не заменил приложение за 20 секунд")
@unittest.skipUnless(sys.platform == "win32", "Тест предназначен для Windows")
def test_windows_helper_rolls_back_when_new_app_exits_early(self) -> None:
def test_windows_helper_replaces_binary(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"
archive = root / "update.zip"
new_executable = (system_root / "System32" / "whoami.exe").read_bytes()
shutil.copy2(system_root / "System32" / "whoami.exe", source)
shutil.copy2(system_root / "System32" / "where.exe", target)
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")
expected_digest = self._digest(source)
self._wait_for_rollback(
target,
expected_digest,
root / "commands/version.txt",
)
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._wait_for_replacement(target, expected_digest)
processes[0].wait(timeout=5)
self._assert_package_preserved(root)
self._wait_for_windows_unlock(target)
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)
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
def test_linux_helper_rolls_back_when_new_app_exits_early(self) -> None:
def test_linux_helper_replaces_binary(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / ".downloaded-update"
target = root / "plate_config_app-linux-x86_64"
archive = root / "update.zip"
new_executable = Path("/bin/true").read_bytes()
shutil.copy2("/bin/true", source)
shutil.copy2("/bin/false", target)
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(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",
)
)
expected_digest = self._digest(target)
processes = self._launch_and_capture(archive, target, "Linux")
self._wait_for_rollback(
target,
expected_digest,
root / "commands/version.txt",
)
self._wait_for_replacement(target, expected_digest)
processes[0].wait(timeout=5)
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)
self.assertFalse(target.with_name(target.name + ".old").exists())
if __name__ == "__main__":