feat: non-blocking updates — download in app, apply at launch
Startup no longer blocks on a download. New split: - services/installer.py: download_and_stage() (runs while the app is open → staging\app.new, verified) and apply_staged() (launcher swaps it into app\ at next launch, when the exe isn't locked). - launcher.py: slimmed to seed + apply_staged + launch; no network at startup. - main_window: a new version lights a dot on the ⚙ 配置 button instead of a blocking banner; the old open-folder banner is removed. - settings_dialog: "检查并更新" downloads + stages, then "下次启动生效". Verified end-to-end over a local HTTP server (download→stage→apply); 101 tests pass (test_installer +11, test_launcher rewritten). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,130 @@
|
||||
"""Tests for services.installer — no GUI/network (download is mocked)."""
|
||||
import hashlib
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
from services import installer
|
||||
from services.update_service import UpdateInfo
|
||||
|
||||
|
||||
def _make_release_zip(dst_zip, version, exe_body, wrap=False):
|
||||
tmp = Path(tempfile.mkdtemp())
|
||||
try:
|
||||
base = tmp / ("CMBot-" + version) if wrap else tmp
|
||||
base.mkdir(parents=True, exist_ok=True)
|
||||
(base / "CMBot.exe").write_text(exe_body, encoding="ascii")
|
||||
(base / "version.txt").write_text(version, encoding="ascii")
|
||||
with zipfile.ZipFile(str(dst_zip), "w", zipfile.ZIP_DEFLATED) as z:
|
||||
for p in base.rglob("*"):
|
||||
z.write(str(p), str(p.relative_to(tmp)))
|
||||
finally:
|
||||
shutil.rmtree(str(tmp), ignore_errors=True)
|
||||
return hashlib.sha256(dst_zip.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
class _Base(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = Path(tempfile.mkdtemp())
|
||||
self.root = self.tmp / "install"
|
||||
self.app = self.root / "app"
|
||||
self.app.mkdir(parents=True)
|
||||
(self.app / "CMBot.exe").write_text("OLD-1.0.0", encoding="ascii")
|
||||
(self.app / "version.txt").write_text("1.0.0", encoding="ascii")
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(str(self.tmp), ignore_errors=True)
|
||||
|
||||
def _info(self, version, zip_path):
|
||||
return UpdateInfo(version=version, source="http://x/{}.zip".format(version),
|
||||
sha256=hashlib.sha256(zip_path.read_bytes()).hexdigest(),
|
||||
size=zip_path.stat().st_size)
|
||||
|
||||
|
||||
class TestDownloadAndStage(_Base):
|
||||
def _copy_download(self, src_zip):
|
||||
def fake(url, dest, user="", password="", timeout=120):
|
||||
shutil.copy(str(src_zip), str(dest))
|
||||
return fake
|
||||
|
||||
def test_stage_flat_zip(self):
|
||||
z = self.tmp / "n.zip"
|
||||
_make_release_zip(z, "1.1.0", "NEW")
|
||||
with patch("services.installer.download", side_effect=self._copy_download(z)):
|
||||
ver = installer.download_and_stage(self.root, self._info("1.1.0", z))
|
||||
self.assertEqual(ver, "1.1.0")
|
||||
self.assertEqual(installer.staged_version(self.root), "1.1.0")
|
||||
self.assertTrue((installer.staged_app(self.root) / "CMBot.exe").exists())
|
||||
# app untouched until apply
|
||||
self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0")
|
||||
|
||||
def test_stage_wrapped_zip(self):
|
||||
z = self.tmp / "n.zip"
|
||||
_make_release_zip(z, "1.1.0", "NEW", wrap=True)
|
||||
with patch("services.installer.download", side_effect=self._copy_download(z)):
|
||||
ver = installer.download_and_stage(self.root, self._info("1.1.0", z))
|
||||
self.assertEqual(ver, "1.1.0")
|
||||
self.assertEqual(installer.staged_version(self.root), "1.1.0")
|
||||
|
||||
def test_sha_mismatch_raises(self):
|
||||
z = self.tmp / "n.zip"
|
||||
_make_release_zip(z, "1.1.0", "NEW")
|
||||
info = self._info("1.1.0", z)
|
||||
info.sha256 = "deadbeef"
|
||||
with patch("services.installer.download", side_effect=self._copy_download(z)):
|
||||
with self.assertRaises(ValueError):
|
||||
installer.download_and_stage(self.root, info)
|
||||
self.assertEqual(installer.staged_version(self.root), "")
|
||||
|
||||
def test_version_mismatch_raises(self):
|
||||
z = self.tmp / "n.zip"
|
||||
_make_release_zip(z, "1.1.0", "NEW")
|
||||
info = self._info("9.9.9", z) # manifest claims a different version
|
||||
with patch("services.installer.download", side_effect=self._copy_download(z)):
|
||||
with self.assertRaises(ValueError):
|
||||
installer.download_and_stage(self.root, info)
|
||||
|
||||
|
||||
class TestApplyStaged(_Base):
|
||||
def _stage(self, version, body):
|
||||
s = installer.staged_app(self.root)
|
||||
s.mkdir(parents=True, exist_ok=True)
|
||||
(s / "CMBot.exe").write_text(body, encoding="ascii")
|
||||
(s / "version.txt").write_text(version, encoding="ascii")
|
||||
|
||||
def test_apply_swaps(self):
|
||||
self._stage("1.1.0", "NEW-1.1.0")
|
||||
applied = installer.apply_staged(self.root)
|
||||
self.assertEqual(applied, "1.1.0")
|
||||
self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.1.0")
|
||||
self.assertEqual((self.app / "CMBot.exe").read_text(encoding="ascii"), "NEW-1.1.0")
|
||||
self.assertEqual((installer.old_dir(self.root) / "version.txt").read_text(encoding="ascii"), "1.0.0")
|
||||
self.assertEqual(installer.staged_version(self.root), "")
|
||||
|
||||
def test_nothing_staged_returns_empty(self):
|
||||
self.assertEqual(installer.apply_staged(self.root), "")
|
||||
self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0")
|
||||
|
||||
def test_not_newer_staged_is_discarded(self):
|
||||
self._stage("0.9.0", "OLDER")
|
||||
self.assertEqual(installer.apply_staged(self.root), "")
|
||||
self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0")
|
||||
self.assertFalse(installer.staged_app(self.root).exists())
|
||||
|
||||
|
||||
class TestHelpers(_Base):
|
||||
def test_is_writable(self):
|
||||
self.assertTrue(installer.is_writable(self.root))
|
||||
|
||||
def test_install_root_for(self):
|
||||
self.assertEqual(installer.install_root_for(self.app), self.root.resolve())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+34
-101
@@ -1,153 +1,86 @@
|
||||
"""Tests for launcher — no GUI/network dependency (download is mocked)."""
|
||||
import hashlib
|
||||
"""Tests for the launcher — no GUI/network. The launcher only seeds config and
|
||||
applies a pre-staged update (downloading lives in services.installer/the app)."""
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
import launcher
|
||||
from services.update_service import UpdateInfo
|
||||
from services import installer
|
||||
|
||||
|
||||
def _make_release_zip(dst_zip, version, exe_body):
|
||||
"""Build a flat release zip (CMBot.exe + version.txt) and return its sha256."""
|
||||
tmp = Path(tempfile.mkdtemp())
|
||||
try:
|
||||
(tmp / "CMBot.exe").write_text(exe_body, encoding="ascii")
|
||||
(tmp / "version.txt").write_text(version, encoding="ascii")
|
||||
with zipfile.ZipFile(str(dst_zip), "w", zipfile.ZIP_DEFLATED) as z:
|
||||
z.write(str(tmp / "CMBot.exe"), "CMBot.exe")
|
||||
z.write(str(tmp / "version.txt"), "version.txt")
|
||||
finally:
|
||||
shutil.rmtree(str(tmp), ignore_errors=True)
|
||||
return hashlib.sha256(dst_zip.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
class _LauncherBase(unittest.TestCase):
|
||||
class _Base(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = Path(tempfile.mkdtemp())
|
||||
self.install = self.tmp / "install"
|
||||
self.app = self.install / "app"
|
||||
self.root = self.tmp / "install"
|
||||
self.app = self.root / "app"
|
||||
(self.app / "config").mkdir(parents=True)
|
||||
(self.app / "CMBot.exe").write_text("OLD-1.0.0", encoding="ascii")
|
||||
(self.app / "version.txt").write_text("1.0.0", encoding="ascii")
|
||||
# factory defaults shipped inside app\config
|
||||
(self.app / "config" / "app_config.json").write_text(
|
||||
json.dumps({"update_source": "http://x/manifest.json", "last_template": "正方形"}),
|
||||
encoding="utf-8")
|
||||
json.dumps({"update_source": "http://x"}), encoding="utf-8")
|
||||
(self.app / "config" / "templates.json").write_text("{}", encoding="utf-8")
|
||||
# data root -> a temp dir via the env override
|
||||
self.data = self.tmp / "data"
|
||||
self._prev_env = os.environ.get("CMBOT_DATA_DIR")
|
||||
self._prev = os.environ.get("CMBOT_DATA_DIR")
|
||||
os.environ["CMBOT_DATA_DIR"] = str(self.data)
|
||||
|
||||
def tearDown(self):
|
||||
if self._prev_env is None:
|
||||
if self._prev is None:
|
||||
os.environ.pop("CMBOT_DATA_DIR", None)
|
||||
else:
|
||||
os.environ["CMBOT_DATA_DIR"] = self._prev_env
|
||||
os.environ["CMBOT_DATA_DIR"] = self._prev
|
||||
shutil.rmtree(str(self.tmp), ignore_errors=True)
|
||||
|
||||
def _stage(self, version, body):
|
||||
s = installer.staged_app(self.root)
|
||||
s.mkdir(parents=True, exist_ok=True)
|
||||
(s / "CMBot.exe").write_text(body, encoding="ascii")
|
||||
(s / "version.txt").write_text(version, encoding="ascii")
|
||||
|
||||
class TestHelpers(_LauncherBase):
|
||||
def test_seed_defaults_copies_when_missing(self):
|
||||
|
||||
class TestSeed(_Base):
|
||||
def test_seed_copies_when_missing(self):
|
||||
launcher.seed_defaults(self.app, self.data)
|
||||
self.assertTrue((self.data / "config" / "app_config.json").exists())
|
||||
self.assertTrue((self.data / "config" / "templates.json").exists())
|
||||
|
||||
def test_seed_defaults_does_not_overwrite(self):
|
||||
def test_seed_does_not_overwrite(self):
|
||||
(self.data / "config").mkdir(parents=True)
|
||||
(self.data / "config" / "app_config.json").write_text(
|
||||
'{"update_source":"USER"}', encoding="utf-8")
|
||||
(self.data / "config" / "app_config.json").write_text('{"update_source":"USER"}', encoding="utf-8")
|
||||
launcher.seed_defaults(self.app, self.data)
|
||||
kept = json.loads((self.data / "config" / "app_config.json").read_text(encoding="utf-8"))
|
||||
self.assertEqual(kept["update_source"], "USER")
|
||||
|
||||
def test_is_writable(self):
|
||||
self.assertTrue(launcher.is_writable(self.install))
|
||||
|
||||
def test_find_package_root_flat(self):
|
||||
d = self.tmp / "flat"
|
||||
d.mkdir()
|
||||
(d / "CMBot.exe").write_text("x", encoding="ascii")
|
||||
self.assertEqual(launcher._find_package_root(d), d)
|
||||
|
||||
def test_find_package_root_wrapped(self):
|
||||
d = self.tmp / "wrap"
|
||||
(d / "CMBot-1.1.0").mkdir(parents=True)
|
||||
(d / "CMBot-1.1.0" / "CMBot.exe").write_text("x", encoding="ascii")
|
||||
self.assertEqual(launcher._find_package_root(d), d / "CMBot-1.1.0")
|
||||
|
||||
def test_find_package_root_missing(self):
|
||||
d = self.tmp / "empty"
|
||||
d.mkdir()
|
||||
self.assertIsNone(launcher._find_package_root(d))
|
||||
|
||||
def test_read_version_tolerates_bom(self):
|
||||
(self.app / "version.txt").write_text("1.0.0", encoding="utf-8-sig")
|
||||
self.assertEqual(launcher._read_version(self.app), "1.0.0")
|
||||
|
||||
|
||||
class TestRun(_LauncherBase):
|
||||
def _info(self, version, zip_path):
|
||||
sha = hashlib.sha256(zip_path.read_bytes()).hexdigest()
|
||||
return UpdateInfo(version=version, source="http://x/{}.zip".format(version),
|
||||
sha256=sha, size=zip_path.stat().st_size)
|
||||
|
||||
def test_update_applied_and_swapped(self):
|
||||
new_zip = self.tmp / "new.zip"
|
||||
_make_release_zip(new_zip, "1.1.0", "NEW-1.1.0")
|
||||
info = self._info("1.1.0", new_zip)
|
||||
|
||||
def fake_download(url, dest, user="", password="", timeout=120):
|
||||
shutil.copy(str(new_zip), str(dest))
|
||||
|
||||
with patch("launcher.check_for_update", return_value=info), \
|
||||
patch("launcher.download", side_effect=fake_download):
|
||||
rc = launcher.run(self.install, no_launch=True)
|
||||
|
||||
class TestRun(_Base):
|
||||
def test_applies_staged_update_then_launches(self):
|
||||
self._stage("1.1.0", "NEW-1.1.0")
|
||||
rc = launcher.run(self.root, no_launch=True)
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.1.0")
|
||||
self.assertEqual((self.app / "CMBot.exe").read_text(encoding="ascii"), "NEW-1.1.0")
|
||||
self.assertEqual(((self.install / "app.old") / "version.txt").read_text(encoding="ascii"), "1.0.0")
|
||||
self.assertFalse((self.install / "staging" / "1.1.0.zip").exists())
|
||||
self.assertEqual((installer.old_dir(self.root) / "version.txt").read_text(encoding="ascii"), "1.0.0")
|
||||
|
||||
def test_no_update_keeps_local(self):
|
||||
with patch("launcher.check_for_update", return_value=None):
|
||||
rc = launcher.run(self.install, no_launch=True)
|
||||
def test_no_staged_just_launches(self):
|
||||
rc = launcher.run(self.root, no_launch=True)
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0")
|
||||
self.assertFalse((self.install / "app.old").exists())
|
||||
|
||||
def test_sha256_mismatch_degrades(self):
|
||||
new_zip = self.tmp / "new.zip"
|
||||
_make_release_zip(new_zip, "1.1.0", "NEW")
|
||||
info = self._info("1.1.0", new_zip)
|
||||
info.sha256 = "deadbeef" # force mismatch
|
||||
|
||||
def fake_download(url, dest, user="", password="", timeout=120):
|
||||
shutil.copy(str(new_zip), str(dest))
|
||||
|
||||
with patch("launcher.check_for_update", return_value=info), \
|
||||
patch("launcher.download", side_effect=fake_download):
|
||||
rc = launcher.run(self.install, no_launch=True)
|
||||
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0")
|
||||
self.assertFalse((self.install / "app.old").exists())
|
||||
self.assertFalse(installer.old_dir(self.root).exists())
|
||||
|
||||
def test_seeds_config_into_data_root(self):
|
||||
with patch("launcher.check_for_update", return_value=None):
|
||||
launcher.run(self.install, no_launch=True)
|
||||
launcher.run(self.root, no_launch=True)
|
||||
self.assertTrue((self.data / "config" / "app_config.json").exists())
|
||||
|
||||
def test_missing_exe_returns_error(self):
|
||||
(self.app / "CMBot.exe").unlink()
|
||||
rc = launcher.run(self.root, no_launch=True)
|
||||
self.assertEqual(rc, 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user