From ef673a76ded5cec22339c1bc81557a88ccca3783 Mon Sep 17 00:00:00 2001 From: pleb Date: Thu, 9 Jul 2026 19:08:08 -0700 Subject: [PATCH] Enhance script for handling updates --- docs/review-07.09.2026.md | 44 +++ src/plugin_helper/cli.py | 60 ++++ src/plugin_helper/models.py | 43 ++- src/plugin_helper/update_runner.py | 236 +++++++++++++++ tests/test_plugin_helper.py | 441 ++++++++++++++++++++++++++++- 5 files changed, 822 insertions(+), 2 deletions(-) create mode 100644 docs/review-07.09.2026.md create mode 100644 src/plugin_helper/update_runner.py diff --git a/docs/review-07.09.2026.md b/docs/review-07.09.2026.md new file mode 100644 index 0000000..a1f73fc --- /dev/null +++ b/docs/review-07.09.2026.md @@ -0,0 +1,44 @@ +**Findings** + +- High: install state is still keyed by instance name, not a concrete installation. [state.py](/home/pleb/ops/beatsaber/plugin-helper/src/plugin_helper/state.py:10) writes everything under `instances/`, and the TUI gives every discovered root the same `runtime.state_root` in [cli.py](/home/pleb/ops/beatsaber/plugin-helper/src/plugin_helper/cli.py:279). That is fine only if Linux and Windows installs never share a state dir. For package-manager ambitions, introduce an `install_id` derived from profile/root/path, and split reusable asset cache from per-install state. + +- High: `apply_plan` is not transactional. It backs up and writes files one by one, then saves state at the end in [installer.py](/home/pleb/ops/beatsaber/plugin-helper/src/plugin_helper/installer.py:39). If extraction/copy fails mid-plan, the game tree can be half-mutated without matching install state. A future package manager wants transactions/generations: stage, apply, record transaction, and have a first-class rollback command. + +- Medium-high: installed state records files but not the installed package identity. [installer.py](/home/pleb/ops/beatsaber/plugin-helper/src/plugin_helper/installer.py:63) stores `path`, `sha256`, and `size`, but not repo/tag/asset/source URL/source hash/reason. The report then uses the current lockfile to describe installed versions, which can drift after a lock update. Record the lock identity into install state at apply time. + +- Medium-high: dependency handling is intentionally thin. `Dependency.constraint` exists in [models.py](/home/pleb/ops/beatsaber/plugin-helper/src/plugin_helper/models.py:10), but [planner.py](/home/pleb/ops/beatsaber/plugin-helper/src/plugin_helper/planner.py:68) only expands required dependency IDs and does not enforce constraints, conflicts, `loadAfter`, `loadBefore`, BeatMods version-id closures, or “provides” relationships. That is the next big unlock for package-manager behavior. + +- Medium: the agent skills have a few stale or conflicting operational rules. The manager skill still tells agents to inspect/edit `registry/plugins.toml` in [.agents/skills/beatsaber-plugin-manager/SKILL.md](/home/pleb/ops/beatsaber/plugin-helper/.agents/skills/beatsaber-plugin-manager/SKILL.md:54), while the project has moved to `registry/plugins/*.toml`. The builder skill says PR checkouts go under `/build` in [.agents/skills/beatsaber-plugin-builder/SKILL.md](/home/pleb/ops/beatsaber/plugin-helper/.agents/skills/beatsaber-plugin-builder/SKILL.md:24), while `AGENTS.md` says GitHub plugin source checkouts should live under `~/src//`. That will make agents inconsistent. + +**Skills** + +I would not collapse the three skills into one. They are conceptually different: + +- `beatsaber-plugin-manager`: mutating install/update/bootstrap workflow. +- `beatsaber-plugin-update-auditor`: read-mostly audit/report workflow. +- `beatsaber-plugin-builder`: source build workflow. + +But I would consolidate shared policy into references, then have each skill import that mental model: + +- `references/repo-workflow.md`: repo root, `.venv`, `PYTHONPATH=src`, dirty worktree rules, validation commands. +- `references/state-and-profiles.md`: `.state`, profiles, install identity, source checkout location. +- `references/artifact-policy.md`: GitHub first, BeatMods metadata/fallback, private sources, checksum policy. +- `references/live-validation.md`: smoketest and process cleanup. + +That removes duplicated drift while preserving good trigger boundaries. + +**Design Direction** + +Yes, the script design is useful. The best parts are exactly the right bones for a real package manager: registry, locks, dry-run plans, hash checks, managed install state, bootstrap as its own phase, update audit, and known-good sets. + +The next shape I’d aim for is: + +- `packages/registry`: identity, source aliases, install strategy, metadata extraction rules. +- `versions/locks`: selected package versions/artifacts/evidence for each Beat Saber version. +- `cache/downloads`: content-addressed artifacts reusable across installs. +- `installs/`: installed packages, transactions, bootstrap, known-good generations, backups. +- commands like `resolve`, `fetch`, `plan`, `apply`, `rollback`, `audit`, `verify`. + +Beat Saber plugins are heterogeneous in packaging, but homogeneous enough in runtime shape (`Plugins`, `Libs`, `IPA/Pending`, root BSIPA files) that this can become a nice small package manager. The trick is to normalize artifacts into a planned file tree before touching the game. + +Validation: `compileall` passed, and `unittest discover` passed: 75 tests, 1 skipped. \ No newline at end of file diff --git a/src/plugin_helper/cli.py b/src/plugin_helper/cli.py index 07018f8..5a31f53 100644 --- a/src/plugin_helper/cli.py +++ b/src/plugin_helper/cli.py @@ -20,6 +20,7 @@ from .planner import create_plan from .reports import installed_plugins_report, print_installed_plugins from .scanner import scan_instance from .state import load_installed_state +from .update_runner import run_update from .updates import check_updates from .userdata import restore_windows_data_repo, sync_windows_data_repo @@ -155,6 +156,21 @@ def build_parser() -> argparse.ArgumentParser: updates.add_argument("--include-prerelease", action="store_true", help="Include prerelease GitHub releases") updates.add_argument("--json", action="store_true", help="Print full JSON update output") + update = subcommands.add_parser( + "update", + help="Download selected update candidates, update the lock, and create an install plan", + parents=[_common_parent()], + ) + update.add_argument("--instance", required=True) + update.add_argument("--registry", default="registry/plugins") + update.add_argument("--lockfile") + update.add_argument("--plugin", action="append", required=True, help="Update this locked plugin id; repeatable") + update.add_argument("--include-prerelease", action="store_true", help="Include prerelease GitHub releases") + update.add_argument("--json", action="store_true", help="Print full JSON update output") + update_mode = update.add_mutually_exclusive_group() + update_mode.add_argument("--apply", action="store_true", help="Apply the generated update plan") + update_mode.add_argument("--dry-run", action="store_true", help="Show selected updates without downloading or writing") + plan = subcommands.add_parser( "plan", help="Create a dry-run install plan from the catalog and version lock", @@ -419,6 +435,50 @@ def run(argv: list[str] | None = None) -> int: print_updates(result) return 2 if result["summary"]["errors"] else 0 + if args.command == "update": + instance = get_instance(inst_roots, args.instance) + root = repo_root() + registry_path = (root / args.registry).resolve() if not Path(args.registry).is_absolute() else Path(args.registry) + lock_path = Path(args.lockfile) if args.lockfile else root / "locks" / f"{args.instance}.lock.toml" + if not lock_path.is_absolute(): + lock_path = (root / lock_path).resolve() + result = run_update( + instance=args.instance, + instance_path=instance.path, + registry=load_registry(registry_path), + lockfile=load_lockfile(lock_path), + lock_path=lock_path, + state_root=st_root, + repo_root=root, + selected=set(args.plugin), + fetch_releases=fetch_releases, + fetch_beatmods=fetch_verified_mods, + include_prerelease=args.include_prerelease, + dry_run=args.dry_run, + apply=args.apply, + ) + if args.json: + _json(result) + else: + action = "Would update" if result["dryRun"] else "Updated" + print(f"{action}: {len(result['updated'])}") + for item in result["updated"]: + print(f" {item['plugin']}: {item.get('fromTag')} -> {item.get('toTag')} ({item.get('asset')})") + if result["refused"]: + print("Refused:") + for item in result["refused"]: + notes = ", ".join(item.get("reviewReasons") or item.get("messages") or ()) + suffix = f": {notes}" if notes else "" + print(f" {item['plugin']}: {item['status']}{suffix}") + if result.get("planPath"): + print(f"Plan: {result['planPath']}") + if not result["dryRun"] and result["updated"]: + print(f"Lockfile: {result['lockfile']}") + if args.apply: + print(f"Applied: {len(result['applied'])}") + print(f"State: {result.get('statePath')}") + return 0 if not result["refused"] else 2 + if args.command == "bootstrap": instance = get_instance(inst_roots, args.instance) root = repo_root() diff --git a/src/plugin_helper/models.py b/src/plugin_helper/models.py index 919f881..78d5ff1 100644 --- a/src/plugin_helper/models.py +++ b/src/plugin_helper/models.py @@ -1,7 +1,7 @@ from __future__ import annotations import tomllib -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from pathlib import Path from typing import Any @@ -53,6 +53,47 @@ class Lockfile: plugins: tuple[LockedPlugin, ...] +def _quote_toml_string(value: str) -> str: + return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' + + +def lockfile_to_toml(lockfile: Lockfile) -> str: + lines = [ + f"beat_saber_version = {_quote_toml_string(lockfile.beat_saber_version)}", + f"instance = {_quote_toml_string(lockfile.instance)}", + "", + ] + for plugin in lockfile.plugins: + lines.append("[[plugins]]") + lines.append(f"id = {_quote_toml_string(plugin.id)}") + if plugin.repo is not None: + lines.append(f"repo = {_quote_toml_string(plugin.repo)}") + if plugin.tag is not None: + lines.append(f"tag = {_quote_toml_string(plugin.tag)}") + if plugin.asset is not None: + lines.append(f"asset = {_quote_toml_string(plugin.asset)}") + if plugin.sha256 is not None: + lines.append(f"sha256 = {_quote_toml_string(plugin.sha256)}") + if plugin.install_strategy is not None: + lines.append(f"install_strategy = {_quote_toml_string(plugin.install_strategy)}") + if plugin.reason is not None: + lines.append(f"reason = {_quote_toml_string(plugin.reason)}") + lines.append("") + return "\n".join(lines) + + +def write_lockfile(path: Path, lockfile: Lockfile) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(lockfile_to_toml(lockfile), encoding="utf-8") + + +def replace_locked_plugins(lockfile: Lockfile, replacements: dict[str, LockedPlugin]) -> Lockfile: + return replace( + lockfile, + plugins=tuple(replacements.get(plugin.id, plugin) for plugin in lockfile.plugins), + ) + + def _load_toml(path: Path) -> dict[str, Any]: with path.open("rb") as handle: return tomllib.load(handle) diff --git a/src/plugin_helper/update_runner.py b/src/plugin_helper/update_runner.py new file mode 100644 index 0000000..526c39f --- /dev/null +++ b/src/plugin_helper/update_runner.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import os +from collections.abc import Callable +from dataclasses import replace +from pathlib import Path +from typing import Any +from urllib.request import Request, urlopen + +from .fsutil import sha256_file +from .installer import apply_plan +from .models import LockedPlugin, Lockfile, Registry, replace_locked_plugins, write_lockfile +from .planner import create_plan +from .state import plugin_downloads_dir +from .updates import FetchBeatMods, FetchReleases, check_updates + + +DownloadAsset = Callable[[str, Path], dict[str, Any]] +ApplyPlan = Callable[[dict[str, Any], Path], dict[str, Any]] + + +def _download_asset(url: str, destination: Path) -> dict[str, Any]: + destination.parent.mkdir(parents=True, exist_ok=True) + headers = {"User-Agent": "plugin-helper"} + token = os.environ.get("GITHUB_TOKEN") + if token: + headers["Authorization"] = f"Bearer {token}" + request = Request(url, headers=headers) + with urlopen(request, timeout=60) as response: + destination.write_bytes(response.read()) + return {"url": url, "path": str(destination), "sha256": sha256_file(destination)} + + +def _reason_for_update(plugin: dict[str, Any], *, actual_sha256: str, beat_saber_version: str) -> str: + if plugin.get("latestBeatModsVersionId") is not None: + return ( + f"BeatMods verified {plugin['name']} " + f"{str(plugin.get('latestTag') or '').removeprefix('beatmods-')} " + f"for Beat Saber {beat_saber_version} as version id {plugin['latestBeatModsVersionId']}, " + f"zipHash {plugin.get('latestBeatModsZipHash')}. " + f"Downloaded from BeatMods CDN and verified SHA-256 {actual_sha256}." + ) + + digest = plugin.get("latestAssetSha256") + digest_note = ( + f"GitHub release digest matched {digest}." + if digest + else f"GitHub release did not publish a digest; computed SHA-256 {actual_sha256}." + ) + release = f" release {plugin.get('latestUrl')}" if plugin.get("latestUrl") else "" + return ( + f"Updated from GitHub {plugin.get('repo')} tag {plugin.get('latestTag')}{release}, " + f"asset {plugin.get('latestAsset')}. {digest_note}" + ) + + +def _updated_locked_plugin( + locked: LockedPlugin, + plugin: dict[str, Any], + *, + actual_sha256: str, + beat_saber_version: str, +) -> LockedPlugin: + return replace( + locked, + tag=plugin.get("latestTag"), + asset=plugin.get("latestAsset"), + sha256=actual_sha256, + reason=_reason_for_update( + plugin, + actual_sha256=actual_sha256, + beat_saber_version=beat_saber_version, + ), + ) + + +def run_update( + *, + instance: str, + instance_path: Path, + registry: Registry, + lockfile: Lockfile, + lock_path: Path, + state_root: Path, + repo_root: Path, + selected: set[str], + fetch_releases: FetchReleases, + fetch_beatmods: FetchBeatMods | None = None, + include_prerelease: bool = False, + dry_run: bool = False, + apply: bool = False, + download_asset: DownloadAsset = _download_asset, + apply_plan_func: ApplyPlan = apply_plan, +) -> dict[str, Any]: + if not selected: + raise ValueError("update requires at least one --plugin") + + report = check_updates( + registry=registry, + lockfile=lockfile, + fetch_releases=fetch_releases, + fetch_beatmods=fetch_beatmods, + selected=selected, + include_prerelease=include_prerelease, + ) + + locked_by_id = {plugin.id: plugin for plugin in lockfile.plugins} + reported_ids = {plugin["id"] for plugin in report["plugins"]} + updated: list[dict[str, Any]] = [] + refused: list[dict[str, Any]] = [] + downloads: list[dict[str, Any]] = [] + replacements: dict[str, LockedPlugin] = {} + + for missing_id in sorted(selected - reported_ids): + refused.append( + { + "plugin": missing_id, + "status": "error", + "messages": ["plugin is not locked for this instance"], + "reviewReasons": [], + } + ) + + for plugin in report["plugins"]: + plugin_id = plugin["id"] + if plugin["status"] != "update": + refused.append( + { + "plugin": plugin_id, + "status": plugin["status"], + "messages": plugin.get("messages", []), + "reviewReasons": plugin.get("reviewReasons", []), + } + ) + continue + if not plugin.get("latestAsset") or not plugin.get("latestAssetUrl") or not plugin.get("latestTag"): + refused.append( + { + "plugin": plugin_id, + "status": "error", + "messages": ["update candidate is missing tag, asset, or download URL"], + "reviewReasons": [], + } + ) + continue + if plugin_id not in locked_by_id: + refused.append( + { + "plugin": plugin_id, + "status": "error", + "messages": ["plugin is not locked for this instance"], + "reviewReasons": [], + } + ) + continue + + if dry_run: + updated.append( + { + "plugin": plugin_id, + "fromTag": plugin.get("currentTag"), + "toTag": plugin.get("latestTag"), + "asset": plugin.get("latestAsset"), + "url": plugin.get("latestAssetUrl"), + "dryRun": True, + } + ) + continue + + destination = plugin_downloads_dir(state_root, instance, plugin_id) / plugin["latestAsset"] + downloaded = download_asset(plugin["latestAssetUrl"], destination) + actual_sha = str(downloaded.get("sha256") or sha256_file(destination)) + expected_sha = plugin.get("latestAssetSha256") + if expected_sha and actual_sha != expected_sha: + destination.unlink(missing_ok=True) + raise ValueError(f"{plugin_id}: downloaded asset sha256 mismatch") + + downloads.append( + { + "plugin": plugin_id, + "asset": plugin["latestAsset"], + "path": str(destination), + "url": plugin["latestAssetUrl"], + "sha256": actual_sha, + } + ) + replacements[plugin_id] = _updated_locked_plugin( + locked_by_id[plugin_id], + plugin, + actual_sha256=actual_sha, + beat_saber_version=report["beatSaberVersion"], + ) + updated.append( + { + "plugin": plugin_id, + "fromTag": plugin.get("currentTag"), + "toTag": plugin.get("latestTag"), + "asset": plugin.get("latestAsset"), + "sha256": actual_sha, + } + ) + + result: dict[str, Any] = { + "instance": report["instance"], + "beatSaberVersion": report["beatSaberVersion"], + "lockfile": str(lock_path), + "updated": updated, + "refused": refused, + "downloads": downloads, + "planPath": None, + "applied": [], + "dryRun": dry_run, + } + if dry_run or not replacements: + return result + + updated_lockfile = replace_locked_plugins(lockfile, replacements) + plan, plan_path = create_plan( + instance=instance, + instance_path=instance_path, + beat_saber_version=updated_lockfile.beat_saber_version, + registry=registry, + lockfile=updated_lockfile, + state_root=state_root, + repo_root=repo_root, + selected=set(replacements), + ) + result["planPath"] = str(plan_path) + + if apply: + apply_result = apply_plan_func(plan, state_root) + result["applied"] = apply_result["applied"] + result["statePath"] = apply_result["statePath"] + + write_lockfile(lock_path, updated_lockfile) + return result diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index 9863093..316236d 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -22,12 +22,13 @@ from plugin_helper.config import is_windows, load_local_config, resolve_runtime_ from plugin_helper.fsutil import sha256_file from plugin_helper.installer import apply_plan, disable_plugin, uninstall_plugin from plugin_helper.instances import get_instance, list_instances -from plugin_helper.models import Dependency, Lockfile, LockedPlugin, Registry, RegistryPlugin, load_registry +from plugin_helper.models import Dependency, Lockfile, LockedPlugin, Registry, RegistryPlugin, load_lockfile, load_registry from plugin_helper.operations import enable_disabled_plugin from plugin_helper.planner import create_plan from plugin_helper.scanner import scan_bootstrap_files, scan_instance from plugin_helper.state import downloads_dir, load_installed_state, plugin_downloads_dir, save_bootstrap_state, save_installed_state from plugin_helper.tui import InstallationChoice, PluginHelperTui +from plugin_helper.update_runner import run_update from plugin_helper.updates import check_updates from plugin_helper.userdata import ( backup_userdata, @@ -40,6 +41,11 @@ from plugin_helper.userdata import ( class PluginHelperTests(unittest.TestCase): + def _write_plugin_zip(self, path: Path, *, content: bytes = b"updated dll") -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with ZipFile(path, "w") as archive: + archive.writestr("Plugins/Example.dll", content) + def test_load_registry_reads_plugin_directory(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) @@ -1826,6 +1832,439 @@ instance = "1.40.8" "compatibility failure history", ]) + def test_update_command_prepares_github_update_and_writes_lock(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + state = work / "state" + instance = work / "instance" + lock_path = work / "locks" / "1.40.8.lock.toml" + source_zip = work / "source" / "Example-v2.zip" + instance.mkdir() + self._write_plugin_zip(source_zip) + source_sha = sha256_file(source_zip) + registry = Registry( + { + "example": RegistryPlugin( + id="example", + name="Example", + repo="owner/example", + asset_patterns=("*.zip",), + install_strategy="bsipa-zip", + ) + } + ) + lockfile = Lockfile( + beat_saber_version="1.40.8", + instance="1.40.8", + plugins=( + LockedPlugin( + id="example", + repo="owner/example", + tag="v1.0.0", + asset="Example-v1.zip", + sha256="old", + install_strategy="bsipa-zip", + reason="old reason", + ), + ), + ) + + def download(_url: str, destination: Path) -> dict[str, object]: + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(source_zip.read_bytes()) + return {"path": str(destination), "sha256": sha256_file(destination)} + + result = run_update( + instance="1.40.8", + instance_path=instance, + registry=registry, + lockfile=lockfile, + lock_path=lock_path, + state_root=state, + repo_root=work, + selected={"example"}, + fetch_releases=lambda repo: [ + { + "tag_name": "v2.0.0", + "html_url": "https://github.com/owner/example/releases/tag/v2.0.0", + "published_at": "2026-06-12T00:00:00Z", + "assets": [ + { + "name": "Example-v2.zip", + "browser_download_url": "https://example.invalid/Example-v2.zip", + "digest": f"sha256:{source_sha}", + } + ], + } + ], + download_asset=download, + ) + + self.assertEqual(result["updated"][0]["plugin"], "example") + self.assertTrue(Path(result["planPath"]).is_file()) + updated_lock = load_lockfile(lock_path) + self.assertEqual(updated_lock.plugins[0].tag, "v2.0.0") + self.assertEqual(updated_lock.plugins[0].asset, "Example-v2.zip") + self.assertEqual(updated_lock.plugins[0].sha256, source_sha) + self.assertIn("GitHub release digest matched", updated_lock.plugins[0].reason or "") + + def test_update_command_prepares_beatmods_update(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + state = work / "state" + instance = work / "instance" + lock_path = work / "locks" / "1.44.1.lock.toml" + source_zip = work / "source" / "SongCore-3.17.0.zip" + instance.mkdir() + self._write_plugin_zip(source_zip) + registry = Registry( + { + "songcore": RegistryPlugin( + id="songcore", + name="SongCore", + repo="Kylemc1413/SongCore", + asset_patterns=("SongCore-*.zip",), + install_strategy="bsipa-zip", + ) + } + ) + lockfile = Lockfile( + beat_saber_version="1.44.1", + instance="1.44.1", + plugins=( + LockedPlugin( + id="songcore", + repo="Kylemc1413/SongCore", + tag="beatmods-3.16.0", + asset="SongCore-3.16.0.zip", + sha256="old", + install_strategy="bsipa-zip", + ), + ), + ) + + def download(url: str, destination: Path) -> dict[str, object]: + self.assertEqual(url, "https://beatmods.com/cdn/mod/abc.zip") + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(source_zip.read_bytes()) + return {"path": str(destination), "sha256": sha256_file(destination)} + + run_update( + instance="1.44.1", + instance_path=instance, + registry=registry, + lockfile=lockfile, + lock_path=lock_path, + state_root=state, + repo_root=work, + selected={"songcore"}, + fetch_releases=lambda repo: [], + fetch_beatmods=lambda game_version: [ + { + "mod": {"id": 1, "name": "SongCore", "gitUrl": "https://github.com/Kylemc1413/SongCore"}, + "latest": {"id": 2600, "modVersion": "3.17.0", "zipHash": "abc"}, + } + ], + download_asset=download, + ) + + updated_lock = load_lockfile(lock_path) + self.assertEqual(updated_lock.plugins[0].tag, "beatmods-3.17.0") + self.assertEqual(updated_lock.plugins[0].asset, "SongCore-3.17.0.zip") + self.assertIn("version id 2600", updated_lock.plugins[0].reason or "") + self.assertIn("zipHash abc", updated_lock.plugins[0].reason or "") + + def test_update_command_dry_run_does_not_download_or_write(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + lock_path = work / "locks" / "1.40.8.lock.toml" + registry = Registry( + { + "example": RegistryPlugin( + id="example", + name="Example", + repo="owner/example", + asset_patterns=("*.zip",), + install_strategy="bsipa-zip", + ) + } + ) + lockfile = Lockfile( + beat_saber_version="1.40.8", + instance="1.40.8", + plugins=(LockedPlugin(id="example", repo="owner/example", tag="v1", asset="old.zip", sha256="old"),), + ) + + result = run_update( + instance="1.40.8", + instance_path=work / "instance", + registry=registry, + lockfile=lockfile, + lock_path=lock_path, + state_root=work / "state", + repo_root=work, + selected={"example"}, + fetch_releases=lambda repo: [ + { + "tag_name": "v2", + "published_at": "2026-06-12T00:00:00Z", + "assets": [{"name": "new.zip", "browser_download_url": "https://example.invalid/new.zip"}], + } + ], + dry_run=True, + download_asset=lambda _url, _destination: self.fail("dry-run should not download"), + ) + + self.assertTrue(result["dryRun"]) + self.assertEqual(result["updated"][0]["toTag"], "v2") + self.assertFalse(lock_path.exists()) + self.assertIsNone(result["planPath"]) + + def test_update_command_refuses_review_plugin(self) -> None: + registry = Registry( + { + "jdfixer": RegistryPlugin( + id="jdfixer", + name="JDFixer", + repo="zeph-yr/JDFixer", + asset_patterns=("JDFixer.dll",), + install_strategy="dll-to-plugins", + ) + } + ) + lockfile = Lockfile( + beat_saber_version="1.44.1", + instance="1.44.1", + plugins=( + LockedPlugin( + id="jdfixer", + repo="zeph-yr/JDFixer", + tag="pr-26-3fce6ce", + asset="JDFixer.dll", + sha256="hash", + reason="Local build from GitHub PR; failed compatibility trial.", + ), + ), + ) + + result = run_update( + instance="1.44.1", + instance_path=Path("/tmp/unused"), + registry=registry, + lockfile=lockfile, + lock_path=Path("/tmp/unused.lock.toml"), + state_root=Path("/tmp/state"), + repo_root=Path("/tmp/repo"), + selected={"jdfixer"}, + fetch_releases=lambda repo: self.fail("review plugins should not query GitHub"), + download_asset=lambda _url, _destination: self.fail("review plugins should not download"), + ) + + self.assertEqual(result["updated"], []) + self.assertEqual(result["refused"][0]["plugin"], "jdfixer") + self.assertEqual(result["refused"][0]["status"], "review") + + def test_update_command_apply_failure_leaves_lock_unchanged(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + state = work / "state" + instance = work / "instance" + lock_path = work / "locks" / "1.40.8.lock.toml" + source_zip = work / "source" / "Example-v2.zip" + instance.mkdir() + self._write_plugin_zip(source_zip) + original_lock_text = """ +beat_saber_version = "1.40.8" +instance = "1.40.8" + +[[plugins]] +id = "example" +repo = "owner/example" +tag = "v1" +asset = "old.zip" +sha256 = "old" +install_strategy = "bsipa-zip" +""".lstrip() + lock_path.parent.mkdir() + lock_path.write_text(original_lock_text, encoding="utf-8") + registry = Registry( + { + "example": RegistryPlugin( + id="example", + name="Example", + repo="owner/example", + asset_patterns=("*.zip",), + install_strategy="bsipa-zip", + ) + } + ) + + def download(_url: str, destination: Path) -> dict[str, object]: + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(source_zip.read_bytes()) + return {"path": str(destination), "sha256": sha256_file(destination)} + + with self.assertRaises(RuntimeError): + run_update( + instance="1.40.8", + instance_path=instance, + registry=registry, + lockfile=load_lockfile(lock_path), + lock_path=lock_path, + state_root=state, + repo_root=work, + selected={"example"}, + fetch_releases=lambda repo: [ + { + "tag_name": "v2", + "published_at": "2026-06-12T00:00:00Z", + "assets": [{"name": "Example-v2.zip", "browser_download_url": "https://example.invalid/new.zip"}], + } + ], + download_asset=download, + apply=True, + apply_plan_func=lambda _plan, _state: (_ for _ in ()).throw(RuntimeError("boom")), + ) + + self.assertEqual(lock_path.read_text(encoding="utf-8"), original_lock_text) + + def test_update_command_updates_multiple_plugins_preserving_lock_order(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + state = work / "state" + instance = work / "instance" + lock_path = work / "locks" / "1.40.8.lock.toml" + instance.mkdir() + zip_a = work / "source" / "Alpha-v2.zip" + zip_b = work / "source" / "Beta-v2.zip" + self._write_plugin_zip(zip_a, content=b"alpha") + self._write_plugin_zip(zip_b, content=b"beta") + registry = Registry( + { + "alpha": RegistryPlugin(id="alpha", name="Alpha", repo="owner/alpha", asset_patterns=("*.zip",), install_strategy="bsipa-zip"), + "beta": RegistryPlugin(id="beta", name="Beta", repo="owner/beta", asset_patterns=("*.zip",), install_strategy="bsipa-zip"), + } + ) + lockfile = Lockfile( + beat_saber_version="1.40.8", + instance="1.40.8", + plugins=( + LockedPlugin(id="alpha", repo="owner/alpha", tag="v1", asset="Alpha-v1.zip", sha256="old-a", install_strategy="bsipa-zip"), + LockedPlugin(id="beta", repo="owner/beta", tag="v1", asset="Beta-v1.zip", sha256="old-b", install_strategy="bsipa-zip"), + ), + ) + + def releases(repo: str) -> list[dict[str, object]]: + name = "Alpha-v2.zip" if repo == "owner/alpha" else "Beta-v2.zip" + return [{"tag_name": "v2", "published_at": "2026-06-12T00:00:00Z", "assets": [{"name": name, "browser_download_url": f"https://example.invalid/{name}"}]}] + + def download(url: str, destination: Path) -> dict[str, object]: + source = zip_a if "Alpha" in url else zip_b + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(source.read_bytes()) + return {"path": str(destination), "sha256": sha256_file(destination)} + + run_update( + instance="1.40.8", + instance_path=instance, + registry=registry, + lockfile=lockfile, + lock_path=lock_path, + state_root=state, + repo_root=work, + selected={"alpha", "beta"}, + fetch_releases=releases, + download_asset=download, + ) + + updated_lock = load_lockfile(lock_path) + self.assertEqual([plugin.id for plugin in updated_lock.plugins], ["alpha", "beta"]) + self.assertEqual([plugin.asset for plugin in updated_lock.plugins], ["Alpha-v2.zip", "Beta-v2.zip"]) + + def test_update_cli_requires_plugin(self) -> None: + with patch("sys.stderr", new_callable=StringIO): + with self.assertRaises(SystemExit): + run(["update", "--instance", "1.40.8"]) + + def test_update_cli_passes_plugins_and_prints_json(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + instance_root = work / "instances" + instance = instance_root / "1.40.8" + state = work / "state" + locks = work / "locks" + registry = work / "registry" + instance.mkdir(parents=True) + (instance / "Beat Saber_Data").mkdir() + locks.mkdir() + registry.mkdir() + (locks / "1.40.8.lock.toml").write_text( + """ +beat_saber_version = "1.40.8" +instance = "1.40.8" + +[[plugins]] +id = "alpha" +repo = "owner/alpha" +tag = "v1" +asset = "Alpha.zip" +sha256 = "old" +""".lstrip(), + encoding="utf-8", + ) + (registry / "plugins.toml").write_text( + """ +[[plugins]] +id = "alpha" +name = "Alpha" +repo = "owner/alpha" +asset_patterns = ["*.zip"] +install_strategy = "bsipa-zip" +""".lstrip(), + encoding="utf-8", + ) + captured: dict[str, object] = {} + + def fake_update(**kwargs: object) -> dict[str, object]: + captured.update(kwargs) + return { + "instance": "1.40.8", + "beatSaberVersion": "1.40.8", + "lockfile": str(locks / "1.40.8.lock.toml"), + "updated": [{"plugin": "alpha"}], + "refused": [], + "downloads": [], + "planPath": str(state / "plan.json"), + "applied": [], + "dryRun": False, + } + + with patch("plugin_helper.cli.repo_root", return_value=work): + with patch("plugin_helper.cli.run_update", side_effect=fake_update): + with patch("sys.stdout", new_callable=StringIO) as stdout: + status = run( + [ + "--instances-root", + str(instance_root), + "--state-dir", + str(state), + "update", + "--instance", + "1.40.8", + "--plugin", + "alpha", + "--plugin", + "beta", + "--json", + ] + ) + + self.assertEqual(status, 0) + self.assertEqual(captured["selected"], {"alpha", "beta"}) + data = json.loads(stdout.getvalue()) + self.assertEqual(data["lockfile"], str(locks / "1.40.8.lock.toml")) + self.assertEqual(data["updated"][0]["plugin"], "alpha") + def _make_tui_fixture( root: Path,