Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 35c9693740 | |||
| 3a4491f0c4 | |||
| add8aff0af | |||
| e163b63666 |
1
.gitignore
vendored
1
.gitignore
vendored
@ -17,6 +17,7 @@ dist/
|
||||
build-linux/
|
||||
dist-linux/
|
||||
.venv-linux/
|
||||
release-builds/
|
||||
*.manifest
|
||||
*.spec.bak
|
||||
|
||||
|
||||
22
BUILD_EXE.md
22
BUILD_EXE.md
@ -4,18 +4,19 @@
|
||||
|
||||
Для Linux используется нативная сборка: Windows-версия PyInstaller не может
|
||||
создать рабочий Linux-бинарник. В Debian 13 установите Python и средства
|
||||
архивации, затем запустите скрипт из корня проекта:
|
||||
сборки, затем запустите скрипт из корня проекта:
|
||||
|
||||
```bash
|
||||
sudo apt-get update
|
||||
sudo apt-get install -y python3 python3-venv python3-pip zip libglib2.0-0t64 libgl1 libegl1 libfontconfig1 libxkbcommon0 libfreetype6 libxcb-cursor0 libxcb-icccm4 libxcb-image0 libxcb-keysyms1 libxcb-render-util0 libxcb-shape0 libxcb-xkb1 libxcb-util1 libxkbcommon-x11-0
|
||||
sudo apt-get install -y python3 python3-venv python3-pip libglib2.0-0t64 libgl1 libegl1 libfontconfig1 libxkbcommon0 libfreetype6 libxcb-cursor0 libxcb-icccm4 libxcb-image0 libxcb-keysyms1 libxcb-render-util0 libxcb-shape0 libxcb-xkb1 libxcb-util1 libxkbcommon-x11-0
|
||||
chmod +x build_linux.sh
|
||||
./build_linux.sh
|
||||
```
|
||||
|
||||
Результат: `dist-linux/plate_config_app-debian13-x86_64.zip`.
|
||||
Архив содержит исполняемый файл `plate_config_app` и папки конфигураций рядом
|
||||
с ним. Для работы с USB-COM пользователь Debian должен состоять в группе
|
||||
Результат: бинарник `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.exe`
|
||||
Готовые файлы: `dist\plate_config_app-windows-x86_64.exe` и
|
||||
`dist\plate_config_app-windows-x86_64.zip`. Для автообновления публикуется
|
||||
ZIP-архив.
|
||||
|
||||
## Команды по отдельности
|
||||
|
||||
@ -50,8 +53,10 @@ python -m pip install -r requirements-build.txt
|
||||
Скрипт автоматически:
|
||||
- очищает `build/`, `dist/`, `__pycache__/`
|
||||
- запускает 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. Только очистка (без сборки)
|
||||
|
||||
@ -93,4 +98,5 @@ Hidden imports: `qtawesome`, `Crypto.Cipher._raw_aes`, `Crypto.Util._raw_api`
|
||||
| `commands/` | Наборы команд (.csv) |
|
||||
|
||||
Эти файлы можно редактировать без пересборки — достаточно изменить их
|
||||
рядом с EXE и перезапустить приложение.
|
||||
рядом с EXE и перезапустить приложение. При установке OTA-обновления каталоги,
|
||||
которые входят в новый ZIP-пакет, полностью заменяются их новыми версиями.
|
||||
|
||||
@ -123,6 +123,11 @@ plate_config_app/
|
||||
Сервисы загружают данные с диска, управляют COM-соединением,
|
||||
выполняют авторизацию. Могут использоваться без GUI.
|
||||
|
||||
`ComConnection` хранит версию текущей прошивки и режим авторизации,
|
||||
определённые по стартовой строке. Маркер `no_auth` проверяется без учёта
|
||||
регистра; для неизвестной версии применяется строгий режим с AES-256
|
||||
handshake. После переподключения и OTA режим определяется заново.
|
||||
|
||||
2. **ui/** -- визуальное оформление. `theme.py` содержит палитры
|
||||
(`ThemePalette`) и QSS-шаблон с подстановкой переменных.
|
||||
`icons.py` -- обёртка над `qtawesome` с fallback при отсутствии
|
||||
@ -196,5 +201,6 @@ health-check) вынесены в `QThread`-подклассы. Связь с UI
|
||||
.\build_exe.ps1 -InstallBuildDeps
|
||||
```
|
||||
|
||||
Результат: `dist\plate_config_app.exe` + папки `dist\light_modes\`,
|
||||
Результат: `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\`.
|
||||
|
||||
51
README.md
51
README.md
@ -10,6 +10,7 @@
|
||||
|
||||
- [Системные требования](#системные-требования)
|
||||
- [Установка и запуск](#установка-и-запуск)
|
||||
- [Обновление приложения](#обновление-приложения)
|
||||
- [Интерфейс приложения](#интерфейс-приложения)
|
||||
- [Главная страница](#главная-страница)
|
||||
- [Подключение к плате](#подключение-к-плате)
|
||||
@ -32,10 +33,11 @@
|
||||
|
||||
## Установка и запуск
|
||||
|
||||
### Из готового EXE
|
||||
### Из готовой сборки
|
||||
|
||||
1. Скопируйте `plate_config_app.exe` в любую папку.
|
||||
2. Запустите двойным кликом.
|
||||
1. Скопируйте сборку для своей платформы в каталог с папками ресурсов.
|
||||
2. На Windows запустите `plate_config_app-windows-x86_64.exe`.
|
||||
3. На Debian запустите `./plate_config_app-linux-x86_64`.
|
||||
|
||||
Файл настроек `settings.json` создается автоматически рядом с EXE при первом
|
||||
изменении настроек.
|
||||
@ -49,6 +51,35 @@
|
||||
|
||||
---
|
||||
|
||||
## Обновление приложения
|
||||
|
||||
При запуске приложение в фоне переходит по адресу
|
||||
`https://git.robotail.pro/Robotail/plate-config-app/releases/latest` и сравнивает
|
||||
тег последнего релиза с текущей версией. Кнопка «Скачать обновление» появляется
|
||||
под настройками только тогда, когда удалённый тег формата `vX.Y.Z` численно
|
||||
новее: сначала сравнивается major, затем minor и patch.
|
||||
|
||||
Приложение скачивает ZIP-пакет текущей платформы, проверяет его целостность и
|
||||
безопасность, завершает старый процесс, заменяет бинарник и все каталоги из
|
||||
пакета, а затем запускается заново. Перед заменой создаётся резервная копия;
|
||||
она удаляется только после подтверждения запуска новой версии. Если запуск
|
||||
не удался, предыдущая версия восстанавливается, а сообщение записывается
|
||||
в `update-error.log`. Файл `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 последующие обновления можно устанавливать кнопкой в приложении.
|
||||
|
||||
---
|
||||
|
||||
## Интерфейс приложения
|
||||
|
||||
### Главная страница
|
||||
@ -59,6 +90,8 @@
|
||||
- **Команды платы** -- справочник поддерживаемых команд платы с поиском.
|
||||
- **Настройки приложения** -- выбор цветовой темы (светлая / темная, зеленый /
|
||||
синий акцент).
|
||||
- **Скачать обновление** -- появляется под настройками, если версия последнего
|
||||
релиза формата `vX.Y.Z` новее версии запущенного приложения.
|
||||
|
||||
Приложение поддерживает несколько одновременных подключений: каждое соединение
|
||||
открывается в отдельной вкладке.
|
||||
@ -118,9 +151,15 @@
|
||||
- Зеленый — подключено, плата готова к работе.
|
||||
|
||||
Кнопка переподключения (↻) позволяет быстро восстановить связь при обрыве.
|
||||
Перед запуском OTA и после успешного обновления приложение автоматически
|
||||
выполняет полную AES-256 авторизацию платы. В терминале команд доступна
|
||||
отдельная кнопка ручной авторизации.
|
||||
При подключении приложение определяет режим по версии в стартовой строке платы.
|
||||
Если версия содержит маркер `no_auth` без учёта регистра, AES-256 авторизация
|
||||
пропускается, а статус отображается как «Готово без авторизации». Для остальных
|
||||
и неизвестных версий сохраняется строгая обязательная авторизация.
|
||||
|
||||
После успешного OTA и перезагрузки приложение повторно читает версию новой
|
||||
прошивки. Полная AES-256 авторизация выполняется только для прошивок без маркера
|
||||
`no_auth`; перед командой `04` повторная авторизация не выполняется. Кнопка
|
||||
ручной авторизации в терминале использует то же правило совместимости.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -1,3 +1,3 @@
|
||||
"""Plate Config App package."""
|
||||
|
||||
__version__ = "1.5.0"
|
||||
__version__ = "2.0.2"
|
||||
|
||||
@ -2,11 +2,13 @@ from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
from PySide6.QtCore import QTimer
|
||||
from PySide6.QtGui import QFont
|
||||
from PySide6.QtWidgets import QApplication
|
||||
|
||||
from .main_window import MainWindow
|
||||
from .services import settings_service
|
||||
from .services.update_service import signal_update_ready
|
||||
from .ui.icons import ICON_COLOR_PRIMARY, icon
|
||||
from .ui.theme import build_stylesheet, get_palette, create_qpalette
|
||||
|
||||
@ -27,4 +29,5 @@ def main() -> None:
|
||||
if not app_icon.isNull():
|
||||
window.setWindowIcon(app_icon)
|
||||
window.show()
|
||||
QTimer.singleShot(0, signal_update_ready)
|
||||
sys.exit(app.exec())
|
||||
|
||||
@ -119,6 +119,8 @@ class MainWindow(QMainWindow):
|
||||
widget = self.pages.widget(i)
|
||||
if isinstance(widget, ConnectionTab):
|
||||
widget.close_serial()
|
||||
elif isinstance(widget, StartTab):
|
||||
widget.shutdown()
|
||||
super().closeEvent(event)
|
||||
|
||||
def _attach_close_button(self, index: int) -> None:
|
||||
|
||||
@ -26,6 +26,7 @@ from Crypto.Cipher import AES
|
||||
from .com_service import ComConnection
|
||||
|
||||
STARTUP_BANNER_PREFIX = "periphery controller ver_"
|
||||
NO_AUTH_MARKER = "no_auth"
|
||||
STARTUP_DELAY_SECONDS = 2.0
|
||||
STARTUP_BANNER_TIMEOUT_SECONDS = 20.0
|
||||
READY_TIMEOUT_SECONDS = 5.0
|
||||
@ -54,6 +55,22 @@ class AuthResult:
|
||||
status: AuthStatus
|
||||
detail: str = ""
|
||||
startup_ready: bool = False
|
||||
firmware_version: str | None = None
|
||||
auth_skipped: bool = False
|
||||
|
||||
|
||||
def firmware_requires_auth(version: str) -> bool:
|
||||
"""Определить режим авторизации по строке версии прошивки."""
|
||||
return NO_AUTH_MARKER not in version.casefold()
|
||||
|
||||
|
||||
def _remember_firmware(conn: ComConnection, banner: str) -> str:
|
||||
version = banner[len(STARTUP_BANNER_PREFIX):].strip()
|
||||
conn.set_firmware_version(
|
||||
version,
|
||||
auth_required=firmware_requires_auth(version),
|
||||
)
|
||||
return version
|
||||
|
||||
|
||||
def _aes_encrypt(plaintext: bytes, key: bytes = _AES_KEY) -> bytes:
|
||||
@ -102,8 +119,10 @@ def wait_for_startup_banner(
|
||||
def check_status(conn: ComConnection, timeout: float = 3.0) -> AuthStatus:
|
||||
"""Отправить команду 99 и проанализировать ответ.
|
||||
|
||||
Только точная строка ``READY!`` означает готовность. Любой другой
|
||||
ответ или таймаут означает, что авторизация ещё не подтверждена.
|
||||
Для прошивки с авторизацией только точная строка ``READY!`` означает
|
||||
готовность. Для обнаруженной прошивки ``no_auth`` достаточно любого
|
||||
непустого ответа: отсутствие ответа считается потерей связи, а не
|
||||
запросом авторизации.
|
||||
"""
|
||||
if not conn.is_connected:
|
||||
return AuthStatus.DISCONNECTED
|
||||
@ -113,12 +132,42 @@ def check_status(conn: ComConnection, timeout: float = 3.0) -> AuthStatus:
|
||||
return AuthStatus.DISCONNECTED
|
||||
|
||||
try:
|
||||
if conn.auth_required is False:
|
||||
response = _read_until(conn, bool, timeout)
|
||||
return AuthStatus.OK if response is not None else AuthStatus.DISCONNECTED
|
||||
|
||||
ready = _read_until(conn, lambda line: line == "READY!", timeout)
|
||||
return AuthStatus.OK if ready is not None else AuthStatus.NEED_AUTH
|
||||
finally:
|
||||
conn.release()
|
||||
|
||||
|
||||
def perform_required_handshake(
|
||||
conn: ComConnection,
|
||||
timeout: float = 2.0,
|
||||
ready_timeout: float = READY_TIMEOUT_SECONDS,
|
||||
) -> AuthResult:
|
||||
"""Выполнить handshake только если он требуется текущей прошивке."""
|
||||
if not conn.is_connected:
|
||||
return AuthResult(AuthStatus.DISCONNECTED, "Порт не открыт")
|
||||
if conn.auth_required is False:
|
||||
version = conn.firmware_version
|
||||
detail = "Авторизация не требуется: обнаружена прошивка no_auth"
|
||||
if version:
|
||||
detail = f"Прошивка {version}: авторизация не требуется"
|
||||
return AuthResult(
|
||||
AuthStatus.OK,
|
||||
detail,
|
||||
firmware_version=version,
|
||||
auth_skipped=True,
|
||||
)
|
||||
return perform_handshake(
|
||||
conn,
|
||||
timeout=timeout,
|
||||
ready_timeout=ready_timeout,
|
||||
)
|
||||
|
||||
|
||||
def perform_handshake(
|
||||
conn: ComConnection,
|
||||
timeout: float = 2.0,
|
||||
@ -210,7 +259,7 @@ def perform_startup_handshake(
|
||||
ready_timeout: float = READY_TIMEOUT_SECONDS,
|
||||
sleep_fn: Callable[[float], None] = time.sleep,
|
||||
) -> AuthResult:
|
||||
"""Дождаться загрузки платы, выдержать паузу и выполнить handshake."""
|
||||
"""Дождаться загрузки и выполнить handshake, если он требуется."""
|
||||
if not conn.is_connected:
|
||||
return AuthResult(AuthStatus.DISCONNECTED, "Порт не открыт")
|
||||
|
||||
@ -225,9 +274,25 @@ def perform_startup_handshake(
|
||||
f"Таймаут ожидания {STARTUP_BANNER_PREFIX}<версия>",
|
||||
)
|
||||
|
||||
version = _remember_firmware(conn, banner)
|
||||
sleep_fn(startup_delay)
|
||||
if not conn.is_connected:
|
||||
return AuthResult(AuthStatus.DISCONNECTED, "Порт закрыт до начала авторизации")
|
||||
return AuthResult(
|
||||
AuthStatus.DISCONNECTED,
|
||||
"Порт закрыт до проверки режима авторизации",
|
||||
startup_ready=True,
|
||||
firmware_version=version,
|
||||
)
|
||||
|
||||
if not firmware_requires_auth(version):
|
||||
conn.release()
|
||||
return AuthResult(
|
||||
AuthStatus.OK,
|
||||
f"{banner}\nАвторизация не требуется для прошивки no_auth",
|
||||
startup_ready=True,
|
||||
firmware_version=version,
|
||||
auth_skipped=True,
|
||||
)
|
||||
|
||||
result = perform_handshake(
|
||||
conn,
|
||||
@ -235,6 +300,7 @@ def perform_startup_handshake(
|
||||
ready_timeout=ready_timeout,
|
||||
)
|
||||
result.startup_ready = True
|
||||
result.firmware_version = version
|
||||
if result.status == AuthStatus.OK:
|
||||
result.detail = f"{banner}\n{result.detail}"
|
||||
return result
|
||||
|
||||
@ -91,6 +91,8 @@ class ComConnection(QObject):
|
||||
# Параметры последнего подключения (для reconnect)
|
||||
self._last_port: str | None = None
|
||||
self._last_params: SerialParams | None = None
|
||||
self._firmware_version: str | None = None
|
||||
self._auth_required: bool | None = None
|
||||
|
||||
# ── Properties ────────────────────────────────────────────────────
|
||||
|
||||
@ -102,6 +104,20 @@ class ComConnection(QObject):
|
||||
def serial(self) -> Serial | None:
|
||||
return self._serial
|
||||
|
||||
@property
|
||||
def firmware_version(self) -> str | None:
|
||||
return self._firmware_version
|
||||
|
||||
@property
|
||||
def auth_required(self) -> bool | None:
|
||||
"""Требуется ли авторизация по данным текущей прошивки."""
|
||||
return self._auth_required
|
||||
|
||||
def set_firmware_version(self, version: str, *, auth_required: bool) -> None:
|
||||
"""Сохранить версию и обнаруженный режим авторизации прошивки."""
|
||||
self._firmware_version = version.strip()
|
||||
self._auth_required = auth_required
|
||||
|
||||
@property
|
||||
def seconds_since_last_command(self) -> float:
|
||||
"""Сколько секунд прошло с последней отправки команды."""
|
||||
@ -169,6 +185,8 @@ class ComConnection(QObject):
|
||||
self._serial = None
|
||||
self._read_buf.clear()
|
||||
self._event_buf.clear()
|
||||
self._firmware_version = None
|
||||
self._auth_required = None
|
||||
|
||||
def reconnect(
|
||||
self,
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
"""Авторизованный запуск OTA-обновления контроллера."""
|
||||
"""Запуск OTA с повторной авторизацией после обновления контроллера."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@ -12,7 +12,6 @@ from .auth_service import (
|
||||
READY_TIMEOUT_SECONDS,
|
||||
STARTUP_BANNER_TIMEOUT_SECONDS,
|
||||
STARTUP_DELAY_SECONDS,
|
||||
perform_handshake,
|
||||
perform_startup_handshake,
|
||||
)
|
||||
from .com_service import ComConnection
|
||||
@ -24,7 +23,6 @@ OTA_TIMEOUT_SECONDS = 60.0
|
||||
class FirmwareUpdateResult:
|
||||
success: bool
|
||||
detail: str
|
||||
pre_auth: AuthResult | None = None
|
||||
post_auth: AuthResult | None = None
|
||||
|
||||
|
||||
@ -39,29 +37,11 @@ def perform_authenticated_ota(
|
||||
ready_timeout: float = READY_TIMEOUT_SECONDS,
|
||||
sleep_fn: Callable[[float], None] = time.sleep,
|
||||
) -> FirmwareUpdateResult:
|
||||
"""Повторно авторизоваться, выполнить ``04`` и авторизоваться после OTA."""
|
||||
"""Выполнить ``04`` и полностью авторизоваться после успешного OTA."""
|
||||
emit = log_message or (lambda _message: None)
|
||||
if not conn.is_connected:
|
||||
return FirmwareUpdateResult(False, "Порт не подключён")
|
||||
|
||||
emit("Авторизация перед запуском OTA…")
|
||||
try:
|
||||
pre_auth = perform_handshake(
|
||||
conn,
|
||||
timeout=handshake_timeout,
|
||||
ready_timeout=ready_timeout,
|
||||
)
|
||||
except Exception as exc:
|
||||
return FirmwareUpdateResult(False, f"Ошибка авторизации перед OTA: {exc}")
|
||||
|
||||
if pre_auth.status != AuthStatus.OK:
|
||||
return FirmwareUpdateResult(
|
||||
False,
|
||||
f"Авторизация перед OTA не выполнена: {pre_auth.detail}",
|
||||
pre_auth=pre_auth,
|
||||
)
|
||||
emit("✓ Авторизация перед OTA выполнена.")
|
||||
|
||||
try:
|
||||
conn.send_command("04")
|
||||
emit("→ 04")
|
||||
@ -82,10 +62,9 @@ def perform_authenticated_ota(
|
||||
return FirmwareUpdateResult(
|
||||
False,
|
||||
"Таймаут ожидания подтверждения успешного обновления",
|
||||
pre_auth=pre_auth,
|
||||
)
|
||||
|
||||
emit("✓ Обновление подтверждено. Ожидание перезагрузки и авторизации…")
|
||||
emit("✓ Обновление подтверждено. Ожидание перезагрузки платы…")
|
||||
post_auth = perform_startup_handshake(
|
||||
conn,
|
||||
banner_timeout=banner_timeout,
|
||||
@ -98,22 +77,24 @@ def perform_authenticated_ota(
|
||||
return FirmwareUpdateResult(
|
||||
False,
|
||||
f"Авторизация после OTA не выполнена: {post_auth.detail}",
|
||||
pre_auth=pre_auth,
|
||||
post_auth=post_auth,
|
||||
)
|
||||
|
||||
emit("✓ Авторизация после OTA выполнена.")
|
||||
if post_auth.auth_skipped:
|
||||
emit("✓ Прошивка no_auth: авторизация после OTA не требуется.")
|
||||
detail = "Прошивка обновлена, плата готова без авторизации"
|
||||
else:
|
||||
emit("✓ Авторизация после OTA выполнена.")
|
||||
detail = "Прошивка обновлена, плата перезагружена и авторизована"
|
||||
return FirmwareUpdateResult(
|
||||
True,
|
||||
"Прошивка обновлена, плата перезагружена и авторизована",
|
||||
pre_auth=pre_auth,
|
||||
detail,
|
||||
post_auth=post_auth,
|
||||
)
|
||||
except Exception as exc:
|
||||
return FirmwareUpdateResult(
|
||||
False,
|
||||
f"Ошибка OTA: {exc}",
|
||||
pre_auth=pre_auth,
|
||||
)
|
||||
finally:
|
||||
conn.release()
|
||||
|
||||
648
app/services/update_service.py
Normal file
648
app/services/update_service.py
Normal file
@ -0,0 +1,648 @@
|
||||
"""Проверка, загрузка и установка обновлений Plate Config App."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
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 typing import BinaryIO
|
||||
from urllib.parse import quote, unquote, urlparse
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
LATEST_RELEASE_URL = (
|
||||
"https://git.robotail.pro/Robotail/plate-config-app/releases/latest"
|
||||
)
|
||||
RELEASE_DOWNLOAD_BASE_URL = (
|
||||
"https://git.robotail.pro/Robotail/plate-config-app/releases/download"
|
||||
)
|
||||
UPDATE_HOST = "git.robotail.pro"
|
||||
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):
|
||||
"""Ошибка проверки, загрузки или запуска обновления."""
|
||||
|
||||
|
||||
class UpdateCancelled(UpdateError):
|
||||
"""Загрузка обновления отменена пользователем или закрытием приложения."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class UpdateInfo:
|
||||
tag: str
|
||||
version: str
|
||||
asset_name: str
|
||||
download_url: str
|
||||
|
||||
|
||||
def normalize_version(value: str) -> str:
|
||||
normalized = value.strip()
|
||||
if normalized[:1].casefold() == "v":
|
||||
normalized = normalized[1:]
|
||||
return normalized
|
||||
|
||||
|
||||
def parse_version(value: str) -> tuple[int, int, int]:
|
||||
"""Преобразовать версию X.Y.Z или vX.Y.Z в числовой кортеж."""
|
||||
match = VERSION_PATTERN.fullmatch(value.strip())
|
||||
if match is None:
|
||||
raise UpdateError(
|
||||
f"Некорректная версия {value!r}: ожидается формат vX.Y.Z"
|
||||
)
|
||||
return tuple(int(component) for component in match.groups())
|
||||
|
||||
|
||||
def platform_executable_name(
|
||||
*,
|
||||
system: str | None = None,
|
||||
machine: str | None = None,
|
||||
) -> str:
|
||||
system_name = (system or platform.system()).casefold()
|
||||
machine_name = (machine or platform.machine()).casefold()
|
||||
if machine_name not in {"amd64", "x86_64"}:
|
||||
raise UpdateError(f"Автообновление не поддерживает архитектуру {machine_name}")
|
||||
if system_name == "windows":
|
||||
return "plate_config_app-windows-x86_64.exe"
|
||||
if system_name == "linux":
|
||||
return "plate_config_app-linux-x86_64"
|
||||
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:
|
||||
raise UpdateError("Сервер релизов перенаправил запрос на неизвестный адрес")
|
||||
parts = [unquote(part) for part in parsed.path.split("/") if part]
|
||||
try:
|
||||
release_index = parts.index("releases")
|
||||
except ValueError as exc:
|
||||
raise UpdateError("Не удалось определить тег последнего релиза") from exc
|
||||
if parts[release_index + 1:release_index + 2] != ["tag"]:
|
||||
raise UpdateError("Не удалось определить тег последнего релиза")
|
||||
tag_parts = parts[release_index + 2:release_index + 3]
|
||||
if not tag_parts or not tag_parts[0].strip():
|
||||
raise UpdateError("Сервер релизов вернул пустой тег")
|
||||
return tag_parts[0].strip()
|
||||
|
||||
|
||||
def build_download_url(tag: str, asset_name: str) -> str:
|
||||
encoded_tag = quote(tag, safe="")
|
||||
encoded_asset = quote(asset_name, safe="")
|
||||
return f"{RELEASE_DOWNLOAD_BASE_URL}/{encoded_tag}/{encoded_asset}"
|
||||
|
||||
|
||||
def check_for_update(
|
||||
current_version: str,
|
||||
*,
|
||||
timeout: float = HTTP_TIMEOUT_SECONDS,
|
||||
opener: Callable[..., BinaryIO] | None = None,
|
||||
system: str | None = None,
|
||||
machine: str | None = None,
|
||||
) -> UpdateInfo | None:
|
||||
"""Перейти по latest URL и вернуть данные отличающегося релиза."""
|
||||
request = Request(
|
||||
LATEST_RELEASE_URL,
|
||||
headers={
|
||||
"Accept": "text/html,application/xhtml+xml",
|
||||
"User-Agent": f"PlateConfigApp/{normalize_version(current_version)}",
|
||||
},
|
||||
)
|
||||
open_request = opener or urlopen
|
||||
try:
|
||||
with open_request(request, timeout=timeout) as response:
|
||||
final_url = response.geturl()
|
||||
except Exception as exc:
|
||||
raise UpdateError(f"Не удалось проверить обновления: {exc}") from exc
|
||||
|
||||
tag = extract_release_tag(final_url)
|
||||
remote_version = parse_version(tag)
|
||||
if remote_version <= parse_version(current_version):
|
||||
return None
|
||||
|
||||
version = normalize_version(tag)
|
||||
asset_name = platform_asset_name(system=system, machine=machine)
|
||||
return UpdateInfo(
|
||||
tag=tag,
|
||||
version=version,
|
||||
asset_name=asset_name,
|
||||
download_url=build_download_url(tag, asset_name),
|
||||
)
|
||||
|
||||
|
||||
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"
|
||||
if not valid:
|
||||
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,
|
||||
*,
|
||||
timeout: float = DOWNLOAD_TIMEOUT_SECONDS,
|
||||
opener: Callable[..., BinaryIO] | None = None,
|
||||
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("Каталог приложения не найден")
|
||||
if not os.access(target_directory, os.W_OK):
|
||||
raise UpdateError("Нет прав на запись в каталог приложения")
|
||||
|
||||
temporary_path = target_directory / (
|
||||
f".{info.asset_name}.download-{uuid.uuid4().hex}"
|
||||
)
|
||||
request = Request(
|
||||
info.download_url,
|
||||
headers={
|
||||
"Accept": "application/octet-stream",
|
||||
"User-Agent": f"PlateConfigAppUpdater/{info.version}",
|
||||
},
|
||||
)
|
||||
open_request = opener or urlopen
|
||||
received = 0
|
||||
try:
|
||||
with open_request(request, timeout=timeout) as response:
|
||||
final_url = urlparse(response.geturl())
|
||||
if final_url.scheme != "https" or final_url.hostname != UPDATE_HOST:
|
||||
raise UpdateError("Загрузка перенаправлена на неизвестный сервер")
|
||||
if final_url.path.rstrip("/") == "/user/login":
|
||||
raise UpdateError("Release-артефакты недоступны без авторизации")
|
||||
content_length = response.headers.get("Content-Length")
|
||||
total = int(content_length) if content_length else 0
|
||||
if total > MAX_DOWNLOAD_BYTES:
|
||||
raise UpdateError("Размер обновления превышает допустимый предел")
|
||||
|
||||
with temporary_path.open("wb") as output:
|
||||
while True:
|
||||
if cancelled is not None and cancelled():
|
||||
raise UpdateCancelled("Загрузка обновления отменена")
|
||||
chunk = response.read(DOWNLOAD_CHUNK_SIZE)
|
||||
if not chunk:
|
||||
break
|
||||
received += len(chunk)
|
||||
if received > MAX_DOWNLOAD_BYTES:
|
||||
raise UpdateError("Размер обновления превышает допустимый предел")
|
||||
output.write(chunk)
|
||||
if progress is not None and total > 0:
|
||||
progress(min(received * 100 // total, 100))
|
||||
output.flush()
|
||||
os.fsync(output.fileno())
|
||||
|
||||
if received == 0:
|
||||
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)
|
||||
if progress is not None:
|
||||
progress(100)
|
||||
return temporary_path
|
||||
except Exception:
|
||||
temporary_path.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
|
||||
def current_executable() -> Path:
|
||||
if not getattr(sys, "frozen", False):
|
||||
raise UpdateError("Самообновление доступно только в собранной версии")
|
||||
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
|
||||
)
|
||||
$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
|
||||
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)
|
||||
}
|
||||
if (-not (Test-Path -LiteralPath $Target -PathType Leaf)) {
|
||||
throw "В пакете обновления отсутствует исполняемый файл"
|
||||
}
|
||||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||||
$previousReadyFile = $env:PLATE_CONFIG_UPDATE_READY_FILE
|
||||
$previousResetEnvironment = $env:PYINSTALLER_RESET_ENVIRONMENT
|
||||
try {
|
||||
$env:PLATE_CONFIG_UPDATE_READY_FILE = $ReadyFile
|
||||
$env:PYINSTALLER_RESET_ENVIRONMENT = "1"
|
||||
$restarted = Start-Process -FilePath $Target -WorkingDirectory $applicationDirectory -PassThru
|
||||
} finally {
|
||||
$env:PLATE_CONFIG_UPDATE_READY_FILE = $previousReadyFile
|
||||
$env:PYINSTALLER_RESET_ENVIRONMENT = $previousResetEnvironment
|
||||
}
|
||||
$deadline = (Get-Date).AddSeconds(60)
|
||||
while (-not (Test-Path -LiteralPath $ReadyFile -PathType Leaf)) {
|
||||
if ($restarted.HasExited) { throw "Новая версия завершилась до запуска интерфейса" }
|
||||
if ((Get-Date) -ge $deadline) { throw "Новая версия не подтвердила запуск" }
|
||||
Start-Sleep -Milliseconds 250
|
||||
}
|
||||
if ($restarted.HasExited) { throw "Новая версия завершилась сразу после запуска" }
|
||||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||||
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
|
||||
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
|
||||
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force
|
||||
} catch {
|
||||
$failure = $_
|
||||
if ($restarted -and -not $restarted.HasExited) {
|
||||
& taskkill.exe /PID $restarted.Id /T /F *> $null
|
||||
Start-Sleep -Milliseconds 500
|
||||
}
|
||||
$rollbackErrors = [System.Collections.Generic.List[string]]::new()
|
||||
foreach ($name in $installed) {
|
||||
$destination = Join-Path $applicationDirectory $name
|
||||
if (Test-Path -LiteralPath $destination) {
|
||||
try {
|
||||
Remove-Item -LiteralPath $destination -Recurse -Force -ErrorAction Stop
|
||||
} catch {
|
||||
$rollbackErrors.Add($_.ToString())
|
||||
}
|
||||
}
|
||||
}
|
||||
foreach ($name in $backedUp) {
|
||||
$backupItem = Join-Path $Backup $name
|
||||
$destination = Join-Path $applicationDirectory $name
|
||||
if (Test-Path -LiteralPath $backupItem) {
|
||||
try {
|
||||
Move-Item -LiteralPath $backupItem -Destination $destination -Force -ErrorAction Stop
|
||||
} catch {
|
||||
$rollbackErrors.Add($_.ToString())
|
||||
}
|
||||
}
|
||||
}
|
||||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||||
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
|
||||
if ($rollbackErrors.Count -eq 0) {
|
||||
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
|
||||
}
|
||||
@($failure.ToString()) + @($rollbackErrors) | Out-File -LiteralPath $errorLog -Encoding utf8
|
||||
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force -ErrorAction SilentlyContinue
|
||||
exit 1
|
||||
}
|
||||
'''
|
||||
script_path.write_text(script, encoding="utf-8-sig", newline="\r\n")
|
||||
|
||||
|
||||
def _write_linux_updater(script_path: Path) -> None:
|
||||
script = r'''#!/bin/sh
|
||||
set -eu
|
||||
application_pid="$1"
|
||||
staging_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
|
||||
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"
|
||||
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
|
||||
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)
|
||||
|
||||
|
||||
def launch_update(
|
||||
downloaded_path: Path,
|
||||
*,
|
||||
executable_path: Path | None = None,
|
||||
process_id: int | None = None,
|
||||
system: str | None = None,
|
||||
) -> None:
|
||||
"""Распаковать обновление и запустить helper для замены пакета."""
|
||||
archive_path = downloaded_path.resolve()
|
||||
target = (executable_path or current_executable()).resolve()
|
||||
if not archive_path.is_file():
|
||||
raise UpdateError("Скачанный архив обновления не найден")
|
||||
if not os.access(target.parent, os.W_OK):
|
||||
raise UpdateError("Нет прав на замену приложения")
|
||||
|
||||
updater_id = uuid.uuid4().hex
|
||||
application_pid = process_id or os.getpid()
|
||||
system_name = (system or platform.system()).casefold()
|
||||
if system_name not in {"windows", "linux"}:
|
||||
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")
|
||||
|
||||
staging_path = target.parent / f".plate_config_update-{updater_id}"
|
||||
backup_path = target.parent / f".plate_config_backup-{updater_id}"
|
||||
ready_path = target.parent / f".plate_config_ready-{updater_id}"
|
||||
script_path: Path | None = None
|
||||
try:
|
||||
_extract_update_archive(archive_path, staging_path, target.name)
|
||||
archive_path.unlink()
|
||||
except Exception:
|
||||
shutil.rmtree(staging_path, ignore_errors=True)
|
||||
raise
|
||||
|
||||
if system_name == "windows":
|
||||
script_path = target.parent / f".plate_config_updater-{updater_id}.ps1"
|
||||
try:
|
||||
_write_windows_updater(script_path)
|
||||
creation_flags = (
|
||||
getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
|
||||
| getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||||
)
|
||||
subprocess.Popen(
|
||||
[
|
||||
"powershell.exe",
|
||||
"-NoProfile",
|
||||
"-ExecutionPolicy",
|
||||
"Bypass",
|
||||
"-File",
|
||||
str(script_path),
|
||||
str(application_pid),
|
||||
str(staging_path),
|
||||
str(target),
|
||||
str(backup_path),
|
||||
str(ready_path),
|
||||
],
|
||||
close_fds=True,
|
||||
creationflags=creation_flags,
|
||||
)
|
||||
except Exception:
|
||||
script_path.unlink(missing_ok=True)
|
||||
shutil.rmtree(staging_path, ignore_errors=True)
|
||||
raise
|
||||
return
|
||||
if system_name == "linux":
|
||||
script_path = target.parent / f".plate_config_updater-{updater_id}.sh"
|
||||
try:
|
||||
_write_linux_updater(script_path)
|
||||
subprocess.Popen(
|
||||
[
|
||||
"/bin/sh",
|
||||
str(script_path),
|
||||
str(application_pid),
|
||||
str(staging_path),
|
||||
str(target),
|
||||
str(backup_path),
|
||||
str(ready_path),
|
||||
],
|
||||
cwd=str(target.parent),
|
||||
close_fds=True,
|
||||
start_new_session=True,
|
||||
stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
except Exception:
|
||||
script_path.unlink(missing_ok=True)
|
||||
shutil.rmtree(staging_path, ignore_errors=True)
|
||||
raise
|
||||
return
|
||||
@ -30,7 +30,7 @@ from ..services.auth_service import (
|
||||
AuthResult,
|
||||
AuthStatus,
|
||||
check_status,
|
||||
perform_handshake,
|
||||
perform_required_handshake,
|
||||
perform_startup_handshake,
|
||||
)
|
||||
from ..services import settings_service
|
||||
@ -669,6 +669,8 @@ class ConnectionTab(QWidget):
|
||||
PlateState.NEED_AUTH: "Требуется авторизация",
|
||||
PlateState.READY: "Готово",
|
||||
}
|
||||
if state == PlateState.READY and self._connection.auth_required is False:
|
||||
_STATE_LABELS[PlateState.READY] = "Готово без авторизации"
|
||||
_STATE_CSS = {
|
||||
PlateState.DISCONNECTED: "disconnected",
|
||||
PlateState.NEED_AUTH: "need-auth",
|
||||
@ -954,5 +956,5 @@ class _HandshakeThread(QThread):
|
||||
if self._wait_for_startup:
|
||||
result = perform_startup_handshake(self._conn)
|
||||
else:
|
||||
result = perform_handshake(self._conn, timeout=2.0)
|
||||
result = perform_required_handshake(self._conn, timeout=2.0)
|
||||
self.result_ready.emit(result)
|
||||
|
||||
@ -18,7 +18,11 @@ from PySide6.QtWidgets import (
|
||||
)
|
||||
|
||||
from ...services.com_service import ComConnection
|
||||
from ...services.auth_service import AuthResult, AuthStatus, perform_handshake
|
||||
from ...services.auth_service import (
|
||||
AuthResult,
|
||||
AuthStatus,
|
||||
perform_required_handshake,
|
||||
)
|
||||
from ...ui.icons import get_icon_color, get_icon_muted, icon
|
||||
from .styled_dialog import StyledDialog
|
||||
|
||||
@ -144,7 +148,7 @@ class CommandTerminalDialog(StyledDialog):
|
||||
|
||||
self._set_command_controls_enabled(False)
|
||||
self._auth_btn.setText("Авторизация…")
|
||||
self._log.append('<span style="color:#facc15;">Запуск полной авторизации…</span>')
|
||||
self._log.append('<span style="color:#facc15;">Проверка авторизации…</span>')
|
||||
self._conn.pause_drain()
|
||||
|
||||
thread = _TerminalHandshakeThread(self._conn, parent=self)
|
||||
@ -158,9 +162,15 @@ class CommandTerminalDialog(StyledDialog):
|
||||
self._set_command_controls_enabled(True)
|
||||
self._auth_btn.setText("Авторизация")
|
||||
if result.status == AuthStatus.OK:
|
||||
self._log.append(
|
||||
'<span style="color:#4ade80;">✓ Авторизация успешно завершена</span>'
|
||||
)
|
||||
if result.auth_skipped:
|
||||
self._log.append(
|
||||
'<span style="color:#4ade80;">✓ Прошивка no_auth: '
|
||||
'авторизация не требуется</span>'
|
||||
)
|
||||
else:
|
||||
self._log.append(
|
||||
'<span style="color:#4ade80;">✓ Авторизация успешно завершена</span>'
|
||||
)
|
||||
else:
|
||||
self._log.append(
|
||||
f'<span style="color:#f87171;">✗ Авторизация: {result.detail}</span>'
|
||||
@ -262,7 +272,7 @@ class _TerminalHandshakeThread(QThread):
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
result = perform_handshake(self._conn, timeout=2.0)
|
||||
result = perform_required_handshake(self._conn, timeout=2.0)
|
||||
except Exception as exc:
|
||||
result = AuthResult(AuthStatus.FAILED, str(exc))
|
||||
self.result_ready.emit(result)
|
||||
|
||||
@ -244,7 +244,7 @@ class FirmwareDialog(StyledDialog):
|
||||
|
||||
self._flash_btn.setEnabled(False)
|
||||
self._upload_btn.setEnabled(False)
|
||||
self._log.append("Запуск авторизованного OTA…")
|
||||
self._log.append("Запуск OTA (04)…")
|
||||
|
||||
from PySide6.QtCore import QCoreApplication
|
||||
|
||||
|
||||
@ -1,17 +1,29 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from PySide6.QtCore import Qt
|
||||
from pathlib import Path
|
||||
|
||||
from PySide6.QtCore import QThread, QTimer, Qt, Signal
|
||||
from PySide6.QtGui import QColor
|
||||
from PySide6.QtWidgets import (
|
||||
QFrame,
|
||||
QGraphicsDropShadowEffect,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QMessageBox,
|
||||
QPushButton,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from .. import __version__
|
||||
from ..services.update_service import (
|
||||
UpdateCancelled,
|
||||
UpdateInfo,
|
||||
check_for_update,
|
||||
current_executable,
|
||||
download_update,
|
||||
launch_update,
|
||||
)
|
||||
from ..ui.icons import ICON_COLOR_PRIMARY, get_icon_color, icon
|
||||
from .dialogs.commands_dialog import CommandsDialog
|
||||
from .dialogs.settings_dialog import SettingsDialog
|
||||
@ -21,6 +33,9 @@ class StartTab(QWidget):
|
||||
def __init__(self, on_new_connection) -> None:
|
||||
super().__init__()
|
||||
self._open_dialog_windows: dict[str, QWidget] = {}
|
||||
self._available_update: UpdateInfo | None = None
|
||||
self._update_check_thread: _UpdateCheckThread | None = None
|
||||
self._update_download_thread: _UpdateDownloadThread | None = None
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(26, 26, 26, 26)
|
||||
layout.addStretch(1)
|
||||
@ -81,17 +96,33 @@ class StartTab(QWidget):
|
||||
self._commands_btn.clicked.connect(self._open_commands)
|
||||
bottom_row.addWidget(self._commands_btn)
|
||||
|
||||
settings_column = QWidget()
|
||||
self._settings_column_layout = QVBoxLayout(settings_column)
|
||||
self._settings_column_layout.setContentsMargins(0, 0, 0, 0)
|
||||
self._settings_column_layout.setSpacing(8)
|
||||
|
||||
self._settings_btn = QPushButton("Настройки приложения")
|
||||
self._settings_btn.setMinimumHeight(38)
|
||||
self._settings_btn.setMinimumWidth(180)
|
||||
self._settings_btn.setIcon(icon("fa5s.cog", color=get_icon_color()))
|
||||
self._settings_btn.clicked.connect(self._open_settings)
|
||||
bottom_row.addWidget(self._settings_btn)
|
||||
self._settings_column_layout.addWidget(self._settings_btn)
|
||||
|
||||
self._update_btn = QPushButton("Скачать обновление")
|
||||
self._update_btn.setProperty("role", "primary")
|
||||
self._update_btn.setMinimumHeight(38)
|
||||
self._update_btn.setMinimumWidth(180)
|
||||
self._update_btn.setIcon(icon("fa5s.download", color=ICON_COLOR_PRIMARY))
|
||||
self._update_btn.clicked.connect(self._start_update_download)
|
||||
self._update_btn.hide()
|
||||
self._settings_column_layout.addWidget(self._update_btn)
|
||||
bottom_row.addWidget(settings_column, alignment=Qt.AlignmentFlag.AlignTop)
|
||||
|
||||
bottom_row.addStretch(1)
|
||||
layout.addLayout(bottom_row)
|
||||
|
||||
layout.addStretch(2)
|
||||
QTimer.singleShot(500, self._start_update_check)
|
||||
|
||||
# ── Public ────────────────────────────────────────────────────
|
||||
|
||||
@ -100,6 +131,14 @@ class StartTab(QWidget):
|
||||
clr = get_icon_color()
|
||||
self._commands_btn.setIcon(icon("fa5s.book", color=clr))
|
||||
self._settings_btn.setIcon(icon("fa5s.cog", color=clr))
|
||||
self._update_btn.setIcon(icon("fa5s.download", color=ICON_COLOR_PRIMARY))
|
||||
|
||||
def shutdown(self) -> None:
|
||||
"""Остановить фоновые задачи перед закрытием приложения."""
|
||||
for thread in (self._update_check_thread, self._update_download_thread):
|
||||
if thread is not None and thread.isRunning():
|
||||
thread.requestInterruption()
|
||||
thread.wait(35_000)
|
||||
|
||||
# ── Handlers ──────────────────────────────────────────────────
|
||||
|
||||
@ -132,3 +171,121 @@ class StartTab(QWidget):
|
||||
|
||||
def _open_settings(self) -> None:
|
||||
self._open_single_dialog("settings", lambda: SettingsDialog(self))
|
||||
|
||||
# ── Обновление приложения ─────────────────────────────────────
|
||||
|
||||
def _start_update_check(self) -> None:
|
||||
if self._update_check_thread is not None:
|
||||
return
|
||||
thread = _UpdateCheckThread(parent=self)
|
||||
thread.result_ready.connect(self._on_update_check_done)
|
||||
thread.finished.connect(self._cleanup_update_check)
|
||||
self._update_check_thread = thread
|
||||
thread.start()
|
||||
|
||||
def _on_update_check_done(self, info: UpdateInfo | None) -> None:
|
||||
self._available_update = info
|
||||
if info is None:
|
||||
self._update_btn.hide()
|
||||
return
|
||||
self._update_btn.setText("Скачать обновление")
|
||||
self._update_btn.setToolTip(f"Доступна версия {info.tag}")
|
||||
self._update_btn.setEnabled(True)
|
||||
self._update_btn.show()
|
||||
|
||||
def _cleanup_update_check(self) -> None:
|
||||
if self._update_check_thread is not None:
|
||||
self._update_check_thread.deleteLater()
|
||||
self._update_check_thread = None
|
||||
|
||||
def _start_update_download(self) -> None:
|
||||
info = self._available_update
|
||||
if info is None or self._update_download_thread is not None:
|
||||
return
|
||||
try:
|
||||
target_directory = current_executable().parent
|
||||
except Exception as exc:
|
||||
QMessageBox.information(self, "Обновление приложения", str(exc))
|
||||
return
|
||||
|
||||
self._update_btn.setEnabled(False)
|
||||
self._update_btn.setText("Скачивание 0%")
|
||||
thread = _UpdateDownloadThread(info, target_directory, parent=self)
|
||||
thread.progress_changed.connect(self._on_update_progress)
|
||||
thread.result_ready.connect(self._on_update_downloaded)
|
||||
thread.failed.connect(self._on_update_failed)
|
||||
thread.finished.connect(self._cleanup_update_download)
|
||||
self._update_download_thread = thread
|
||||
thread.start()
|
||||
|
||||
def _on_update_progress(self, percent: int) -> None:
|
||||
self._update_btn.setText(f"Скачивание {percent}%")
|
||||
|
||||
def _on_update_downloaded(self, path: Path) -> None:
|
||||
self._update_btn.setText("Перезапуск…")
|
||||
try:
|
||||
launch_update(path)
|
||||
except Exception as exc:
|
||||
path.unlink(missing_ok=True)
|
||||
self._on_update_failed(str(exc))
|
||||
return
|
||||
self.window().close()
|
||||
|
||||
def _on_update_failed(self, detail: str) -> None:
|
||||
self._update_btn.setText("Скачать обновление")
|
||||
self._update_btn.setEnabled(True)
|
||||
QMessageBox.warning(
|
||||
self,
|
||||
"Обновление приложения",
|
||||
f"Не удалось установить обновление:\n{detail}",
|
||||
)
|
||||
|
||||
def _cleanup_update_download(self) -> None:
|
||||
if self._update_download_thread is not None:
|
||||
self._update_download_thread.deleteLater()
|
||||
self._update_download_thread = None
|
||||
|
||||
|
||||
class _UpdateCheckThread(QThread):
|
||||
result_ready = Signal(object)
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
info = check_for_update(__version__)
|
||||
except Exception:
|
||||
info = None
|
||||
if not self.isInterruptionRequested():
|
||||
self.result_ready.emit(info)
|
||||
|
||||
|
||||
class _UpdateDownloadThread(QThread):
|
||||
progress_changed = Signal(int)
|
||||
result_ready = Signal(object)
|
||||
failed = Signal(str)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
info: UpdateInfo,
|
||||
target_directory: Path,
|
||||
parent: QWidget | None = None,
|
||||
) -> None:
|
||||
super().__init__(parent)
|
||||
self._info = info
|
||||
self._target_directory = target_directory
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
path = download_update(
|
||||
self._info,
|
||||
self._target_directory,
|
||||
progress=self.progress_changed.emit,
|
||||
cancelled=self.isInterruptionRequested,
|
||||
)
|
||||
except UpdateCancelled:
|
||||
return
|
||||
except Exception as exc:
|
||||
if not self.isInterruptionRequested():
|
||||
self.failed.emit(str(exc))
|
||||
return
|
||||
if not self.isInterruptionRequested():
|
||||
self.result_ready.emit(path)
|
||||
|
||||
@ -54,16 +54,36 @@ if ($InstallBuildDeps) {
|
||||
# ── Сборка ─────────────────────────────────────────────────────────
|
||||
Write-Host ""
|
||||
Write-Host "=== Сборка EXE ==="
|
||||
& $venvPython -m PyInstaller --noconfirm --clean plate_config_app.spec
|
||||
|
||||
if ($LASTEXITCODE -ne 0) {
|
||||
throw "PyInstaller завершился с ошибкой (код $LASTEXITCODE)."
|
||||
$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
|
||||
}
|
||||
|
||||
$exe = Join-Path $projectRoot "dist\plate_config_app.exe"
|
||||
if (-not (Test-Path $exe)) {
|
||||
if ($buildExitCode -ne 0) {
|
||||
throw "PyInstaller завершился с ошибкой (код $buildExitCode)."
|
||||
}
|
||||
|
||||
$builtExe = Join-Path $projectRoot "dist\plate_config_app.exe"
|
||||
if (-not (Test-Path $builtExe)) {
|
||||
throw "EXE не найден после сборки."
|
||||
}
|
||||
$exe = Join-Path $projectRoot "dist\plate_config_app-windows-x86_64.exe"
|
||||
Move-Item -LiteralPath $builtExe -Destination $exe -Force
|
||||
|
||||
# ── Копирование ресурсных папок рядом с EXE ───────────────────────
|
||||
Write-Host ""
|
||||
@ -80,6 +100,18 @@ foreach ($folder in $resourceFolders) {
|
||||
}
|
||||
}
|
||||
|
||||
$archive = Join-Path $projectRoot "dist\plate_config_app-windows-x86_64.zip"
|
||||
$archiveEntries = @($exe)
|
||||
foreach ($folder in $resourceFolders) {
|
||||
$resourcePath = Join-Path $projectRoot "dist\$folder"
|
||||
if (Test-Path $resourcePath) {
|
||||
$archiveEntries += $resourcePath
|
||||
}
|
||||
}
|
||||
Compress-Archive -LiteralPath $archiveEntries -DestinationPath $archive -Force
|
||||
|
||||
$size = [math]::Round((Get-Item $exe).Length / 1MB, 1)
|
||||
$archiveSize = [math]::Round((Get-Item $archive).Length / 1MB, 1)
|
||||
Write-Host ""
|
||||
Write-Host "Сборка завершена: $exe ($size MB)"
|
||||
Write-Host "ZIP-архив: $archive ($archiveSize MB)"
|
||||
|
||||
@ -6,8 +6,8 @@ project_root="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
|
||||
venv_dir="$project_root/.venv-linux"
|
||||
build_dir="$project_root/build-linux"
|
||||
dist_dir="$project_root/dist-linux"
|
||||
package_name="plate_config_app-debian13-x86_64"
|
||||
package_dir="$dist_dir/$package_name"
|
||||
artifact_name="plate_config_app-linux-x86_64"
|
||||
archive_name="plate_config_app-linux-x86_64.zip"
|
||||
|
||||
cd "$project_root"
|
||||
|
||||
@ -25,17 +25,17 @@ python3 -m venv "$venv_dir"
|
||||
--distpath "$dist_dir" \
|
||||
plate_config_app.spec
|
||||
|
||||
mkdir -p "$package_dir"
|
||||
install -m 0755 "$dist_dir/plate_config_app" "$package_dir/plate_config_app"
|
||||
install -m 0755 "$dist_dir/plate_config_app" "$dist_dir/$artifact_name"
|
||||
rm -f "$dist_dir/plate_config_app"
|
||||
for resource_dir in light_modes input_mappings commands; do
|
||||
cp -a "$resource_dir" "$package_dir/$resource_dir"
|
||||
cp -a "$resource_dir" "$dist_dir/$resource_dir"
|
||||
done
|
||||
|
||||
cat > "$package_dir/README.txt" <<'EOF'
|
||||
cat > "$dist_dir/README.txt" <<'EOF'
|
||||
Plate Config App для Debian 13 x86_64
|
||||
|
||||
Запуск:
|
||||
./plate_config_app
|
||||
./plate_config_app-linux-x86_64
|
||||
|
||||
Папки light_modes, input_mappings и commands можно изменять без пересборки.
|
||||
Для доступа к USB-COM добавьте пользователя в группу dialout и войдите в систему заново:
|
||||
@ -44,9 +44,16 @@ EOF
|
||||
|
||||
(
|
||||
cd "$dist_dir"
|
||||
zip -qr "$package_name.zip" "$package_name"
|
||||
sha256sum "$package_name.zip" > "$package_name.zip.sha256"
|
||||
"$venv_dir/bin/python" -m zipfile -c \
|
||||
"$archive_name" \
|
||||
"$artifact_name" \
|
||||
light_modes \
|
||||
input_mappings \
|
||||
commands \
|
||||
README.txt
|
||||
)
|
||||
|
||||
printf 'Готово: %s\n' "$dist_dir/$package_name.zip"
|
||||
sha256sum "$dist_dir/$package_name.zip"
|
||||
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"
|
||||
|
||||
@ -8,7 +8,9 @@ from app.services.auth_service import (
|
||||
AuthStatus,
|
||||
_aes_encrypt,
|
||||
check_status,
|
||||
firmware_requires_auth,
|
||||
perform_handshake,
|
||||
perform_required_handshake,
|
||||
perform_startup_handshake,
|
||||
wait_for_startup_banner,
|
||||
)
|
||||
@ -20,6 +22,8 @@ class FakeConnection:
|
||||
self._responses = deque(responses)
|
||||
self.events: list[str] = []
|
||||
self.release_count = 0
|
||||
self.firmware_version: str | None = None
|
||||
self.auth_required: bool | None = None
|
||||
|
||||
def read_response(self, timeout: float | None = None) -> str | None:
|
||||
response = self._responses.popleft() if self._responses else None
|
||||
@ -35,6 +39,10 @@ class FakeConnection:
|
||||
def release(self) -> None:
|
||||
self.release_count += 1
|
||||
|
||||
def set_firmware_version(self, version: str, *, auth_required: bool) -> None:
|
||||
self.firmware_version = version
|
||||
self.auth_required = auth_required
|
||||
|
||||
|
||||
class StartupHandshakeTests(unittest.TestCase):
|
||||
_HOST_CHALLENGE = bytes.fromhex("00112233445566778899AABBCCDDEEFF")
|
||||
@ -91,6 +99,45 @@ class StartupHandshakeTests(unittest.TestCase):
|
||||
index for index, event in enumerate(conn.events) if event.startswith("send:07")
|
||||
)
|
||||
self.assertLess(sleep_index, send_index)
|
||||
self.assertEqual("2.3.6 release", conn.firmware_version)
|
||||
self.assertTrue(conn.auth_required)
|
||||
|
||||
def test_no_auth_banner_skips_handshake(self) -> None:
|
||||
conn = FakeConnection([
|
||||
"boot log",
|
||||
"periphery controller ver_2.3.6_No_Auth_prizes",
|
||||
])
|
||||
sleep_calls: list[float] = []
|
||||
|
||||
result = perform_startup_handshake(
|
||||
conn,
|
||||
startup_delay=2.0,
|
||||
sleep_fn=sleep_calls.append,
|
||||
)
|
||||
|
||||
self.assertEqual(AuthStatus.OK, result.status)
|
||||
self.assertTrue(result.startup_ready)
|
||||
self.assertTrue(result.auth_skipped)
|
||||
self.assertEqual("2.3.6_No_Auth_prizes", result.firmware_version)
|
||||
self.assertFalse(conn.auth_required)
|
||||
self.assertEqual([2.0], sleep_calls)
|
||||
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
|
||||
self.assertEqual(1, conn.release_count)
|
||||
|
||||
def test_auth_mode_marker_is_case_insensitive(self) -> None:
|
||||
self.assertFalse(firmware_requires_auth("2.3.6_no_auth_prizes"))
|
||||
self.assertFalse(firmware_requires_auth("2.3.6_NO_AUTH"))
|
||||
self.assertTrue(firmware_requires_auth("2.3.6_release"))
|
||||
|
||||
def test_manual_handshake_is_skipped_for_known_no_auth_firmware(self) -> None:
|
||||
conn = FakeConnection([])
|
||||
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
|
||||
|
||||
result = perform_required_handshake(conn, timeout=0.1)
|
||||
|
||||
self.assertEqual(AuthStatus.OK, result.status)
|
||||
self.assertTrue(result.auth_skipped)
|
||||
self.assertFalse(any(event.startswith("send:") for event in conn.events))
|
||||
|
||||
def test_banner_requires_non_empty_version_suffix(self) -> None:
|
||||
conn = FakeConnection(["periphery controller ver_", None])
|
||||
@ -174,6 +221,42 @@ class StartupHandshakeTests(unittest.TestCase):
|
||||
self.assertEqual(AuthStatus.OK, check_status(ready_conn, timeout=0.1))
|
||||
self.assertEqual(AuthStatus.NEED_AUTH, check_status(not_ready_conn, timeout=0.1))
|
||||
|
||||
def test_no_auth_status_accepts_any_non_empty_response(self) -> None:
|
||||
conn = FakeConnection(["OK"])
|
||||
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
|
||||
|
||||
self.assertEqual(AuthStatus.OK, check_status(conn, timeout=0.1))
|
||||
|
||||
def test_no_auth_status_timeout_is_connection_failure(self) -> None:
|
||||
conn = FakeConnection([None])
|
||||
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
|
||||
|
||||
self.assertEqual(AuthStatus.DISCONNECTED, check_status(conn, timeout=0.1))
|
||||
|
||||
def test_new_secure_banner_reenables_handshake(self) -> None:
|
||||
conn = FakeConnection([
|
||||
"periphery controller ver_2.0.0_release",
|
||||
self._CHAL,
|
||||
self._RAES,
|
||||
"READY!",
|
||||
])
|
||||
conn.set_firmware_version("1.0.0_no_auth", auth_required=False)
|
||||
|
||||
with patch(
|
||||
"app.services.auth_service.os.urandom",
|
||||
return_value=self._HOST_CHALLENGE,
|
||||
):
|
||||
result = perform_startup_handshake(
|
||||
conn,
|
||||
startup_delay=0.0,
|
||||
sleep_fn=lambda _seconds: None,
|
||||
)
|
||||
|
||||
self.assertEqual(AuthStatus.OK, result.status)
|
||||
self.assertFalse(result.auth_skipped)
|
||||
self.assertTrue(conn.auth_required)
|
||||
self.assertTrue(any(event.startswith("send:07") for event in conn.events))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@ -51,6 +51,9 @@ class FakeSerial:
|
||||
def reset_input_buffer(self) -> None:
|
||||
self.events.append("reset_input_buffer")
|
||||
|
||||
def close(self) -> None:
|
||||
self.is_open = False
|
||||
|
||||
|
||||
class ComConnectionTests(unittest.TestCase):
|
||||
def test_read_response_keeps_serial_timeout_unchanged(self) -> None:
|
||||
@ -77,6 +80,16 @@ class ComConnectionTests(unittest.TestCase):
|
||||
serial.events,
|
||||
)
|
||||
|
||||
def test_close_clears_detected_firmware_mode(self) -> None:
|
||||
conn = ComConnection()
|
||||
conn._serial = FakeSerial()
|
||||
conn.set_firmware_version("2.0.0_no_auth", auth_required=False)
|
||||
|
||||
conn.close()
|
||||
|
||||
self.assertIsNone(conn.firmware_version)
|
||||
self.assertIsNone(conn.auth_required)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@ -14,6 +14,8 @@ class FakeConnection:
|
||||
self._responses = deque(responses)
|
||||
self.events: list[str] = []
|
||||
self.release_count = 0
|
||||
self.firmware_version: str | None = None
|
||||
self.auth_required: bool | None = None
|
||||
|
||||
def read_response(self, timeout: float | None = None) -> str | None:
|
||||
response = self._responses.popleft() if self._responses else None
|
||||
@ -29,6 +31,10 @@ class FakeConnection:
|
||||
def release(self) -> None:
|
||||
self.release_count += 1
|
||||
|
||||
def set_firmware_version(self, version: str, *, auth_required: bool) -> None:
|
||||
self.firmware_version = version
|
||||
self.auth_required = auth_required
|
||||
|
||||
|
||||
class AuthenticatedOtaTests(unittest.TestCase):
|
||||
_HOST_CHALLENGE = bytes.fromhex("00112233445566778899AABBCCDDEEFF")
|
||||
@ -50,11 +56,8 @@ class AuthenticatedOtaTests(unittest.TestCase):
|
||||
sleep_fn=lambda _seconds: None,
|
||||
)
|
||||
|
||||
def test_authenticates_before_04_and_after_successful_update(self) -> None:
|
||||
def test_sends_04_before_authentication_after_successful_update(self) -> None:
|
||||
conn = FakeConnection([
|
||||
self._CHAL,
|
||||
self._RAES,
|
||||
"READY!",
|
||||
"Writing 100%",
|
||||
"Successful update",
|
||||
"periphery controller ver_2.4.0 release",
|
||||
@ -66,27 +69,37 @@ class AuthenticatedOtaTests(unittest.TestCase):
|
||||
result = self._run(conn)
|
||||
|
||||
self.assertTrue(result.success)
|
||||
self.assertEqual(AuthStatus.OK, result.pre_auth.status)
|
||||
self.assertEqual(AuthStatus.OK, result.post_auth.status)
|
||||
send_events = [event for event in conn.events if event.startswith("send:")]
|
||||
self.assertEqual("send:04", send_events[1])
|
||||
self.assertTrue(send_events[0].startswith("send:07"))
|
||||
self.assertTrue(send_events[2].startswith("send:07"))
|
||||
self.assertEqual("send:04", send_events[0])
|
||||
self.assertTrue(send_events[1].startswith("send:07"))
|
||||
|
||||
def test_does_not_send_04_when_pre_authentication_fails(self) -> None:
|
||||
def test_sends_04_without_pre_authentication(self) -> None:
|
||||
conn = FakeConnection([None])
|
||||
|
||||
result = self._run(conn)
|
||||
|
||||
self.assertFalse(result.success)
|
||||
self.assertEqual(AuthStatus.TIMEOUT, result.pre_auth.status)
|
||||
self.assertFalse(any(event == "send:04" for event in conn.events))
|
||||
self.assertEqual("send:04", conn.events[0])
|
||||
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
|
||||
|
||||
def test_no_auth_firmware_skips_post_ota_handshake(self) -> None:
|
||||
conn = FakeConnection([
|
||||
"Successful update",
|
||||
"periphery controller ver_2.0.0_no_auth",
|
||||
])
|
||||
|
||||
result = self._run(conn)
|
||||
|
||||
self.assertTrue(result.success)
|
||||
self.assertEqual(AuthStatus.OK, result.post_auth.status)
|
||||
self.assertTrue(result.post_auth.auth_skipped)
|
||||
self.assertFalse(conn.auth_required)
|
||||
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
|
||||
self.assertIn("без авторизации", result.detail)
|
||||
|
||||
def test_reports_post_authentication_failure_after_update(self) -> None:
|
||||
conn = FakeConnection([
|
||||
self._CHAL,
|
||||
self._RAES,
|
||||
"READY!",
|
||||
"Successful update",
|
||||
None,
|
||||
])
|
||||
|
||||
160
tests/test_start_tab.py
Normal file
160
tests/test_start_tab.py
Normal file
@ -0,0 +1,160 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
from PySide6.QtWidgets import QApplication
|
||||
|
||||
from app.services.update_service import UpdateError, UpdateInfo
|
||||
from app.widgets.start_tab import StartTab, _UpdateCheckThread, _UpdateDownloadThread
|
||||
|
||||
|
||||
class StartTabUpdateTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls._app = QApplication.instance() or QApplication([])
|
||||
|
||||
def test_update_button_is_below_settings_and_hidden_initially(self) -> None:
|
||||
tab = StartTab(lambda: None)
|
||||
|
||||
self.assertTrue(tab._update_btn.isHidden())
|
||||
self.assertLess(
|
||||
tab._settings_column_layout.indexOf(tab._settings_btn),
|
||||
tab._settings_column_layout.indexOf(tab._update_btn),
|
||||
)
|
||||
tab.shutdown()
|
||||
|
||||
def test_available_update_shows_download_button(self) -> None:
|
||||
tab = StartTab(lambda: None)
|
||||
info = UpdateInfo(
|
||||
tag="v2.1.0",
|
||||
version="2.1.0",
|
||||
asset_name="plate_config_app-windows-x86_64.zip",
|
||||
download_url="https://git.robotail.pro/download/test",
|
||||
)
|
||||
|
||||
tab._on_update_check_done(info)
|
||||
|
||||
self.assertFalse(tab._update_btn.isHidden())
|
||||
self.assertEqual("Скачать обновление", tab._update_btn.text())
|
||||
self.assertIn("v2.1.0", tab._update_btn.toolTip())
|
||||
tab.shutdown()
|
||||
|
||||
def test_no_update_hides_previously_visible_button(self) -> None:
|
||||
tab = StartTab(lambda: None)
|
||||
tab._on_update_check_done(self._update_info())
|
||||
|
||||
tab._on_update_check_done(None)
|
||||
|
||||
self.assertTrue(tab._update_btn.isHidden())
|
||||
self.assertIsNone(tab._available_update)
|
||||
tab.shutdown()
|
||||
|
||||
def test_progress_updates_button_text(self) -> None:
|
||||
tab = StartTab(lambda: None)
|
||||
|
||||
tab._on_update_progress(57)
|
||||
|
||||
self.assertEqual("Скачивание 57%", tab._update_btn.text())
|
||||
tab.shutdown()
|
||||
|
||||
def test_download_failure_resets_button_and_warns(self) -> None:
|
||||
tab = StartTab(lambda: None)
|
||||
tab._update_btn.setEnabled(False)
|
||||
tab._update_btn.setText("Скачивание 42%")
|
||||
|
||||
with patch("app.widgets.start_tab.QMessageBox.warning") as warning:
|
||||
tab._on_update_failed("ошибка сети")
|
||||
|
||||
self.assertTrue(tab._update_btn.isEnabled())
|
||||
self.assertEqual("Скачать обновление", tab._update_btn.text())
|
||||
warning.assert_called_once()
|
||||
self.assertIn("ошибка сети", warning.call_args.args[2])
|
||||
tab.shutdown()
|
||||
|
||||
def test_launch_failure_removes_download_and_warns(self) -> None:
|
||||
tab = StartTab(lambda: None)
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
downloaded = Path(directory) / ".downloaded-update"
|
||||
downloaded.write_bytes(b"MZupdate")
|
||||
|
||||
with (
|
||||
patch(
|
||||
"app.widgets.start_tab.launch_update",
|
||||
side_effect=UpdateError("не удалось запустить helper"),
|
||||
),
|
||||
patch("app.widgets.start_tab.QMessageBox.warning") as warning,
|
||||
):
|
||||
tab._on_update_downloaded(downloaded)
|
||||
|
||||
self.assertFalse(downloaded.exists())
|
||||
self.assertEqual("Скачать обновление", tab._update_btn.text())
|
||||
self.assertTrue(tab._update_btn.isEnabled())
|
||||
warning.assert_called_once()
|
||||
tab.shutdown()
|
||||
|
||||
def test_successful_download_launches_update_and_closes_window(self) -> None:
|
||||
tab = StartTab(lambda: None)
|
||||
downloaded = Path("downloaded-update")
|
||||
|
||||
with (
|
||||
patch("app.widgets.start_tab.launch_update") as launcher,
|
||||
patch.object(tab, "close") as close_window,
|
||||
):
|
||||
tab._on_update_downloaded(downloaded)
|
||||
|
||||
launcher.assert_called_once_with(downloaded)
|
||||
close_window.assert_called_once_with()
|
||||
self.assertEqual("Перезапуск…", tab._update_btn.text())
|
||||
tab.shutdown()
|
||||
|
||||
def test_update_check_thread_hides_network_errors(self) -> None:
|
||||
thread = _UpdateCheckThread()
|
||||
results: list[UpdateInfo | None] = []
|
||||
thread.result_ready.connect(results.append)
|
||||
|
||||
with patch(
|
||||
"app.widgets.start_tab.check_for_update",
|
||||
side_effect=UpdateError("offline"),
|
||||
):
|
||||
thread.run()
|
||||
|
||||
self.assertEqual([None], results)
|
||||
|
||||
def test_download_thread_emits_progress_and_result(self) -> None:
|
||||
info = self._update_info()
|
||||
result_path = Path("downloaded-update")
|
||||
thread = _UpdateDownloadThread(info, Path("."))
|
||||
progress_values: list[int] = []
|
||||
results: list[Path] = []
|
||||
thread.progress_changed.connect(progress_values.append)
|
||||
thread.result_ready.connect(results.append)
|
||||
|
||||
def fake_download(_info, _directory, *, progress, cancelled):
|
||||
self.assertFalse(cancelled())
|
||||
progress(75)
|
||||
return result_path
|
||||
|
||||
with patch("app.widgets.start_tab.download_update", side_effect=fake_download):
|
||||
thread.run()
|
||||
|
||||
self.assertEqual([75], progress_values)
|
||||
self.assertEqual([result_path], results)
|
||||
|
||||
@staticmethod
|
||||
def _update_info() -> UpdateInfo:
|
||||
return UpdateInfo(
|
||||
tag="v2.1.0",
|
||||
version="2.1.0",
|
||||
asset_name="plate_config_app-windows-x86_64.zip",
|
||||
download_url="https://git.robotail.pro/download/test",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
716
tests/test_update_service.py
Normal file
716
tests/test_update_service.py
Normal file
@ -0,0 +1,716 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import io
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from app.services import update_service
|
||||
from app.services.update_service import (
|
||||
UpdateCancelled,
|
||||
UpdateError,
|
||||
UpdateInfo,
|
||||
archive_executable_name,
|
||||
build_download_url,
|
||||
check_for_update,
|
||||
download_update,
|
||||
extract_release_tag,
|
||||
launch_update,
|
||||
normalize_version,
|
||||
parse_version,
|
||||
platform_asset_name,
|
||||
platform_executable_name,
|
||||
signal_update_ready,
|
||||
)
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(
|
||||
self,
|
||||
url: str,
|
||||
data: bytes = b"",
|
||||
*,
|
||||
content_length: int | None = None,
|
||||
) -> None:
|
||||
self._url = url
|
||||
self._stream = io.BytesIO(data)
|
||||
self.headers: dict[str, str] = {}
|
||||
if content_length is not None:
|
||||
self.headers["Content-Length"] = str(content_length)
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *_args) -> None:
|
||||
return None
|
||||
|
||||
def geturl(self) -> str:
|
||||
return self._url
|
||||
|
||||
def read(self, size: int = -1) -> bytes:
|
||||
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 "))
|
||||
self.assertEqual("2.0.0", normalize_version("V2.0.0"))
|
||||
|
||||
def test_parses_version_into_numeric_components(self) -> None:
|
||||
self.assertEqual((12, 34, 56), parse_version("v12.34.56"))
|
||||
self.assertEqual((2, 0, 0), parse_version(" 2.0.0 "))
|
||||
|
||||
def test_rejects_non_standard_versions(self) -> None:
|
||||
invalid_versions = (
|
||||
"v2.0",
|
||||
"v2.0.0.1",
|
||||
"v2.0.0-beta",
|
||||
"v2.01.0",
|
||||
"release-2.0.0",
|
||||
)
|
||||
for version in invalid_versions:
|
||||
with self.subTest(version=version), self.assertRaises(UpdateError):
|
||||
parse_version(version)
|
||||
|
||||
def test_extracts_tag_from_latest_redirect(self) -> None:
|
||||
tag = extract_release_tag(
|
||||
"https://git.robotail.pro/Robotail/plate-config-app/releases/tag/v2.1.0"
|
||||
)
|
||||
|
||||
self.assertEqual("v2.1.0", tag)
|
||||
|
||||
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",
|
||||
)
|
||||
for url in invalid_urls:
|
||||
with self.subTest(url=url), self.assertRaises(UpdateError):
|
||||
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")
|
||||
|
||||
self.assertTrue(url.endswith("/v2.1.0%20rc%2F1/plate%20config.zip"))
|
||||
|
||||
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)
|
||||
|
||||
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"),
|
||||
("2.5.3", "v2.5.4"),
|
||||
)
|
||||
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="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_rejects_invalid_current_or_remote_version(self) -> None:
|
||||
for current, remote in (("development", "v2.1.0"), ("2.0.0", "v2.1")):
|
||||
with self.subTest(current=current, remote=remote), self.assertRaises(
|
||||
UpdateError
|
||||
):
|
||||
check_for_update(
|
||||
current,
|
||||
opener=self._release_opener(remote),
|
||||
system="Linux",
|
||||
machine="x86_64",
|
||||
)
|
||||
|
||||
def test_wraps_network_error_during_update_check(self) -> None:
|
||||
def opener(_request, timeout: float):
|
||||
raise OSError("offline")
|
||||
|
||||
with self.assertRaisesRegex(UpdateError, "offline") as raised:
|
||||
check_for_update("2.0.0", opener=opener)
|
||||
|
||||
self.assertIsInstance(raised.exception.__cause__, OSError)
|
||||
|
||||
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.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_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_signals_restart_only_when_ready_file_is_requested(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
marker = Path(directory) / "ready"
|
||||
with patch.dict(os.environ, {update_service.UPDATE_READY_ENV: str(marker)}):
|
||||
signal_update_ready()
|
||||
self.assertEqual(str(os.getpid()), marker.read_text(encoding="ascii"))
|
||||
self.assertNotIn(update_service.UPDATE_READY_ENV, os.environ)
|
||||
marker.unlink()
|
||||
signal_update_ready()
|
||||
self.assertFalse(marker.exists())
|
||||
|
||||
def test_downloads_and_validates_windows_zip(self) -> None:
|
||||
info = self._update_info()
|
||||
data = make_update_archive(
|
||||
"plate_config_app-windows-x86_64.exe",
|
||||
b"MZnew",
|
||||
extra_files={"commands/default.csv": b"04;Update"},
|
||||
)
|
||||
progress_values: list[int] = []
|
||||
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
result = download_update(
|
||||
info,
|
||||
Path(directory),
|
||||
opener=lambda _request, timeout: FakeResponse(
|
||||
info.download_url,
|
||||
data,
|
||||
content_length=len(data),
|
||||
),
|
||||
progress=progress_values.append,
|
||||
)
|
||||
|
||||
self.assertEqual(data, result.read_bytes())
|
||||
self.assertEqual(100, progress_values[-1])
|
||||
|
||||
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),
|
||||
)
|
||||
|
||||
with zipfile.ZipFile(result) as archive:
|
||||
self.assertEqual(
|
||||
["plate_config_app-linux-x86_64"],
|
||||
archive.namelist(),
|
||||
)
|
||||
|
||||
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",
|
||||
)
|
||||
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,
|
||||
target,
|
||||
opener=lambda _request, timeout, payload=data, length=content_length: (
|
||||
FakeResponse(
|
||||
info.download_url,
|
||||
payload,
|
||||
content_length=length,
|
||||
)
|
||||
),
|
||||
)
|
||||
self.assertEqual([], list(target.iterdir()))
|
||||
|
||||
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,
|
||||
):
|
||||
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:
|
||||
target = Path(directory)
|
||||
with self.assertRaises(UpdateCancelled):
|
||||
download_update(
|
||||
info,
|
||||
target,
|
||||
opener=lambda _request, timeout: FakeResponse(
|
||||
info.download_url,
|
||||
b"partial",
|
||||
),
|
||||
cancelled=lambda: True,
|
||||
)
|
||||
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:
|
||||
with self.assertRaisesRegex(UpdateError, "не найден"):
|
||||
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_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)
|
||||
target = root / "plate_config_app-windows-x86_64.exe"
|
||||
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(
|
||||
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("$Staging", script)
|
||||
self.assertIn("$backedUp", script)
|
||||
self.assertIn("AddSeconds(30)", script)
|
||||
self.assertIn("$restarted.HasExited", script)
|
||||
self.assertIn("$ReadyFile", script)
|
||||
self.assertIn('$env:PYINSTALLER_RESET_ENVIRONMENT = "1"', script)
|
||||
popen.assert_called_once()
|
||||
|
||||
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
|
||||
def test_prepares_linux_package_replacement_helper(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
target = root / "plate_config_app-linux-x86_64"
|
||||
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(
|
||||
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("staging_path", script)
|
||||
self.assertIn("rollback", script)
|
||||
self.assertIn("PLATE_CONFIG_UPDATE_READY_FILE", script)
|
||||
self.assertIn("PYINSTALLER_RESET_ENVIRONMENT=1", script)
|
||||
popen.assert_called_once()
|
||||
|
||||
def test_extracts_external_archive_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"
|
||||
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,
|
||||
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])
|
||||
|
||||
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"
|
||||
archive = root / "update.zip"
|
||||
target.write_bytes(b"old")
|
||||
|
||||
with self.assertRaisesRegex(UpdateError, "не найден"):
|
||||
launch_update(archive, executable_path=target, system="Linux")
|
||||
|
||||
archive.write_bytes(b"not-a-zip")
|
||||
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()
|
||||
|
||||
@staticmethod
|
||||
def _update_info(*, system: str = "windows") -> UpdateInfo:
|
||||
asset_name = f"plate_config_app-{system}-x86_64.zip"
|
||||
return UpdateInfo(
|
||||
tag="v2.1.0",
|
||||
version="2.1.0",
|
||||
asset_name=asset_name,
|
||||
download_url=(
|
||||
"https://git.robotail.pro/Robotail/plate-config-app/releases/"
|
||||
f"download/v2.1.0/{asset_name}"
|
||||
),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _release_opener(tag: str):
|
||||
def opener(_request, timeout: float):
|
||||
return FakeResponse(
|
||||
"https://git.robotail.pro/Robotail/plate-config-app/"
|
||||
f"releases/tag/{tag}"
|
||||
)
|
||||
|
||||
return opener
|
||||
|
||||
|
||||
class ReplacementIntegrationTests(unittest.TestCase):
|
||||
@staticmethod
|
||||
def _digest(path: Path) -> str:
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
@staticmethod
|
||||
def _wait_for_rollback(
|
||||
target: Path,
|
||||
expected_digest: str,
|
||||
replaced_resource: Path,
|
||||
) -> None:
|
||||
deadline = time.monotonic() + 20
|
||||
while time.monotonic() < deadline:
|
||||
if (
|
||||
target.is_file()
|
||||
and ReplacementIntegrationTests._digest(target) == expected_digest
|
||||
and replaced_resource.is_file()
|
||||
and replaced_resource.read_text(encoding="utf-8") == "old"
|
||||
and not list(target.parent.glob(".plate_config_updater-*"))
|
||||
and not list(target.parent.glob(".plate_config_update-*"))
|
||||
and not list(target.parent.glob(".plate_config_backup-*"))
|
||||
and (target.parent / "update-error.log").is_file()
|
||||
):
|
||||
return
|
||||
time.sleep(0.1)
|
||||
raise AssertionError("Внешний helper не откатил пакет за 20 секунд")
|
||||
|
||||
@unittest.skipUnless(sys.platform == "win32", "Тест предназначен для Windows")
|
||||
def test_windows_helper_rolls_back_when_new_app_exits_early(self) -> None:
|
||||
system_root = Path(os.environ.get("SystemRoot", r"C:\Windows"))
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
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" / "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")
|
||||
|
||||
self._wait_for_rollback(
|
||||
target,
|
||||
expected_digest,
|
||||
root / "commands/version.txt",
|
||||
)
|
||||
processes[0].wait(timeout=5)
|
||||
self._assert_package_preserved(root)
|
||||
self._wait_for_windows_unlock(target)
|
||||
|
||||
@unittest.skipUnless(sys.platform.startswith("linux"), "Тест предназначен для Linux")
|
||||
def test_linux_helper_rolls_back_when_new_app_exits_early(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
target = root / "plate_config_app-linux-x86_64"
|
||||
archive = root / "update.zip"
|
||||
new_executable = Path("/bin/true").read_bytes()
|
||||
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(target)
|
||||
processes = self._launch_and_capture(archive, target, "Linux")
|
||||
|
||||
self._wait_for_rollback(
|
||||
target,
|
||||
expected_digest,
|
||||
root / "commands/version.txt",
|
||||
)
|
||||
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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
x
Reference in New Issue
Block a user