260 lines
8.5 KiB
Python
260 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from ptlinux.paths import AppPaths
|
|
from ptlinux.self_update import (
|
|
LauncherUpdateCheck,
|
|
UpdateManifest,
|
|
_download_archive,
|
|
_record_launcher_operation,
|
|
check_for_launcher_update,
|
|
install_launcher_update,
|
|
rollback_launcher_update,
|
|
)
|
|
|
|
|
|
def release_manifest(version: str, channel: str = "stable") -> UpdateManifest:
|
|
return UpdateManifest(
|
|
schema=1,
|
|
channel=channel,
|
|
version=version,
|
|
minimum_updater_version="0.5.0",
|
|
archive_url=f"https://updates.example.test/PristonTale-EU-{version}.tar.gz",
|
|
archive_sha256="a" * 64,
|
|
archive_size=123,
|
|
published_at="2026-09-08T12:00:00Z",
|
|
notes_url=f"https://updates.example.test/releases/tag/v{version}",
|
|
tag=f"v{version}",
|
|
manifest_sha256="b" * 64,
|
|
)
|
|
|
|
|
|
def app_paths(root: Path) -> AppPaths:
|
|
return AppPaths.from_environment({"HOME": str(root / "home")})
|
|
|
|
|
|
def make_release(path: Path, version: str) -> None:
|
|
files = {
|
|
"release.json": json.dumps(
|
|
{
|
|
"channel": "prerelease" if "-" in version else "stable",
|
|
"config_schema": 1,
|
|
"minimum_config_schema": 1,
|
|
"schema": 1,
|
|
"tag": f"v{version}",
|
|
"version": version,
|
|
}
|
|
).encode(),
|
|
"ptlinux/__init__.py": f'__version__ = "{version}"\n'.encode(),
|
|
"ptlinux/__main__.py": b"fixture\n",
|
|
"ptlinux/cli.py": b"fixture\n",
|
|
"ptlinux/self_update.py": b"fixture\n",
|
|
"ptlinux/launcher-update-public.pem": b"fixture\n",
|
|
"ptdiag/__init__.py": b"fixture\n",
|
|
"ptdiag/assbin.py": b"fixture\n",
|
|
"ptdiag/logs.py": b"fixture\n",
|
|
}
|
|
for relative, data in files.items():
|
|
target = path / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(data)
|
|
target.chmod(0o644)
|
|
for directory in (path, *path.rglob("*")):
|
|
if directory.is_dir():
|
|
directory.chmod(0o755)
|
|
|
|
|
|
class UpdateTelemetryTests(unittest.TestCase):
|
|
def test_local_operation_history_is_bounded_to_thirty_records(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
paths = app_paths(Path(temporary))
|
|
for index in range(35):
|
|
self.assertTrue(
|
|
_record_launcher_operation(
|
|
paths,
|
|
from_version=f"1.0.{index}",
|
|
to_version=f"1.0.{index + 1}",
|
|
from_channel="stable",
|
|
to_channel="stable",
|
|
selected_channel="stable",
|
|
result="success",
|
|
rollback=False,
|
|
reason="automatic_update",
|
|
)
|
|
)
|
|
value = json.loads(
|
|
paths.launcher_update_operations.read_text(encoding="utf-8")
|
|
)
|
|
self.assertEqual(len(value["operations"]), 30)
|
|
self.assertEqual(value["operations"][0]["from"], "1.0.5")
|
|
self.assertEqual(value["operations"][-1]["to"], "1.0.35")
|
|
|
|
def test_rollback_download_emits_success_and_failure_families(self) -> None:
|
|
content = b"authenticated rollback archive"
|
|
manifest = replace(
|
|
release_manifest("1.7.4"),
|
|
archive_size=len(content),
|
|
archive_sha256=hashlib.sha256(content).hexdigest(),
|
|
)
|
|
|
|
class Response(io.BytesIO):
|
|
def __init__(self, data: bytes) -> None:
|
|
super().__init__(data)
|
|
self.headers = {"Content-Length": str(len(data))}
|
|
|
|
def geturl(self) -> str:
|
|
return manifest.archive_url
|
|
|
|
def __enter__(self) -> Response:
|
|
return self
|
|
|
|
def __exit__(self, *_args: object) -> None:
|
|
self.close()
|
|
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
metadata = {
|
|
"from_version": "1.8.0",
|
|
"to_version": "1.7.4",
|
|
"rollback": True,
|
|
}
|
|
telemetry = MagicMock()
|
|
with (
|
|
patch("ptlinux.self_update.urlopen", return_value=Response(content)),
|
|
patch("ptlinux.self_update.current_telemetry", return_value=telemetry),
|
|
):
|
|
_download_archive(
|
|
manifest,
|
|
root / "success.tar.gz",
|
|
timeout_seconds=1,
|
|
event_family="rollback",
|
|
event_metadata=metadata,
|
|
)
|
|
self.assertEqual(
|
|
[call.args[0] for call in telemetry.send_event.call_args_list],
|
|
["rollback_download_started", "rollback_download_success"],
|
|
)
|
|
telemetry.reset_mock()
|
|
with (
|
|
patch("ptlinux.self_update.urlopen", return_value=Response(b"x" * len(content))),
|
|
patch("ptlinux.self_update.current_telemetry", return_value=telemetry),
|
|
self.assertRaisesRegex(RuntimeError, "SHA-256"),
|
|
):
|
|
_download_archive(
|
|
manifest,
|
|
root / "failure.tar.gz",
|
|
timeout_seconds=1,
|
|
event_family="rollback",
|
|
event_metadata=metadata,
|
|
)
|
|
self.assertEqual(
|
|
[call.args[0] for call in telemetry.send_event.call_args_list],
|
|
["rollback_download_started", "rollback_download_failed"],
|
|
)
|
|
|
|
def test_discovery_emits_release_identity_without_release_notes(self) -> None:
|
|
telemetry = MagicMock()
|
|
manifest = release_manifest("1.8.0")
|
|
with (
|
|
patch("ptlinux.self_update._fetch_authenticated_manifest", return_value=manifest),
|
|
patch("ptlinux.self_update.current_telemetry", return_value=telemetry),
|
|
):
|
|
check_for_launcher_update(
|
|
current_version="1.7.4",
|
|
manifest_url="https://updates.example.test/stable.json",
|
|
channel="stable",
|
|
)
|
|
|
|
call = telemetry.send_event.call_args
|
|
self.assertEqual(call.args[0], "update_release_discovered")
|
|
metadata = call.kwargs["metadata"]
|
|
self.assertEqual(metadata["current_version"], "1.7.4")
|
|
self.assertEqual(metadata["target_version"], "1.8.0")
|
|
self.assertEqual(metadata["current_channel"], "stable")
|
|
self.assertEqual(metadata["target_channel"], "stable")
|
|
self.assertEqual(metadata["selected_update_channel"], "stable")
|
|
self.assertEqual(metadata["release_tag"], "v1.8.0")
|
|
self.assertEqual(metadata["manifest_version"], 1)
|
|
self.assertEqual(metadata["manifest_hash"], "b" * 64)
|
|
self.assertFalse(metadata["rollback"])
|
|
self.assertNotIn("notes", metadata)
|
|
|
|
def test_update_apply_emits_started_and_success_with_shared_metadata(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
paths = app_paths(Path(temporary))
|
|
manifest = release_manifest("1.8.0")
|
|
check = LauncherUpdateCheck(
|
|
current_version="1.7.4",
|
|
latest_version="1.8.0",
|
|
available=True,
|
|
supported=True,
|
|
reason="Launcher 1.8.0 is available.",
|
|
manifest_url="https://updates.example.test/stable.json",
|
|
manifest=manifest,
|
|
selected_channel="stable",
|
|
)
|
|
release = paths.launcher_releases / "1.8.0"
|
|
telemetry = MagicMock()
|
|
with (
|
|
patch("ptlinux.self_update.check_for_launcher_update", return_value=check),
|
|
patch("ptlinux.self_update._precheck_release_install"),
|
|
patch("ptlinux.self_update._download_archive"),
|
|
patch("ptlinux.self_update._stage_and_install_release", return_value=release),
|
|
patch("ptlinux.self_update._activate_release", return_value=None),
|
|
patch("ptlinux.self_update.current_telemetry", return_value=telemetry),
|
|
):
|
|
result = install_launcher_update(paths, current_version="1.7.4")
|
|
|
|
self.assertTrue(result.changed)
|
|
events = [call.args[0] for call in telemetry.send_event.call_args_list]
|
|
self.assertEqual(events, ["update_apply_started", "update_apply_success"])
|
|
for call in telemetry.send_event.call_args_list:
|
|
metadata = call.kwargs["metadata"]
|
|
self.assertEqual(metadata["current_version"], "1.7.4")
|
|
self.assertEqual(metadata["target_version"], "1.8.0")
|
|
self.assertFalse(metadata["rollback"])
|
|
|
|
def test_local_rollback_emits_requested_and_apply_lifecycle(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
paths = app_paths(Path(temporary))
|
|
current = paths.launcher_releases / "1.8.0"
|
|
previous = paths.launcher_releases / "1.7.4"
|
|
for release in (current, previous):
|
|
make_release(release, release.name)
|
|
paths.launcher_current.symlink_to(Path("releases") / current.name)
|
|
paths.launcher_previous.symlink_to(Path("releases") / previous.name)
|
|
telemetry = MagicMock()
|
|
with patch("ptlinux.self_update.current_telemetry", return_value=telemetry):
|
|
rollback_launcher_update(paths, reason="support_recommendation")
|
|
|
|
events = [call.args[0] for call in telemetry.send_event.call_args_list]
|
|
self.assertEqual(
|
|
events,
|
|
[
|
|
"rollback_requested",
|
|
"rollback_apply_started",
|
|
"rollback_apply_success",
|
|
],
|
|
)
|
|
metadata = telemetry.send_event.call_args_list[0].kwargs["metadata"]
|
|
self.assertEqual(metadata["current_version"], "1.8.0")
|
|
self.assertEqual(metadata["target_version"], "1.7.4")
|
|
self.assertEqual(metadata["from_version"], "1.8.0")
|
|
self.assertEqual(metadata["to_version"], "1.7.4")
|
|
self.assertEqual(metadata["from_channel"], "stable")
|
|
self.assertEqual(metadata["to_channel"], "stable")
|
|
self.assertEqual(metadata["reason"], "support_recommendation")
|
|
self.assertTrue(metadata["rollback"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|