Danil Iarosh
35c9693740
- Исправлен перезапуск приложения после замены файлов и добавлено подтверждение запуска новой версии. - Обеспечены откат при ошибке и корректное завершение фоновых задач. - Исключены посторонние DLL из Windows-сборки, обновлены тесты и инструкция ручного перехода с v2.0.1.
717 lines
27 KiB
Python
717 lines
27 KiB
Python
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()
|