loop-uh
e28d92309f
Some checks failed
Windows project source guards / test (push) Failing after 27s
После обновления программа сообщает серверу, какой ступенью скачала (files — хранилище, parts — куски архива, setup — архив целиком) и сколько; при неудаче «lock» — код программы, державшей файл. Ядро — копия из Zapret 2 GUI. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
106 lines
5.3 KiB
Python
106 lines
5.3 KiB
Python
from __future__ import annotations
|
||
|
||
"""Лестница загрузки обновления: хранилище с заплатками → куски архива → архив целиком."""
|
||
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
from xray_fluent.updates import app_updater
|
||
from zapret_update import manifest as update_manifest
|
||
|
||
|
||
def _manifest(*, with_store: bool):
|
||
return update_manifest.parse({
|
||
"schema": 1, "product": "zapret-kvn", "channel": "stable", "version": "9.9.9", "created": 1, "key": "k",
|
||
"files": [{"path": "ZapretKVN.exe", "size": 3, "sha256": "a" * 64, "zsize": 2 if with_store else 0}],
|
||
})
|
||
|
||
|
||
class StoreStepTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self._tmp = tempfile.TemporaryDirectory()
|
||
self.addCleanup(self._tmp.cleanup)
|
||
self.app = Path(self._tmp.name) / "app"
|
||
self.app.mkdir()
|
||
|
||
def _downloader(self, manifest):
|
||
update = app_updater.AppUpdate("9.9.9", "v9.9.9", "https://git2.zapret.moe/x.zip", 100, "", "b" * 64, (), manifest)
|
||
return app_updater.UpdateDownloader(update, app_dir=self.app)
|
||
|
||
def test_release_without_a_store_skips_the_step(self) -> None:
|
||
with patch("zapret_update.fetch.fetch_plan") as fetch:
|
||
self.assertIsNone(self._downloader(_manifest(with_store=False))._try_store())
|
||
fetch.assert_not_called()
|
||
|
||
def test_store_builds_the_new_version_tree(self) -> None:
|
||
tree = self.app / "data" / "runtime" / "app_update" / "store-9.9.9" / "tree"
|
||
|
||
def build(planned, *, root, ready, stage_dir):
|
||
stage_dir.mkdir(parents=True)
|
||
(stage_dir / "ZapretKVN.exe").write_bytes(b"new")
|
||
return stage_dir
|
||
|
||
with patch("zapret_update.fetch.fetch_plan", return_value={}) as fetch, patch(
|
||
"zapret_update.stage.build_stage", side_effect=build
|
||
), patch("zapret_update.store.codecs", return_value=frozenset({"zstd"})):
|
||
self.assertEqual(self._downloader(_manifest(with_store=True))._try_store(), tree)
|
||
# Скачанное лежит в постоянной папке: после перезапуска загрузка продолжится.
|
||
self.assertEqual(fetch.call_args.kwargs["work_dir"], tree.parent)
|
||
self.assertIn("zapretkvn/generic/zapret-kvn-store/1/", fetch.call_args.kwargs["sources"][0].url_for("x.zst"))
|
||
|
||
def test_log_names_patch_kinds(self) -> None:
|
||
import hashlib
|
||
|
||
old = b"old"
|
||
(self.app / "ZapretKVN.exe").write_bytes(old)
|
||
listed = update_manifest.parse({
|
||
"schema": 1, "product": "zapret-kvn", "channel": "stable", "version": "9.9.9", "created": 1, "key": "k",
|
||
"files": [{"path": "ZapretKVN.exe", "size": 3, "sha256": "a" * 64, "zsize": 50}],
|
||
"patches": [{"from": hashlib.sha256(old).hexdigest(), "to": "a" * 64, "size": 7}],
|
||
})
|
||
|
||
def build(planned, *, root, ready, stage_dir):
|
||
stage_dir.mkdir(parents=True)
|
||
(stage_dir / "ZapretKVN.exe").write_bytes(b"new")
|
||
return stage_dir
|
||
|
||
with patch("zapret_update.fetch.fetch_plan", return_value={}), patch(
|
||
"zapret_update.stage.build_stage", side_effect=build
|
||
), patch("zapret_update.store.codecs", return_value=frozenset({"zstd"})), self.assertLogs(
|
||
app_updater._log, "INFO"
|
||
) as logs:
|
||
downloader = self._downloader(listed)
|
||
self.assertIsNotNone(downloader._try_store())
|
||
self.assertIn("скачано 7 из 100 байт (файлов 1, заплаток 1 (zstd 1), без изменений 0)", "\n".join(logs.output))
|
||
# Ступень и объём уходят серверу вместе с «дошла».
|
||
self.assertEqual((downloader.via, downloader.downloaded), ("files", 7))
|
||
|
||
def test_without_zstd_unpacker_nothing_is_downloaded(self) -> None:
|
||
# 0.8.26 собрана на Python 3.12 без zstandard: качала файл и падала на распаковке.
|
||
with patch("zapret_update.fetch.fetch_plan") as fetch, patch(
|
||
"zapret_update.store.codecs", return_value=frozenset({"bsdiff"})
|
||
):
|
||
self.assertIsNone(self._downloader(_manifest(with_store=True))._try_store())
|
||
fetch.assert_not_called()
|
||
|
||
def test_failed_store_hands_over_to_the_next_step(self) -> None:
|
||
from zapret_update.fetch import FetchError
|
||
|
||
with patch("zapret_update.fetch.fetch_plan", side_effect=FetchError("нет сети")) as fetch, patch(
|
||
"zapret_update.store.codecs", return_value=frozenset({"zstd"})
|
||
):
|
||
self.assertIsNone(self._downloader(_manifest(with_store=True))._try_store())
|
||
fetch.assert_called_once()
|
||
|
||
def test_store_comes_before_archive_pieces(self) -> None:
|
||
downloader = self._downloader(_manifest(with_store=True))
|
||
order = []
|
||
downloader._try_store = lambda: order.append("store") or None
|
||
downloader._try_delta = lambda work_dir: order.append("delta") or None
|
||
downloader._fetch = lambda archive: (order.append("zip"), (_ for _ in ()).throw(app_updater.DownloadFailed("стоп")))
|
||
with patch.object(app_updater, "_purge_stale_update_dirs"):
|
||
downloader.run()
|
||
self.assertEqual(order, ["store", "delta", "zip"])
|