diff --git a/.gitignore b/.gitignore
index 0de6235..c725230 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,4 +1,5 @@
-/__pycache__/
+**/__pycache__/
+/.superpowers/
/venv/
BlueStacksRootGUI*.spec
/dist/
diff --git a/adb_handler.py b/adb_handler.py
index 8d2a265..6309f92 100644
--- a/adb_handler.py
+++ b/adb_handler.py
@@ -160,3 +160,26 @@ def _p(msg):
"\"Apps and ADB\" and try again. The zip was also copied to the "
"instance's Download folder -- or flash it there: Modules -> Install "
"from storage -> Download." % (out or "unknown error"))
+
+
+def list_running_instances(adb_exe: str, instances, runner: Runner = _run) -> dict:
+ """Which of ``instances`` are currently reachable over ADB.
+
+ ``instances`` is an iterable of (unique_id, config_path, original_name).
+ Returns {unique_id: port} for every instance whose configured ADB port
+ is currently connected. Instances with no recorded port (never booted)
+ are skipped without spawning a process.
+ """
+ running = {}
+ for unique_id, config_path, name in instances:
+ port = instance_adb_port(config_path, name)
+ if port is None:
+ continue
+ try:
+ cp = runner([adb_exe, "connect", "127.0.0.1:%d" % port])
+ out = (cp.stdout or "") + (cp.stderr or "")
+ if "connected" in out.lower():
+ running[unique_id] = port
+ except Exception as exc: # noqa: BLE001 - one bad instance mustn't abort the rest
+ logger.warning("ADB probe failed for %s: %s", unique_id, exc)
+ return running
diff --git a/conftest.py b/conftest.py
new file mode 100644
index 0000000..fc69b95
--- /dev/null
+++ b/conftest.py
@@ -0,0 +1,28 @@
+# The empty-on-purpose part: this file's presence at the repo root makes pytest
+# insert the repo root onto sys.path (prepend import mode), so tests can
+# `import constants`, `import adb_handler`, and `from views import ...` the same
+# way main.py does, without a src/ layout or package install.
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _no_real_bluestacks_detection(monkeypatch):
+ """Make the live registry probe a no-op for every test.
+
+ MainWindow.__init__ schedules initialize_paths_and_instances() via
+ QTimer.singleShot(0, ...). It never runs during a test's synchronous body,
+ but pytest-qt spins the Qt event loop at teardown, which fires it. On a
+ machine WITHOUT BlueStacks that init returns early (no installations found);
+ on a dev machine WITH BlueStacks it proceeds to read the real engine state --
+ draining tests' monkeypatched finite _engine_state iterators (StopIteration)
+ and byte-scanning real binaries. That made the suite pass or fail depending
+ on the host. Stubbing detection to "nothing installed" makes the stray init
+ harmless everywhere; tests that need installations set them on the window
+ directly after construction.
+ """
+ monkeypatch.setattr(
+ "registry_handler.get_all_bluestacks_installations",
+ lambda: [],
+ raising=True,
+ )
diff --git a/main.py b/main.py
index 515afdc..2a7d2ca 100644
--- a/main.py
+++ b/main.py
@@ -5,35 +5,13 @@
import os
import logging
import tempfile
-from typing import Any
-from PyQt5.QtWidgets import (
- QApplication,
- QWidget,
- QPushButton,
- QLabel,
- QVBoxLayout,
- QHBoxLayout,
- QGridLayout,
- QGroupBox,
- QCheckBox,
- QMessageBox,
- QFileDialog,
-)
-from PyQt5.QtCore import Qt, QTimer, QThread, QObject, pyqtSignal, pyqtSlot
-from PyQt5.QtGui import QIcon
+from PyQt5.QtWidgets import QApplication
-
-import constants
-import registry_handler
-import config_handler
-import instance_handler
-import root_persistence
-import integrity_patch
-import su_patch_offline
-import ext4_symlink
-import adb_handler
import admin
+import constants
+from views import theme
+from views.main_window import MainWindow
# Log to console AND a file in the local temp dir. The file is important when
# the app is relaunched elevated: that process has its own (often invisible)
@@ -50,615 +28,6 @@
handlers=_handlers)
logger = logging.getLogger(__name__)
-def resource_path(relative_path):
- try:
- base_path = sys._MEIPASS
- except AttributeError:
- base_path = os.path.abspath(".")
- return os.path.join(base_path, relative_path)
-
-
-class _OpWorker(QObject):
- """Runs a blocking job(progress) on a worker thread, relaying progress text."""
- progress = pyqtSignal(str)
- done = pyqtSignal(bool, str)
-
- def __init__(self, job):
- super().__init__()
- self._job = job
-
- @pyqtSlot()
- def run(self):
- try:
- summary = self._job(self.progress.emit)
- self.done.emit(True, summary)
- except Exception as exc: # noqa: BLE001
- logger.exception("Background operation failed")
- self.done.emit(False, str(exc))
-
-
-class BluestacksRootToggle(QWidget):
- """Main application window for toggling BlueStacks root and R/W settings."""
-
- # Emitted from the worker thread to surface a modal notice on the UI thread.
- show_notice = pyqtSignal(str, str) # (title, message)
-
- def __init__(self):
- super().__init__()
- self.installations: list[registry_handler.Installation] = []
- self.instance_data: dict[str, dict[str, Any]] = {}
- self.instance_checkboxes: dict[str, dict[str, Any]] = {}
-
- # Queued (cross-thread) so background jobs can pop a dialog safely.
- self.show_notice.connect(self._show_notice)
- self.setWindowTitle(f"{constants.APP_NAME} v{constants.APP_VERSION}")
- self._set_icon()
- self.status_refresh_timer = QTimer(self)
- self.status_refresh_timer.timeout.connect(
- lambda: self.update_instance_statuses(preserve_selection=True)
- )
- self.init_ui()
- QTimer.singleShot(0, self.initialize_paths_and_instances)
-
- def _set_icon(self):
- try:
- icon_path = resource_path(constants.ICON_FILENAME)
- app_icon = QIcon(icon_path)
- if not app_icon.isNull(): self.setWindowIcon(app_icon)
- except Exception as e:
- logger.error(f"Error setting window icon: {e}")
-
- def init_ui(self) -> None:
- main_layout = QVBoxLayout()
- self.path_label = QLabel("BlueStacks Path: Loading...")
- self.path_label.setWordWrap(True)
- main_layout.addWidget(self.path_label)
-
- self.instance_group = QGroupBox("Instances")
- self.instance_layout = QGridLayout()
- self.instance_layout.setColumnStretch(0, 4)
- self.instance_layout.setColumnStretch(1, 1)
- self.instance_layout.setColumnStretch(2, 1)
- self.instance_layout.setHorizontalSpacing(15)
- self.instance_group.setLayout(self.instance_layout)
- main_layout.addWidget(self.instance_group)
-
- button_layout = QHBoxLayout()
- self.root_toggle_button = QPushButton("Toggle Root")
- self.root_toggle_button.clicked.connect(self.handle_toggle_root)
- self.rw_toggle_button = QPushButton("Toggle R/W")
- self.rw_toggle_button.clicked.connect(self.handle_toggle_rw)
- button_layout.addWidget(self.root_toggle_button)
- button_layout.addWidget(self.rw_toggle_button)
- main_layout.addLayout(button_layout)
-
- # --- Engine patch (5.22.150.1014+ only) ----------------------------
- # One button that both SHOWS the engine state and acts on it: when the
- # engine isn't patched it offers "Patch"; once patched it turns green and
- # offers "Undo". The engine patch flips _isDiskVerificationRequired() in
- # HD-Player.exe to 0 (disables the "illegally tampered" shutdown so a
- # rooted /system can boot). It's per-install, shared by every instance.
- # Hidden entirely on older builds (classic conf rooting).
- self.engine_button = QPushButton("")
- self.engine_button.clicked.connect(self.handle_engine_button)
- main_layout.addWidget(self.engine_button)
- self.engine_button.setVisible(False) # shown once installations are read
-
- # --- Install a Magisk/Kitsune module directly ---------------------
- # Pushes a module .zip into a RUNNING instance and flashes it over an ADB
- # root shell (magisk --install-module) -- sidesteps BlueStacks' file
- # picker handing Magisk an "Invalid Uri" it can't open. Falls back to
- # dropping the zip in Download if the root shell isn't reachable.
- self.sideload_button = QPushButton("Install Magisk Module (.zip) into a running instance")
- self.sideload_button.setToolTip(
- "Select one running instance above, choose a module .zip, and it's "
- "pushed in and flashed via Magisk automatically. Close and reopen the "
- "instance afterwards to activate it. The instance must be running."
- )
- self.sideload_button.clicked.connect(self.handle_sideload_module)
- main_layout.addWidget(self.sideload_button)
-
- self.status_label = QLabel("Ready")
- self.status_label.setAlignment(Qt.AlignCenter)
- main_layout.addWidget(self.status_label)
-
- self.setLayout(main_layout)
- self.setMinimumWidth(550)
-
- def initialize_paths_and_instances(self) -> None:
- logger.info("Initializing BlueStacks paths and instances...")
- self._clear_instance_widgets()
- self.installations = registry_handler.get_all_bluestacks_installations()
- if not self.installations:
- self.path_label.setText("No BlueStacks installations found.")
- return
-
- path_details = ["Installations Found:"]
- for inst in self.installations:
- ver = ".".join(map(str, inst["version"])) if inst.get("version") else "?"
- path_details.append(f" - {inst['source']} v{ver}: {inst['user_path']}")
- self.path_label.setText("\n".join(path_details))
-
- self._refresh_patch_ui()
- self.update_instance_statuses(preserve_selection=False)
- self.status_refresh_timer.start(constants.REFRESH_INTERVAL_MS)
-
- def _refresh_patch_ui(self) -> None:
- """Drive the single engine button from the current patch state.
-
- Shown only when a 5.22.150.1014+ install exists. The button's label,
- colour, and action all reflect state: unpatched -> "Patch", patched ->
- green "Undo", partial -> "re-patch", unknown -> disabled.
- """
- has_patch_build = any(i.get("patch_mode") for i in self.installations)
- self.engine_button.setVisible(has_patch_build)
- if not has_patch_build:
- return
- state = self._engine_state()
- text, color, tip, enabled = {
- "patched": ("✓ Engine patched — click to Undo (restore originals)",
- "#2e7d32", "Restores HD-Player.exe + HD-MultiInstanceManager.exe "
- "from the .prepatch.bak backups.", True),
- "unpatched": ("Patch BlueStacks Engine (required for root)",
- "#c62828", "Patches HD-Player.exe (+ HD-MultiInstanceManager.exe) "
- "to disable the integrity shutdown so rooted instances boot. Do "
- "this once, then Toggle Root per instance.", True),
- "partial": ("⚠ Engine partially patched — click to finish patching",
- "#e65100", "Some engine binaries aren't patched yet — re-run the "
- "patch to bring them all up to date.", True),
- "unknown": ("Engine status unknown (unrecognized build)",
- "#616161", "Couldn't read the engine patch state for this build.",
- False),
- }[state]
- self.engine_button.setText(text)
- self.engine_button.setToolTip(tip)
- self.engine_button.setStyleSheet(f"color: {color}; font-weight: bold;")
- # Remember the action for the click handler; disabled only on unknown
- # builds (busy-state disabling is handled by _set_busy during ops).
- self._engine_action = "restore" if state == "patched" else "patch"
- self.engine_button.setEnabled(enabled)
-
- def _engine_state(self) -> str:
- """One of 'patched' | 'unpatched' | 'partial' | 'unknown' across installs."""
- states = [integrity_patch.installation_patched(i["install_path"])
- for i in self.installations
- if i.get("patch_mode") and i.get("install_path")
- and os.path.isdir(i["install_path"])]
- if states and all(s is True for s in states):
- return "patched"
- if states and all(s is False for s in states):
- return "unpatched"
- if any(s is True for s in states):
- return "partial"
- return "unknown"
-
- def handle_engine_button(self) -> None:
- """Patch or restore the engine, per the current (freshly-read) state."""
- # Re-read state at click time so a stale label can't send us the wrong way.
- self._refresh_patch_ui()
- if getattr(self, "_engine_action", "patch") == "restore":
- self.handle_restore_patches()
- else:
- self.handle_apply_patches()
-
- def update_instance_data(self) -> None:
- if not self.installations: return
-
- all_found_instances: dict[str, dict[str, Any]] = {}
- for inst in self.installations:
- source_id, config_path, data_path = inst["source"], inst["config_path"], inst["data_path"]
- install_path = inst.get("install_path") # for locating HD-Adb.exe (sideload)
- patch_mode = inst.get("patch_mode", False) # 5.22.150.1014+ uses the patches
- root_info = config_handler.get_complete_root_statuses(config_path)
- instance_root_statuses = root_info['instance_statuses']
-
- disk_instances = {entry for entry in (os.listdir(data_path) if os.path.isdir(data_path) else []) if os.path.isdir(os.path.join(data_path, entry))}
- all_instance_names = set(instance_root_statuses.keys()) | disk_instances
-
- for name in sorted(all_instance_names):
- unique_id = f"{name} ({source_id})"
- instance_dir_path = os.path.join(data_path, name)
-
- rw_mode = constants.MODE_UNKNOWN
- if os.path.isdir(instance_dir_path):
- is_readonly = instance_handler.is_instance_readonly(instance_dir_path)
- if is_readonly is True: rw_mode = constants.MODE_READONLY
- elif is_readonly is False: rw_mode = constants.MODE_READWRITE
-
- individual_root_on = instance_root_statuses.get(name, False)
- if patch_mode:
- # 5.22.150.1014+: app root = the guest su binary actually patched
- # (tracked by the Data.vhdx sidecar). enable_root_access only makes
- # /system/xbin/su appear; apps still need the su patched because
- # they cannot reach /dev/bstvmsg for the dev-mode escape, so the
- # sidecar -- not the conf flag -- is the real indicator.
- effective_root_status = su_patch_offline.instance_root_state(instance_dir_path)
- else:
- # Classic builds: the per-instance enable_root_access flag is the
- # persistent root control (it exposes the guest `su`). Do NOT gate
- # status on bst.feature.rooting -- BlueStacks resets that global flag
- # to 0 on launch while root stays live, which would wrongly show a
- # rooted instance as "Off". The toggle re-sets feature.rooting when
- # enabling.
- effective_root_status = individual_root_on
-
- all_found_instances[unique_id] = {
- "original_name": name,
- "config_path": config_path,
- "data_path": instance_dir_path,
- "install_path": install_path,
- "rw_mode": rw_mode,
- "root_enabled": effective_root_status,
- "individual_root_status": individual_root_on,
- "patch_mode": patch_mode,
- }
-
- self.instance_data = {
- uid: data for uid, data in all_found_instances.items()
- if data["rw_mode"] != constants.MODE_UNKNOWN
- }
-
- logger.debug(f"Instance data updated. Displaying {len(self.instance_data)} instances.")
-
- def _clear_instance_widgets(self):
- while self.instance_layout.count():
- item = self.instance_layout.takeAt(0)
- if item and item.widget():
- item.widget().deleteLater()
-
- def update_instance_checkboxes(self, preserve_selection: bool = True):
- previous_selection = {uid for uid, w in self.instance_checkboxes.items() if w["checkbox"].isChecked()} if preserve_selection else set()
-
- self._clear_instance_widgets()
- self.instance_checkboxes = {}
-
- for row, unique_id in enumerate(sorted(self.instance_data.keys())):
- data = self.instance_data[unique_id]
- checkbox = QCheckBox(unique_id)
- checkbox.setChecked(unique_id in previous_selection)
-
- root_text = "On" if data["root_enabled"] else "Off"
- rw_text = "On" if data["rw_mode"] == constants.MODE_READWRITE else "Off"
-
- self.instance_layout.addWidget(checkbox, row, 0)
- self.instance_layout.addWidget(QLabel(f"Root: {root_text}"), row, 1)
- self.instance_layout.addWidget(QLabel(f"R/W: {rw_text}"), row, 2)
- self.instance_checkboxes[unique_id] = {"checkbox": checkbox}
-
- def _toggle_single_instance_root(self, unique_id, progress=None):
- """Dispatch root toggle by build: 5.22.150.1014+ needs the su-binary patch
- AND enable_root_access; older builds just toggle the conf flag."""
- if self.instance_data[unique_id].get("patch_mode"):
- self._toggle_root_patchmode(unique_id, progress)
- else:
- self._toggle_root_conf(unique_id, progress)
-
- def _toggle_root_patchmode(self, unique_id, progress=None):
- """5.22.150.1014+ app root needs BOTH halves:
- 1. enable_root_access=1, so BlueStacks exposes /system/xbin/su, and
- 2. the guest su binary patched (isDeveloperMode->true) inside Data.vhdx.
- An app cannot reach /dev/bstvmsg, so the host-side dev-mode escape (the
- engine patch) only grants a root *shell* -- apps stay denied until the su
- binary itself is patched. The engine patch is still required (integrity
- bypass + keeping enable_root_access from being reset).
-
- su only exists in Data.vhdx after the instance has booted once with root
- enabled; if it isn't there yet, su_patch_offline reports that and you
- re-toggle after a single boot.
- """
- instance = self.instance_data[unique_id]
- turn_on = not instance["root_enabled"]
- config_path = instance["config_path"]
- key = f"{constants.INSTANCE_PREFIX}{instance['original_name']}{constants.ENABLE_ROOT_KEY}"
- if turn_on:
- if progress:
- progress("Part 1/2: enabling root access in bluestacks.conf...")
- config_handler.modify_config_file(config_path, key, "1")
- config_handler.modify_config_file(config_path, constants.FEATURE_ROOTING_KEY, "1")
- if progress:
- progress("Part 2/2: patching guest su in Data.vhdx...")
- results = su_patch_offline.set_instance_root(instance["data_path"], True, progress)
- # If no guest su was patched (sidecar not written), Android hasn't
- # generated su yet -- the instance must be booted once first. This is
- # easy to miss in the status bar, so tell the user with a dialog.
- if not su_patch_offline.instance_root_state(instance["data_path"]):
- self.show_notice.emit(
- "Boot this instance once first",
- "%s hasn't generated its root files yet, so there was nothing "
- "to patch.\n\nStart this instance in BlueStacks, let it fully "
- "reach the home screen, then close it completely and click "
- "\"Toggle Root\" again." % unique_id,
- )
- else:
- if progress:
- progress("Part 1/2: restoring guest su in Data.vhdx...")
- results = su_patch_offline.set_instance_root(instance["data_path"], False, progress)
- if progress:
- progress("Part 2/2: disabling root access in bluestacks.conf...")
- config_handler.modify_config_file(config_path, key, "0")
- logger.info("Root %s (patch-mode) for %s: %s", "ON" if turn_on else "OFF",
- unique_id, " | ".join(results))
-
- def _toggle_root_conf(self, unique_id, progress=None):
- """Root toggle via bluestacks.conf (enable_root_access + feature.rooting).
-
- Classic builds expose only /system/xbin/bstk/su (a root *shell*); apps
- cannot see it because it is not on PATH. So we also add a
- /system/xbin/su -> bstk/su symlink offline in Root.vhd for app-visible
- root. That step is best-effort: if it fails (tools missing, no Root.vhd,
- unexpected layout) the conf-based shell root still applies.
- """
- if progress:
- progress("updating bluestacks.conf...")
- instance = self.instance_data[unique_id]
- config_path, original_name = instance["config_path"], instance["original_name"]
- is_currently_on = instance["root_enabled"]
- setting_key = f"{constants.INSTANCE_PREFIX}{original_name}{constants.ENABLE_ROOT_KEY}"
- if is_currently_on:
- config_handler.modify_config_file(config_path, setting_key, "0")
- any_other_rooted = any(
- d.get("individual_root_status", False)
- for uid, d in self.instance_data.items()
- if uid != unique_id and d["config_path"] == config_path)
- if not any_other_rooted:
- config_handler.modify_config_file(config_path, constants.FEATURE_ROOTING_KEY, "0")
- self._set_classic_app_su(instance, False, progress)
- else:
- config_handler.modify_config_file(config_path, setting_key, "1")
- config_handler.modify_config_file(config_path, constants.FEATURE_ROOTING_KEY, "1")
- self._set_classic_app_su(instance, True, progress)
- logger.info(f"Root toggle (conf) processed for {unique_id}")
-
- def _set_classic_app_su(self, instance, turn_on, progress=None):
- """Best-effort /system/xbin/su symlink in Root.vhd for app-visible root.
-
- Only meaningful on classic builds whose Root.vhd ships bstk/su. Any
- failure is logged and surfaced via progress but never fails the toggle.
- """
- if not ext4_symlink.tools_available():
- logger.info("app-su: bundled e2fsprogs not present; skipping symlink")
- return
- try:
- if turn_on:
- results = ext4_symlink.add_su_symlink(instance["data_path"], progress)
- else:
- results = ext4_symlink.remove_su_symlink(instance["data_path"], progress)
- logger.info("app-su %s: %s", "ON" if turn_on else "OFF", " | ".join(results))
- except Exception as exc: # noqa: BLE001 - never break the conf toggle
- logger.warning("app-su symlink step failed: %s", exc)
- if progress:
- progress("app-root symlink skipped: %s" % exc)
-
- def _toggle_single_instance_rw(self, unique_id, progress=None):
- instance = self.instance_data[unique_id]
- new_mode = constants.MODE_READONLY if instance["rw_mode"] == constants.MODE_READWRITE else constants.MODE_READWRITE
- if progress:
- progress("setting disk to %s..." % new_mode)
- instance_handler.modify_instance_files(instance["data_path"], new_mode)
- logger.info(f"R/W toggled for instance: {unique_id} to {new_mode}")
-
- def _perform_operation(self, operation_func, operation_name):
- selected_ids = [uid for uid, w in self.instance_checkboxes.items() if w["checkbox"].isChecked()]
- if not selected_ids:
- QMessageBox.information(self, "No Selection", f"No instances selected to toggle {operation_name}.")
- return
- total = len(selected_ids)
-
- def job(progress):
- progress("Closing BlueStacks...")
- logger.info("Terminating BlueStacks before %s of %d instance(s)", operation_name, total)
- instance_handler.terminate_bluestacks()
- QThread.msleep(constants.PROCESS_TERMINATION_WAIT_MS)
- for idx, uid in enumerate(selected_ids, 1):
- prefix = "Part %d/%d (%s)" % (idx, total, uid)
- logger.info("---- %s: %s ----", prefix, operation_name)
- progress("%s: %s..." % (prefix, operation_name))
- operation_func(uid, lambda m, _p=prefix: progress("%s: %s" % (_p, m)))
- return "%s complete for %d instance(s). Restart the instance(s)." % (operation_name, total)
-
- self._run_async(job, f"Toggling {operation_name}...")
-
- def handle_toggle_root(self): self._perform_operation(self._toggle_single_instance_root, "Root")
- def handle_toggle_rw(self): self._perform_operation(self._toggle_single_instance_rw, "R/W")
-
- def handle_sideload_module(self) -> None:
- """Push a chosen module .zip into one running instance and flash it.
-
- Unlike every other action here, this needs the instance RUNNING (its ADB
- port must be open). We target exactly one selected instance so the module
- installs into the right guest.
- """
- selected = [uid for uid, w in self.instance_checkboxes.items() if w["checkbox"].isChecked()]
- if len(selected) != 1:
- QMessageBox.information(
- self, "Select one instance",
- "Tick exactly one (running) instance to receive the module, then "
- "click Sideload again.")
- return
- uid = selected[0]
- instance = self.instance_data[uid]
-
- install_dirs = [i.get("install_path") for i in self.installations]
- adb_exe = adb_handler.find_adb(install_dirs)
- if not adb_exe:
- QMessageBox.warning(
- self, "ADB not found",
- "Couldn't find HD-Adb.exe in the BlueStacks install folder, so the "
- "module can't be pushed.")
- return
-
- zip_path, _ = QFileDialog.getOpenFileName(
- self, "Select Magisk/Kitsune module", "", "Module archives (*.zip)")
- if not zip_path:
- return
-
- port = adb_handler.instance_adb_port(instance["config_path"], instance["original_name"])
-
- def job(progress):
- msg = adb_handler.install_module(adb_exe, port, zip_path, progress=progress)
- self.show_notice.emit("Module installed", msg)
- return "Module installed. Close and reopen the instance to activate it."
-
- self._run_async(job, "Installing %s..." % os.path.basename(zip_path))
-
- # ---- Background-thread plumbing (keeps the UI responsive) -------------
- def _action_buttons(self):
- return [self.root_toggle_button, self.rw_toggle_button,
- self.engine_button, self.sideload_button]
-
- def _set_busy(self, busy):
- for b in self._action_buttons():
- b.setEnabled(not busy)
- if busy:
- self.status_refresh_timer.stop()
-
- def _run_async(self, job, start_text):
- """Run job(progress) on a worker thread; relay progress to the status bar."""
- if getattr(self, "_op_thread", None) is not None:
- QMessageBox.information(self, "Busy", "An operation is already running.")
- return
- self._set_busy(True)
- self.status_label.setText(start_text)
- logger.info("==== %s ====", start_text)
- self._op_thread = QThread(self)
- self._op_worker = _OpWorker(job)
- self._op_worker.moveToThread(self._op_thread)
- self._op_thread.started.connect(self._op_worker.run)
- self._op_worker.progress.connect(self._on_async_progress)
- self._op_worker.done.connect(self._on_async_done)
- self._op_worker.done.connect(self._op_thread.quit)
- self._op_thread.finished.connect(self._cleanup_async)
- self._op_thread.start()
-
- def _on_async_progress(self, msg):
- self.status_label.setText(msg)
-
- @pyqtSlot(str, str)
- def _show_notice(self, title, message):
- """Show a modal info dialog. Runs on the UI thread via a queued signal."""
- QMessageBox.information(self, title, message)
-
- def _on_async_done(self, ok, summary):
- self._set_busy(False)
- self.status_label.setText(summary if ok else f"Error: {summary}")
- logger.info("Operation finished (ok=%s): %s", ok, summary)
- self.update_instance_statuses(preserve_selection=True)
- self._refresh_patch_ui() # reflect the new engine-patch state after a patch/restore
- self.status_refresh_timer.start(constants.REFRESH_INTERVAL_MS)
-
- def _cleanup_async(self):
- if getattr(self, "_op_worker", None) is not None:
- self._op_worker.deleteLater()
- if getattr(self, "_op_thread", None) is not None:
- self._op_thread.deleteLater()
- self._op_worker = None
- self._op_thread = None
-
- # ---- Engine patch (5.22.150.1014+) ------------------------------------
- def _install_dirs_or_warn(self) -> list[str] | None:
- """Return BlueStacks install dirs, after ensuring admin rights.
-
- Returns None (and shows the appropriate dialog) if not elevated or no
- install directory was found.
- """
- # Writing into Program Files needs admin. Offer to relaunch elevated.
- if not admin.is_admin():
- choice = QMessageBox.question(
- self, "Administrator required",
- "Patching the BlueStacks binaries requires administrator rights.\n\n"
- "Relaunch this app as administrator now?",
- QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes)
- if choice == QMessageBox.Yes and admin.relaunch_as_admin():
- QApplication.quit()
- return None
-
- # Only patch 5.22.150.1014+ installs (older builds don't need it).
- install_dirs = sorted({inst.get("install_path") for inst in self.installations
- if inst.get("patch_mode") and inst.get("install_path")
- and os.path.isdir(inst["install_path"])})
- if not install_dirs:
- QMessageBox.warning(self, "Not applicable",
- "No BlueStacks 5.22.150.1014+ install found that needs the "
- "engine patch (older builds use classic conf rooting).")
- return None
- return install_dirs
-
- def handle_apply_patches(self) -> None:
- """One click: patch BlueStacks so apps get root and root stays on.
-
- * HD-Player.exe -> _isDiskVerificationRequired() returns 0, which both
- disables the "illegally tampered" shutdown AND turns on Developer Mode
- (so the guest `su` grants root to every app -- no /system edits needed).
- * HD-MultiInstanceManager.exe -> stop resetting enable_root_access to 0.
- """
- install_dirs = self._install_dirs_or_warn()
- if install_dirs is None:
- return
-
- confirm = QMessageBox.question(
- self, "Enable Root",
- "This patches your BlueStacks install to enable root:\n\n"
- " - Unlocks root for apps (Developer Mode) and removes the\n"
- " \"illegally tampered\" shutdown [HD-Player.exe]\n"
- " - Keeps root enabled across launches [HD-MultiInstanceManager.exe]\n\n"
- "No guest /system changes are needed. A .prepatch.bak backup is made "
- "next to each binary (use \"Undo Root Patch\" to revert). All BlueStacks "
- "processes will be closed first.\n\nContinue?",
- QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
- if confirm != QMessageBox.Yes:
- return
-
- total = len(install_dirs)
-
- def job(progress):
- progress("Closing BlueStacks...")
- instance_handler.terminate_bluestacks()
- QThread.msleep(constants.PROCESS_TERMINATION_WAIT_MS)
- all_results = []
- for i, install_dir in enumerate(install_dirs, 1):
- progress("Patching engine %d/%d: HD-Player.exe..." % (i, total))
- all_results.extend(integrity_patch.patch_installation(install_dir))
- progress("Patching engine %d/%d: HD-MultiInstanceManager.exe..." % (i, total))
- all_results.extend(root_persistence.patch_root_persistence(install_dir))
- for line in all_results:
- logger.info(" %s", line)
- logger.info("Engine patched. Next: Toggle Root per instance, then restart. "
- "Disable BstHdUpdaterSvc so an update doesn't re-lock it.")
- return "Engine patched. Now Toggle Root per instance, then restart."
-
- self._run_async(job, "Patching BlueStacks engine...")
-
- def handle_restore_patches(self) -> None:
- """Restore HD-Player.exe + HD-MultiInstanceManager.exe from backups."""
- install_dirs = self._install_dirs_or_warn()
- if install_dirs is None:
- return
- if QMessageBox.question(
- self, "Undo Root Patch",
- "Restore the original BlueStacks binaries from their .prepatch.bak "
- "backups? All BlueStacks processes will be closed first.",
- QMessageBox.Yes | QMessageBox.No, QMessageBox.No) != QMessageBox.Yes:
- return
-
- total = len(install_dirs)
-
- def job(progress):
- progress("Closing BlueStacks...")
- instance_handler.terminate_bluestacks()
- QThread.msleep(constants.PROCESS_TERMINATION_WAIT_MS)
- all_results = []
- for i, install_dir in enumerate(install_dirs, 1):
- progress("Restoring engine %d/%d..." % (i, total))
- all_results.extend(integrity_patch.patch_installation(install_dir, restore=True))
- all_results.extend(root_persistence.patch_root_persistence(install_dir, restore=True))
- for line in all_results:
- logger.info(" %s", line)
- return "Engine binaries restored from backup."
-
- self._run_async(job, "Restoring BlueStacks engine...")
- def update_instance_statuses(self, preserve_selection: bool = True): self.update_instance_data(); self.update_instance_checkboxes(preserve_selection)
- def closeEvent(self, event): self.status_refresh_timer.stop(); event.accept()
-
if __name__ == "__main__":
# Patching Program Files binaries and killing BlueStacks processes need
# admin rights. If not elevated, request elevation via UAC and relaunch;
@@ -669,7 +38,8 @@ def closeEvent(self, event): self.status_refresh_timer.stop(); event.accept()
logger.info("Starting %s (admin=%s, log=%s)",
constants.APP_NAME, admin.is_admin(), LOG_PATH)
app = QApplication(sys.argv)
- window = BluestacksRootToggle()
+ theme.apply_theme(app, theme.load_saved_theme())
+ window = MainWindow()
window.show()
sys.exit(app.exec_())
except SystemExit:
diff --git a/requirements-dev.txt b/requirements-dev.txt
new file mode 100644
index 0000000..bf785db
--- /dev/null
+++ b/requirements-dev.txt
@@ -0,0 +1,2 @@
+pytest>=8.0
+pytest-qt>=4.4
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/test_adb_handler_running.py b/tests/test_adb_handler_running.py
new file mode 100644
index 0000000..5a0a762
--- /dev/null
+++ b/tests/test_adb_handler_running.py
@@ -0,0 +1,49 @@
+from types import SimpleNamespace
+
+from adb_handler import list_running_instances
+
+
+def _fake_runner(responses):
+ calls = []
+
+ def runner(cmd):
+ calls.append(cmd)
+ key = cmd[-1] if len(cmd) >= 2 and cmd[1] == "connect" else None
+ stdout = responses.get(key, "")
+ return SimpleNamespace(stdout=stdout, stderr="", returncode=0)
+
+ runner.calls = calls
+ return runner
+
+
+def test_list_running_instances_returns_connected_ports(tmp_path):
+ config_path = tmp_path / "bluestacks.conf"
+ config_path.write_text(
+ 'bst.instance.Pie64.status.adb_port="5555"\n'
+ 'bst.instance.Idle.status.adb_port="5565"\n'
+ )
+ responses = {
+ "127.0.0.1:5555": "connected to 127.0.0.1:5555",
+ "127.0.0.1:5565": "failed to connect to 127.0.0.1:5565",
+ }
+ runner = _fake_runner(responses)
+ instances = [
+ ("Pie64 (Normal)", str(config_path), "Pie64"),
+ ("Idle (Normal)", str(config_path), "Idle"),
+ ]
+
+ result = list_running_instances("HD-Adb.exe", instances, runner=runner)
+
+ assert result == {"Pie64 (Normal)": 5555}
+
+
+def test_list_running_instances_skips_instances_with_no_recorded_port(tmp_path):
+ config_path = tmp_path / "bluestacks.conf"
+ config_path.write_text("") # no adb_port keys at all
+ runner = _fake_runner({})
+ instances = [("NeverBooted (Normal)", str(config_path), "NeverBooted")]
+
+ result = list_running_instances("HD-Adb.exe", instances, runner=runner)
+
+ assert result == {}
+ assert runner.calls == [] # never attempted an adb connect
diff --git a/tests/test_dashboard_page.py b/tests/test_dashboard_page.py
new file mode 100644
index 0000000..d014288
--- /dev/null
+++ b/tests/test_dashboard_page.py
@@ -0,0 +1,64 @@
+from PyQt5.QtCore import Qt
+
+from views.dashboard_page import DashboardPage
+
+
+def test_alert_hidden_by_default(qtbot):
+ page = DashboardPage()
+ qtbot.addWidget(page)
+ page.show()
+ assert page.alert_label.isVisible() is False
+ assert page.repatch_button.isVisible() is False
+
+
+def test_set_update_reverted_shows_alert(qtbot):
+ page = DashboardPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_update_reverted(True)
+ assert page.alert_label.isVisible() is True
+ assert page.repatch_button.isVisible() is True
+
+
+def test_clicking_repatch_emits_signal(qtbot):
+ page = DashboardPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_update_reverted(True)
+ with qtbot.waitSignal(page.repatch_requested, timeout=1000):
+ qtbot.mouseClick(page.repatch_button, Qt.LeftButton)
+
+
+def test_set_engine_state_updates_button(qtbot):
+ page = DashboardPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_engine_state(True, "Patch BlueStacks Engine", "tip", "#c62828", True)
+ assert page.engine_button.isVisible() is True
+ assert page.engine_button.text() == "Patch BlueStacks Engine"
+ assert page.engine_button.isEnabled() is True
+
+
+def test_clicking_engine_button_emits_signal(qtbot):
+ page = DashboardPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_engine_state(True, "Patch", "tip", "#c62828", True)
+ with qtbot.waitSignal(page.patch_engine_requested, timeout=1000):
+ qtbot.mouseClick(page.engine_button, Qt.LeftButton)
+
+
+def test_set_rooted_count_updates_label(qtbot):
+ page = DashboardPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_rooted_count(2, 5)
+ assert page.stat_label.text() == "2 / 5 instances rooted"
+
+
+def test_set_paths_text_updates_label(qtbot):
+ page = DashboardPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_paths_text("Installations Found:\n - NXT v5.22.232.1002: C:/x")
+ assert page.path_label.text() == "Installations Found:\n - NXT v5.22.232.1002: C:/x"
diff --git a/tests/test_engine_rules.py b/tests/test_engine_rules.py
new file mode 100644
index 0000000..a44d7fe
--- /dev/null
+++ b/tests/test_engine_rules.py
@@ -0,0 +1,66 @@
+from views.engine_rules import blocked_for_root_toggle, update_was_reverted
+
+
+def test_blocked_for_root_toggle_blocks_unpatched_patch_mode_instances():
+ selected = {
+ "Pie64 (Normal)": {"patch_mode": True},
+ "MSI_Pie (MSI)": {"patch_mode": False},
+ }
+ assert blocked_for_root_toggle(selected, "unpatched") == ["Pie64 (Normal)"]
+
+
+def test_blocked_for_root_toggle_allows_everything_when_patched():
+ selected = {
+ "Pie64 (Normal)": {"patch_mode": True},
+ "MSI_Pie (MSI)": {"patch_mode": False},
+ }
+ assert blocked_for_root_toggle(selected, "patched") == []
+
+
+def test_blocked_for_root_toggle_ignores_classic_instances():
+ selected = {"MSI_Pie (MSI)": {"patch_mode": False}}
+ assert blocked_for_root_toggle(selected, "unpatched") == []
+
+
+def test_blocked_for_root_toggle_blocks_turning_root_on():
+ selected = {"Pie64 (Normal)": {"patch_mode": True, "root_enabled": False}}
+ assert blocked_for_root_toggle(selected, "unpatched") == ["Pie64 (Normal)"]
+
+
+def test_blocked_for_root_toggle_allows_turning_root_off():
+ # A currently-rooted patch-mode instance being toggled OFF is not blocked:
+ # disabling root needs no engine patch, so a stale root after an
+ # auto-update revert must still be clearable.
+ selected = {"Pie64 (Normal)": {"patch_mode": True, "root_enabled": True}}
+ assert blocked_for_root_toggle(selected, "unpatched") == []
+
+
+def test_blocked_for_root_toggle_blocks_on_partial():
+ selected = {"Pie64 (Normal)": {"patch_mode": True, "root_enabled": False}}
+ assert blocked_for_root_toggle(selected, "partial") == ["Pie64 (Normal)"]
+
+
+def test_blocked_for_root_toggle_allows_when_engine_state_unknown():
+ # Unrecognized build: we can't read patch state, so don't trap the user.
+ selected = {"Pie64 (Normal)": {"patch_mode": True, "root_enabled": False}}
+ assert blocked_for_root_toggle(selected, "unknown") == []
+
+
+def test_update_was_reverted_true_when_patch_lost_with_root_active():
+ assert update_was_reverted("patched", "unpatched", any_rooted=True) is True
+
+
+def test_update_was_reverted_false_when_nothing_was_rooted():
+ assert update_was_reverted("patched", "unpatched", any_rooted=False) is False
+
+
+def test_update_was_reverted_false_when_still_patched():
+ assert update_was_reverted("patched", "patched", any_rooted=True) is False
+
+
+def test_update_was_reverted_false_when_never_was_patched():
+ assert update_was_reverted("unpatched", "unpatched", any_rooted=True) is False
+
+
+def test_update_was_reverted_false_on_first_ever_check():
+ assert update_was_reverted(None, "unpatched", any_rooted=True) is False
diff --git a/tests/test_instances_page.py b/tests/test_instances_page.py
new file mode 100644
index 0000000..4ec3c6c
--- /dev/null
+++ b/tests/test_instances_page.py
@@ -0,0 +1,82 @@
+from PyQt5.QtCore import Qt
+
+import constants
+from views.instances_page import InstancesPage
+
+
+def test_banner_hidden_by_default(qtbot):
+ page = InstancesPage()
+ qtbot.addWidget(page)
+ page.show()
+ assert page.banner_label.isVisible() is False
+
+
+def test_set_engine_locked_banner_shows_banner(qtbot):
+ page = InstancesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_engine_locked_banner(True)
+ assert page.banner_label.isVisible() is True
+ assert page.banner_fix_button.isVisible() is True
+
+
+def test_set_instances_creates_one_row_per_instance(qtbot):
+ page = InstancesPage()
+ qtbot.addWidget(page)
+ page.show()
+ data = {
+ "Pie64 (Normal)": {"root_enabled": True, "rw_mode": constants.MODE_READWRITE},
+ "MSI_Pie (MSI)": {"root_enabled": False, "rw_mode": constants.MODE_READONLY},
+ }
+ page.set_instances(data)
+ assert set(page.checkboxes.keys()) == {"Pie64 (Normal)", "MSI_Pie (MSI)"}
+
+
+def test_selected_ids_reflects_checked_boxes(qtbot):
+ page = InstancesPage()
+ qtbot.addWidget(page)
+ page.show()
+ data = {"Pie64 (Normal)": {"root_enabled": True, "rw_mode": constants.MODE_READWRITE}}
+ page.set_instances(data)
+ page.checkboxes["Pie64 (Normal)"].setChecked(True)
+ assert page.selected_ids() == ["Pie64 (Normal)"]
+
+
+def test_set_instances_preserves_selection_across_refresh(qtbot):
+ page = InstancesPage()
+ qtbot.addWidget(page)
+ page.show()
+ data = {"Pie64 (Normal)": {"root_enabled": True, "rw_mode": constants.MODE_READWRITE}}
+ page.set_instances(data)
+ page.checkboxes["Pie64 (Normal)"].setChecked(True)
+
+ page.set_instances(data, preserve_selection=True)
+ assert page.checkboxes["Pie64 (Normal)"].isChecked() is True
+
+
+def test_set_instances_can_clear_selection(qtbot):
+ page = InstancesPage()
+ qtbot.addWidget(page)
+ page.show()
+ data = {"Pie64 (Normal)": {"root_enabled": True, "rw_mode": constants.MODE_READWRITE}}
+ page.set_instances(data)
+ page.checkboxes["Pie64 (Normal)"].setChecked(True)
+
+ page.set_instances(data, preserve_selection=False)
+ assert page.checkboxes["Pie64 (Normal)"].isChecked() is False
+
+
+def test_clicking_toggle_root_emits_signal(qtbot):
+ page = InstancesPage()
+ qtbot.addWidget(page)
+ page.show()
+ with qtbot.waitSignal(page.toggle_root_requested, timeout=1000):
+ qtbot.mouseClick(page.root_toggle_button, Qt.LeftButton)
+
+
+def test_clicking_fix_it_emits_go_to_dashboard(qtbot):
+ page = InstancesPage()
+ qtbot.addWidget(page)
+ page.show()
+ with qtbot.waitSignal(page.go_to_dashboard_requested, timeout=1000):
+ qtbot.mouseClick(page.banner_fix_button, Qt.LeftButton)
diff --git a/tests/test_main_bootstrap.py b/tests/test_main_bootstrap.py
new file mode 100644
index 0000000..8809c44
--- /dev/null
+++ b/tests/test_main_bootstrap.py
@@ -0,0 +1,7 @@
+import importlib
+
+
+def test_main_module_imports_without_running_gui():
+ main = importlib.import_module("main")
+ assert main.LOG_PATH.endswith("BlueStacksRootGUI.log")
+ assert hasattr(main, "MainWindow")
diff --git a/tests/test_main_window_close_guard.py b/tests/test_main_window_close_guard.py
new file mode 100644
index 0000000..8b6a6bf
--- /dev/null
+++ b/tests/test_main_window_close_guard.py
@@ -0,0 +1,34 @@
+from unittest.mock import MagicMock
+
+from PyQt5.QtWidgets import QMessageBox
+
+from views.main_window import MainWindow
+
+
+def test_close_blocked_while_operation_running(qtbot, monkeypatch):
+ # Closing mid-operation would kill a thread writing real binaries/disk
+ # images. closeEvent must warn and refuse instead of tearing down.
+ window = MainWindow()
+ qtbot.addWidget(window)
+ window._op_thread = object() # stand-in for an in-flight operation
+ warned = MagicMock()
+ monkeypatch.setattr(QMessageBox, "warning", warned)
+ event = MagicMock()
+
+ window.closeEvent(event)
+
+ warned.assert_called_once()
+ event.ignore.assert_called_once()
+ event.accept.assert_not_called()
+
+
+def test_close_allowed_when_idle(qtbot):
+ window = MainWindow()
+ qtbot.addWidget(window)
+ window._op_thread = None
+ event = MagicMock()
+
+ window.closeEvent(event)
+
+ event.accept.assert_called_once()
+ event.ignore.assert_not_called()
diff --git a/tests/test_main_window_gating.py b/tests/test_main_window_gating.py
new file mode 100644
index 0000000..69f8fcd
--- /dev/null
+++ b/tests/test_main_window_gating.py
@@ -0,0 +1,95 @@
+from unittest.mock import MagicMock
+
+from PyQt5.QtWidgets import QMessageBox
+
+from views.main_window import MainWindow
+
+# Note: MainWindow() is safe to construct in tests without touching the
+# registry -- initialize_paths_and_instances only runs via
+# QTimer.singleShot(0, ...), which needs an active event loop (app.exec_())
+# to fire. Nothing in these tests calls that, so it never fires.
+
+
+def test_toggle_root_blocked_when_patch_mode_unpatched(qtbot, monkeypatch):
+ window = MainWindow()
+ qtbot.addWidget(window)
+ window.instance_data = {
+ "Pie64 (Normal)": {"patch_mode": True, "root_enabled": False,
+ "config_path": "c", "original_name": "Pie64"},
+ }
+ window.instances_page.set_instances(window.instance_data)
+ window.instances_page.checkboxes["Pie64 (Normal)"].setChecked(True)
+ monkeypatch.setattr(window, "_engine_state", lambda: "unpatched")
+
+ warned = MagicMock()
+ monkeypatch.setattr(QMessageBox, "warning", warned)
+ ran = MagicMock()
+ monkeypatch.setattr(window, "_run_async", ran)
+
+ window.handle_toggle_root()
+
+ warned.assert_called_once()
+ ran.assert_not_called()
+
+
+def test_toggle_root_proceeds_when_engine_patched(qtbot, monkeypatch):
+ window = MainWindow()
+ qtbot.addWidget(window)
+ window.instance_data = {
+ "Pie64 (Normal)": {"patch_mode": True, "root_enabled": False,
+ "config_path": "c", "original_name": "Pie64"},
+ }
+ window.instances_page.set_instances(window.instance_data)
+ window.instances_page.checkboxes["Pie64 (Normal)"].setChecked(True)
+ monkeypatch.setattr(window, "_engine_state", lambda: "patched")
+
+ ran = MagicMock()
+ monkeypatch.setattr(window, "_run_async", ran)
+
+ window.handle_toggle_root()
+
+ ran.assert_called_once()
+
+
+def test_toggle_root_off_allowed_when_engine_unpatched(qtbot, monkeypatch):
+ # A patch-mode instance that is currently rooted must be togglable OFF even
+ # when the engine is unpatched (e.g. an auto-update reverted the patch) --
+ # disabling root doesn't need the engine patch, so it must not be blocked.
+ window = MainWindow()
+ qtbot.addWidget(window)
+ window.instance_data = {
+ "Pie64 (Normal)": {"patch_mode": True, "root_enabled": True,
+ "config_path": "c", "original_name": "Pie64"},
+ }
+ window.instances_page.set_instances(window.instance_data)
+ window.instances_page.checkboxes["Pie64 (Normal)"].setChecked(True)
+ monkeypatch.setattr(window, "_engine_state", lambda: "unpatched")
+
+ warned = MagicMock()
+ monkeypatch.setattr(QMessageBox, "warning", warned)
+ ran = MagicMock()
+ monkeypatch.setattr(window, "_run_async", ran)
+
+ window.handle_toggle_root()
+
+ warned.assert_not_called()
+ ran.assert_called_once()
+
+
+def test_toggle_root_unaffected_for_classic_instance(qtbot, monkeypatch):
+ window = MainWindow()
+ qtbot.addWidget(window)
+ window.instance_data = {
+ "MSI_Pie (MSI)": {"patch_mode": False, "root_enabled": False,
+ "config_path": "c", "original_name": "MSI_Pie"},
+ }
+ window.instances_page.set_instances(window.instance_data)
+ window.instances_page.checkboxes["MSI_Pie (MSI)"].setChecked(True)
+ monkeypatch.setattr(window, "_engine_state", lambda: "unpatched")
+
+ ran = MagicMock()
+ monkeypatch.setattr(window, "_run_async", ran)
+
+ window.handle_toggle_root()
+
+ ran.assert_called_once()
diff --git a/tests/test_main_window_progress_wiring.py b/tests/test_main_window_progress_wiring.py
new file mode 100644
index 0000000..c889a94
--- /dev/null
+++ b/tests/test_main_window_progress_wiring.py
@@ -0,0 +1,17 @@
+from views.main_window import MainWindow
+
+
+def test_on_async_progress_maps_negative_pct_to_indeterminate(qtbot):
+ window = MainWindow()
+ qtbot.addWidget(window)
+ window._on_async_progress("Working...", -1)
+ assert window.progress_bar._bar.minimum() == 0
+ assert window.progress_bar._bar.maximum() == 0
+
+
+def test_on_async_progress_sets_determinate_value(qtbot):
+ window = MainWindow()
+ qtbot.addWidget(window)
+ window._on_async_progress("Step 2", 50)
+ assert window.progress_bar._bar.maximum() == 100
+ assert window.progress_bar._bar.value() == 50
diff --git a/tests/test_main_window_revert_alert.py b/tests/test_main_window_revert_alert.py
new file mode 100644
index 0000000..7c3f8e7
--- /dev/null
+++ b/tests/test_main_window_revert_alert.py
@@ -0,0 +1,105 @@
+from views.main_window import MainWindow
+
+
+def _setup_window(qtbot, window):
+ """Helper to set up MainWindow for testing dashboard alerts."""
+ qtbot.addWidget(window)
+ window.pages.setCurrentIndex(0) # Show dashboard page
+ window.show()
+
+
+def test_no_alert_before_any_poll(qtbot):
+ window = MainWindow()
+ _setup_window(qtbot, window)
+ assert window.dashboard_page.alert_label.isVisible() is False
+
+
+def test_alert_shows_when_engine_state_drops_while_rooted(qtbot, monkeypatch):
+ window = MainWindow()
+ _setup_window(qtbot, window)
+ window.instance_data = {"Pie64 (Normal)": {"root_enabled": True, "patch_mode": True}}
+ states = iter(["patched", "unpatched"])
+ monkeypatch.setattr(window, "_engine_state", lambda: next(states))
+
+ window._check_for_reverted_patch() # seeds _last_engine_state = "patched"
+ assert window.dashboard_page.alert_label.isVisible() is False
+
+ window._check_for_reverted_patch() # drops to "unpatched" with root active
+ assert window.dashboard_page.alert_label.isVisible() is True
+
+
+def test_no_alert_when_nothing_was_rooted(qtbot, monkeypatch):
+ window = MainWindow()
+ _setup_window(qtbot, window)
+ window.instance_data = {"Pie64 (Normal)": {"root_enabled": False}}
+ states = iter(["patched", "unpatched"])
+ monkeypatch.setattr(window, "_engine_state", lambda: next(states))
+
+ window._check_for_reverted_patch()
+ window._check_for_reverted_patch()
+ assert window.dashboard_page.alert_label.isVisible() is False
+
+
+def test_no_alert_when_only_classic_instance_rooted(qtbot, monkeypatch):
+ # A rooted classic (conf/MSI) instance does not depend on the engine patch,
+ # so a patch-mode engine flipping patched -> unpatched must NOT raise the
+ # "your engine patch was reverted" alert on its account.
+ window = MainWindow()
+ _setup_window(qtbot, window)
+ window.instance_data = {"MSI_Pie (MSI)": {"root_enabled": True, "patch_mode": False}}
+ states = iter(["patched", "unpatched"])
+ monkeypatch.setattr(window, "_engine_state", lambda: next(states))
+
+ window._check_for_reverted_patch()
+ window._check_for_reverted_patch()
+ assert window.dashboard_page.alert_label.isVisible() is False
+
+
+def test_repatch_via_on_async_done_clears_alert_immediately(qtbot, monkeypatch):
+ # With the alert already showing, finishing a successful (re)patch should
+ # clear it right away rather than waiting for the next timer tick.
+ window = MainWindow()
+ _setup_window(qtbot, window)
+ window.instance_data = {"Pie64 (Normal)": {"root_enabled": True, "patch_mode": True}}
+ window.dashboard_page.set_update_reverted(True)
+ assert window.dashboard_page.alert_label.isVisible() is True
+ monkeypatch.setattr(window, "_engine_state", lambda: "patched")
+
+ window._on_async_done(True, "Engine patched.")
+ assert window.dashboard_page.alert_label.isVisible() is False
+
+
+def test_alert_clears_once_repatched(qtbot, monkeypatch):
+ window = MainWindow()
+ _setup_window(qtbot, window)
+ window.instance_data = {"Pie64 (Normal)": {"root_enabled": True, "patch_mode": True}}
+ states = iter(["patched", "unpatched", "patched"])
+ monkeypatch.setattr(window, "_engine_state", lambda: next(states))
+
+ window._check_for_reverted_patch()
+ window._check_for_reverted_patch()
+ assert window.dashboard_page.alert_label.isVisible() is True
+ window._check_for_reverted_patch()
+ assert window.dashboard_page.alert_label.isVisible() is False
+
+
+def test_on_async_done_resyncs_state_so_manual_undo_does_not_false_alert(qtbot, monkeypatch):
+ """Regression test: 'Undo Root Patch' (handle_restore_patches) legitimately
+ reverts the engine patch but leaves per-instance root_enabled=True. Before
+ the fix, _last_engine_state stayed "patched" after the operation finished,
+ so the next status-timer tick saw patched -> unpatched with a rooted
+ instance and misreported it as an auto-update revert. _on_async_done must
+ resync _last_engine_state immediately so that doesn't happen."""
+ window = MainWindow()
+ _setup_window(qtbot, window)
+ window.instance_data = {"Pie64 (Normal)": {"root_enabled": True, "patch_mode": True}}
+ window._last_engine_state = "patched"
+ monkeypatch.setattr(window, "_engine_state", lambda: "unpatched")
+
+ window._on_async_done(True, "Restored 1 installation(s).")
+ assert window._last_engine_state == "unpatched"
+ assert window.dashboard_page.alert_label.isVisible() is False
+
+ # Next timer tick should see no change from the now-resynced state.
+ window._check_for_reverted_patch()
+ assert window.dashboard_page.alert_label.isVisible() is False
diff --git a/tests/test_modules_page.py b/tests/test_modules_page.py
new file mode 100644
index 0000000..ae95901
--- /dev/null
+++ b/tests/test_modules_page.py
@@ -0,0 +1,114 @@
+from PyQt5.QtCore import Qt
+
+from views.modules_page import ModulesPage
+
+
+def test_no_running_label_visible_when_list_empty(qtbot):
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_running_instances([])
+ assert page.no_running_label.isVisible() is True
+
+
+def test_no_running_label_hidden_when_instances_present(qtbot):
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_running_instances(["Pie64 (Normal)"])
+ assert page.no_running_label.isVisible() is False
+
+
+def test_push_button_stays_disabled_while_busy(qtbot):
+ # Even with a running instance selected and a zip chosen, the push button
+ # must stay disabled while the app is busy with a background operation.
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_running_instances(["Pie64 (Normal)"])
+ page._radios["Pie64 (Normal)"].setChecked(True)
+ page.set_zip_path("C:/mod.zip")
+ assert page.push_button.isEnabled() is True
+
+ page.set_busy(True)
+ assert page.push_button.isEnabled() is False
+
+ # A zip/radio change mid-operation re-runs _update_push_enabled; it must
+ # still stay disabled while busy.
+ page.set_zip_path("C:/other.zip")
+ assert page.push_button.isEnabled() is False
+
+ page.set_busy(False)
+ assert page.push_button.isEnabled() is True
+
+
+def test_set_scanning_shows_hint_and_clears_radios(qtbot):
+ # While the background ADB probe runs, the tab shows a "checking..." hint
+ # instead of a stale list, and no instance can be selected.
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_running_instances(["Pie64 (Normal)"])
+ assert len(page._radios) == 1
+
+ page.set_scanning()
+ assert page._radios == {}
+ assert page.no_running_label.isVisible() is True
+ assert page.no_running_label.text() == ModulesPage._SCANNING_TEXT
+ assert page.push_button.isEnabled() is False
+
+
+def test_empty_result_after_scanning_restores_empty_text(qtbot):
+ # A scan that finds nothing must reset the label back to the "no instance
+ # running" copy, not leave the transient "checking..." text stuck.
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_scanning()
+ assert page.no_running_label.text() == ModulesPage._SCANNING_TEXT
+
+ page.set_running_instances([])
+ assert page.no_running_label.isVisible() is True
+ assert page.no_running_label.text() == ModulesPage._EMPTY_TEXT
+
+
+def test_push_disabled_until_instance_and_zip_chosen(qtbot):
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_running_instances(["Pie64 (Normal)"])
+ assert page.push_button.isEnabled() is False
+
+ page._radios["Pie64 (Normal)"].setChecked(True)
+ assert page.push_button.isEnabled() is False # zip still missing
+
+ page.set_zip_path("C:/mods/module.zip")
+ assert page.push_button.isEnabled() is True
+
+
+def test_selected_instance_id_reflects_radio_choice(qtbot):
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_running_instances(["A (Normal)", "B (Normal)"])
+ page._radios["B (Normal)"].setChecked(True)
+ assert page.selected_instance_id() == "B (Normal)"
+
+
+def test_clicking_browse_emits_signal(qtbot):
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ with qtbot.waitSignal(page.browse_zip_requested, timeout=1000):
+ qtbot.mouseClick(page.browse_button, Qt.LeftButton)
+
+
+def test_clicking_push_emits_signal_when_enabled(qtbot):
+ page = ModulesPage()
+ qtbot.addWidget(page)
+ page.show()
+ page.set_running_instances(["Pie64 (Normal)"])
+ page._radios["Pie64 (Normal)"].setChecked(True)
+ page.set_zip_path("C:/mods/module.zip")
+ with qtbot.waitSignal(page.push_requested, timeout=1000):
+ qtbot.mouseClick(page.push_button, Qt.LeftButton)
diff --git a/tests/test_nav_rail.py b/tests/test_nav_rail.py
new file mode 100644
index 0000000..d4f635c
--- /dev/null
+++ b/tests/test_nav_rail.py
@@ -0,0 +1,37 @@
+from PyQt5.QtCore import Qt
+
+from views.nav_rail import NavRail, DASHBOARD, INSTANCES, MODULES
+
+
+def test_nav_rail_starts_on_dashboard(qtbot):
+ rail = NavRail()
+ qtbot.addWidget(rail)
+ assert rail.current() == DASHBOARD
+ assert rail._buttons[DASHBOARD].isChecked() is True
+
+
+def test_clicking_instances_emits_navigate_and_updates_active_state(qtbot):
+ rail = NavRail()
+ qtbot.addWidget(rail)
+ with qtbot.waitSignal(rail.navigate, timeout=1000) as blocker:
+ qtbot.mouseClick(rail._buttons[INSTANCES], Qt.LeftButton)
+ assert blocker.args == [INSTANCES]
+ assert rail.current() == INSTANCES
+ assert rail._buttons[DASHBOARD].isChecked() is False
+
+
+def test_clicking_modules_then_dashboard_only_one_active(qtbot):
+ rail = NavRail()
+ qtbot.addWidget(rail)
+ qtbot.mouseClick(rail._buttons[MODULES], Qt.LeftButton)
+ qtbot.mouseClick(rail._buttons[DASHBOARD], Qt.LeftButton)
+ assert rail.current() == DASHBOARD
+ checked = [k for k, b in rail._buttons.items() if b.isChecked()]
+ assert checked == [DASHBOARD]
+
+
+def test_select_updates_state_without_a_click(qtbot):
+ rail = NavRail()
+ qtbot.addWidget(rail)
+ rail.select(MODULES)
+ assert rail.current() == MODULES
diff --git a/tests/test_op_worker_progress.py b/tests/test_op_worker_progress.py
new file mode 100644
index 0000000..e666496
--- /dev/null
+++ b/tests/test_op_worker_progress.py
@@ -0,0 +1,17 @@
+from views.main_window import _OpWorker
+
+
+def test_op_worker_progress_emits_text_and_percent(qapp):
+ worker = _OpWorker(lambda progress: (progress("halfway", 50), "done")[-1])
+ received = []
+ worker.progress.connect(lambda msg, pct: received.append((msg, pct)))
+ worker.run()
+ assert received == [("halfway", 50)]
+
+
+def test_op_worker_progress_accepts_unknown_percent(qapp):
+ worker = _OpWorker(lambda progress: (progress("working", -1), "done")[-1])
+ received = []
+ worker.progress.connect(lambda msg, pct: received.append((msg, pct)))
+ worker.run()
+ assert received == [("working", -1)]
diff --git a/tests/test_progress.py b/tests/test_progress.py
new file mode 100644
index 0000000..ccb849f
--- /dev/null
+++ b/tests/test_progress.py
@@ -0,0 +1,54 @@
+from views.progress import step_percent, OperationProgressBar
+
+
+def test_step_percent_midpoint():
+ assert step_percent(1, 4) == 25
+
+
+def test_step_percent_final_step_is_100():
+ assert step_percent(4, 4) == 100
+
+
+def test_step_percent_zero_total_is_zero():
+ assert step_percent(1, 0) == 0
+
+
+def test_progress_bar_starts_with_ready_text_and_hidden_bar(qtbot):
+ bar = OperationProgressBar()
+ qtbot.addWidget(bar)
+ assert bar._label.text() == "Ready"
+ assert bar._bar.isVisible() is False
+
+
+def test_start_shows_bar_and_sets_text(qtbot):
+ bar = OperationProgressBar()
+ qtbot.addWidget(bar)
+ bar.start("Working...")
+ assert bar._bar.isVisible() is True
+ assert bar._label.text() == "Working..."
+ assert bar._bar.minimum() == 0 and bar._bar.maximum() == 0
+
+
+def test_set_progress_determinate(qtbot):
+ bar = OperationProgressBar()
+ qtbot.addWidget(bar)
+ bar.set_progress("Step 2", 50)
+ assert bar._bar.maximum() == 100
+ assert bar._bar.value() == 50
+
+
+def test_set_progress_indeterminate(qtbot):
+ bar = OperationProgressBar()
+ qtbot.addWidget(bar)
+ bar.set_progress("Working...", None)
+ assert bar._bar.maximum() == 0
+ assert bar._bar.minimum() == 0
+
+
+def test_finish_hides_bar_and_sets_summary_text(qtbot):
+ bar = OperationProgressBar()
+ qtbot.addWidget(bar)
+ bar.start("Working...")
+ bar.finish("Done. 3 instances updated.")
+ assert bar._bar.isVisible() is False
+ assert bar._label.text() == "Done. 3 instances updated."
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
new file mode 100644
index 0000000..cdf6115
--- /dev/null
+++ b/tests/test_smoke.py
@@ -0,0 +1,6 @@
+import constants
+
+
+def test_constants_importable_and_sane():
+ assert constants.APP_NAME == "BlueStacks Root GUI"
+ assert constants.MODE_READWRITE == "Normal"
diff --git a/tests/test_theme.py b/tests/test_theme.py
new file mode 100644
index 0000000..4e3174f
--- /dev/null
+++ b/tests/test_theme.py
@@ -0,0 +1,39 @@
+import pytest
+
+from views import theme
+
+
+@pytest.fixture(autouse=True)
+def _isolated_settings(tmp_path, monkeypatch):
+ # Route persistence at a throwaway org/app name per test so tests never
+ # touch the real registry entry the actual app would use.
+ monkeypatch.setattr(theme, "_ORG", "BlueStacksRootGUI-Test")
+ monkeypatch.setattr(theme, "_APP", "theme-test-%s" % tmp_path.name)
+ yield
+
+
+def test_stylesheet_for_light_contains_light_background():
+ assert "#f3f3f3" in theme.stylesheet_for(theme.LIGHT)
+
+
+def test_stylesheet_for_dark_contains_dark_background():
+ assert "#202020" in theme.stylesheet_for(theme.DARK)
+
+
+def test_stylesheet_for_unknown_theme_raises():
+ with pytest.raises(ValueError):
+ theme.stylesheet_for("solarized")
+
+
+def test_apply_theme_sets_app_stylesheet(qapp):
+ theme.apply_theme(qapp, theme.DARK)
+ assert qapp.styleSheet() == theme.stylesheet_for(theme.DARK)
+
+
+def test_load_saved_theme_defaults_to_light():
+ assert theme.load_saved_theme() == theme.LIGHT
+
+
+def test_apply_theme_persists_choice(qapp):
+ theme.apply_theme(qapp, theme.DARK)
+ assert theme.load_saved_theme() == theme.DARK
diff --git a/views/__init__.py b/views/__init__.py
new file mode 100644
index 0000000..364dd95
--- /dev/null
+++ b/views/__init__.py
@@ -0,0 +1 @@
+"""Views package."""
diff --git a/views/dashboard_page.py b/views/dashboard_page.py
new file mode 100644
index 0000000..6ccf40e
--- /dev/null
+++ b/views/dashboard_page.py
@@ -0,0 +1,58 @@
+"""Dashboard page: install paths, engine-patch state, rooted-count stat,
+update-revert alert."""
+from __future__ import annotations
+
+from PyQt5.QtCore import pyqtSignal
+from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QPushButton
+
+
+class DashboardPage(QWidget):
+ patch_engine_requested = pyqtSignal()
+ repatch_requested = pyqtSignal()
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ layout = QVBoxLayout(self)
+
+ self.path_label = QLabel("BlueStacks Path: Loading...")
+ self.path_label.setWordWrap(True)
+ layout.addWidget(self.path_label)
+
+ self.alert_label = QLabel(
+ "An auto-update reverted your engine patch. Rooted instances will "
+ "fail the integrity check on next boot until you re-patch."
+ )
+ self.alert_label.setWordWrap(True)
+ self.alert_label.setObjectName("UpdateRevertedAlert")
+ self.repatch_button = QPushButton("Re-patch now")
+ self.repatch_button.clicked.connect(self.repatch_requested.emit)
+ layout.addWidget(self.alert_label)
+ layout.addWidget(self.repatch_button)
+ self.set_update_reverted(False)
+
+ self.engine_button = QPushButton("")
+ self.engine_button.clicked.connect(self.patch_engine_requested.emit)
+ self.engine_button.setVisible(False)
+ layout.addWidget(self.engine_button)
+
+ self.stat_label = QLabel("0 / 0 instances rooted")
+ layout.addWidget(self.stat_label)
+ layout.addStretch(1)
+
+ def set_paths_text(self, text: str) -> None:
+ self.path_label.setText(text)
+
+ def set_update_reverted(self, reverted: bool) -> None:
+ self.alert_label.setVisible(reverted)
+ self.repatch_button.setVisible(reverted)
+
+ def set_engine_state(self, visible: bool, text: str, tooltip: str,
+ color: str, enabled: bool) -> None:
+ self.engine_button.setVisible(visible)
+ self.engine_button.setText(text)
+ self.engine_button.setToolTip(tooltip)
+ self.engine_button.setStyleSheet("color: %s; font-weight: bold;" % color)
+ self.engine_button.setEnabled(enabled)
+
+ def set_rooted_count(self, rooted: int, total: int) -> None:
+ self.stat_label.setText("%d / %d instances rooted" % (rooted, total))
diff --git a/views/engine_rules.py b/views/engine_rules.py
new file mode 100644
index 0000000..ae789d9
--- /dev/null
+++ b/views/engine_rules.py
@@ -0,0 +1,34 @@
+"""Pure decision logic for patch-gating and update-revert detection.
+
+Kept free of Qt so it can be unit-tested without a QApplication.
+"""
+from __future__ import annotations
+
+
+def blocked_for_root_toggle(selected: dict, engine_state: str) -> list[str]:
+ """Names of selected instances that must not have root ENABLED yet.
+
+ An instance is blocked only when the user is turning root ON (it is
+ currently off), it needs the engine patch (`patch_mode` True), and the
+ engine is in a known-not-patched state ("unpatched" or "partial").
+
+ Turning root OFF is never blocked -- disabling root does not depend on
+ the engine patch, so a user must always be able to clear a stale root
+ (e.g. after an auto-update reverts the patch). An "unknown" engine
+ state (unrecognized build we can't read) blocks nothing either, rather
+ than trapping the user behind a Dashboard button that is itself
+ disabled for "unknown". ``selected`` maps unique_id -> instance dict
+ (needs at least ``patch_mode`` and ``root_enabled``).
+ """
+ if engine_state not in ("unpatched", "partial"):
+ return []
+ return [uid for uid, data in selected.items()
+ if data.get("patch_mode") and not data.get("root_enabled")]
+
+
+def update_was_reverted(previous_state: str | None, current_state: str,
+ any_rooted: bool) -> bool:
+ """True when the engine flips from patched to not-patched while a
+ rooted instance still exists -- the signature of an auto-update
+ silently undoing the engine patch."""
+ return previous_state == "patched" and current_state != "patched" and any_rooted
diff --git a/views/instances_page.py b/views/instances_page.py
new file mode 100644
index 0000000..bd927eb
--- /dev/null
+++ b/views/instances_page.py
@@ -0,0 +1,84 @@
+"""Instances page: instance grid + Toggle Root/R-W + patch-gating banner."""
+from __future__ import annotations
+
+from PyQt5.QtCore import pyqtSignal
+from PyQt5.QtWidgets import (
+ QWidget, QVBoxLayout, QGridLayout, QGroupBox, QCheckBox, QLabel,
+ QPushButton, QHBoxLayout,
+)
+
+import constants
+
+
+class InstancesPage(QWidget):
+ toggle_root_requested = pyqtSignal()
+ toggle_rw_requested = pyqtSignal()
+ go_to_dashboard_requested = pyqtSignal()
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ layout = QVBoxLayout(self)
+
+ self.banner_label = QLabel(
+ "Patch-mode root is locked. Patch the engine to root the "
+ "5.22.150+ instances. (The MSI classic instance roots without it.)"
+ )
+ self.banner_label.setWordWrap(True)
+ self.banner_fix_button = QPushButton("Fix it")
+ self.banner_fix_button.clicked.connect(self.go_to_dashboard_requested.emit)
+ banner_row = QHBoxLayout()
+ banner_row.addWidget(self.banner_label, 1)
+ banner_row.addWidget(self.banner_fix_button)
+ layout.addLayout(banner_row)
+ self.set_engine_locked_banner(False)
+
+ self.instance_group = QGroupBox("Instances")
+ self.instance_layout = QGridLayout()
+ self.instance_layout.setColumnStretch(0, 4)
+ self.instance_layout.setColumnStretch(1, 1)
+ self.instance_layout.setColumnStretch(2, 1)
+ self.instance_layout.setHorizontalSpacing(15)
+ self.instance_group.setLayout(self.instance_layout)
+ layout.addWidget(self.instance_group)
+
+ button_row = QHBoxLayout()
+ self.root_toggle_button = QPushButton("Toggle Root")
+ self.root_toggle_button.clicked.connect(self.toggle_root_requested.emit)
+ self.rw_toggle_button = QPushButton("Toggle R/W")
+ self.rw_toggle_button.clicked.connect(self.toggle_rw_requested.emit)
+ button_row.addWidget(self.root_toggle_button)
+ button_row.addWidget(self.rw_toggle_button)
+ layout.addLayout(button_row)
+
+ self.checkboxes: dict[str, QCheckBox] = {}
+
+ def set_engine_locked_banner(self, locked: bool) -> None:
+ self.banner_label.setVisible(locked)
+ self.banner_fix_button.setVisible(locked)
+
+ def set_instances(self, instance_data: dict, preserve_selection: bool = True) -> None:
+ """Rebuild the grid from ``instance_data`` (unique_id -> dict with
+ at least ``root_enabled`` and ``rw_mode``)."""
+ previous_selection = (
+ {uid for uid, cb in self.checkboxes.items() if cb.isChecked()}
+ if preserve_selection else set()
+ )
+ while self.instance_layout.count():
+ item = self.instance_layout.takeAt(0)
+ if item and item.widget():
+ item.widget().deleteLater()
+ self.checkboxes = {}
+
+ for row, unique_id in enumerate(sorted(instance_data.keys())):
+ data = instance_data[unique_id]
+ checkbox = QCheckBox(unique_id)
+ checkbox.setChecked(unique_id in previous_selection)
+ root_text = "On" if data.get("root_enabled") else "Off"
+ rw_text = "On" if data.get("rw_mode") == constants.MODE_READWRITE else "Off"
+ self.instance_layout.addWidget(checkbox, row, 0)
+ self.instance_layout.addWidget(QLabel("Root: %s" % root_text), row, 1)
+ self.instance_layout.addWidget(QLabel("R/W: %s" % rw_text), row, 2)
+ self.checkboxes[unique_id] = checkbox
+
+ def selected_ids(self) -> list[str]:
+ return [uid for uid, cb in self.checkboxes.items() if cb.isChecked()]
diff --git a/views/main_window.py b/views/main_window.py
new file mode 100644
index 0000000..9af8552
--- /dev/null
+++ b/views/main_window.py
@@ -0,0 +1,751 @@
+# views/main_window.py
+"""Main application window: nav rail + Dashboard/Instances/Modules pages."""
+from __future__ import annotations
+
+import os
+import sys
+import logging
+from typing import Any
+
+from PyQt5.QtWidgets import (
+ QWidget, QVBoxLayout, QHBoxLayout, QStackedWidget, QPushButton,
+ QMessageBox, QFileDialog, QApplication,
+)
+from PyQt5.QtCore import Qt, QTimer, QThread, QObject, pyqtSignal, pyqtSlot
+from PyQt5.QtGui import QIcon
+
+import constants
+import registry_handler
+import config_handler
+import instance_handler
+import root_persistence
+import integrity_patch
+import su_patch_offline
+import ext4_symlink
+import adb_handler
+import admin
+
+from views.nav_rail import (
+ NavRail, DASHBOARD as NAV_DASHBOARD, INSTANCES as NAV_INSTANCES,
+ MODULES as NAV_MODULES,
+)
+from views.dashboard_page import DashboardPage
+from views.instances_page import InstancesPage
+from views.modules_page import ModulesPage
+from views.progress import OperationProgressBar, step_percent
+from views import theme
+from views import engine_rules
+
+logger = logging.getLogger(__name__)
+
+
+def resource_path(relative_path):
+ try:
+ base_path = sys._MEIPASS
+ except AttributeError:
+ base_path = os.path.abspath(".")
+ return os.path.join(base_path, relative_path)
+
+
+class _OpWorker(QObject):
+ """Runs a blocking job(progress) on a worker thread, relaying progress
+ text and an optional percent complete (-1 for unknown)."""
+ progress = pyqtSignal(str, int)
+ done = pyqtSignal(bool, str)
+
+ def __init__(self, job):
+ super().__init__()
+ self._job = job
+
+ @pyqtSlot()
+ def run(self):
+ try:
+ summary = self._job(self.progress.emit)
+ self.done.emit(True, summary)
+ except Exception as exc: # noqa: BLE001
+ logger.exception("Background operation failed")
+ self.done.emit(False, str(exc))
+
+
+class _RunningScanWorker(QObject):
+ """Probes which instances are reachable over ADB, on a worker thread.
+
+ The probe shells out to adb once per instance (adb server start + a per-port
+ ``connect`` that blocks until it succeeds or times out), which can take
+ several seconds and must never run on the UI thread.
+ """
+ finished = pyqtSignal(list) # sorted list of running unique_ids
+
+ def __init__(self, adb_exe, instances):
+ super().__init__()
+ self._adb_exe = adb_exe
+ self._instances = instances
+
+ @pyqtSlot()
+ def run(self):
+ try:
+ running = adb_handler.list_running_instances(self._adb_exe, self._instances)
+ self.finished.emit(sorted(running.keys()))
+ except Exception: # noqa: BLE001 - a probe failure just means "none found"
+ logger.exception("Running-instance ADB probe failed")
+ self.finished.emit([])
+
+
+class MainWindow(QWidget):
+ """Main application window for toggling BlueStacks root and R/W settings."""
+
+ show_notice = pyqtSignal(str, str) # (title, message)
+
+ def __init__(self):
+ super().__init__()
+ self.installations: list = []
+ self.instance_data: dict[str, dict[str, Any]] = {}
+
+ self.show_notice.connect(self._show_notice)
+ self.setWindowTitle(f"{constants.APP_NAME} v{constants.APP_VERSION}")
+ self._set_icon()
+ self._last_engine_state = None
+ # Background ADB probe for the Modules tab (see _refresh_running_instances).
+ self._scan_thread = None
+ self._scan_worker = None
+ self._scan_pending = False
+ self.status_refresh_timer = QTimer(self)
+ self.status_refresh_timer.timeout.connect(self._on_status_timer)
+ self.init_ui()
+ QTimer.singleShot(0, self.initialize_paths_and_instances)
+
+ def _set_icon(self):
+ try:
+ icon_path = resource_path(constants.ICON_FILENAME)
+ app_icon = QIcon(icon_path)
+ if not app_icon.isNull():
+ self.setWindowIcon(app_icon)
+ except Exception as e:
+ logger.error(f"Error setting window icon: {e}")
+
+ def init_ui(self) -> None:
+ root_layout = QVBoxLayout(self)
+ root_layout.setContentsMargins(0, 0, 0, 0)
+ root_layout.setSpacing(0)
+
+ toolbar = QHBoxLayout()
+ toolbar.setContentsMargins(8, 6, 8, 6)
+ toolbar.addStretch(1)
+ self.theme_button = QPushButton("Toggle theme")
+ self.theme_button.clicked.connect(self._handle_toggle_theme)
+ toolbar.addWidget(self.theme_button)
+ root_layout.addLayout(toolbar)
+
+ body = QHBoxLayout()
+ body.setContentsMargins(0, 0, 0, 0)
+ body.setSpacing(0)
+
+ self.nav_rail = NavRail()
+ self.nav_rail.navigate.connect(self._handle_navigate)
+ body.addWidget(self.nav_rail)
+
+ self.pages = QStackedWidget()
+ self.dashboard_page = DashboardPage()
+ self.instances_page = InstancesPage()
+ self.modules_page = ModulesPage()
+ self.pages.addWidget(self.dashboard_page)
+ self.pages.addWidget(self.instances_page)
+ self.pages.addWidget(self.modules_page)
+ self._pages_by_key = {
+ NAV_DASHBOARD: self.dashboard_page,
+ NAV_INSTANCES: self.instances_page,
+ NAV_MODULES: self.modules_page,
+ }
+ body.addWidget(self.pages, 1)
+
+ root_layout.addLayout(body, 1)
+
+ self.progress_bar = OperationProgressBar()
+ root_layout.addWidget(self.progress_bar)
+
+ self.dashboard_page.patch_engine_requested.connect(self.handle_engine_button)
+ self.dashboard_page.repatch_requested.connect(self.handle_apply_patches)
+ self.instances_page.toggle_root_requested.connect(self.handle_toggle_root)
+ self.instances_page.toggle_rw_requested.connect(self.handle_toggle_rw)
+ self.instances_page.go_to_dashboard_requested.connect(
+ lambda: self.nav_rail.select(NAV_DASHBOARD))
+ self.modules_page.browse_zip_requested.connect(self._handle_browse_zip)
+ self.modules_page.push_requested.connect(self._handle_push_module)
+
+ self.setMinimumWidth(700)
+ self.setMinimumHeight(480)
+
+ def _handle_navigate(self, key: str) -> None:
+ self.pages.setCurrentWidget(self._pages_by_key[key])
+ if key == NAV_MODULES:
+ self._refresh_running_instances()
+
+ def _handle_toggle_theme(self) -> None:
+ current = theme.load_saved_theme()
+ next_theme = theme.DARK if current == theme.LIGHT else theme.LIGHT
+ theme.apply_theme(QApplication.instance(), next_theme)
+
+ def initialize_paths_and_instances(self) -> None:
+ logger.info("Initializing BlueStacks paths and instances...")
+ self.installations = registry_handler.get_all_bluestacks_installations()
+ if not self.installations:
+ self.dashboard_page.set_paths_text("No BlueStacks installations found.")
+ return
+
+ path_details = ["Installations Found:"]
+ for inst in self.installations:
+ ver = ".".join(map(str, inst["version"])) if inst.get("version") else "?"
+ path_details.append(f" - {inst['source']} v{ver}: {inst['user_path']}")
+ self.dashboard_page.set_paths_text("\n".join(path_details))
+
+ # Populate instance_data BEFORE refreshing the patch UI: the Dashboard
+ # "N / M instances rooted" stat is derived from instance_data, so
+ # refreshing first would render "0 / 0" until the next timer tick.
+ self.update_instance_statuses(preserve_selection=False)
+ self._refresh_patch_ui()
+ self.status_refresh_timer.start(constants.REFRESH_INTERVAL_MS)
+
+ def _refresh_patch_ui(self, state: str | None = None) -> None:
+ has_patch_build = any(i.get("patch_mode") for i in self.installations)
+ if not has_patch_build:
+ self.dashboard_page.set_engine_state(False, "", "", "#000000", False)
+ self.instances_page.set_engine_locked_banner(False)
+ else:
+ if state is None:
+ state = self._engine_state()
+ text, color, tip, enabled = {
+ "patched": ("Engine patched (click to Undo)",
+ "#2e7d32", "Restores HD-Player.exe and HD-MultiInstanceManager.exe "
+ "from the .prepatch.bak backups.", True),
+ "unpatched": ("Patch BlueStacks Engine (required for root)",
+ "#c62828", "Patches HD-Player.exe (and HD-MultiInstanceManager.exe) "
+ "to disable the integrity shutdown so rooted instances boot. Do "
+ "this once, then Toggle Root per instance.", True),
+ "partial": ("Engine partially patched (click to finish)",
+ "#e65100", "Some engine binaries aren't patched yet. Re-run the "
+ "patch to bring them all up to date.", True),
+ "unknown": ("Engine status unknown (unrecognized build)",
+ "#616161", "Couldn't read the engine patch state for this build.",
+ False),
+ }[state]
+ self._engine_action = "restore" if state == "patched" else "patch"
+ self.dashboard_page.set_engine_state(True, text, tip, color, enabled)
+ # Only surface the "locked" banner for states we can act on from
+ # the Dashboard (unpatched/partial). "unknown" leaves the engine
+ # button disabled, so a banner pointing there would be a dead end.
+ self.instances_page.set_engine_locked_banner(state in ("unpatched", "partial"))
+
+ rooted = sum(1 for d in self.instance_data.values() if d.get("root_enabled"))
+ self.dashboard_page.set_rooted_count(rooted, len(self.instance_data))
+
+ def _engine_state(self) -> str:
+ states = [integrity_patch.installation_patched(i["install_path"])
+ for i in self.installations
+ if i.get("patch_mode") and i.get("install_path")
+ and os.path.isdir(i["install_path"])]
+ if states and all(s is True for s in states):
+ return "patched"
+ if states and all(s is False for s in states):
+ return "unpatched"
+ if any(s is True for s in states):
+ return "partial"
+ return "unknown"
+
+ def _on_status_timer(self) -> None:
+ self.update_instance_statuses(preserve_selection=True)
+ # _engine_state() reads and byte-scans HD-Player.exe from disk; compute
+ # it once per tick and share it with both consumers.
+ state = self._engine_state()
+ self._refresh_patch_ui(state)
+ self._check_for_reverted_patch(state)
+
+ def _check_for_reverted_patch(self, current_state: str | None = None) -> None:
+ if current_state is None:
+ current_state = self._engine_state()
+ # Only patch-mode instances depend on the engine patch; a rooted classic
+ # (MSI/conf) instance is unaffected by a revert, so don't let it trigger
+ # a false "your engine patch was reverted" alert.
+ any_rooted = any(d.get("root_enabled") for d in self.instance_data.values()
+ if d.get("patch_mode"))
+ if engine_rules.update_was_reverted(self._last_engine_state, current_state, any_rooted):
+ self.dashboard_page.set_update_reverted(True)
+ logger.warning("Engine patch appears to have been reverted (was %s, now %s) "
+ "while a rooted instance exists.", self._last_engine_state, current_state)
+ elif current_state == "patched":
+ self.dashboard_page.set_update_reverted(False)
+ self._last_engine_state = current_state
+
+ def handle_engine_button(self) -> None:
+ self._refresh_patch_ui()
+ if getattr(self, "_engine_action", "patch") == "restore":
+ self.handle_restore_patches()
+ else:
+ self.handle_apply_patches()
+
+ def update_instance_data(self) -> None:
+ if not self.installations:
+ return
+
+ all_found_instances: dict[str, dict[str, Any]] = {}
+ for inst in self.installations:
+ source_id, config_path, data_path = inst["source"], inst["config_path"], inst["data_path"]
+ install_path = inst.get("install_path")
+ patch_mode = inst.get("patch_mode", False)
+ root_info = config_handler.get_complete_root_statuses(config_path)
+ instance_root_statuses = root_info['instance_statuses']
+
+ disk_instances = set()
+ if os.path.isdir(data_path):
+ try:
+ disk_instances = {
+ entry for entry in os.listdir(data_path)
+ if os.path.isdir(os.path.join(data_path, entry))
+ }
+ except OSError:
+ # Runs on the status-refresh timer; a PermissionError here
+ # must not take down the refresh loop.
+ logger.warning("Could not list %s", data_path, exc_info=True)
+ all_instance_names = set(instance_root_statuses.keys()) | disk_instances
+
+ for name in sorted(all_instance_names):
+ unique_id = f"{name} ({source_id})"
+ instance_dir_path = os.path.join(data_path, name)
+
+ rw_mode = constants.MODE_UNKNOWN
+ if os.path.isdir(instance_dir_path):
+ is_readonly = instance_handler.is_instance_readonly(instance_dir_path)
+ if is_readonly is True:
+ rw_mode = constants.MODE_READONLY
+ elif is_readonly is False:
+ rw_mode = constants.MODE_READWRITE
+
+ individual_root_on = instance_root_statuses.get(name, False)
+ if patch_mode:
+ effective_root_status = su_patch_offline.instance_root_state(instance_dir_path)
+ else:
+ effective_root_status = individual_root_on
+
+ all_found_instances[unique_id] = {
+ "original_name": name,
+ "config_path": config_path,
+ "data_path": instance_dir_path,
+ "install_path": install_path,
+ "rw_mode": rw_mode,
+ "root_enabled": effective_root_status,
+ "individual_root_status": individual_root_on,
+ "patch_mode": patch_mode,
+ }
+
+ self.instance_data = {
+ uid: data for uid, data in all_found_instances.items()
+ if data["rw_mode"] != constants.MODE_UNKNOWN
+ }
+
+ logger.debug(f"Instance data updated. Displaying {len(self.instance_data)} instances.")
+
+ def update_instance_checkboxes(self, preserve_selection: bool = True) -> None:
+ self.instances_page.set_instances(self.instance_data, preserve_selection)
+
+ def _toggle_single_instance_root(self, unique_id, progress=None):
+ if self.instance_data[unique_id].get("patch_mode"):
+ self._toggle_root_patchmode(unique_id, progress)
+ else:
+ self._toggle_root_conf(unique_id, progress)
+
+ def _toggle_root_patchmode(self, unique_id, progress=None):
+ instance = self.instance_data[unique_id]
+ turn_on = not instance["root_enabled"]
+ config_path = instance["config_path"]
+ key = f"{constants.INSTANCE_PREFIX}{instance['original_name']}{constants.ENABLE_ROOT_KEY}"
+ if turn_on:
+ if progress:
+ progress("Part 1/2: enabling root access in bluestacks.conf...")
+ config_handler.modify_config_file(config_path, key, "1")
+ config_handler.modify_config_file(config_path, constants.FEATURE_ROOTING_KEY, "1")
+ if progress:
+ progress("Part 2/2: patching guest su in Data.vhdx...")
+ results = su_patch_offline.set_instance_root(instance["data_path"], True, progress)
+ if not su_patch_offline.instance_root_state(instance["data_path"]):
+ self.show_notice.emit(
+ "Boot this instance once first",
+ "%s hasn't generated its root files yet, so there was nothing "
+ "to patch.\n\nStart this instance in BlueStacks, let it fully "
+ "reach the home screen, then close it completely and click "
+ "\"Toggle Root\" again." % unique_id,
+ )
+ else:
+ if progress:
+ progress("Part 1/2: restoring guest su in Data.vhdx...")
+ results = su_patch_offline.set_instance_root(instance["data_path"], False, progress)
+ if progress:
+ progress("Part 2/2: disabling root access in bluestacks.conf...")
+ config_handler.modify_config_file(config_path, key, "0")
+ logger.info("Root %s (patch-mode) for %s: %s", "ON" if turn_on else "OFF",
+ unique_id, " | ".join(results))
+
+ def _toggle_root_conf(self, unique_id, progress=None):
+ if progress:
+ progress("updating bluestacks.conf...")
+ instance = self.instance_data[unique_id]
+ config_path, original_name = instance["config_path"], instance["original_name"]
+ is_currently_on = instance["root_enabled"]
+ setting_key = f"{constants.INSTANCE_PREFIX}{original_name}{constants.ENABLE_ROOT_KEY}"
+ if is_currently_on:
+ config_handler.modify_config_file(config_path, setting_key, "0")
+ any_other_rooted = any(
+ d.get("individual_root_status", False)
+ for uid, d in self.instance_data.items()
+ if uid != unique_id and d["config_path"] == config_path)
+ if not any_other_rooted:
+ config_handler.modify_config_file(config_path, constants.FEATURE_ROOTING_KEY, "0")
+ self._set_classic_app_su(instance, False, progress)
+ else:
+ config_handler.modify_config_file(config_path, setting_key, "1")
+ config_handler.modify_config_file(config_path, constants.FEATURE_ROOTING_KEY, "1")
+ self._set_classic_app_su(instance, True, progress)
+ logger.info(f"Root toggle (conf) processed for {unique_id}")
+
+ def _set_classic_app_su(self, instance, turn_on, progress=None):
+ if not ext4_symlink.tools_available():
+ logger.info("app-su: bundled e2fsprogs not present; skipping symlink")
+ return
+ try:
+ if turn_on:
+ results = ext4_symlink.add_su_symlink(instance["data_path"], progress)
+ else:
+ results = ext4_symlink.remove_su_symlink(instance["data_path"], progress)
+ logger.info("app-su %s: %s", "ON" if turn_on else "OFF", " | ".join(results))
+ except Exception as exc: # noqa: BLE001
+ logger.warning("app-su symlink step failed: %s", exc)
+ if progress:
+ progress("app-root symlink skipped: %s" % exc)
+
+ def _toggle_single_instance_rw(self, unique_id, progress=None):
+ instance = self.instance_data[unique_id]
+ new_mode = constants.MODE_READONLY if instance["rw_mode"] == constants.MODE_READWRITE else constants.MODE_READWRITE
+ if progress:
+ progress("setting disk to %s..." % new_mode)
+ instance_handler.modify_instance_files(instance["data_path"], new_mode)
+ logger.info(f"R/W toggled for instance: {unique_id} to {new_mode}")
+
+ def _perform_operation(self, operation_func, operation_name):
+ selected_ids = self.instances_page.selected_ids()
+ if not selected_ids:
+ QMessageBox.information(self, "No Selection", f"No instances selected to toggle {operation_name}.")
+ return
+
+ if operation_name == "Root":
+ blocked = engine_rules.blocked_for_root_toggle(
+ {uid: self.instance_data[uid] for uid in selected_ids},
+ self._engine_state(),
+ )
+ if blocked:
+ QMessageBox.warning(
+ self, "Patch the engine first",
+ "%s need%s the engine patched before root will work.\n\n"
+ "Patch it from the Dashboard, then try again."
+ % (", ".join(blocked), "" if len(blocked) > 1 else "s"))
+ return
+
+ total = len(selected_ids)
+
+ def job(progress):
+ progress("Closing BlueStacks...", 0)
+ logger.info("Terminating BlueStacks before %s of %d instance(s)", operation_name, total)
+ instance_handler.terminate_bluestacks()
+ QThread.msleep(constants.PROCESS_TERMINATION_WAIT_MS)
+ for idx, uid in enumerate(selected_ids, 1):
+ prefix = "Part %d/%d (%s)" % (idx, total, uid)
+ logger.info("---- %s: %s ----", prefix, operation_name)
+ pct = step_percent(idx - 1, total)
+ progress("%s: %s..." % (prefix, operation_name), pct)
+ operation_func(uid, lambda m, _p=prefix, _pct=pct: progress("%s: %s" % (_p, m), _pct))
+ return ("%s complete for %d instance(s). They're now closed. Start "
+ "them from BlueStacks to use the change." % (operation_name, total))
+
+ self._run_async(job, f"Toggling {operation_name}...")
+
+ def handle_toggle_root(self):
+ self._perform_operation(self._toggle_single_instance_root, "Root")
+
+ def handle_toggle_rw(self):
+ self._perform_operation(self._toggle_single_instance_rw, "R/W")
+
+ def _refresh_running_instances(self) -> None:
+ """Populate the Modules tab's running-instance list.
+
+ The ADB probe is slow (seconds), so it runs on a worker thread and the
+ list is filled in when it returns -- switching to the Modules tab stays
+ instant. Only one probe runs at a time; a request that arrives while one
+ is in flight is coalesced and re-run once it finishes.
+ """
+ install_dirs = [i.get("install_path") for i in self.installations]
+ adb_exe = adb_handler.find_adb(install_dirs)
+ if not adb_exe:
+ self.modules_page.set_running_instances([])
+ return
+ if self._scan_thread is not None:
+ self._scan_pending = True
+ return
+ instances = [
+ (uid, data["config_path"], data["original_name"])
+ for uid, data in self.instance_data.items()
+ ]
+ self._scan_pending = False
+ self.modules_page.set_scanning()
+ self._scan_thread = QThread(self)
+ self._scan_worker = _RunningScanWorker(adb_exe, instances)
+ self._scan_worker.moveToThread(self._scan_thread)
+ self._scan_thread.started.connect(self._scan_worker.run)
+ self._scan_worker.finished.connect(self._on_scan_finished)
+ self._scan_worker.finished.connect(self._scan_thread.quit)
+ # Delete the worker from inside its own still-running event loop; a
+ # deleteLater() issued after the thread has stopped never gets processed.
+ self._scan_worker.finished.connect(self._scan_worker.deleteLater)
+ self._scan_thread.finished.connect(self._cleanup_scan)
+ self._scan_thread.start()
+
+ def _on_scan_finished(self, running_ids: list) -> None:
+ self.modules_page.set_running_instances(running_ids)
+
+ def _cleanup_scan(self) -> None:
+ # The worker deletes itself via its finished -> deleteLater connection.
+ if self._scan_thread is not None:
+ self._scan_thread.deleteLater()
+ self._scan_worker = None
+ self._scan_thread = None
+ # A refresh was requested while the probe was running (e.g. instance data
+ # changed after an operation). Re-run it now against the current data.
+ if self._scan_pending:
+ self._refresh_running_instances()
+
+ def _handle_browse_zip(self) -> None:
+ zip_path, _ = QFileDialog.getOpenFileName(
+ self, "Select Magisk/Kitsune module", "", "Module archives (*.zip)")
+ if zip_path:
+ self.modules_page.set_zip_path(zip_path)
+
+ def _handle_push_module(self) -> None:
+ uid = self.modules_page.selected_instance_id()
+ zip_path = self.modules_page.zip_path()
+ if not uid or not zip_path:
+ return
+ instance = self.instance_data[uid]
+ install_dirs = [i.get("install_path") for i in self.installations]
+ adb_exe = adb_handler.find_adb(install_dirs)
+ if not adb_exe:
+ QMessageBox.warning(
+ self, "ADB not found",
+ "Couldn't find HD-Adb.exe in the BlueStacks install folder, so the "
+ "module can't be pushed.")
+ return
+ port = adb_handler.instance_adb_port(instance["config_path"], instance["original_name"])
+
+ def job(progress):
+ def adb_progress(msg):
+ progress(msg, -1)
+ msg = adb_handler.install_module(adb_exe, port, zip_path, progress=adb_progress)
+ self.show_notice.emit("Module installed", msg)
+ return "Module installed. Close and reopen the instance to activate it."
+
+ self._run_async(job, "Installing %s..." % os.path.basename(zip_path))
+
+ def _action_buttons(self):
+ return [self.instances_page.root_toggle_button, self.instances_page.rw_toggle_button,
+ self.dashboard_page.engine_button, self.modules_page.push_button]
+
+ def _set_busy(self, busy):
+ for b in self._action_buttons():
+ b.setEnabled(not busy)
+ # The Modules push button re-derives its own enabled state on every
+ # radio/zip change, so it needs the busy flag explicitly or it would
+ # re-enable itself mid-operation.
+ self.modules_page.set_busy(busy)
+ if busy:
+ self.status_refresh_timer.stop()
+
+ def _run_async(self, job, start_text):
+ if getattr(self, "_op_thread", None) is not None:
+ QMessageBox.information(self, "Busy", "An operation is already running.")
+ return
+ self._set_busy(True)
+ self.progress_bar.start(start_text)
+ logger.info("==== %s ====", start_text)
+ self._op_thread = QThread(self)
+ self._op_worker = _OpWorker(job)
+ self._op_worker.moveToThread(self._op_thread)
+ self._op_thread.started.connect(self._op_worker.run)
+ self._op_worker.progress.connect(self._on_async_progress)
+ self._op_worker.done.connect(self._on_async_done)
+ self._op_worker.done.connect(self._op_thread.quit)
+ # Delete the worker from inside its own still-running event loop (see
+ # _refresh_running_instances for why a post-stop deleteLater leaks).
+ self._op_worker.done.connect(self._op_worker.deleteLater)
+ self._op_thread.finished.connect(self._cleanup_async)
+ self._op_thread.start()
+
+ def _on_async_progress(self, msg, pct):
+ self.progress_bar.set_progress(msg, None if pct < 0 else pct)
+
+ @pyqtSlot(str, str)
+ def _show_notice(self, title, message):
+ QMessageBox.information(self, title, message)
+
+ def _confirm(self, title: str, text: str, informative_html: str) -> bool:
+ """A Yes/No confirmation, defaulting to No.
+
+ ``informative_html`` is rendered as rich text so bullet lists and
+ emphasis lay out consistently -- plain-text messages with hand-drawn
+ indentation render ragged in a proportional font.
+ """
+ box = QMessageBox(self)
+ box.setIcon(QMessageBox.Question)
+ box.setWindowTitle(title)
+ box.setText(text)
+ box.setTextFormat(Qt.RichText)
+ box.setInformativeText(informative_html)
+ box.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
+ box.setDefaultButton(QMessageBox.No)
+ return box.exec_() == QMessageBox.Yes
+
+ def _on_async_done(self, ok, summary):
+ self._set_busy(False)
+ self.progress_bar.finish(summary if ok else f"Error: {summary}")
+ logger.info("Operation finished (ok=%s): %s", ok, summary)
+ self.update_instance_statuses(preserve_selection=True)
+ state = self._engine_state()
+ self._refresh_patch_ui(state)
+ # A successful (re)patch clears any standing "reverted" alert right away
+ # instead of waiting for the next timer tick. We only ever CLEAR here:
+ # a user-initiated Undo (state now unpatched) must not re-raise the
+ # alert, and resyncing _last_engine_state keeps the next tick from
+ # misreading that deliberate change as an auto-revert.
+ if state == "patched":
+ self.dashboard_page.set_update_reverted(False)
+ self._last_engine_state = state
+ if self.nav_rail.current() == NAV_MODULES:
+ self._refresh_running_instances()
+ self.status_refresh_timer.start(constants.REFRESH_INTERVAL_MS)
+
+ def _cleanup_async(self):
+ # The worker deletes itself via its done -> deleteLater connection.
+ if getattr(self, "_op_thread", None) is not None:
+ self._op_thread.deleteLater()
+ self._op_worker = None
+ self._op_thread = None
+
+ def _install_dirs_or_warn(self):
+ if not admin.is_admin():
+ choice = QMessageBox.question(
+ self, "Administrator required",
+ "Patching the BlueStacks binaries requires administrator rights.\n\n"
+ "Relaunch this app as administrator now?",
+ QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes)
+ if choice == QMessageBox.Yes and admin.relaunch_as_admin():
+ QApplication.quit()
+ return None
+
+ install_dirs = sorted({inst.get("install_path") for inst in self.installations
+ if inst.get("patch_mode") and inst.get("install_path")
+ and os.path.isdir(inst["install_path"])})
+ if not install_dirs:
+ QMessageBox.warning(self, "Not applicable",
+ "No BlueStacks 5.22.150.1014+ install found that needs the "
+ "engine patch (older builds use classic conf rooting).")
+ return None
+ return install_dirs
+
+ def handle_apply_patches(self) -> None:
+ install_dirs = self._install_dirs_or_warn()
+ if install_dirs is None:
+ return
+
+ if not self._confirm(
+ "Enable Root",
+ "Patch the BlueStacks engine to enable root?",
+ "This modifies two BlueStacks program files:
"
+ ""
+ "- Unlocks root for apps and removes the \"illegally tampered\" "
+ "shutdown (HD-Player.exe)
"
+ "- Keeps root enabled across launches "
+ "(HD-MultiInstanceManager.exe)
"
+ "
"
+ "A backup is saved automatically, so you can undo this at any "
+ "time. All BlueStacks processes will be closed first.
"):
+ return
+
+ total = len(install_dirs)
+
+ def job(progress):
+ progress("Closing BlueStacks...", 0)
+ instance_handler.terminate_bluestacks()
+ QThread.msleep(constants.PROCESS_TERMINATION_WAIT_MS)
+ all_results = []
+ for i, install_dir in enumerate(install_dirs, 1):
+ pct = step_percent(i - 1, total)
+ progress("Patching engine %d/%d: HD-Player.exe..." % (i, total), pct)
+ all_results.extend(integrity_patch.patch_installation(install_dir))
+ progress("Patching engine %d/%d: HD-MultiInstanceManager.exe..." % (i, total), pct)
+ all_results.extend(root_persistence.patch_root_persistence(install_dir))
+ for line in all_results:
+ logger.info(" %s", line)
+ logger.info("Engine patched. Next: Toggle Root per instance, then start "
+ "BlueStacks. Disable BstHdUpdaterSvc so an update doesn't "
+ "re-lock it.")
+ return "Engine patched. Now Toggle Root per instance, then start BlueStacks."
+
+ self._run_async(job, "Patching BlueStacks engine...")
+
+ def handle_restore_patches(self) -> None:
+ install_dirs = self._install_dirs_or_warn()
+ if install_dirs is None:
+ return
+ if not self._confirm(
+ "Undo Root Patch",
+ "Restore the original BlueStacks binaries?",
+ "This restores HD-Player.exe and "
+ "HD-MultiInstanceManager.exe from their .prepatch.bak "
+ "backups, undoing the root patch.
"
+ "All BlueStacks processes will be closed first.
"):
+ return
+
+ total = len(install_dirs)
+
+ def job(progress):
+ progress("Closing BlueStacks...", 0)
+ instance_handler.terminate_bluestacks()
+ QThread.msleep(constants.PROCESS_TERMINATION_WAIT_MS)
+ all_results = []
+ for i, install_dir in enumerate(install_dirs, 1):
+ progress("Restoring engine %d/%d..." % (i, total), step_percent(i - 1, total))
+ all_results.extend(integrity_patch.patch_installation(install_dir, restore=True))
+ all_results.extend(root_persistence.patch_root_persistence(install_dir, restore=True))
+ for line in all_results:
+ logger.info(" %s", line)
+ return "Engine binaries restored from backup."
+
+ self._run_async(job, "Restoring BlueStacks engine...")
+
+ def update_instance_statuses(self, preserve_selection: bool = True):
+ self.update_instance_data()
+ self.update_instance_checkboxes(preserve_selection)
+
+ def closeEvent(self, event):
+ # A background engine/root operation writes real binaries and disk
+ # images; tearing the app down mid-write can crash on exit or corrupt
+ # those files. Refuse to close until it finishes rather than killing it.
+ if getattr(self, "_op_thread", None) is not None:
+ QMessageBox.warning(
+ self, "Operation in progress",
+ "A background operation is still running. Please wait for it to "
+ "finish before closing.")
+ event.ignore()
+ return
+ self.status_refresh_timer.stop()
+ # Don't let a running ADB probe outlive the window (QThread would warn
+ # "destroyed while still running"). It's bounded by adb's own timeout.
+ self._scan_pending = False
+ if self._scan_thread is not None:
+ self._scan_thread.quit()
+ self._scan_thread.wait(2000)
+ event.accept()
diff --git a/views/modules_page.py b/views/modules_page.py
new file mode 100644
index 0000000..6d20fbf
--- /dev/null
+++ b/views/modules_page.py
@@ -0,0 +1,111 @@
+"""Modules page: pick a running instance, pick a module zip, push & flash."""
+from __future__ import annotations
+
+from PyQt5.QtCore import pyqtSignal
+from PyQt5.QtWidgets import (
+ QWidget, QVBoxLayout, QLabel, QRadioButton, QButtonGroup, QPushButton,
+)
+
+
+class ModulesPage(QWidget):
+ browse_zip_requested = pyqtSignal()
+ push_requested = pyqtSignal()
+
+ _EMPTY_TEXT = ("No instance is running. Start one from the Instances tab, then "
+ "it will appear here.")
+ _SCANNING_TEXT = "Checking which instances are running..."
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ layout = QVBoxLayout(self)
+
+ layout.addWidget(QLabel("1. Choose a running instance"))
+ self.running_group = QButtonGroup(self)
+ self.running_group.setExclusive(True)
+ self._running_layout = QVBoxLayout()
+ layout.addLayout(self._running_layout)
+ self.no_running_label = QLabel(self._EMPTY_TEXT)
+ self.no_running_label.setWordWrap(True)
+ self.no_running_label.hide()
+ layout.addWidget(self.no_running_label)
+
+ layout.addWidget(QLabel("2. Choose module archive"))
+ self.zip_label = QLabel("No file chosen")
+ self.zip_label.setWordWrap(True)
+ self.browse_button = QPushButton("Browse...")
+ self.browse_button.clicked.connect(self.browse_zip_requested.emit)
+ layout.addWidget(self.zip_label)
+ layout.addWidget(self.browse_button)
+
+ self.push_button = QPushButton("Push and flash module")
+ self.push_button.clicked.connect(self.push_requested.emit)
+ self.push_button.setEnabled(False)
+ layout.addWidget(self.push_button)
+ layout.addStretch(1)
+
+ self._radios: dict[str, QRadioButton] = {}
+ self._busy = False
+
+ def set_busy(self, busy: bool) -> None:
+ """Force the push button disabled while a background op runs, so a
+ radio/zip change mid-operation can't re-enable it out from under the
+ app's busy state."""
+ self._busy = busy
+ self._update_push_enabled()
+
+ def _clear_radios(self) -> None:
+ for radio in self._radios.values():
+ self.running_group.removeButton(radio)
+ radio.deleteLater()
+ self._radios = {}
+ while self._running_layout.count():
+ item = self._running_layout.takeAt(0)
+ if item and item.widget():
+ item.widget().deleteLater()
+
+ def set_scanning(self) -> None:
+ """Show a transient 'checking...' hint while the ADB probe runs on a
+ background thread, so switching to this tab stays instant."""
+ self._clear_radios()
+ self.no_running_label.setText(self._SCANNING_TEXT)
+ self.no_running_label.show()
+ self._update_push_enabled()
+
+ def set_running_instances(self, unique_ids: list) -> None:
+ """Rebuild the radio list. Selection is cleared on every rebuild."""
+ self._clear_radios()
+
+ for uid in unique_ids:
+ radio = QRadioButton(uid)
+ radio.toggled.connect(self._update_push_enabled)
+ self.running_group.addButton(radio)
+ self._running_layout.addWidget(radio)
+ self._radios[uid] = radio
+
+ if unique_ids:
+ self.no_running_label.hide()
+ else:
+ self.no_running_label.setText(self._EMPTY_TEXT)
+ self.no_running_label.show()
+ self._update_push_enabled()
+
+ def set_zip_path(self, path: str) -> None:
+ self.zip_label.setText(path or "No file chosen")
+ self._update_push_enabled()
+
+ def selected_instance_id(self):
+ for uid, radio in self._radios.items():
+ if radio.isChecked():
+ return uid
+ return None
+
+ def zip_path(self) -> str:
+ text = self.zip_label.text()
+ return "" if text == "No file chosen" else text
+
+ def _update_push_enabled(self) -> None:
+ self.push_button.setEnabled(
+ not self._busy
+ and bool(self.selected_instance_id())
+ and bool(self.zip_path())
+ )
diff --git a/views/nav_rail.py b/views/nav_rail.py
new file mode 100644
index 0000000..d8d5ebe
--- /dev/null
+++ b/views/nav_rail.py
@@ -0,0 +1,47 @@
+"""Left navigation rail: Dashboard / Instances / Modules."""
+from __future__ import annotations
+
+from PyQt5.QtCore import pyqtSignal
+from PyQt5.QtWidgets import QFrame, QPushButton, QVBoxLayout
+
+DASHBOARD = "dashboard"
+INSTANCES = "instances"
+MODULES = "modules"
+
+_DESTINATIONS = [
+ (DASHBOARD, "Dashboard"),
+ (INSTANCES, "Instances"),
+ (MODULES, "Modules"),
+]
+
+
+class NavRail(QFrame):
+ """Emits ``navigate(str)`` with one of DASHBOARD/INSTANCES/MODULES."""
+
+ navigate = pyqtSignal(str)
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ self.setObjectName("NavRail")
+ self._buttons: dict[str, QPushButton] = {}
+ layout = QVBoxLayout(self)
+ layout.setContentsMargins(8, 8, 8, 12)
+ for key, label in _DESTINATIONS:
+ btn = QPushButton(label)
+ btn.setCheckable(True)
+ btn.clicked.connect(lambda _checked, k=key: self.select(k))
+ layout.addWidget(btn)
+ self._buttons[key] = btn
+ layout.addStretch(1)
+ self.select(DASHBOARD)
+
+ def select(self, key: str) -> None:
+ for k, btn in self._buttons.items():
+ btn.setChecked(k == key)
+ self.navigate.emit(key)
+
+ def current(self) -> str:
+ for key, btn in self._buttons.items():
+ if btn.isChecked():
+ return key
+ return DASHBOARD
diff --git a/views/progress.py b/views/progress.py
new file mode 100644
index 0000000..054de38
--- /dev/null
+++ b/views/progress.py
@@ -0,0 +1,57 @@
+"""Docked status/progress indicator for long-running operations."""
+from __future__ import annotations
+
+from typing import Optional
+
+from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QProgressBar
+
+
+def step_percent(index: int, total: int) -> int:
+ """Percent complete for step ``index`` of ``total``.
+
+ Clamped to [0, 100]; ``total <= 0`` returns 0 to avoid a
+ divide-by-zero when a job has no steps yet.
+ """
+ if total <= 0:
+ return 0
+ pct = int((index / total) * 100)
+ return max(0, min(100, pct))
+
+
+class OperationProgressBar(QWidget):
+ """A status label that's always visible, plus a QProgressBar that only
+ shows itself while an operation is running (determinate or
+ indeterminate/busy)."""
+
+ def __init__(self, parent: Optional[QWidget] = None):
+ super().__init__(parent)
+ layout = QVBoxLayout(self)
+ layout.setContentsMargins(8, 4, 8, 4)
+ self._label = QLabel("Ready")
+ # Status text can be a full sentence; wrap it so it never clips at the
+ # window edge instead of running off to the right.
+ self._label.setWordWrap(True)
+ self._bar = QProgressBar()
+ self._bar.setTextVisible(False)
+ self._bar.setVisible(False)
+ layout.addWidget(self._label)
+ layout.addWidget(self._bar)
+
+ def start(self, text: str) -> None:
+ self.show() # Ensure parent is visible
+ self._bar.setVisible(True)
+ self.set_progress(text, None)
+
+ def set_progress(self, text: str, pct: Optional[int]) -> None:
+ self._label.setText(text)
+ if pct is None:
+ self._bar.setRange(0, 0) # indeterminate/busy animation
+ else:
+ self._bar.setRange(0, 100)
+ self._bar.setValue(max(0, min(100, pct)))
+
+ def finish(self, text: str) -> None:
+ self._label.setText(text)
+ self._bar.setVisible(False)
+ self._bar.setRange(0, 100)
+ self._bar.setValue(0)
diff --git a/views/theme.py b/views/theme.py
new file mode 100644
index 0000000..9662b23
--- /dev/null
+++ b/views/theme.py
@@ -0,0 +1,51 @@
+"""Light/dark QSS themes and persistence."""
+from __future__ import annotations
+
+from PyQt5.QtCore import QSettings
+
+_ORG = "RobThePCGuy"
+_APP = "BlueStacksRootGUI"
+_SETTINGS_KEY = "theme"
+
+LIGHT = "light"
+DARK = "dark"
+
+_LIGHT_QSS = """
+QWidget { background-color: #f3f3f3; color: #1a1a1a; }
+QPushButton { background-color: #ffffff; border: 1px solid rgba(0,0,0,0.13); border-radius: 7px; padding: 6px 14px; }
+QPushButton:hover { background-color: #f6f6f6; }
+QPushButton:checked { background-color: #005fb8; color: #ffffff; }
+QProgressBar { border: 1px solid rgba(0,0,0,0.13); border-radius: 3px; background: #eaeef2; }
+QProgressBar::chunk { background-color: #005fb8; border-radius: 3px; }
+"""
+
+_DARK_QSS = """
+QWidget { background-color: #202020; color: #ffffff; }
+QPushButton { background-color: #2b2b2b; border: 1px solid rgba(255,255,255,0.11); border-radius: 7px; padding: 6px 14px; color: #ffffff; }
+QPushButton:hover { background-color: #303030; }
+QPushButton:checked { background-color: #60cdff; color: #0a0a0a; }
+QProgressBar { border: 1px solid rgba(255,255,255,0.11); border-radius: 3px; background: #262626; }
+QProgressBar::chunk { background-color: #60cdff; border-radius: 3px; }
+"""
+
+_THEMES = {LIGHT: _LIGHT_QSS, DARK: _DARK_QSS}
+
+
+def stylesheet_for(theme: str) -> str:
+ """QSS text for ``theme`` ("light" or "dark"). Raises ValueError otherwise."""
+ try:
+ return _THEMES[theme]
+ except KeyError:
+ raise ValueError("Unknown theme: %r" % theme) from None
+
+
+def apply_theme(app, theme: str) -> None:
+ """Apply ``theme`` to ``app`` (a QApplication) and persist the choice."""
+ app.setStyleSheet(stylesheet_for(theme))
+ QSettings(_ORG, _APP).setValue(_SETTINGS_KEY, theme)
+
+
+def load_saved_theme() -> str:
+ """The last-persisted theme, defaulting to light if none was saved."""
+ value = QSettings(_ORG, _APP).value(_SETTINGS_KEY, LIGHT)
+ return value if value in _THEMES else LIGHT