From c86e51b85a1069e6a63417c987ed703d1da0ed29 Mon Sep 17 00:00:00 2001 From: pleb Date: Wed, 1 Jul 2026 22:55:47 -0700 Subject: [PATCH] Auto-bootstrap BSIPA when re-enabling plugins Run bootstrap automatically on enable paths when health is bad, skip bsipa in disable-all, and add tests for the shared bootstrap gate. Co-authored-by: Cursor --- README.md | 5 +- src/plugin_helper/bootstrap.py | 45 ++++++- src/plugin_helper/bsipa.py | 11 ++ src/plugin_helper/cli.py | 2 + src/plugin_helper/operations.py | 114 ++++++++++++++-- src/plugin_helper/planner.py | 9 +- src/plugin_helper/tui.py | 51 +++++--- tests/test_plugin_helper.py | 225 +++++++++++++++++++++++++++++++- 8 files changed, 425 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index ff0c0cd..1001cbf 100644 --- a/README.md +++ b/README.md @@ -127,8 +127,9 @@ from an interactive terminal. The menu reads `plugin-helper.local.toml` when present, shows each discovered Beat Saber install with its resolved state directory, and lets you toggle managed plugins with arrow keys and Space. In the plugin table, use `d` to -disable all currently enabled managed plugins and `e` to enable all currently -disabled managed plugins. +disable all currently enabled managed plugins (except `bsipa`) and `e` to enable all +currently disabled managed plugins. Re-enabling a plugin auto-bootstraps BSIPA when +needed. The individual subcommands are mostly for automation and debugging. If you use them, pass `--state-dir` directly only when you intentionally want to override diff --git a/src/plugin_helper/bootstrap.py b/src/plugin_helper/bootstrap.py index 94725a8..e832238 100644 --- a/src/plugin_helper/bootstrap.py +++ b/src/plugin_helper/bootstrap.py @@ -8,7 +8,7 @@ from pathlib import Path from typing import Any, Callable from urllib.request import Request, urlopen -from .bsipa import BSIPA_PLUGIN_ID, check_bsipa_health +from .bsipa import BSIPA_PLUGIN_ID, bootstrap_health_error, check_bsipa_health, planning_requires_bootstrap from .config import is_windows from .fsutil import sha256_file from .installer import apply_plan @@ -296,3 +296,46 @@ def run_bootstrap( if completed["returncode"] != 0: raise RuntimeError(f"IPA.exe -n failed with exit code {completed['returncode']}; state written to {state['statePath']}") return state + + +def ensure_healthy_bootstrap( + *, + instance: str, + instance_path: Path, + beat_saber_version: str, + registry: Registry, + lockfile: Lockfile, + state_root: Path, + repo_root: Path, + selected_ids: set[str], + proton: Path | None = None, + native: bool | None = None, + progress: Callable[[str], None] | None = None, + ipa_timeout_seconds: int = 120, +) -> None: + if not planning_requires_bootstrap(lockfile.plugins, selected_ids): + return + + health = check_bsipa_health(instance_path, state_root, instance) + if health["ok"]: + return + + tell = progress or (lambda _message: None) + tell("BSIPA bootstrap is unhealthy; running bootstrap") + run_bootstrap( + instance=instance, + instance_path=instance_path, + beat_saber_version=beat_saber_version, + registry=registry, + lockfile=lockfile, + state_root=state_root, + repo_root=repo_root, + proton=proton, + native=native, + progress=progress, + ipa_timeout_seconds=ipa_timeout_seconds, + ) + + health = check_bsipa_health(instance_path, state_root, instance) + if not health["ok"]: + raise ValueError(bootstrap_health_error(health)) diff --git a/src/plugin_helper/bsipa.py b/src/plugin_helper/bsipa.py index b32f99f..aa4865d 100644 --- a/src/plugin_helper/bsipa.py +++ b/src/plugin_helper/bsipa.py @@ -10,6 +10,17 @@ from .state import bootstrap_state_path, load_bootstrap_state BSIPA_PLUGIN_ID = "bsipa" +def planning_requires_bootstrap(lockfile_plugins: tuple[Any, ...] | list[Any], selected_ids: set[str]) -> bool: + has_locked_bsipa = any(plugin.id == BSIPA_PLUGIN_ID for plugin in lockfile_plugins) + planning_ordinary_plugins = any(plugin_id != BSIPA_PLUGIN_ID for plugin_id in selected_ids) + return has_locked_bsipa and planning_ordinary_plugins + + +def bootstrap_health_error(health: dict[str, Any]) -> str: + joined = "; ".join(health["messages"]) + return f"BSIPA bootstrap is not healthy; run bootstrap first: {joined}" + + def latest_log_path(instance_path: Path) -> Path: return instance_path / "Logs" / "_latest.log" diff --git a/src/plugin_helper/cli.py b/src/plugin_helper/cli.py index 69d9ce0..f0091b1 100644 --- a/src/plugin_helper/cli.py +++ b/src/plugin_helper/cli.py @@ -504,6 +504,7 @@ def run(argv: list[str] | None = None) -> int: if args.command == "enable": instance = get_instance(inst_roots, args.instance) + progress = (lambda message: print(f" {message}", flush=True)) result = enable_disabled_plugin( instance=args.instance, instance_path=instance.path, @@ -511,6 +512,7 @@ def run(argv: list[str] | None = None) -> int: plugin_id=args.plugin, registry=args.registry, lockfile=args.lockfile, + progress=progress, ) print(f"Enabled: {args.plugin}") print(f"Plan: {result['planPath']}") diff --git a/src/plugin_helper/operations.py b/src/plugin_helper/operations.py index 501f140..94c4283 100644 --- a/src/plugin_helper/operations.py +++ b/src/plugin_helper/operations.py @@ -1,15 +1,34 @@ from __future__ import annotations +from collections.abc import Callable from pathlib import Path from typing import Any +from .bootstrap import ensure_healthy_bootstrap from .config import repo_root from .installer import apply_plan -from .models import load_lockfile, load_registry +from .models import Lockfile, Registry, load_lockfile, load_registry from .planner import create_plan from .state import load_installed_state +def _resolve_paths( + *, + instance: str, + registry: str, + lockfile: str | None, + repo: Path | None, +) -> tuple[Path, Path, Path, Lockfile, Registry]: + root = repo or repo_root() + registry_path = (root / registry).resolve() if not Path(registry).is_absolute() else Path(registry) + lock_path = Path(lockfile) if lockfile else root / "locks" / f"{instance}.lock.toml" + if not lock_path.is_absolute(): + lock_path = (root / lock_path).resolve() + loaded_lockfile = load_lockfile(lock_path) + loaded_registry = load_registry(registry_path) + return root, registry_path, lock_path, loaded_lockfile, loaded_registry + + def enable_disabled_plugin( *, instance: str, @@ -19,25 +38,38 @@ def enable_disabled_plugin( registry: str = "registry/plugins.toml", lockfile: str | None = None, repo: Path | None = None, + progress: Callable[[str], None] | None = None, ) -> dict[str, Any]: installed_state = load_installed_state(state_root, instance) if plugin_id not in installed_state.get("disabledPlugins", {}): raise KeyError(f"plugin is not recorded as disabled: {plugin_id}") - root = repo or repo_root() - registry_path = (root / registry).resolve() if not Path(registry).is_absolute() else Path(registry) - lock_path = Path(lockfile) if lockfile else root / "locks" / f"{instance}.lock.toml" - if not lock_path.is_absolute(): - lock_path = (root / lock_path).resolve() - loaded_lockfile = load_lockfile(lock_path) + root, registry_path, _lock_path, loaded_lockfile, loaded_registry = _resolve_paths( + instance=instance, + registry=registry, + lockfile=lockfile, + repo=repo, + ) if not any(plugin.id == plugin_id for plugin in loaded_lockfile.plugins): raise KeyError(f"plugin is disabled but not locked for this instance: {plugin_id}") + ensure_healthy_bootstrap( + instance=instance, + instance_path=instance_path, + beat_saber_version=loaded_lockfile.beat_saber_version, + registry=loaded_registry, + lockfile=loaded_lockfile, + state_root=state_root, + repo_root=root, + selected_ids={plugin_id}, + progress=progress, + ) + plan, path = create_plan( instance=instance, instance_path=instance_path, beat_saber_version=loaded_lockfile.beat_saber_version, - registry=load_registry(registry_path), + registry=loaded_registry, lockfile=loaded_lockfile, state_root=state_root, repo_root=root, @@ -45,3 +77,69 @@ def enable_disabled_plugin( ) result = apply_plan(plan, state_root) return {"planPath": str(path), **result} + + +def enable_disabled_plugins( + *, + instance: str, + instance_path: Path, + state_root: Path, + plugin_ids: list[str], + registry: str = "registry/plugins.toml", + lockfile: str | None = None, + repo: Path | None = None, + progress: Callable[[str], None] | None = None, +) -> dict[str, Any]: + if not plugin_ids: + return {"enabled": [], "errors": []} + + root, _registry_path, _lock_path, loaded_lockfile, loaded_registry = _resolve_paths( + instance=instance, + registry=registry, + lockfile=lockfile, + repo=repo, + ) + installed_state = load_installed_state(state_root, instance) + disabled_plugins = installed_state.get("disabledPlugins", {}) + selected_ids = set(plugin_ids) + + ensure_healthy_bootstrap( + instance=instance, + instance_path=instance_path, + beat_saber_version=loaded_lockfile.beat_saber_version, + registry=loaded_registry, + lockfile=loaded_lockfile, + state_root=state_root, + repo_root=root, + selected_ids=selected_ids, + progress=progress, + ) + + enabled: list[dict[str, Any]] = [] + errors: list[dict[str, str]] = [] + for plugin_id in plugin_ids: + if plugin_id not in disabled_plugins: + errors.append({"plugin": plugin_id, "error": f"plugin is not recorded as disabled: {plugin_id}"}) + continue + if not any(plugin.id == plugin_id for plugin in loaded_lockfile.plugins): + errors.append( + {"plugin": plugin_id, "error": f"plugin is disabled but not locked for this instance: {plugin_id}"} + ) + continue + try: + plan, path = create_plan( + instance=instance, + instance_path=instance_path, + beat_saber_version=loaded_lockfile.beat_saber_version, + registry=loaded_registry, + lockfile=loaded_lockfile, + state_root=state_root, + repo_root=root, + selected={plugin_id}, + ) + result = apply_plan(plan, state_root) + enabled.append({"plugin": plugin_id, "planPath": str(path), "applied": len(result["applied"])}) + except Exception as exc: + errors.append({"plugin": plugin_id, "error": str(exc)}) + + return {"enabled": enabled, "errors": errors} diff --git a/src/plugin_helper/planner.py b/src/plugin_helper/planner.py index 04620d7..292c714 100644 --- a/src/plugin_helper/planner.py +++ b/src/plugin_helper/planner.py @@ -9,7 +9,7 @@ from zipfile import ZipFile from .fsutil import ensure_relative, sha256_bytes, sha256_file from .models import Lockfile, Registry, VALID_STRATEGIES -from .bsipa import BSIPA_PLUGIN_ID, check_bsipa_health +from .bsipa import bootstrap_health_error, check_bsipa_health, planning_requires_bootstrap from .state import downloads_dir, plans_dir, plugin_downloads_dir @@ -81,13 +81,10 @@ def create_plan( changes: list[dict[str, Any]] = [] warnings: list[str] = [] - has_locked_bsipa = any(plugin.id == BSIPA_PLUGIN_ID for plugin in lockfile.plugins) - planning_ordinary_plugins = any(plugin_id != BSIPA_PLUGIN_ID for plugin_id in selected_ids) - if require_bootstrap and has_locked_bsipa and planning_ordinary_plugins: + if require_bootstrap and planning_requires_bootstrap(lockfile.plugins, selected_ids): health = check_bsipa_health(instance_path, state_root, instance) if not health["ok"]: - joined = "; ".join(health["messages"]) - raise ValueError(f"BSIPA bootstrap is not healthy; run bootstrap first: {joined}") + raise ValueError(bootstrap_health_error(health)) for locked in lockfile.plugins: if locked.id not in selected_ids: diff --git a/src/plugin_helper/tui.py b/src/plugin_helper/tui.py index 0ceed4c..9e69a43 100644 --- a/src/plugin_helper/tui.py +++ b/src/plugin_helper/tui.py @@ -9,9 +9,10 @@ from textual.app import App, ComposeResult from textual.binding import Binding from textual.widgets import DataTable, Footer, Header, Static +from .bsipa import BSIPA_PLUGIN_ID from .installer import disable_plugin from .models import load_lockfile, load_registry -from .operations import enable_disabled_plugin +from .operations import enable_disabled_plugin, enable_disabled_plugins from .reports import installed_plugins_report from .state import load_installed_state @@ -124,6 +125,7 @@ class PluginHelperTui(App[int]): state_root=target.state_root, plugin_id=plugin_id, repo=self.repo_root, + progress=lambda message: self._set_status(f"Bootstrapping: {message}"), ) self._set_status(f"Enabled {plugin_id}; applied {len(result['applied'])} files.") else: @@ -138,7 +140,14 @@ class PluginHelperTui(App[int]): if self.mode != "plugins" or self.selected_installation is None: return target = self.selected_installation - enabled = [plugin for plugin in self.plugin_rows if plugin["status"] == "enabled"] + enabled = [ + plugin + for plugin in self.plugin_rows + if plugin["status"] == "enabled" and plugin["id"] != BSIPA_PLUGIN_ID + ] + skipped_bsipa = any( + plugin["status"] == "enabled" and plugin["id"] == BSIPA_PLUGIN_ID for plugin in self.plugin_rows + ) changed = 0 errors: list[str] = [] for plugin in enabled: @@ -157,7 +166,10 @@ class PluginHelperTui(App[int]): errors.append(f"{plugin_id}: {self._format_skipped(result['skipped'])}") except Exception as exc: errors.append(f"{plugin_id}: {exc}") - self._set_bulk_status("Disabled", changed, errors) + if skipped_bsipa and not errors: + self._set_status(f"Disabled {changed} plugins (bsipa kept enabled).") + else: + self._set_bulk_status("Disabled", changed, errors) self._show_plugins(preserve_status=True) def action_enable_all_plugins(self) -> None: @@ -165,21 +177,24 @@ class PluginHelperTui(App[int]): return target = self.selected_installation disabled = [plugin for plugin in self.plugin_rows if plugin["status"] == "disabled"] - changed = 0 - errors: list[str] = [] - for plugin in disabled: - plugin_id = plugin["id"] - try: - enable_disabled_plugin( - instance=target.instance_name, - instance_path=target.instance_path, - state_root=target.state_root, - plugin_id=plugin_id, - repo=self.repo_root, - ) - changed += 1 - except Exception as exc: - errors.append(f"{plugin_id}: {exc}") + if not disabled: + self._set_status("No disabled plugins to enable.") + return + plugin_ids = [plugin["id"] for plugin in disabled] + try: + result = enable_disabled_plugins( + instance=target.instance_name, + instance_path=target.instance_path, + state_root=target.state_root, + plugin_ids=plugin_ids, + repo=self.repo_root, + progress=lambda message: self._set_status(f"Bootstrapping: {message}"), + ) + except Exception as exc: + self._set_status(f"Could not enable plugins: {exc}") + return + changed = len(result["enabled"]) + errors = [f"{item['plugin']}: {item['error']}" for item in result["errors"]] self._set_bulk_status("Enabled", changed, errors) self._show_plugins(preserve_status=True) diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index 19ab5b8..e589649 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -13,8 +13,8 @@ from rich.text import Text from textual.coordinate import Coordinate from textual.widgets import DataTable -from plugin_helper.bootstrap import _run_ipa, build_bootstrap_command -from plugin_helper.bsipa import check_bsipa_health +from plugin_helper.bootstrap import _run_ipa, build_bootstrap_command, ensure_healthy_bootstrap +from plugin_helper.bsipa import check_bsipa_health, planning_requires_bootstrap from plugin_helper.beatmods import by_version_id, normalize_mods from plugin_helper.checker import check_lock from plugin_helper.cli import installed_plugins_report, run @@ -23,6 +23,7 @@ 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 Lockfile, LockedPlugin, Registry, RegistryPlugin +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 @@ -380,6 +381,226 @@ state_dir = ".state" self.assertFalse(result["ok"]) self.assertIn("missing Logs/_latest.log", result["messages"]) + def test_planning_requires_bootstrap(self) -> None: + lockfile_plugins = ( + LockedPlugin(id="bsipa", repo=None, tag="4.3.7", asset="BSIPA.zip", sha256=None), + LockedPlugin(id="example", repo=None, tag="v1.0.0", asset="Example.dll", sha256=None), + ) + self.assertTrue(planning_requires_bootstrap(lockfile_plugins, {"example"})) + self.assertFalse(planning_requires_bootstrap(lockfile_plugins, {"bsipa"})) + self.assertFalse(planning_requires_bootstrap((), {"example"})) + + def test_ensure_healthy_bootstrap_noop_when_healthy(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + instance = work / "1.44.1" + state = work / "state" + instance.mkdir(parents=True) + (instance / "IPA").mkdir() + (instance / "Libs").mkdir() + (instance / "IPA.exe").write_bytes(b"ipa") + (instance / "winhttp.dll").write_bytes(b"proxy") + save_bootstrap_state( + state, + "1.44.1", + { + "bootstrapMode": "native", + "ipaExitCode": 0, + "ipaTimedOut": False, + "files": scan_bootstrap_files(instance), + }, + ) + lockfile = Lockfile( + beat_saber_version="1.44.1", + instance="1.44.1", + plugins=( + LockedPlugin(id="bsipa", repo=None, tag="4.3.7", asset="BSIPA.zip", sha256=None), + LockedPlugin(id="example", repo=None, tag="v1.0.0", asset="Example.dll", sha256=None), + ), + ) + registry = Registry( + { + "bsipa": RegistryPlugin(id="bsipa", name="BSIPA", repo=None, install_strategy="root-zip"), + "example": RegistryPlugin( + id="example", + name="Example", + repo=None, + install_strategy="dll-to-plugins", + ), + } + ) + with patch("plugin_helper.bootstrap.run_bootstrap") as run_bootstrap: + ensure_healthy_bootstrap( + instance="1.44.1", + instance_path=instance, + beat_saber_version="1.44.1", + registry=registry, + lockfile=lockfile, + state_root=state, + repo_root=work, + selected_ids={"example"}, + ) + run_bootstrap.assert_not_called() + + def test_ensure_healthy_bootstrap_runs_when_unhealthy(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + instance = work / "1.44.1" + state = work / "state" + instance.mkdir(parents=True) + lockfile = Lockfile( + beat_saber_version="1.44.1", + instance="1.44.1", + plugins=( + LockedPlugin(id="bsipa", repo=None, tag="4.3.7", asset="BSIPA.zip", sha256=None), + LockedPlugin(id="example", repo=None, tag="v1.0.0", asset="Example.dll", sha256=None), + ), + ) + registry = Registry( + { + "bsipa": RegistryPlugin(id="bsipa", name="BSIPA", repo=None, install_strategy="root-zip"), + "example": RegistryPlugin( + id="example", + name="Example", + repo=None, + install_strategy="dll-to-plugins", + ), + } + ) + healthy = { + "ok": True, + "messages": [], + "statePath": str(state / "instances" / "1.44.1" / "bootstrap.json"), + "logPath": str(instance / "Logs" / "_latest.log"), + } + unhealthy = {**healthy, "ok": False, "messages": ["missing IPA.exe"]} + with patch("plugin_helper.bootstrap.check_bsipa_health", side_effect=[unhealthy, healthy]) as check_health: + with patch("plugin_helper.bootstrap.run_bootstrap") as run_bootstrap: + ensure_healthy_bootstrap( + instance="1.44.1", + instance_path=instance, + beat_saber_version="1.44.1", + registry=registry, + lockfile=lockfile, + state_root=state, + repo_root=work, + selected_ids={"example"}, + ) + self.assertEqual(check_health.call_count, 2) + run_bootstrap.assert_called_once() + + def test_ensure_healthy_bootstrap_skips_for_bsipa_only(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + instance = work / "1.44.1" + state = work / "state" + instance.mkdir(parents=True) + lockfile = Lockfile( + beat_saber_version="1.44.1", + instance="1.44.1", + plugins=(LockedPlugin(id="bsipa", repo=None, tag="4.3.7", asset="BSIPA.zip", sha256=None),), + ) + registry = Registry( + {"bsipa": RegistryPlugin(id="bsipa", name="BSIPA", repo=None, install_strategy="root-zip")} + ) + with patch("plugin_helper.bootstrap.check_bsipa_health") as check_health: + with patch("plugin_helper.bootstrap.run_bootstrap") as run_bootstrap: + ensure_healthy_bootstrap( + instance="1.44.1", + instance_path=instance, + beat_saber_version="1.44.1", + registry=registry, + lockfile=lockfile, + state_root=state, + repo_root=work, + selected_ids={"bsipa"}, + ) + check_health.assert_not_called() + run_bootstrap.assert_not_called() + + def test_enable_disabled_plugin_bootstraps_before_plan(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + work = Path(tmp) + instance = work / "instances" / "1.44.1" + state = work / "state" + instance.mkdir(parents=True) + (instance / "Beat Saber_Data").mkdir() + (instance / "Plugins").mkdir() + asset = plugin_downloads_dir(state, "1.44.1", "example") / "Example.dll" + asset.write_bytes(b"managed dll") + save_installed_state( + state, + "1.44.1", + { + "disabledPlugins": { + "example": { + "disabledAt": "2026-01-01T00:00:00Z", + "files": [{"path": "Plugins/Example.dll", "sha256": sha256_file(asset)}], + } + } + }, + ) + (work / "locks").mkdir() + (work / "registry").mkdir() + lock_path = work / "locks" / "1.44.1.lock.toml" + lock_path.write_text( + """ +beat_saber_version = "1.44.1" +instance = "1.44.1" + +[[plugins]] +id = "bsipa" +tag = "4.3.7" +asset = "BSIPA.zip" + +[[plugins]] +id = "example" +tag = "v1.0.0" +asset = "Example.dll" +sha256 = "%s" +""" % sha256_file(asset), + encoding="utf-8", + ) + registry_path = work / "registry" / "plugins.toml" + registry_path.write_text( + """ +[bsipa] +name = "BSIPA" +install_strategy = "root-zip" + +[example] +name = "Example" +install_strategy = "dll-to-plugins" +""", + encoding="utf-8", + ) + call_order: list[str] = [] + + def _ensure(**_kwargs: object) -> None: + call_order.append("ensure") + + def _create_plan(**_kwargs: object) -> tuple[dict[str, object], Path]: + call_order.append("plan") + return {"changes": [], "instance": "1.44.1", "instancePath": str(instance)}, work / "plan.json" + + def _apply_plan(_plan: dict[str, object], _state_root: Path) -> dict[str, object]: + call_order.append("apply") + return {"applied": [], "statePath": str(state / "instances" / "1.44.1" / "installed.json")} + + with patch("plugin_helper.operations.ensure_healthy_bootstrap", side_effect=_ensure): + with patch("plugin_helper.operations.create_plan", side_effect=_create_plan): + with patch("plugin_helper.operations.apply_plan", side_effect=_apply_plan): + enable_disabled_plugin( + instance="1.44.1", + instance_path=instance, + state_root=state, + plugin_id="example", + registry=str(registry_path), + lockfile=str(lock_path), + repo=work, + ) + self.assertEqual(call_order, ["ensure", "plan", "apply"]) + def test_plan_apply_and_uninstall_dll(self) -> None: with tempfile.TemporaryDirectory() as tmp: work = Path(tmp)