Robotail/plate-config-app
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 9 Wiki Activity

feat: add ZIP package updates

Browse Source 
- Сборки Windows и Linux теперь дополнительно формируют ZIP-архивы с бинарником и ресурсами.

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

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

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

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

- Версия приложения повышена до 2.0.1, документация и кроссплатформенные тесты обновлены.
This commit is contained in:
Данил Ярош 2026-09-16 14:07:01 +03:00
parent add8aff0af
commit 3a4491f0c4
9 changed files with 712 additions and 405 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
BUILD_EXE.md
View File

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

5
DEVELOPMENT.md
View File

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

14
README.md
View File

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

2
app/__init__.py
View File

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

368
app/services/update_service.py
View File

@ -6,12 +6,14 @@ import os
import platform
import re
import shutil
import stat
import subprocess
import sys
import uuid
import zipfile
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from pathlib import Path, PurePosixPath
from typing import BinaryIO
from urllib.parse import quote, unquote, urlparse
from urllib.request import Request, urlopen
@ -27,9 +29,12 @@ HTTP_TIMEOUT_SECONDS = 10.0
DOWNLOAD_TIMEOUT_SECONDS = 30.0
DOWNLOAD_CHUNK_SIZE = 1024 * 1024
MAX_DOWNLOAD_BYTES = 300 * 1024 * 1024
MAX_UNPACKED_BYTES = 600 * 1024 * 1024
MAX_ARCHIVE_MEMBERS = 10_000
VERSION_PATTERN = re.compile(
r"[vV]?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)"
)
TOP_LEVEL_NAME_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*")
class UpdateError(RuntimeError):
@ -65,7 +70,7 @@ def parse_version(value: str) -> tuple[int, int, int]:
return tuple(int(component) for component in match.groups())
def platform_asset_name(
def platform_executable_name(
*,
system: str | None = None,
machine: str | None = None,
@ -81,6 +86,30 @@ def platform_asset_name(
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")
def platform_asset_name(
*,
system: str | None = None,
machine: str | None = None,
) -> str:
executable_name = platform_executable_name(system=system, machine=machine)
if executable_name.casefold().endswith(".exe"):
executable_name = executable_name[:-4]
return f"{executable_name}.zip"
def archive_executable_name(asset_name: str) -> str:
names = {
"plate_config_app-windows-x86_64.zip": (
"plate_config_app-windows-x86_64.exe"
),
"plate_config_app-linux-x86_64.zip": "plate_config_app-linux-x86_64",
}
try:
return names[asset_name]
except KeyError as exc:
raise UpdateError("Неизвестное имя архива обновления") from exc
def extract_release_tag(final_url: str) -> str:
parsed = urlparse(final_url)
if parsed.scheme != "https" or parsed.hostname != UPDATE_HOST:
@ -142,10 +171,8 @@ def check_for_update(
)
def _validate_download(path: Path, asset_name: str) -> None:
with path.open("rb") as stream:
signature = stream.read(4)
if asset_name.casefold().endswith(".exe"):
def _validate_executable_signature(signature: bytes, executable_name: str) -> None:
if executable_name.casefold().endswith(".exe"):
valid = signature.startswith(b"MZ")
else:
valid = signature == b"\x7fELF"
@ -153,6 +180,103 @@ def _validate_download(path: Path, asset_name: str) -> None:
raise UpdateError("Сервер вернул файл, не похожий на исполняемый артефакт")
def _validated_archive_members(
archive: zipfile.ZipFile,
executable_name: str,
) -> list[zipfile.ZipInfo]:
members = archive.infolist()
if len(members) > MAX_ARCHIVE_MEMBERS:
raise UpdateError("В архиве обновления слишком много файлов")
total_size = 0
normalized_names: set[str] = set()
executable_member: zipfile.ZipInfo | None = None
for member in members:
raw_name = member.filename
if not raw_name or "\\" in raw_name or "\x00" in raw_name:
raise UpdateError("Архив обновления содержит небезопасный путь")
relative_path = PurePosixPath(raw_name.rstrip("/"))
if (
relative_path.is_absolute()
or not relative_path.parts
or any(part in {"", ".", ".."} for part in relative_path.parts)
):
raise UpdateError("Архив обновления содержит небезопасный путь")
if not TOP_LEVEL_NAME_PATTERN.fullmatch(relative_path.parts[0]):
raise UpdateError("Архив обновления содержит недопустимое имя")
normalized_name = relative_path.as_posix().casefold()
if normalized_name in normalized_names:
raise UpdateError("Архив обновления содержит повторяющиеся пути")
normalized_names.add(normalized_name)
unix_mode = member.external_attr >> 16
file_type = stat.S_IFMT(unix_mode)
if file_type not in {0, stat.S_IFREG, stat.S_IFDIR}:
raise UpdateError("Архив обновления содержит ссылку или специальный файл")
if member.flag_bits & 0x1:
raise UpdateError("Зашифрованные архивы обновления не поддерживаются")
if not member.is_dir():
total_size += member.file_size
if total_size > MAX_UNPACKED_BYTES:
raise UpdateError("Распакованный размер обновления превышает предел")
if relative_path.as_posix() == executable_name:
if member.is_dir():
raise UpdateError("Исполняемый файл обновления имеет неверный тип")
executable_member = member
if executable_member is None:
raise UpdateError("В архиве обновления не найден исполняемый файл")
with archive.open(executable_member, "r") as executable:
_validate_executable_signature(executable.read(4), executable_name)
return members
def _validate_update_archive(path: Path, executable_name: str) -> None:
try:
with zipfile.ZipFile(path, "r") as archive:
_validated_archive_members(archive, executable_name)
bad_member = archive.testzip()
except (OSError, zipfile.BadZipFile, RuntimeError) as exc:
raise UpdateError(f"Не удалось проверить ZIP-архив обновления: {exc}") from exc
if bad_member is not None:
raise UpdateError(f"Повреждён файл в архиве обновления: {bad_member}")
def _extract_update_archive(
archive_path: Path,
staging_directory: Path,
executable_name: str,
) -> Path:
try:
staging_directory.mkdir()
with zipfile.ZipFile(archive_path, "r") as archive:
members = _validated_archive_members(archive, executable_name)
for member in members:
relative_path = PurePosixPath(member.filename.rstrip("/"))
destination = staging_directory.joinpath(*relative_path.parts)
if member.is_dir():
destination.mkdir(parents=True, exist_ok=True)
continue
destination.parent.mkdir(parents=True, exist_ok=True)
with (
archive.open(member, "r") as source,
destination.open("xb") as output,
):
shutil.copyfileobj(source, output)
executable_path = staging_directory / executable_name
with executable_path.open("rb") as executable:
_validate_executable_signature(executable.read(4), executable_name)
if not executable_name.casefold().endswith(".exe"):
executable_path.chmod(0o755)
return executable_path
except Exception:
shutil.rmtree(staging_directory, ignore_errors=True)
raise
def download_update(
info: UpdateInfo,
target_directory: Path,
@ -162,7 +286,7 @@ def download_update(
progress: Callable[[int], None] | None = None,
cancelled: Callable[[], bool] | None = None,
) -> Path:
"""Скачать и проверить новый бинарник рядом с текущим приложением."""
"""Скачать и проверить ZIP-пакет обновления рядом с приложением."""
target_directory = target_directory.resolve()
if not target_directory.is_dir():
raise UpdateError("Каталог приложения не найден")
@ -213,9 +337,8 @@ def download_update(
raise UpdateError("Сервер вернул пустой файл обновления")
if total > 0 and received != total:
raise UpdateError("Файл обновления скачан не полностью")
_validate_download(temporary_path, info.asset_name)
if not info.asset_name.casefold().endswith(".exe"):
temporary_path.chmod(0o755)
executable_name = archive_executable_name(info.asset_name)
_validate_update_archive(temporary_path, executable_name)
if progress is not None:
progress(100)
return temporary_path
@ -233,26 +356,51 @@ def current_executable() -> Path:
def _write_windows_updater(script_path: Path) -> None:
script = r'''param(
[int]$ApplicationProcessId,
[string]$Source,
[string]$Target
[string]$Staging,
[string]$Target,
[string]$Backup
)
$ErrorActionPreference = "Stop"
$backup = "$Target.old"
$errorLog = Join-Path (Split-Path -Parent $Target) "update-error.log"
$applicationDirectory = Split-Path -Parent $Target
$errorLog = Join-Path $applicationDirectory "update-error.log"
$installed = [System.Collections.Generic.List[string]]::new()
$backedUp = [System.Collections.Generic.List[string]]::new()
try {
Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $backup -Force -ErrorAction SilentlyContinue
if (Test-Path -LiteralPath $Target) {
Move-Item -LiteralPath $Target -Destination $backup -Force
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
New-Item -ItemType Directory -Path $Backup | Out-Null
$entries = @(Get-ChildItem -LiteralPath $Staging -Force)
foreach ($entry in $entries) {
$destination = Join-Path $applicationDirectory $entry.Name
if (Test-Path -LiteralPath $destination) {
Move-Item -LiteralPath $destination -Destination $Backup -Force
$backedUp.Add($entry.Name)
}
Move-Item -LiteralPath $entry.FullName -Destination $destination -Force
$installed.Add($entry.Name)
}
Move-Item -LiteralPath $Source -Destination $Target -Force
Start-Process -FilePath $Target -WorkingDirectory (Split-Path -Parent $Target)
Remove-Item -LiteralPath $backup -Force -ErrorAction SilentlyContinue
if (-not (Test-Path -LiteralPath $Target -PathType Leaf)) {
throw "В пакете обновления отсутствует исполняемый файл"
}
Start-Process -FilePath $Target -WorkingDirectory $applicationDirectory
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force
} catch {
if (-not (Test-Path -LiteralPath $Target) -and (Test-Path -LiteralPath $backup)) {
Move-Item -LiteralPath $backup -Destination $Target -Force
foreach ($name in $installed) {
$destination = Join-Path $applicationDirectory $name
Remove-Item -LiteralPath $destination -Recurse -Force -ErrorAction SilentlyContinue
}
foreach ($name in $backedUp) {
$backupItem = Join-Path $Backup $name
$destination = Join-Path $applicationDirectory $name
Remove-Item -LiteralPath $destination -Recurse -Force -ErrorAction SilentlyContinue
if (Test-Path -LiteralPath $backupItem) {
Move-Item -LiteralPath $backupItem -Destination $destination -Force
}
}
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
$_ | Out-File -LiteralPath $errorLog -Encoding utf8
exit 1
}
@ -264,28 +412,60 @@ def _write_linux_updater(script_path: Path) -> None:
script = r'''#!/bin/sh
set -eu
application_pid="$1"
source_path="$2"
staging_path="$2"
target_path="$3"
backup_path="${target_path}.old"
error_log="$(dirname -- "$target_path")/update-error.log"
backup_path="$4"
application_dir="$(dirname -- "$target_path")"
error_log="$application_dir/update-error.log"
installed_file="$backup_path/.installed"
backed_up_file="$backup_path/.backed_up"
rollback() {
exit_code="$?"
trap - EXIT HUP INT TERM
if [ -f "$installed_file" ]; then
while IFS= read -r name; do
rm -rf -- "$application_dir/$name"
done < "$installed_file"
fi
if [ -f "$backed_up_file" ]; then
while IFS= read -r name; do
rm -rf -- "$application_dir/$name"
if [ -e "$backup_path/$name" ]; then
mv -f -- "$backup_path/$name" "$application_dir/$name"
fi
done < "$backed_up_file"
fi
rm -rf -- "$staging_path" "$backup_path"
printf '%s\n' "Не удалось заменить приложение" >"$error_log"
if [ "$exit_code" -eq 0 ]; then
exit_code=1
fi
exit "$exit_code"
}
trap rollback EXIT HUP INT TERM
while kill -0 "$application_pid" 2>/dev/null; do
sleep 0.2
done
rm -f -- "$backup_path"
if [ -e "$target_path" ]; then
mv -f -- "$target_path" "$backup_path"
fi
if mv -f -- "$source_path" "$target_path"; then
chmod 755 -- "$target_path"
"$target_path" >/dev/null 2>&1 &
rm -f -- "$backup_path" "$0"
else
if [ ! -e "$target_path" ] && [ -e "$backup_path" ]; then
mv -f -- "$backup_path" "$target_path"
rm -rf -- "$backup_path"
mkdir -p -- "$backup_path"
: >"$installed_file"
: >"$backed_up_file"
for source_path in "$staging_path"/*; do
name="$(basename -- "$source_path")"
destination="$application_dir/$name"
if [ -e "$destination" ]; then
mv -f -- "$destination" "$backup_path/$name"
printf '%s\n' "$name" >>"$backed_up_file"
fi
printf '%s\n' "Не удалось заменить приложение" >"$error_log"
exit 1
fi
mv -f -- "$source_path" "$destination"
printf '%s\n' "$name" >>"$installed_file"
done
test -f "$target_path"
chmod 755 -- "$target_path"
"$target_path" >/dev/null 2>&1 &
trap - EXIT HUP INT TERM
rm -rf -- "$backup_path" "$staging_path"
rm -f -- "$0"
'''
script_path.write_text(script, encoding="utf-8", newline="\n")
script_path.chmod(0o700)
@ -298,61 +478,81 @@ def launch_update(
process_id: int | None = None,
system: str | None = None,
) -> None:
"""Запустить внешний helper, который заменит бинарник после выхода."""
source = downloaded_path.resolve()
"""Распаковать обновление и запустить helper для замены пакета."""
archive_path = downloaded_path.resolve()
target = (executable_path or current_executable()).resolve()
if not source.is_file():
raise UpdateError("Скачанный файл обновления не найден")
if source.parent != target.parent:
staged = target.parent / source.name
shutil.move(str(source), str(staged))
source = staged.resolve()
if not archive_path.is_file():
raise UpdateError("Скачанный архив обновления не найден")
if not os.access(target.parent, os.W_OK):
raise UpdateError("Нет прав на замену приложения")
updater_id = uuid.uuid4().hex
application_pid = process_id or os.getpid()
system_name = (system or platform.system()).casefold()
if system_name not in {"windows", "linux"}:
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")
staging_path = target.parent / f".plate_config_update-{updater_id}"
backup_path = target.parent / f".plate_config_backup-{updater_id}"
script_path: Path | None = None
try:
_extract_update_archive(archive_path, staging_path, target.name)
archive_path.unlink()
except Exception:
shutil.rmtree(staging_path, ignore_errors=True)
raise
if system_name == "windows":
script_path = target.parent / f".plate_config_updater-{updater_id}.ps1"
_write_windows_updater(script_path)
creation_flags = (
getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
| getattr(subprocess, "CREATE_NO_WINDOW", 0)
)
subprocess.Popen(
[
"powershell.exe",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
str(script_path),
str(application_pid),
str(source),
str(target),
],
close_fds=True,
creationflags=creation_flags,
)
try:
_write_windows_updater(script_path)
creation_flags = (
getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
| getattr(subprocess, "CREATE_NO_WINDOW", 0)
)
subprocess.Popen(
[
"powershell.exe",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
str(script_path),
str(application_pid),
str(staging_path),
str(target),
str(backup_path),
],
close_fds=True,
creationflags=creation_flags,
)
except Exception:
script_path.unlink(missing_ok=True)
shutil.rmtree(staging_path, ignore_errors=True)
raise
return
if system_name == "linux":
script_path = target.parent / f".plate_config_updater-{updater_id}.sh"
_write_linux_updater(script_path)
subprocess.Popen(
[
"/bin/sh",
str(script_path),
str(application_pid),
str(source),
str(target),
],
cwd=str(target.parent),
close_fds=True,
start_new_session=True,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
try:
_write_linux_updater(script_path)
subprocess.Popen(
[
"/bin/sh",
str(script_path),
str(application_pid),
str(staging_path),
str(target),
str(backup_path),
],
cwd=str(target.parent),
close_fds=True,
start_new_session=True,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except Exception:
script_path.unlink(missing_ok=True)
shutil.rmtree(staging_path, ignore_errors=True)
raise
return
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")

12
build_exe.ps1
View File

@ -82,6 +82,18 @@ foreach ($folder in $resourceFolders) {
}
}
$archive = Join-Path $projectRoot "dist\plate_config_app-windows-x86_64.zip"
$archiveEntries = @($exe)
foreach ($folder in $resourceFolders) {
$resourcePath = Join-Path $projectRoot "dist\$folder"
if (Test-Path $resourcePath) {
$archiveEntries += $resourcePath
}
}
Compress-Archive -LiteralPath $archiveEntries -DestinationPath $archive -Force
$size = [math]::Round((Get-Item $exe).Length / 1MB, 1)
$archiveSize = [math]::Round((Get-Item $archive).Length / 1MB, 1)
Write-Host ""
Write-Host "Сборка завершена: $exe ($size MB)"
Write-Host "ZIP-архив: $archive ($archiveSize MB)"

14
build_linux.sh
View File

@ -7,6 +7,7 @@ venv_dir="$project_root/.venv-linux"
build_dir="$project_root/build-linux"
dist_dir="$project_root/dist-linux"
artifact_name="plate_config_app-linux-x86_64"
archive_name="plate_config_app-linux-x86_64.zip"
cd "$project_root"
@ -41,5 +42,18 @@ Plate Config App для Debian 13 x86_64
sudo usermod -aG dialout "$USER"
EOF
(
cd "$dist_dir"
"$venv_dir/bin/python" -m zipfile -c \
"$archive_name" \
"$artifact_name" \
light_modes \
input_mappings \
commands \
README.txt
)
printf 'Готово: %s\n' "$dist_dir/$artifact_name"
sha256sum "$dist_dir/$artifact_name"
printf 'ZIP-архив: %s\n' "$dist_dir/$archive_name"
sha256sum "$dist_dir/$archive_name"

4
tests/test_start_tab.py
View File

@ -34,7 +34,7 @@ class StartTabUpdateTests(unittest.TestCase):
info = UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name="plate_config_app-windows-x86_64.exe",
asset_name="plate_config_app-windows-x86_64.zip",
download_url="https://git.robotail.pro/download/test",
)
@ -153,7 +153,7 @@ class StartTabUpdateTests(unittest.TestCase):
return UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name="plate_config_app-windows-x86_64.exe",
asset_name="plate_config_app-windows-x86_64.zip",
download_url="https://git.robotail.pro/download/test",
)

679
tests/test_update_service.py
View File

@ -10,6 +10,7 @@ import sys
import tempfile
import time
import unittest
import zipfile
from pathlib import Path
from unittest.mock import patch
@ -18,6 +19,7 @@ from app.services.update_service import (
UpdateCancelled,
UpdateError,
UpdateInfo,
archive_executable_name,
build_download_url,
check_for_update,
download_update,
@ -26,6 +28,7 @@ from app.services.update_service import (
normalize_version,
parse_version,
platform_asset_name,
platform_executable_name,
)
@ -56,6 +59,20 @@ class FakeResponse:
return self._stream.read(size)
def make_update_archive(
executable_name: str,
executable_data: bytes,
*,
extra_files: dict[str, bytes] | None = None,
) -> bytes:
stream = io.BytesIO()
with zipfile.ZipFile(stream, "w", zipfile.ZIP_DEFLATED) as archive:
archive.writestr(executable_name, executable_data)
for name, data in (extra_files or {}).items():
archive.writestr(name, data)
return stream.getvalue()
class UpdateServiceTests(unittest.TestCase):
def test_normalizes_version_prefix_and_whitespace(self) -> None:
self.assertEqual("2.0.0", normalize_version(" v2.0.0 "))
@ -84,12 +101,9 @@ class UpdateServiceTests(unittest.TestCase):
self.assertEqual("v2.1.0", tag)
def test_rejects_login_redirect(self) -> None:
with self.assertRaises(UpdateError):
extract_release_tag("https://git.robotail.pro/user/login")
def test_rejects_insecure_or_foreign_release_redirect(self) -> None:
def test_rejects_invalid_release_redirects(self) -> None:
invalid_urls = (
"https://git.robotail.pro/user/login",
"http://git.robotail.pro/Robotail/plate-config-app/releases/tag/v2.1.0",
"https://example.com/Robotail/plate-config-app/releases/tag/v2.1.0",
)
@ -98,49 +112,28 @@ class UpdateServiceTests(unittest.TestCase):
extract_release_tag(url)
def test_build_download_url_escapes_tag_and_asset(self) -> None:
url = build_download_url("v2.1.0 rc/1", "plate config.exe")
url = build_download_url("v2.1.0 rc/1", "plate config.zip")
self.assertTrue(url.endswith("/v2.1.0%20rc%2F1/plate%20config.exe"))
self.assertTrue(url.endswith("/v2.1.0%20rc%2F1/plate%20config.zip"))
def test_equal_tag_does_not_offer_update(self) -> None:
def opener(_request, timeout: float):
self.assertGreater(timeout, 0)
return FakeResponse(
"https://git.robotail.pro/Robotail/plate-config-app/"
"releases/tag/v2.0.0"
)
result = check_for_update(
"2.0.0",
opener=opener,
system="Windows",
machine="AMD64",
def test_equal_or_older_tag_does_not_offer_update(self) -> None:
cases = (
("2.0.0", "v2.0.0"),
("2.0.0", "v1.99.99"),
("2.5.0", "v2.4.99"),
("2.5.4", "v2.5.3"),
)
for current, remote in cases:
with self.subTest(current=current, remote=remote):
result = check_for_update(
current,
opener=self._release_opener(remote),
system="Windows",
machine="AMD64",
)
self.assertIsNone(result)
self.assertIsNone(result)
def test_different_tag_selects_windows_asset(self) -> None:
def opener(_request, timeout: float):
self.assertGreater(timeout, 0)
return FakeResponse(
"https://git.robotail.pro/Robotail/plate-config-app/"
"releases/tag/v2.1.0"
)
result = check_for_update(
"2.0.0",
opener=opener,
system="Windows",
machine="x86_64",
)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("v2.1.0", result.tag)
self.assertEqual("plate_config_app-windows-x86_64.exe", result.asset_name)
self.assertTrue(result.download_url.endswith("/v2.1.0/" + result.asset_name))
def test_newer_major_minor_and_patch_versions_offer_update(self) -> None:
def test_newer_major_minor_and_patch_versions_offer_zip(self) -> None:
cases = (
("1.9.9", "v2.0.0"),
("2.4.9", "v2.5.0"),
@ -151,33 +144,16 @@ class UpdateServiceTests(unittest.TestCase):
result = check_for_update(
current,
opener=self._release_opener(remote),
system="Linux",
system="Windows",
machine="x86_64",
)
self.assertIsNotNone(result)
def test_older_major_minor_and_patch_versions_do_not_offer_update(self) -> None:
cases = (
("2.0.0", "v1.99.99"),
("2.5.0", "v2.4.99"),
("2.5.4", "v2.5.3"),
)
for current, remote in cases:
with self.subTest(current=current, remote=remote):
result = check_for_update(
current,
opener=self._release_opener(remote),
system="Linux",
machine="x86_64",
)
self.assertIsNone(result)
assert result is not None
self.assertEqual("plate_config_app-windows-x86_64.zip", result.asset_name)
self.assertTrue(result.download_url.endswith("/" + result.asset_name))
def test_rejects_invalid_current_or_remote_version(self) -> None:
cases = (
("development", "v2.1.0"),
("2.0.0", "v2.1"),
)
for current, remote in cases:
for current, remote in (("development", "v2.1.0"), ("2.0.0", "v2.1")):
with self.subTest(current=current, remote=remote), self.assertRaises(
UpdateError
):
@ -197,32 +173,38 @@ class UpdateServiceTests(unittest.TestCase):
self.assertIsInstance(raised.exception.__cause__, OSError)
def test_platform_asset_names(self) -> None:
def test_platform_names(self) -> None:
self.assertEqual(
"plate_config_app-windows-x86_64.exe",
platform_executable_name(system="Windows", machine="AMD64"),
)
self.assertEqual(
"plate_config_app-windows-x86_64.zip",
platform_asset_name(system="Windows", machine="AMD64"),
)
self.assertEqual(
"plate_config_app-linux-x86_64",
"plate_config_app-linux-x86_64.zip",
platform_asset_name(system="Linux", machine="x86_64"),
)
self.assertEqual(
"plate_config_app-linux-x86_64",
archive_executable_name("plate_config_app-linux-x86_64.zip"),
)
def test_rejects_unsupported_platform_or_architecture(self) -> None:
def test_rejects_unsupported_platform_architecture_or_asset(self) -> None:
with self.assertRaisesRegex(UpdateError, "архитектуру"):
platform_asset_name(system="Linux", machine="arm64")
with self.assertRaisesRegex(UpdateError, "платформу"):
platform_asset_name(system="Darwin", machine="x86_64")
with self.assertRaisesRegex(UpdateError, "Неизвестное"):
archive_executable_name("update.zip")
def test_downloads_and_validates_windows_binary(self) -> None:
data = b"MZ" + b"test-update"
info = UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name="plate_config_app-windows-x86_64.exe",
download_url=(
"https://git.robotail.pro/Robotail/plate-config-app/releases/"
"download/v2.1.0/plate_config_app-windows-x86_64.exe"
),
def test_downloads_and_validates_windows_zip(self) -> None:
info = self._update_info()
data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
extra_files={"commands/default.csv": b"04;Update"},
)
progress_values: list[int] = []
@ -241,271 +223,317 @@ class UpdateServiceTests(unittest.TestCase):
self.assertEqual(data, result.read_bytes())
self.assertEqual(100, progress_values[-1])
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
def test_downloads_linux_binary_and_makes_it_executable(self) -> None:
data = b"\x7fELF" + b"test-update"
def test_downloads_and_validates_linux_zip(self) -> None:
info = self._update_info(system="linux")
data = make_update_archive("plate_config_app-linux-x86_64", b"\x7fELFnew")
with tempfile.TemporaryDirectory() as directory:
result = download_update(
info,
Path(directory),
opener=lambda _request, timeout: FakeResponse(
info.download_url,
data,
content_length=len(data),
),
opener=lambda _request, timeout: FakeResponse(info.download_url, data),
)
self.assertEqual(data, result.read_bytes())
self.assertTrue(result.stat().st_mode & stat.S_IXUSR)
def test_rejects_html_instead_of_binary(self) -> None:
info = UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name="plate_config_app-linux-x86_64",
download_url=(
"https://git.robotail.pro/Robotail/plate-config-app/releases/"
"download/v2.1.0/plate_config_app-linux-x86_64"
),
)
with tempfile.TemporaryDirectory() as directory:
path = Path(directory)
with self.assertRaises(UpdateError):
download_update(
info,
path,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
b"<html>login</html>",
),
)
self.assertEqual([], list(path.iterdir()))
def test_rejects_incomplete_download(self) -> None:
data = b"MZpartial"
info = UpdateInfo(
tag="v2.1.0",
version="2.1.0",
asset_name="plate_config_app-windows-x86_64.exe",
download_url=(
"https://git.robotail.pro/Robotail/plate-config-app/releases/"
"download/v2.1.0/plate_config_app-windows-x86_64.exe"
),
)
with tempfile.TemporaryDirectory() as directory:
with self.assertRaises(UpdateError):
download_update(
info,
Path(directory),
opener=lambda _request, timeout: FakeResponse(
info.download_url,
data,
content_length=len(data) + 10,
),
with zipfile.ZipFile(result) as archive:
self.assertEqual(
["plate_config_app-linux-x86_64"],
archive.namelist(),
)
def test_rejects_login_or_foreign_download_redirect(self) -> None:
def test_rejects_invalid_or_incomplete_download(self) -> None:
info = self._update_info()
redirects = (
"https://git.robotail.pro/user/login",
"https://example.com/update.exe",
valid_data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
)
for redirect in redirects:
with self.subTest(redirect=redirect), tempfile.TemporaryDirectory() as directory:
path = Path(directory)
cases = (
(b"<html>login</html>", None),
(valid_data, len(valid_data) + 10),
(b"", None),
)
for data, content_length in cases:
with self.subTest(size=len(data)), tempfile.TemporaryDirectory() as directory:
target = Path(directory)
with self.assertRaises(UpdateError):
download_update(
info,
path,
opener=lambda _request, timeout, url=redirect: FakeResponse(
url,
b"MZupdate",
target,
opener=lambda _request, timeout, payload=data, length=content_length: (
FakeResponse(
info.download_url,
payload,
content_length=length,
)
),
)
self.assertEqual([], list(path.iterdir()))
self.assertEqual([], list(target.iterdir()))
def test_rejects_oversized_declared_download(self) -> None:
def test_rejects_login_or_foreign_download_redirect(self) -> None:
info = self._update_info()
data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
)
redirects = (
"https://git.robotail.pro/user/login",
"https://example.com/update.zip",
)
for redirect in redirects:
with self.subTest(redirect=redirect), tempfile.TemporaryDirectory() as directory:
with self.assertRaises(UpdateError):
download_update(
info,
Path(directory),
opener=lambda _request, timeout, url=redirect: FakeResponse(
url,
data,
),
)
def test_rejects_oversized_downloads(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory, patch.object(
update_service,
"MAX_DOWNLOAD_BYTES",
4,
):
path = Path(directory)
with self.assertRaisesRegex(UpdateError, "превышает"):
download_update(
info,
path,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
b"MZupdate",
content_length=8,
),
)
self.assertEqual([], list(path.iterdir()))
def test_rejects_oversized_stream_without_content_length(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory, patch.object(
update_service,
"MAX_DOWNLOAD_BYTES",
4,
):
path = Path(directory)
with self.assertRaisesRegex(UpdateError, "превышает"):
download_update(
info,
path,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
b"MZupdate",
),
)
self.assertEqual([], list(path.iterdir()))
def test_rejects_empty_download(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory:
path = Path(directory)
with self.assertRaisesRegex(UpdateError, "пустой"):
download_update(
info,
path,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
),
)
self.assertEqual([], list(path.iterdir()))
for content_length in (8, None):
with self.subTest(content_length=content_length), self.assertRaisesRegex(
UpdateError,
"превышает",
):
download_update(
info,
Path(directory),
opener=lambda _request, timeout, length=content_length: FakeResponse(
info.download_url,
b"12345678",
content_length=length,
),
)
def test_cancels_download_and_removes_partial_file(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory:
path = Path(directory)
target = Path(directory)
with self.assertRaises(UpdateCancelled):
download_update(
info,
path,
target,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
b"MZupdate",
b"partial",
),
cancelled=lambda: True,
)
self.assertEqual([], list(path.iterdir()))
self.assertEqual([], list(target.iterdir()))
def test_rejects_missing_or_read_only_target_directory(self) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory:
missing = Path(directory) / "missing"
with self.assertRaisesRegex(UpdateError, "не найден"):
download_update(info, missing)
download_update(info, Path(directory) / "missing")
with patch("app.services.update_service.os.access", return_value=False):
with self.assertRaisesRegex(UpdateError, "Нет прав"):
download_update(info, Path(directory))
def test_prepares_windows_replacement_helper(self) -> None:
def test_rejects_archive_without_executable(self) -> None:
data = self._custom_archive({"commands/default.csv": b"04;Update"})
self._assert_invalid_archive(data, "исполняемый файл")
def test_rejects_archive_path_traversal(self) -> None:
data = self._custom_archive(
{
"plate_config_app-windows-x86_64.exe": b"MZnew",
"../outside.txt": b"unsafe",
}
)
self._assert_invalid_archive(data, "небезопасный путь")
def test_rejects_archive_symlink(self) -> None:
stream = io.BytesIO()
with zipfile.ZipFile(stream, "w") as archive:
archive.writestr("plate_config_app-windows-x86_64.exe", b"MZnew")
link = zipfile.ZipInfo("commands/link")
link.create_system = 3
link.external_attr = (stat.S_IFLNK | 0o777) << 16
archive.writestr(link, "../../outside")
self._assert_invalid_archive(stream.getvalue(), "ссылку")
def test_rejects_duplicate_archive_paths(self) -> None:
stream = io.BytesIO()
with zipfile.ZipFile(stream, "w") as archive:
archive.writestr("plate_config_app-windows-x86_64.exe", b"MZnew")
archive.writestr("commands/Test.csv", b"one")
archive.writestr("commands/test.csv", b"two")
self._assert_invalid_archive(stream.getvalue(), "повторяющиеся")
def test_rejects_excessive_unpacked_size(self) -> None:
data = make_update_archive(
"plate_config_app-windows-x86_64.exe",
b"MZnew",
extra_files={"commands/large.csv": b"12345678"},
)
with patch.object(update_service, "MAX_UNPACKED_BYTES", 8):
self._assert_invalid_archive(data, "Распакованный размер")
def test_prepares_windows_package_replacement_helper(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / ".downloaded-update"
target = root / "plate_config_app-windows-x86_64.exe"
source.write_bytes(b"MZnew")
archive = root / ".downloaded-update.zip"
target.write_bytes(b"MZold")
archive.write_bytes(
make_update_archive(
target.name,
b"MZnew",
extra_files={"commands/default.csv": b"new"},
)
)
with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update(
source,
archive,
executable_path=target,
process_id=123,
system="Windows",
)
scripts = list(root.glob(".plate_config_updater-*.ps1"))
staging = list(root.glob(".plate_config_update-*"))
self.assertEqual(1, len(scripts))
self.assertEqual(1, len(staging))
self.assertFalse(archive.exists())
self.assertEqual(b"new", (staging[0] / "commands/default.csv").read_bytes())
script = scripts[0].read_text(encoding="utf-8-sig")
self.assertIn("Wait-Process", script)
self.assertIn("Move-Item", script)
self.assertIn("$Staging", script)
self.assertIn("$backedUp", script)
popen.assert_called_once()
creation_flags = popen.call_args.kwargs["creationflags"]
expected_flags = getattr(update_service.subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
expected_flags |= getattr(update_service.subprocess, "CREATE_NO_WINDOW", 0)
self.assertEqual(expected_flags, creation_flags)
def test_prepares_linux_replacement_helper(self) -> None:
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
def test_prepares_linux_package_replacement_helper(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / ".downloaded-update"
target = root / "plate_config_app-linux-x86_64"
source.write_bytes(b"\x7fELFnew")
archive = root / ".downloaded-update.zip"
target.write_bytes(b"\x7fELFold")
archive.write_bytes(
make_update_archive(
target.name,
b"\x7fELFnew",
extra_files={"commands/default.csv": b"new"},
)
)
with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update(
source,
archive,
executable_path=target,
process_id=123,
system="Linux",
)
scripts = list(root.glob(".plate_config_updater-*.sh"))
staging = list(root.glob(".plate_config_update-*"))
self.assertEqual(1, len(scripts))
self.assertEqual(1, len(staging))
self.assertTrue((staging[0] / target.name).stat().st_mode & stat.S_IXUSR)
script = scripts[0].read_text(encoding="utf-8")
self.assertIn("kill -0", script)
self.assertIn("mv -f", script)
self.assertIn("staging_path", script)
self.assertIn("rollback", script)
popen.assert_called_once()
def test_stages_download_next_to_executable(self) -> None:
def test_extracts_external_archive_next_to_executable(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
download_directory = root / "downloads"
application_directory = root / "application"
download_directory.mkdir()
application_directory.mkdir()
source = download_directory / ".downloaded-update"
target = application_directory / "plate_config_app-linux-x86_64"
source.write_bytes(b"\x7fELFnew")
downloads = root / "downloads"
application = root / "application"
downloads.mkdir()
application.mkdir()
target = application / "plate_config_app-linux-x86_64"
archive = downloads / "update.zip"
target.write_bytes(b"\x7fELFold")
archive.write_bytes(make_update_archive(target.name, b"\x7fELFnew"))
with patch("app.services.update_service.subprocess.Popen") as popen:
launch_update(
source,
archive,
executable_path=target,
process_id=123,
system="Linux",
)
staged = application_directory / source.name
self.assertFalse(source.exists())
self.assertTrue(staged.exists())
self.assertIn(str(staged.resolve()), popen.call_args.args[0])
staging = list(application.glob(".plate_config_update-*"))
self.assertEqual(1, len(staging))
self.assertFalse(archive.exists())
self.assertIn(str(staging[0]), popen.call_args.args[0])
def test_rejects_missing_download_or_unsupported_platform(self) -> None:
def test_failed_helper_launch_removes_staging_and_script(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app-windows-x86_64.exe"
archive = root / "update.zip"
target.write_bytes(b"MZold")
archive.write_bytes(make_update_archive(target.name, b"MZnew"))
with patch(
"app.services.update_service.subprocess.Popen",
side_effect=OSError("helper failed"),
), self.assertRaisesRegex(OSError, "helper failed"):
launch_update(
archive,
executable_path=target,
process_id=123,
system="Windows",
)
self.assertEqual([], list(root.glob(".plate_config_update-*")))
self.assertEqual([], list(root.glob(".plate_config_updater-*")))
def test_rejects_missing_archive_or_unsupported_platform(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "plate_config_app"
source = root / ".downloaded-update"
archive = root / "update.zip"
target.write_bytes(b"old")
with self.assertRaisesRegex(UpdateError, "не найден"):
launch_update(source, executable_path=target, system="Linux")
launch_update(archive, executable_path=target, system="Linux")
source.write_bytes(b"new")
archive.write_bytes(b"not-a-zip")
with self.assertRaisesRegex(UpdateError, "платформу"):
launch_update(source, executable_path=target, system="Darwin")
launch_update(archive, executable_path=target, system="Darwin")
def _assert_invalid_archive(self, data: bytes, message: str) -> None:
info = self._update_info()
with tempfile.TemporaryDirectory() as directory:
target = Path(directory)
with self.assertRaisesRegex(UpdateError, message):
download_update(
info,
target,
opener=lambda _request, timeout: FakeResponse(
info.download_url,
data,
),
)
self.assertEqual([], list(target.iterdir()))
@staticmethod
def _custom_archive(files: dict[str, bytes]) -> bytes:
stream = io.BytesIO()
with zipfile.ZipFile(stream, "w", zipfile.ZIP_DEFLATED) as archive:
for name, data in files.items():
archive.writestr(name, data)
return stream.getvalue()
@staticmethod
def _update_info(*, system: str = "windows") -> UpdateInfo:
extension = ".exe" if system == "windows" else ""
asset_name = f"plate_config_app-{system}-x86_64{extension}"
asset_name = f"plate_config_app-{system}-x86_64.zip"
return UpdateInfo(
tag="v2.1.0",
version="2.1.0",
@ -533,90 +561,135 @@ class ReplacementIntegrationTests(unittest.TestCase):
return hashlib.sha256(path.read_bytes()).hexdigest()
@staticmethod
def _wait_for_replacement(target: Path, expected_digest: str) -> None:
def _wait_for_replacement(
target: Path,
expected_digest: str,
replaced_resource: Path,
) -> None:
deadline = time.monotonic() + 20
while time.monotonic() < deadline:
if target.is_file() and ReplacementIntegrationTests._digest(target) == expected_digest:
if not list(target.parent.glob(".plate_config_updater-*")):
return
if (
target.is_file()
and ReplacementIntegrationTests._digest(target) == expected_digest
and replaced_resource.read_text(encoding="utf-8") == "new"
and not list(target.parent.glob(".plate_config_updater-*"))
and not list(target.parent.glob(".plate_config_update-*"))
and not list(target.parent.glob(".plate_config_backup-*"))
):
return
time.sleep(0.1)
raise AssertionError("Внешний helper не заменил приложение за 20 секунд")
raise AssertionError("Внешний helper не заменил пакет за 20 секунд")
@unittest.skipUnless(sys.platform == "win32", "Тест предназначен для Windows")
def test_windows_helper_replaces_binary(self) -> None:
def test_windows_helper_replaces_full_package(self) -> None:
system_root = Path(os.environ.get("SystemRoot", r"C:\Windows"))
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / ".downloaded-update.exe"
target = root / "plate_config_app-windows-x86_64.exe"
shutil.copy2(system_root / "System32" / "whoami.exe", source)
archive = root / "update.zip"
new_executable = (system_root / "System32" / "whoami.exe").read_bytes()
shutil.copy2(system_root / "System32" / "where.exe", target)
expected_digest = self._digest(source)
processes: list[subprocess.Popen] = []
real_popen = subprocess.Popen
def start_process(*args, **kwargs):
process = real_popen(*args, **kwargs)
processes.append(process)
return process
with patch(
"app.services.update_service.subprocess.Popen",
side_effect=start_process,
):
launch_update(
source,
executable_path=target,
process_id=2_147_483_647,
system="Windows",
self._prepare_old_files(root)
archive.write_bytes(
make_update_archive(
target.name,
new_executable,
extra_files={"commands/version.txt": b"new"},
)
)
expected_digest = hashlib.sha256(new_executable).hexdigest()
processes = self._launch_and_capture(archive, target, "Windows")
self._wait_for_replacement(target, expected_digest)
self._wait_for_replacement(
target,
expected_digest,
root / "commands/version.txt",
)
processes[0].wait(timeout=5)
self.assertFalse(target.with_suffix(target.suffix + ".old").exists())
deadline = time.monotonic() + 10
while True:
try:
target.unlink()
break
except PermissionError:
if time.monotonic() >= deadline:
raise
time.sleep(0.1)
self._assert_package_replaced(root)
self._wait_for_windows_unlock(target)
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
def test_linux_helper_replaces_binary(self) -> None:
def test_linux_helper_replaces_full_package(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / ".downloaded-update"
target = root / "plate_config_app-linux-x86_64"
shutil.copy2("/bin/true", source)
archive = root / "update.zip"
new_executable = Path("/bin/true").read_bytes()
shutil.copy2("/bin/false", target)
expected_digest = self._digest(source)
processes: list[subprocess.Popen] = []
real_popen = subprocess.Popen
def start_process(*args, **kwargs):
process = real_popen(*args, **kwargs)
processes.append(process)
return process
with patch(
"app.services.update_service.subprocess.Popen",
side_effect=start_process,
):
launch_update(
source,
executable_path=target,
process_id=2_147_483_647,
system="Linux",
self._prepare_old_files(root)
archive.write_bytes(
make_update_archive(
target.name,
new_executable,
extra_files={"commands/version.txt": b"new"},
)
)
expected_digest = hashlib.sha256(new_executable).hexdigest()
processes = self._launch_and_capture(archive, target, "Linux")
self._wait_for_replacement(target, expected_digest)
self._wait_for_replacement(
target,
expected_digest,
root / "commands/version.txt",
)
processes[0].wait(timeout=5)
self.assertFalse(target.with_name(target.name + ".old").exists())
self._assert_package_replaced(root)
self.assertTrue(target.stat().st_mode & stat.S_IXUSR)
@staticmethod
def _prepare_old_files(root: Path) -> None:
commands = root / "commands"
commands.mkdir()
(commands / "version.txt").write_text("old", encoding="utf-8")
(commands / "obsolete.txt").write_text("remove", encoding="utf-8")
(root / "settings.json").write_text("preserve", encoding="utf-8")
@staticmethod
def _assert_package_replaced(root: Path) -> None:
self_settings = (root / "settings.json").read_text(encoding="utf-8")
if self_settings != "preserve":
raise AssertionError("settings.json не сохранён")
if (root / "commands/obsolete.txt").exists():
raise AssertionError("Старое содержимое каталога не удалено")
@staticmethod
def _launch_and_capture(
archive: Path,
target: Path,
system: str,
) -> list[subprocess.Popen]:
processes: list[subprocess.Popen] = []
real_popen = subprocess.Popen
def start_process(*args, **kwargs):
process = real_popen(*args, **kwargs)
processes.append(process)
return process
with patch(
"app.services.update_service.subprocess.Popen",
side_effect=start_process,
):
launch_update(
archive,
executable_path=target,
process_id=2_147_483_647,
system=system,
)
return processes
@staticmethod
def _wait_for_windows_unlock(target: Path) -> None:
deadline = time.monotonic() + 10
while True:
try:
target.unlink()
return
except PermissionError:
if time.monotonic() >= deadline:
raise
time.sleep(0.1)
if __name__ == "__main__":