Enhance script for handling updates

This commit is contained in:
pleb
2026-07-09 19:08:08 -07:00
parent 1dbe80a36f
commit ef673a76de
5 changed files with 822 additions and 2 deletions
+44
View File
@@ -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/<instance>`, 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 `<state_dir>/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/<owner>/<repo>`. 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 Id 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/<install_id>`: 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.
+60
View File
@@ -20,6 +20,7 @@ from .planner import create_plan
from .reports import installed_plugins_report, print_installed_plugins from .reports import installed_plugins_report, print_installed_plugins
from .scanner import scan_instance from .scanner import scan_instance
from .state import load_installed_state from .state import load_installed_state
from .update_runner import run_update
from .updates import check_updates from .updates import check_updates
from .userdata import restore_windows_data_repo, sync_windows_data_repo 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("--include-prerelease", action="store_true", help="Include prerelease GitHub releases")
updates.add_argument("--json", action="store_true", help="Print full JSON update output") 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 = subcommands.add_parser(
"plan", "plan",
help="Create a dry-run install plan from the catalog and version lock", 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) print_updates(result)
return 2 if result["summary"]["errors"] else 0 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": if args.command == "bootstrap":
instance = get_instance(inst_roots, args.instance) instance = get_instance(inst_roots, args.instance)
root = repo_root() root = repo_root()
+42 -1
View File
@@ -1,7 +1,7 @@
from __future__ import annotations from __future__ import annotations
import tomllib import tomllib
from dataclasses import dataclass, field from dataclasses import dataclass, field, replace
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -53,6 +53,47 @@ class Lockfile:
plugins: tuple[LockedPlugin, ...] 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]: def _load_toml(path: Path) -> dict[str, Any]:
with path.open("rb") as handle: with path.open("rb") as handle:
return tomllib.load(handle) return tomllib.load(handle)
+236
View File
@@ -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
+440 -1
View File
@@ -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.fsutil import sha256_file
from plugin_helper.installer import apply_plan, disable_plugin, uninstall_plugin from plugin_helper.installer import apply_plan, disable_plugin, uninstall_plugin
from plugin_helper.instances import get_instance, list_instances 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.operations import enable_disabled_plugin
from plugin_helper.planner import create_plan from plugin_helper.planner import create_plan
from plugin_helper.scanner import scan_bootstrap_files, scan_instance 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.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.tui import InstallationChoice, PluginHelperTui
from plugin_helper.update_runner import run_update
from plugin_helper.updates import check_updates from plugin_helper.updates import check_updates
from plugin_helper.userdata import ( from plugin_helper.userdata import (
backup_userdata, backup_userdata,
@@ -40,6 +41,11 @@ from plugin_helper.userdata import (
class PluginHelperTests(unittest.TestCase): 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: def test_load_registry_reads_plugin_directory(self) -> None:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) root = Path(tmp)
@@ -1826,6 +1832,439 @@ instance = "1.40.8"
"compatibility failure history", "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( def _make_tui_fixture(
root: Path, root: Path,