pleb/plugin-helper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add plugin disable and enable commands

Browse Source
This commit is contained in:
pleb
2026-07-01 10:42:18 -07:00
parent de0b20fa61
commit 13b1840ba0
3 changed files with 240 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/plugin_helper/cli.py
+83 -4
View File
@@ -11,7 +11,7 @@ from .bootstrap import run_bootstrap
from .bsipa import check_bsipa_health
from .checker import check_lock
from .github import fetch_releases
from .installer import apply_plan, uninstall_plugin
from .installer import apply_plan, disable_plugin, uninstall_plugin
from .instances import get_instance, list_instances
from .models import load_lockfile, load_registry
from .models import Lockfile, Registry
@@ -34,7 +34,15 @@ def installed_plugins_report(
) -> dict[str, Any]:
locked_by_id = {plugin.id: plugin for plugin in lockfile.plugins}
plugins: list[dict[str, Any]] = []
for plugin_id, plugin_state in sorted(installed_state.get("plugins", {}).items()):
state_plugins = [
(plugin_id, plugin_state, "enabled")
for plugin_id, plugin_state in installed_state.get("plugins", {}).items()
]
state_plugins.extend(
(plugin_id, plugin_state, "disabled")
for plugin_id, plugin_state in installed_state.get("disabledPlugins", {}).items()
)
for plugin_id, plugin_state, status in sorted(state_plugins):
registry_plugin = registry.get(plugin_id)
locked = locked_by_id.get(plugin_id)
files = plugin_state.get("files", [])
@@ -48,6 +56,8 @@ def installed_plugins_report(
or (registry_plugin.repo if registry_plugin else None)
or "(unknown)",
"installedAt": plugin_state.get("installedAt", "(unknown)"),
"disabledAt": plugin_state.get("disabledAt"),
"status": status,
"fileCount": len(files),
"files": files,
}
@@ -66,14 +76,15 @@ def print_installed_plugins(report: dict[str, Any]) -> None:
print("No plugins have been installed by plugin-helper yet.")
return
headers = ("Plugin", "Version", "Asset", "Files", "Installed")
headers = ("Plugin", "Status", "Version", "Asset", "Files", "Installed")
rows = [
(
f"{plugin['name']} ({plugin['id']})",
plugin["status"],
plugin["version"],
plugin["asset"],
str(plugin["fileCount"]),
plugin["installedAt"],
plugin["disabledAt"] or plugin["installedAt"],
)
for plugin in plugins
]
@@ -236,6 +247,25 @@ def build_parser() -> argparse.ArgumentParser:
uninstall.add_argument("plugin")
uninstall.add_argument("--force", action="store_true", help="Delete even when current file hashes differ")
disable = subcommands.add_parser(
"disable",
help="Remove a managed plugin's files from the game but keep state/assets for re-enable",
parents=[_common_parent()],
)
disable.add_argument("--instance", required=True)
disable.add_argument("plugin")
disable.add_argument("--force", action="store_true", help="Disable even when current file hashes differ")
enable = subcommands.add_parser(
"enable",
help="Reinstall a disabled locked plugin from local assets",
parents=[_common_parent()],
)
enable.add_argument("--instance", required=True)
enable.add_argument("--registry", default="registry/plugins.toml")
enable.add_argument("--lockfile")
enable.add_argument("plugin")
backup = subcommands.add_parser(
"backup-userdata",
help="Copy UserData and Windows AppData into the adjacent backups repo",
@@ -310,6 +340,8 @@ def _run_menu(inst_roots: list[Path], st_root: Path, input_func: Callable[[str],
("bootstrap-check", "Check BSIPA bootstrap", "Verifies recorded bootstrap state and the latest BSIPA log evidence."),
("plan", "Create install plan", "Writes a dry-run JSON plan for locked plugin files before anything is applied."),
("apply", "Apply a plan by path", "Installs exactly the file changes from a previously generated plan JSON."),
("disable", "Disable managed plugin", "Removes one recorded plugin from the game while keeping helper state and assets."),
("enable", "Enable managed plugin", "Reinstalls one disabled locked plugin from local assets."),
("backup-userdata", "Back up UserData", "Copies UserData and AppData into the adjacent backups repo."),
("restore-userdata", "Restore UserData", "Restores UserData and AppData from the backups repo into an instance."),
("change", "Choose another version", "Returns to the Beat Saber version picker."),
@@ -356,6 +388,12 @@ def _run_menu(inst_roots: list[Path], st_root: Path, input_func: Callable[[str],
print("No plan path entered.")
continue
command.append(plan_path)
elif selected_action in {"disable", "enable"}:
plugin_id = ask("Plugin id> ").strip()
if not plugin_id:
print("No plugin id entered.")
continue
command.extend(["--instance", selected_instance.name, plugin_id])
else:
command.extend(["--instance", selected_instance.name])
@@ -566,6 +604,47 @@ def run(argv: list[str] | None = None) -> int:
print(f" {item['path']}: {item['reason']}")
return 0 if result["stateUpdated"] else 2
if args.command == "disable":
instance = get_instance(inst_roots, args.instance)
result = disable_plugin(args.instance, instance.path, st_root, args.plugin, force=args.force)
print(f"Disabled: {args.plugin}")
print(f"Removed: {len(result['removed'])}")
if result["skipped"]:
print("Skipped:")
for item in result["skipped"]:
print(f" {item['path']}: {item['reason']}")
return 0 if result["stateUpdated"] else 2
if args.command == "enable":
instance = get_instance(inst_roots, args.instance)
installed_state = load_installed_state(st_root, args.instance)
if args.plugin not in installed_state.get("disabledPlugins", {}):
raise KeyError(f"plugin is not recorded as disabled: {args.plugin}")
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()
lockfile = load_lockfile(lock_path)
if not any(plugin.id == args.plugin for plugin in lockfile.plugins):
raise KeyError(f"plugin is disabled but not locked for this instance: {args.plugin}")
plan, path = create_plan(
instance=args.instance,
instance_path=instance.path,
beat_saber_version=lockfile.beat_saber_version,
registry=load_registry(registry_path),
lockfile=lockfile,
state_root=st_root,
repo_root=root,
selected={args.plugin},
)
result = apply_plan(plan, st_root)
print(f"Enabled: {args.plugin}")
print(f"Plan: {path}")
print(f"Applied: {len(result['applied'])}")
print(f"State: {result['statePath']}")
return 0
if args.command == "backup-userdata":
instance = get_instance(inst_roots, args.instance)
root = repo_root()
src/plugin_helper/installer.py
+35
View File
@@ -77,6 +77,7 @@ def apply_plan(plan: dict[str, Any], state_root: Path) -> dict[str, Any]:
"size": target.stat().st_size,
}
)
installed_state.setdefault("disabledPlugins", {}).pop(change["plugin"], None)
applied.append({"path": rel_target, "plugin": change["plugin"], "backup": backup})
save_installed_state(state_root, instance, installed_state)
@@ -110,3 +111,37 @@ def uninstall_plugin(instance: str, instance_path: Path, state_root: Path, plugi
installed_state.get("plugins", {}).pop(plugin_id, None)
save_installed_state(state_root, instance, installed_state)
return {"removed": removed, "skipped": skipped, "stateUpdated": True}
def disable_plugin(instance: str, instance_path: Path, state_root: Path, plugin_id: str, force: bool = False) -> dict[str, Any]:
installed_state = load_installed_state(state_root, instance)
plugin_state = installed_state.get("plugins", {}).get(plugin_id)
if not plugin_state:
if plugin_id in installed_state.get("disabledPlugins", {}):
raise KeyError(f"plugin is already disabled: {plugin_id}")
raise KeyError(f"plugin is not recorded in install state: {plugin_id}")
removed: list[str] = []
skipped: list[dict[str, str]] = []
for item in plugin_state.get("files", []):
rel_path = ensure_relative(item["path"]).as_posix()
target = ensure_inside(instance_path, instance_path / rel_path)
if not target.exists():
removed.append(rel_path)
continue
current_sha = sha256_file(target)
if current_sha != item.get("sha256") and not force:
skipped.append({"path": rel_path, "reason": "hash mismatch"})
continue
target.unlink()
removed.append(rel_path)
if skipped and not force:
return {"removed": removed, "skipped": skipped, "stateUpdated": False}
disabled_state = dict(plugin_state)
disabled_state["disabledAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
installed_state.setdefault("disabledPlugins", {})[plugin_id] = disabled_state
installed_state.get("plugins", {}).pop(plugin_id, None)
save_installed_state(state_root, instance, installed_state)
return {"removed": removed, "skipped": skipped, "stateUpdated": True}
tests/test_plugin_helper.py
+122 -1
View File
@@ -14,7 +14,7 @@ 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
from plugin_helper.fsutil import sha256_file
from plugin_helper.installer import apply_plan, 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.models import Lockfile, LockedPlugin, Registry, RegistryPlugin
from plugin_helper.planner import create_plan
@@ -242,6 +242,127 @@ class PluginHelperTests(unittest.TestCase):
self.assertEqual(removed["removed"], ["Plugins/Example.dll"])
self.assertFalse((instance / "Plugins" / "Example.dll").exists())
def test_disable_plugin_removes_files_but_keeps_disabled_state(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)
instance = work / "instances" / "1.40.8"
state = work / "state"
instance.mkdir(parents=True)
(instance / "Beat Saber_Data").mkdir()
(instance / "Plugins").mkdir()
target = instance / "Plugins" / "Example.dll"
target.write_bytes(b"managed dll")
installed = {
"instance": "1.40.8",
"plugins": {
"example": {
"installedAt": "2026-06-14T17:18:40Z",
"files": [
{
"path": "Plugins/Example.dll",
"sha256": sha256_file(target),
"size": target.stat().st_size,
}
],
}
},
}
from plugin_helper.state import save_installed_state
save_installed_state(state, "1.40.8", installed)
result = disable_plugin("1.40.8", instance, state, "example")
self.assertEqual(result["removed"], ["Plugins/Example.dll"])
self.assertFalse(target.exists())
updated = load_installed_state(state, "1.40.8")
self.assertNotIn("example", updated["plugins"])
self.assertIn("example", updated["disabledPlugins"])
self.assertEqual(updated["disabledPlugins"]["example"]["files"][0]["path"], "Plugins/Example.dll")
def test_enable_command_reinstalls_disabled_plugin_from_asset(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)
instance_root = work / "instances"
instance = instance_root / "1.40.8"
state = work / "state"
registry_dir = work / "registry"
locks_dir = work / "locks"
registry_dir.mkdir()
locks_dir.mkdir()
instance.mkdir(parents=True)
(instance / "Beat Saber_Data").mkdir()
(instance / "Plugins").mkdir()
asset = plugin_downloads_dir(state, "1.40.8", "example") / "Example.dll"
asset.write_bytes(b"managed dll")
(registry_dir / "plugins.toml").write_text(
"""
[[plugins]]
id = "example"
name = "Example"
repo = "owner/example"
asset_patterns = ["*.dll"]
install_strategy = "dll-to-plugins"
""".lstrip(),
encoding="utf-8",
)
(locks_dir / "1.40.8.lock.toml").write_text(
f"""
beat_saber_version = "1.40.8"
instance = "1.40.8"
[[plugins]]
id = "example"
repo = "owner/example"
tag = "v1.0.0"
asset = "Example.dll"
sha256 = "{sha256_file(asset)}"
""".lstrip(),
encoding="utf-8",
)
disabled = {
"instance": "1.40.8",
"plugins": {},
"disabledPlugins": {
"example": {
"installedAt": "2026-06-14T17:18:40Z",
"disabledAt": "2026-06-14T17:20:00Z",
"files": [
{
"path": "Plugins/Example.dll",
"sha256": sha256_file(asset),
"size": asset.stat().st_size,
}
],
}
},
}
from plugin_helper.state import save_installed_state
save_installed_state(state, "1.40.8", disabled)
with patch("plugin_helper.cli.repo_root", return_value=work):
status = run(
[
"--instances-root",
str(instance_root),
"--state-dir",
str(state),
"enable",
"--instance",
"1.40.8",
"example",
]
)
self.assertEqual(status, 0)
self.assertEqual((instance / "Plugins" / "Example.dll").read_bytes(), b"managed dll")
updated = load_installed_state(state, "1.40.8")
self.assertIn("example", updated["plugins"])
self.assertNotIn("example", updated["disabledPlugins"])
def test_zip_to_pending_targets_ipa_pending(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)