Add native Windows compatibility for config, bootstrap, and CLI.

Support profile-based config resolution with auto-selection of a sole profile,
native IPA.exe -n bootstrap on Windows, platform-aware process cleanup, and
Windows-friendly userdata backup path inference.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
pleb
2026-07-01 22:33:24 -07:00
parent 6af1b03c35
commit c67878cac0
7 changed files with 313 additions and 47 deletions
+5
View File
@@ -0,0 +1,5 @@
[[profiles]]
id = "windows"
label = "Native Windows BSManager"
instances_root = "~/BSManager/BSInstances"
state_dir = ".state"
+61 -18
View File
@@ -9,6 +9,7 @@ from typing import Any, Callable
from urllib.request import Request, urlopen from urllib.request import Request, urlopen
from .bsipa import BSIPA_PLUGIN_ID, check_bsipa_health from .bsipa import BSIPA_PLUGIN_ID, check_bsipa_health
from .config import is_windows
from .fsutil import sha256_file from .fsutil import sha256_file
from .installer import apply_plan from .installer import apply_plan
from .models import LockedPlugin, Lockfile, Registry from .models import LockedPlugin, Lockfile, Registry
@@ -130,31 +131,63 @@ def fetch_locked_bsipa_archive(lockfile: Lockfile, state_root: Path) -> dict[str
return result return result
def build_bootstrap_command(
ipa: Path,
*,
proton: Path | None = None,
native: bool | None = None,
) -> list[str]:
use_native = is_windows() if native is None else native
if use_native:
return [str(ipa), "-n"]
proton_path = proton or _default_proton()
return [str(proton_path), "run", str(ipa), "-n"]
def _terminate_process(process: subprocess.Popen[str]) -> None:
if is_windows():
process.terminate()
else:
os.killpg(process.pid, signal.SIGTERM)
def _kill_process(process: subprocess.Popen[str]) -> None:
if is_windows():
process.kill()
else:
os.killpg(process.pid, signal.SIGKILL)
def _run_ipa( def _run_ipa(
*, *,
command: list[str], command: list[str],
instance_path: Path, instance_path: Path,
timeout_seconds: int, timeout_seconds: int,
native: bool = False,
) -> dict[str, Any]: ) -> dict[str, Any]:
process = subprocess.Popen( popen_kwargs: dict[str, Any] = {
command, "cwd": instance_path,
cwd=instance_path, "text": True,
env=_proton_env(instance_path), "stdout": subprocess.PIPE,
text=True, "stderr": subprocess.PIPE,
stdout=subprocess.PIPE, }
stderr=subprocess.PIPE, if native:
start_new_session=True, popen_kwargs["env"] = os.environ.copy()
) else:
popen_kwargs["env"] = _proton_env(instance_path)
popen_kwargs["start_new_session"] = True
process = subprocess.Popen(command, **popen_kwargs)
try: try:
stdout, stderr = process.communicate(timeout=timeout_seconds) stdout, stderr = process.communicate(timeout=timeout_seconds)
timed_out = False timed_out = False
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
timed_out = True timed_out = True
os.killpg(process.pid, signal.SIGTERM) _terminate_process(process)
try: try:
stdout, stderr = process.communicate(timeout=5) stdout, stderr = process.communicate(timeout=5)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
os.killpg(process.pid, signal.SIGKILL) _kill_process(process)
stdout, stderr = process.communicate() stdout, stderr = process.communicate()
return { return {
"returncode": process.returncode, "returncode": process.returncode,
@@ -175,10 +208,14 @@ def run_bootstrap(
state_root: Path, state_root: Path,
repo_root: Path, repo_root: Path,
proton: Path | None = None, proton: Path | None = None,
native: bool | None = None,
progress: Callable[[str], None] | None = None, progress: Callable[[str], None] | None = None,
ipa_timeout_seconds: int = 120, ipa_timeout_seconds: int = 120,
) -> dict[str, Any]: ) -> dict[str, Any]:
tell = progress or (lambda _message: None) tell = progress or (lambda _message: None)
use_native = is_windows() if native is None else native
bootstrap_mode = "native" if use_native else "proton"
tell("Fetching locked BSIPA archive") tell("Fetching locked BSIPA archive")
fetched = fetch_locked_bsipa_archive(lockfile, state_root) fetched = fetch_locked_bsipa_archive(lockfile, state_root)
if fetched.get("cached"): if fetched.get("cached"):
@@ -210,30 +247,34 @@ def run_bootstrap(
if not ipa.is_file(): if not ipa.is_file():
raise FileNotFoundError(f"BSIPA archive did not install IPA.exe: {ipa}") raise FileNotFoundError(f"BSIPA archive did not install IPA.exe: {ipa}")
proton_path = proton or _default_proton() command = build_bootstrap_command(ipa, proton=proton, native=use_native)
if not proton_path.is_file(): if use_native:
raise FileNotFoundError(f"Proton executable not found: {proton_path}") tell(f"Running IPA.exe -n natively; timeout {ipa_timeout_seconds}s")
else:
proton_path = proton or _default_proton()
if not proton_path.is_file():
raise FileNotFoundError(f"Proton executable not found: {proton_path}")
tell(f"Running IPA.exe -n through Proton; timeout {ipa_timeout_seconds}s")
command = [str(proton_path), "run", str(ipa), "-n"]
tell(f"Running IPA.exe -n through Proton; timeout {ipa_timeout_seconds}s")
completed = _run_ipa( completed = _run_ipa(
command=command, command=command,
instance_path=instance_path, instance_path=instance_path,
timeout_seconds=ipa_timeout_seconds, timeout_seconds=ipa_timeout_seconds,
native=use_native,
) )
tell("Scanning bootstrap files after IPA.exe -n") tell("Scanning bootstrap files after IPA.exe -n")
after = scan_bootstrap_files(instance_path) after = scan_bootstrap_files(instance_path)
delta = _files_delta(before, after) delta = _files_delta(before, after)
state = { state: dict[str, Any] = {
"schemaVersion": 1, "schemaVersion": 1,
"instance": instance, "instance": instance,
"beatSaberVersion": beat_saber_version, "beatSaberVersion": beat_saber_version,
"bootstrappedAt": _now_iso(), "bootstrappedAt": _now_iso(),
"plugin": BSIPA_PLUGIN_ID, "plugin": BSIPA_PLUGIN_ID,
"bootstrapMode": bootstrap_mode,
"archive": fetched, "archive": fetched,
"planPath": str(plan_path), "planPath": str(plan_path),
"applied": apply_result["applied"], "applied": apply_result["applied"],
"proton": str(proton_path),
"command": command, "command": command,
"ipaExitCode": completed["returncode"], "ipaExitCode": completed["returncode"],
"ipaTimedOut": completed["timedOut"], "ipaTimedOut": completed["timedOut"],
@@ -244,6 +285,8 @@ def run_bootstrap(
"delta": delta, "delta": delta,
"health": {}, "health": {},
} }
if not use_native:
state["proton"] = str(proton or _default_proton())
save_bootstrap_state(state_root, instance, state) save_bootstrap_state(state_root, instance, state)
state["health"] = check_bsipa_health(instance_path, state_root, instance) state["health"] = check_bsipa_health(instance_path, state_root, instance)
save_bootstrap_state(state_root, instance, state) save_bootstrap_state(state_root, instance, state)
+14 -2
View File
@@ -14,6 +14,14 @@ def latest_log_path(instance_path: Path) -> Path:
return instance_path / "Logs" / "_latest.log" return instance_path / "Logs" / "_latest.log"
def _native_bootstrap_satisfied(state: dict[str, Any]) -> bool:
return (
state.get("bootstrapMode") == "native"
and state.get("ipaExitCode") == 0
and not state.get("ipaTimedOut")
)
def check_bsipa_health(instance_path: Path, state_root: Path, instance: str) -> dict[str, Any]: def check_bsipa_health(instance_path: Path, state_root: Path, instance: str) -> dict[str, Any]:
state = load_bootstrap_state(state_root, instance) state = load_bootstrap_state(state_root, instance)
messages: list[str] = [] messages: list[str] = []
@@ -27,12 +35,15 @@ def check_bsipa_health(instance_path: Path, state_root: Path, instance: str) ->
messages.append(f"missing {rel}/") messages.append(f"missing {rel}/")
log_path = latest_log_path(instance_path) log_path = latest_log_path(instance_path)
native_bootstrap = _native_bootstrap_satisfied(state)
if not log_path.is_file(): if not log_path.is_file():
messages.append("missing Logs/_latest.log") if not native_bootstrap:
messages.append("missing Logs/_latest.log")
else: else:
text = log_path.read_text(encoding="utf-8", errors="replace") text = log_path.read_text(encoding="utf-8", errors="replace")
if "Beat Saber IPA (BSIPA):" not in text and "Beat Saber IPA" not in text: if "Beat Saber IPA (BSIPA):" not in text and "Beat Saber IPA" not in text:
messages.append("Logs/_latest.log does not show BSIPA startup") if not native_bootstrap:
messages.append("Logs/_latest.log does not show BSIPA startup")
if not state: if not state:
messages.append(f"missing bootstrap state: {bootstrap_state_path(state_root, instance)}") messages.append(f"missing bootstrap state: {bootstrap_state_path(state_root, instance)}")
@@ -44,4 +55,5 @@ def check_bsipa_health(instance_path: Path, state_root: Path, instance: str) ->
"logPath": str(log_path), "logPath": str(log_path),
"logSha256": sha256_file(log_path) if log_path.is_file() else None, "logSha256": sha256_file(log_path) if log_path.is_file() else None,
"bootstrapRecordedAt": state.get("updatedAt"), "bootstrapRecordedAt": state.get("updatedAt"),
"bootstrapMode": state.get("bootstrapMode"),
} }
+8 -2
View File
@@ -61,6 +61,8 @@ def print_updates(report: dict[str, Any]) -> None:
def _add_common(parser: argparse.ArgumentParser, *, suppress_default: bool = False) -> None: def _add_common(parser: argparse.ArgumentParser, *, suppress_default: bool = False) -> None:
default = argparse.SUPPRESS if suppress_default else None default = argparse.SUPPRESS if suppress_default else None
parser.add_argument("--config", default=default, help="plugin-helper config TOML path")
parser.add_argument("--profile", default=default, help="Config profile id from [[profiles]]")
parser.add_argument("--instances-root", default=default, help="BSManager instances root") parser.add_argument("--instances-root", default=default, help="BSManager instances root")
parser.add_argument("--state-dir", default=default, help="plugin-helper state directory") parser.add_argument("--state-dir", default=default, help="plugin-helper state directory")
@@ -120,13 +122,14 @@ def build_parser() -> argparse.ArgumentParser:
bootstrap = subcommands.add_parser( bootstrap = subcommands.add_parser(
"bootstrap", "bootstrap",
help="Install locked BSIPA, run IPA.exe -n through Proton, and record bootstrap files", help="Install locked BSIPA, run IPA.exe -n, and record bootstrap files",
parents=[_common_parent()], parents=[_common_parent()],
) )
bootstrap.add_argument("--instance", required=True) bootstrap.add_argument("--instance", required=True)
bootstrap.add_argument("--registry", default="registry/plugins.toml") bootstrap.add_argument("--registry", default="registry/plugins.toml")
bootstrap.add_argument("--lockfile") bootstrap.add_argument("--lockfile")
bootstrap.add_argument("--proton", help="Path to Proton executable") bootstrap.add_argument("--proton", help="Path to Proton executable (Linux/Proton installs)")
bootstrap.add_argument("--native", action="store_true", help="Run IPA.exe -n natively instead of through Proton")
bootstrap.add_argument("--json", action="store_true", help="Print full JSON bootstrap output") bootstrap.add_argument("--json", action="store_true", help="Print full JSON bootstrap output")
bootstrap_check = subcommands.add_parser( bootstrap_check = subcommands.add_parser(
@@ -282,6 +285,8 @@ def run(argv: list[str] | None = None) -> int:
runtime = resolve_runtime_config( runtime = resolve_runtime_config(
instances_root_value=getattr(args, "instances_root", None), instances_root_value=getattr(args, "instances_root", None),
state_dir_value=getattr(args, "state_dir", None), state_dir_value=getattr(args, "state_dir", None),
config_path_value=getattr(args, "config", None),
profile_id=getattr(args, "profile", None),
) )
inst_roots = runtime.instances_roots inst_roots = runtime.instances_roots
st_root = runtime.state_root st_root = runtime.state_root
@@ -409,6 +414,7 @@ def run(argv: list[str] | None = None) -> int:
state_root=st_root, state_root=st_root,
repo_root=root, repo_root=root,
proton=Path(args.proton).expanduser() if args.proton else None, proton=Path(args.proton).expanduser() if args.proton else None,
native=True if args.native else None,
progress=lambda message: print(f" {message}", flush=True), progress=lambda message: print(f" {message}", flush=True),
) )
if args.json: if args.json:
+82 -7
View File
@@ -1,6 +1,7 @@
from __future__ import annotations from __future__ import annotations
import os import os
import sys
import tomllib import tomllib
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -10,12 +11,22 @@ from typing import Any
LOCAL_INSTANCES_ROOT = Path.home() / ".local/share/BSManager/BSInstances" LOCAL_INSTANCES_ROOT = Path.home() / ".local/share/BSManager/BSInstances"
DEFAULT_INSTANCES_ROOT = LOCAL_INSTANCES_ROOT DEFAULT_INSTANCES_ROOT = LOCAL_INSTANCES_ROOT
LOCAL_CONFIG_NAME = "plugin-helper.local.toml" LOCAL_CONFIG_NAME = "plugin-helper.local.toml"
WINDOWS_CONFIG_NAME = "plugin-helper.windows.toml"
@dataclass(frozen=True)
class Profile:
id: str
label: str | None
instances_roots: list[Path]
state_root: Path
@dataclass(frozen=True) @dataclass(frozen=True)
class LocalConfig: class LocalConfig:
instances_roots: list[Path] | None instances_roots: list[Path] | None
state_root: Path | None state_root: Path | None
profiles: tuple[Profile, ...]
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -24,6 +35,12 @@ class RuntimeConfig:
state_root: Path state_root: Path
config_path: Path config_path: Path
config_loaded: bool config_loaded: bool
profile_id: str | None = None
profile_label: str | None = None
def is_windows() -> bool:
return sys.platform == "win32"
def repo_root() -> Path: def repo_root() -> Path:
@@ -38,7 +55,7 @@ def instances_roots(value: str | None = None, *, base: Path | None = None) -> li
raw = value or os.environ.get("PLUGIN_HELPER_INSTANCES_ROOT") raw = value or os.environ.get("PLUGIN_HELPER_INSTANCES_ROOT")
if raw: if raw:
return _resolve_path_list(raw, base or repo_root()) return _resolve_path_list(raw, base or repo_root())
return [DEFAULT_INSTANCES_ROOT] return _default_instances_roots()
def state_root(value: str | None = None) -> Path: def state_root(value: str | None = None) -> Path:
@@ -47,13 +64,16 @@ def state_root(value: str | None = None) -> Path:
env_state = os.environ.get("PLUGIN_HELPER_STATE_DIR") env_state = os.environ.get("PLUGIN_HELPER_STATE_DIR")
if env_state: if env_state:
return _resolve_path(env_state, repo_root()) return _resolve_path(env_state, repo_root())
xdg_state = os.environ.get("XDG_STATE_HOME") return _default_state_root()
base = Path(xdg_state).expanduser() if xdg_state else Path.home() / ".local" / "state"
return base / "plugin-helper"
def default_config_path(root: Path | None = None) -> Path: def default_config_path(root: Path | None = None) -> Path:
return (root or repo_root()) / LOCAL_CONFIG_NAME repo = root or repo_root()
if is_windows():
windows_config = repo / WINDOWS_CONFIG_NAME
if windows_config.is_file():
return windows_config
return repo / LOCAL_CONFIG_NAME
def _resolve_path(value: str | Path, base: Path) -> Path: def _resolve_path(value: str | Path, base: Path) -> Path:
@@ -67,13 +87,34 @@ def _resolve_path_list(value: str, base: Path) -> list[Path]:
return [_resolve_path(item, base) for item in value.split(os.pathsep) if item] return [_resolve_path(item, base) for item in value.split(os.pathsep) if item]
def _parse_profiles(data: dict[str, Any], base: Path) -> tuple[Profile, ...]:
profiles: list[Profile] = []
for entry in data.get("profiles", []):
if not isinstance(entry, dict):
continue
profile_id = entry.get("id")
instances_value = entry.get("instances_root")
state_value = entry.get("state_dir")
if not profile_id or not instances_value or not state_value:
raise ValueError(f"profile {profile_id!r} needs id, instances_root, and state_dir")
profiles.append(
Profile(
id=str(profile_id),
label=entry.get("label"),
instances_roots=_resolve_path_list(str(instances_value), base),
state_root=_resolve_path(str(state_value), base),
)
)
return tuple(profiles)
def load_local_config(config_path: str | Path | None = None, *, root: Path | None = None) -> tuple[LocalConfig, Path, bool]: def load_local_config(config_path: str | Path | None = None, *, root: Path | None = None) -> tuple[LocalConfig, Path, bool]:
repo = root or repo_root() repo = root or repo_root()
path = _resolve_path(config_path, repo) if config_path else default_config_path(repo) path = _resolve_path(config_path, repo) if config_path else default_config_path(repo)
if not path.exists(): if not path.exists():
if config_path: if config_path:
raise FileNotFoundError(f"plugin-helper config not found: {path}") raise FileNotFoundError(f"plugin-helper config not found: {path}")
return LocalConfig(instances_roots=None, state_root=None), path, False return LocalConfig(instances_roots=None, state_root=None, profiles=()), path, False
with path.open("rb") as handle: with path.open("rb") as handle:
data: dict[str, Any] = tomllib.load(handle) data: dict[str, Any] = tomllib.load(handle)
@@ -81,10 +122,12 @@ def load_local_config(config_path: str | Path | None = None, *, root: Path | Non
base = path.parent base = path.parent
instances_value = data.get("instances_root") instances_value = data.get("instances_root")
state_value = data.get("state_dir") state_value = data.get("state_dir")
profiles = _parse_profiles(data, base)
return ( return (
LocalConfig( LocalConfig(
instances_roots=_resolve_path_list(instances_value, base) if instances_value else None, instances_roots=_resolve_path_list(instances_value, base) if instances_value else None,
state_root=_resolve_path(state_value, base) if state_value else None, state_root=_resolve_path(state_value, base) if state_value else None,
profiles=profiles,
), ),
path, path,
True, True,
@@ -102,28 +145,57 @@ def _env_state_root(root: Path) -> Path | None:
def _default_state_root() -> Path: def _default_state_root() -> Path:
if is_windows():
localappdata = os.environ.get("LOCALAPPDATA")
if localappdata:
return Path(localappdata) / "plugin-helper"
return Path.home() / "AppData" / "Local" / "plugin-helper"
xdg_state = os.environ.get("XDG_STATE_HOME") xdg_state = os.environ.get("XDG_STATE_HOME")
base = Path(xdg_state).expanduser() if xdg_state else Path.home() / ".local" / "state" base = Path(xdg_state).expanduser() if xdg_state else Path.home() / ".local" / "state"
return base / "plugin-helper" return base / "plugin-helper"
def _default_instances_roots() -> list[Path]: def _default_instances_roots() -> list[Path]:
if is_windows():
return [Path.home() / "BSManager" / "BSInstances"]
return [DEFAULT_INSTANCES_ROOT] return [DEFAULT_INSTANCES_ROOT]
def _select_profile(local_config: LocalConfig, profile_id: str | None) -> Profile | None:
if profile_id:
for profile in local_config.profiles:
if profile.id == profile_id:
return profile
available = ", ".join(profile.id for profile in local_config.profiles) or "(none)"
raise ValueError(f"unknown profile {profile_id!r}; available profiles: {available}")
if (
local_config.profiles
and local_config.instances_roots is None
and local_config.state_root is None
and len(local_config.profiles) == 1
):
return local_config.profiles[0]
return None
def resolve_runtime_config( def resolve_runtime_config(
*, *,
instances_root_value: str | None = None, instances_root_value: str | None = None,
state_dir_value: str | None = None, state_dir_value: str | None = None,
config_path_value: str | None = None,
profile_id: str | None = None,
root: Path | None = None, root: Path | None = None,
) -> RuntimeConfig: ) -> RuntimeConfig:
repo = root or repo_root() repo = root or repo_root()
local_config, loaded_path, loaded = load_local_config(root=repo) local_config, loaded_path, loaded = load_local_config(config_path_value, root=repo)
profile = _select_profile(local_config, profile_id)
resolved_instances = ( resolved_instances = (
_resolve_path_list(instances_root_value, repo) _resolve_path_list(instances_root_value, repo)
if instances_root_value if instances_root_value
else _env_instances_roots(repo) else _env_instances_roots(repo)
or (profile.instances_roots if profile else None)
or local_config.instances_roots or local_config.instances_roots
or _default_instances_roots() or _default_instances_roots()
) )
@@ -131,6 +203,7 @@ def resolve_runtime_config(
_resolve_path(state_dir_value, repo) _resolve_path(state_dir_value, repo)
if state_dir_value if state_dir_value
else _env_state_root(repo) else _env_state_root(repo)
or (profile.state_root if profile else None)
or local_config.state_root or local_config.state_root
or _default_state_root() or _default_state_root()
) )
@@ -140,4 +213,6 @@ def resolve_runtime_config(
state_root=resolved_state, state_root=resolved_state,
config_path=loaded_path, config_path=loaded_path,
config_loaded=loaded, config_loaded=loaded,
profile_id=profile.id if profile else None,
profile_label=profile.label if profile else None,
) )
+7 -9
View File
@@ -6,7 +6,6 @@ import shutil
import tarfile import tarfile
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any, Callable from typing import Any, Callable
from .fsutil import sha256_file from .fsutil import sha256_file
@@ -62,25 +61,24 @@ def backup_userdata(instance: str, instance_path: Path, state_root: Path) -> dic
} }
destination.parent.mkdir(parents=True, exist_ok=True) destination.parent.mkdir(parents=True, exist_ok=True)
manifest_path = destination.parent / f".{destination.name}.manifest.json"
manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
with tarfile.open(destination, "w:gz") as archive: with tarfile.open(destination, "w:gz") as archive:
archive.add(source, arcname="UserData") archive.add(source, arcname="UserData")
with NamedTemporaryFile("w", encoding="utf-8", suffix=".json") as handle: archive.add(manifest_path, arcname="manifest.json")
json.dump(manifest, handle, indent=2, sort_keys=True) manifest_path.unlink(missing_ok=True)
handle.write("\n")
handle.flush()
archive.add(handle.name, arcname="manifest.json")
return {"archive": str(destination), "manifest": manifest} return {"archive": str(destination), "manifest": manifest}
def infer_windows_appdata_path(instance_path: Path) -> Path: def infer_windows_appdata_path(instance_path: Path) -> Path:
parts = instance_path.resolve().parts parts = instance_path.resolve().parts
try: try:
users_index = parts.index("Users") bsmanager_index = parts.index("BSManager")
except ValueError as exc: except ValueError as exc:
raise ValueError(f"cannot infer Windows user profile from instance path: {instance_path}") from exc raise ValueError(f"cannot infer Windows user profile from instance path: {instance_path}") from exc
if users_index + 1 >= len(parts): if bsmanager_index < 2 or parts[bsmanager_index - 2] != "Users":
raise ValueError(f"cannot infer Windows user profile from instance path: {instance_path}") raise ValueError(f"cannot infer Windows user profile from instance path: {instance_path}")
profile = Path(*parts[: users_index + 2]) profile = Path(*parts[:bsmanager_index])
return profile / "AppData" / "LocalLow" / "Hyperbolic Magnetism" / "Beat Saber" return profile / "AppData" / "LocalLow" / "Hyperbolic Magnetism" / "Beat Saber"
+136 -9
View File
@@ -13,11 +13,12 @@ from rich.text import Text
from textual.coordinate import Coordinate from textual.coordinate import Coordinate
from textual.widgets import DataTable from textual.widgets import DataTable
from plugin_helper.bootstrap import _run_ipa from plugin_helper.bootstrap import _run_ipa, build_bootstrap_command
from plugin_helper.bsipa import check_bsipa_health
from plugin_helper.beatmods import by_version_id, normalize_mods from plugin_helper.beatmods import by_version_id, normalize_mods
from plugin_helper.checker import check_lock from plugin_helper.checker import check_lock
from plugin_helper.cli import installed_plugins_report, run from plugin_helper.cli import installed_plugins_report, run
from plugin_helper.config import load_local_config, resolve_runtime_config from plugin_helper.config import is_windows, load_local_config, resolve_runtime_config
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
@@ -218,7 +219,10 @@ state_dir = "config-state"
root = Path(tmp) root = Path(tmp)
xdg = root / "xdg-state" xdg = root / "xdg-state"
with patch.dict(os.environ, {"XDG_STATE_HOME": str(xdg)}, clear=True): with (
patch.dict(os.environ, {"XDG_STATE_HOME": str(xdg), "HOME": str(root)}, clear=True),
patch("plugin_helper.config.is_windows", return_value=False),
):
runtime = resolve_runtime_config(root=root) runtime = resolve_runtime_config(root=root)
self.assertEqual(runtime.state_root, xdg / "plugin-helper") self.assertEqual(runtime.state_root, xdg / "plugin-helper")
@@ -226,11 +230,15 @@ state_dir = "config-state"
def test_runtime_default_state_uses_home_local_state(self) -> None: def test_runtime_default_state_uses_home_local_state(self) -> None:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) root = Path(tmp)
env = {"HOME": str(root), "USERPROFILE": str(root)}
with patch.dict(os.environ, {}, clear=True): with (
patch.dict(os.environ, env, clear=True),
patch("plugin_helper.config.is_windows", return_value=False),
):
runtime = resolve_runtime_config(root=root) runtime = resolve_runtime_config(root=root)
self.assertEqual(runtime.state_root, Path.home() / ".local" / "state" / "plugin-helper") self.assertEqual(runtime.state_root, Path(root) / ".local" / "state" / "plugin-helper")
def test_no_args_prints_help_when_not_interactive(self) -> None: def test_no_args_prints_help_when_not_interactive(self) -> None:
output = StringIO() output = StringIO()
@@ -253,6 +261,8 @@ state_dir = "config-state"
run_menu.assert_called_once() run_menu.assert_called_once()
def test_run_ipa_timeout_returns_control(self) -> None: def test_run_ipa_timeout_returns_control(self) -> None:
if is_windows():
self.skipTest("POSIX process-group timeout behavior is Linux-specific")
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
result = _run_ipa( result = _run_ipa(
command=["python", "-c", "import time; time.sleep(30)"], command=["python", "-c", "import time; time.sleep(30)"],
@@ -263,6 +273,113 @@ state_dir = "config-state"
self.assertTrue(result["timedOut"]) self.assertTrue(result["timedOut"])
self.assertNotEqual(result["returncode"], 0) self.assertNotEqual(result["returncode"], 0)
def test_run_ipa_timeout_returns_control_on_windows(self) -> None:
if not is_windows():
self.skipTest("Windows subprocess timeout behavior is Windows-specific")
with tempfile.TemporaryDirectory() as tmp:
result = _run_ipa(
command=["python", "-c", "import time; time.sleep(30)"],
instance_path=Path(tmp),
timeout_seconds=1,
native=True,
)
self.assertTrue(result["timedOut"])
self.assertNotEqual(result["returncode"], 0)
def test_build_bootstrap_command_native(self) -> None:
ipa = Path("C:/Games/Beat Saber/IPA.exe")
self.assertEqual(build_bootstrap_command(ipa, native=True), [str(ipa), "-n"])
def test_build_bootstrap_command_proton(self) -> None:
ipa = Path("/tmp/1.44.1/IPA.exe")
proton = Path("/tmp/proton")
self.assertEqual(
build_bootstrap_command(ipa, proton=proton, native=False),
[str(proton), "run", str(ipa), "-n"],
)
def test_profile_config_resolves_windows_paths(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
config = root / "plugin-helper.windows.toml"
config.write_text(
"""
[[profiles]]
id = "windows"
label = "Native Windows BSManager"
instances_root = "~/BSInstances"
state_dir = ".state"
""".lstrip(),
encoding="utf-8",
)
local_config, loaded_path, loaded = load_local_config(config, root=root)
runtime = resolve_runtime_config(config_path_value=str(config), profile_id="windows", root=root)
auto_runtime = resolve_runtime_config(config_path_value=str(config), root=root)
self.assertTrue(loaded)
self.assertEqual(loaded_path, config)
self.assertEqual(len(local_config.profiles), 1)
self.assertEqual(local_config.profiles[0].id, "windows")
self.assertEqual(runtime.instances_roots, [Path("~/BSInstances").expanduser()])
self.assertEqual(runtime.state_root, root / ".state")
self.assertEqual(runtime.profile_id, "windows")
self.assertEqual(auto_runtime.state_root, root / ".state")
self.assertEqual(auto_runtime.profile_id, "windows")
def test_native_bootstrap_health_accepts_state_without_log(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),
},
)
result = check_bsipa_health(instance, state, "1.44.1")
self.assertTrue(result["ok"])
self.assertEqual(result["bootstrapMode"], "native")
def test_proton_bootstrap_health_still_requires_log(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": "proton",
"ipaExitCode": 0,
"ipaTimedOut": False,
"files": scan_bootstrap_files(instance),
},
)
result = check_bsipa_health(instance, state, "1.44.1")
self.assertFalse(result["ok"])
self.assertIn("missing Logs/_latest.log", result["messages"])
def test_plan_apply_and_uninstall_dll(self) -> None: def test_plan_apply_and_uninstall_dll(self) -> None:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp) work = Path(tmp)
@@ -689,13 +806,21 @@ sha256 = "{sha256_file(asset)}"
self.assertTrue(Path(result["archive"]).exists()) self.assertTrue(Path(result["archive"]).exists())
self.assertEqual(result["manifest"]["fileCount"], 1) self.assertEqual(result["manifest"]["fileCount"], 1)
def test_infer_windows_appdata_path_from_native_instance(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
instance = root / "Users" / "pleb" / "BSManager" / "BSInstances" / "1.44.1"
instance.mkdir(parents=True)
expected = root / "Users" / "pleb" / "AppData" / "LocalLow" / "Hyperbolic Magnetism" / "Beat Saber"
self.assertEqual(infer_windows_appdata_path(instance), expected)
def test_infer_windows_appdata_path_from_mounted_instance(self) -> None: def test_infer_windows_appdata_path_from_mounted_instance(self) -> None:
instance = Path("/home/pleb/Windows/Users/pleb/BSManager/BSInstances/1.44.1") instance = Path("/home/pleb/Windows/Users/pleb/BSManager/BSInstances/1.44.1")
expected = Path("/home/pleb/Windows/Users/pleb/AppData/LocalLow/Hyperbolic Magnetism/Beat Saber")
self.assertEqual( if is_windows():
infer_windows_appdata_path(instance), self.skipTest("POSIX mount paths are Linux-specific")
Path("/home/pleb/Windows/Users/pleb/AppData/LocalLow/Hyperbolic Magnetism/Beat Saber"), self.assertEqual(infer_windows_appdata_path(instance), expected)
)
def test_sync_windows_data_repo_copies_into_stable_backup_root(self) -> None: def test_sync_windows_data_repo_copies_into_stable_backup_root(self) -> None:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
@@ -747,6 +872,8 @@ sha256 = "{sha256_file(asset)}"
) )
def test_infer_appdata_path_uses_windows_or_proton(self) -> None: def test_infer_appdata_path_uses_windows_or_proton(self) -> None:
if is_windows():
self.skipTest("POSIX mount paths are Linux-specific")
windows_instance = Path("/home/pleb/Windows/Users/pleb/BSManager/BSInstances/1.44.1") windows_instance = Path("/home/pleb/Windows/Users/pleb/BSManager/BSInstances/1.44.1")
linux_instance = Path("/home/pleb/.local/share/BSManager/BSInstances/1.44.1") linux_instance = Path("/home/pleb/.local/share/BSManager/BSInstances/1.44.1")