diff options
| -rw-r--r-- | py_modules/lsfg_vk/flatpak_service.py | 466 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/plugin.py | 41 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/steam_service.py | 39 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/wrapper_service.py | 210 | ||||
| -rw-r--r-- | tests/test_flatpak_service.py | 341 | ||||
| -rw-r--r-- | tests/test_plugin_flatpak_lifecycle.py | 62 | ||||
| -rw-r--r-- | tests/test_plugin_migration.py | 5 | ||||
| -rw-r--r-- | tests/test_steam_service.py | 120 | ||||
| -rw-r--r-- | tests/test_wrapper_service.py | 145 |
9 files changed, 593 insertions, 836 deletions
diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index f486b74..842d5ca 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,4 +1,4 @@ -"""Flatpak runtime support for classified Steam targets.""" +"""Plugin-level Flatpak capability for the bundled lsfg-vk Vulkan layers.""" from __future__ import annotations @@ -10,7 +10,7 @@ import shutil import subprocess import threading from pathlib import Path -from typing import Dict, Optional, Set +from typing import Dict, Optional, Set, Tuple from .base_service import BaseService from .constants import ( @@ -22,15 +22,12 @@ from .constants import ( class FlatpakService(BaseService): + """Install the shared Flatpak layer and own only the grants we add.""" + EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk" SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08") - DERIVED_RUNTIME_IDS = {"org.gnome.Platform", "org.kde.Platform"} - RUNTIME_METADATA_SECTION = "Extension org.freedesktop.Platform.GL" OWNERSHIP_FILENAME = "flatpak_extensions.json" - OWNERSHIP_VERSION = 1 - APP_ID_PATTERN = re.compile( - r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$" - ) + OWNERSHIP_VERSION = 2 def __init__(self, logger=None): super().__init__(logger) @@ -74,12 +71,6 @@ class FlatpakService(BaseService): return subprocess.run(command, env=env, **kwargs) @classmethod - def _validate_app_id(cls, app_id: str) -> str: - if not isinstance(app_id, str) or not cls.APP_ID_PATTERN.fullmatch(app_id): - raise ValueError("Invalid Flatpak application ID") - return app_id - - @classmethod def _validate_runtime(cls, branch: str) -> str: if branch not in cls.SUPPORTED_RUNTIMES: raise ValueError( @@ -89,33 +80,6 @@ class FlatpakService(BaseService): return branch @classmethod - def runtime_branch_from_ref(cls, runtime_ref: str) -> str: - parts = runtime_ref.strip().split("/") if isinstance(runtime_ref, str) else [] - if len(parts) != 3 or parts[0] != "org.freedesktop.Platform": - raise ValueError(f"Unsupported Flatpak runtime reference: {runtime_ref}") - return cls._validate_runtime(parts[2]) - - @classmethod - def runtime_branch_from_metadata(cls, metadata: str) -> str: - section = None - versions = [] - for raw_line in metadata.splitlines() if isinstance(metadata, str) else []: - line = raw_line.strip() - if line.startswith("[") and line.endswith("]"): - section = line[1:-1].strip() - continue - if section != cls.RUNTIME_METADATA_SECTION: - continue - key, separator, value = line.partition("=") - if separator and key.strip() == "versions": - versions.extend(part.strip() for part in value.split(";")) - for value in versions: - for branch in cls.SUPPORTED_RUNTIMES: - if value == branch or value.startswith(f"{branch}-"): - return branch - raise ValueError("Could not determine a supported Freedesktop base runtime from Flatpak metadata") - - @classmethod def _extension_ref(cls, branch: str) -> str: return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(branch)}" @@ -145,26 +109,44 @@ class FlatpakService(BaseService): installed.add(fields[2]) return installed - def _owned_branches(self) -> Set[str]: + @staticmethod + def _validate_filesystem_path(value: object) -> str: + if not isinstance(value, str) or not value or "\x00" in value: + raise ValueError("Flatpak filesystem ownership entries must be non-empty strings") + path = Path(value) + if not path.is_absolute() or str(path) != value: + raise ValueError("Flatpak filesystem ownership entries must be normalized absolute paths") + return value + + def _read_ownership(self) -> Tuple[Set[str], Set[str]]: path = self.ownership_path if not path.exists() and not path.is_symlink(): - return set() + return set(), set() if path.is_symlink() or not path.is_file(): raise RuntimeError("Flatpak ownership metadata is not a regular file") try: data = json.loads(path.read_text(encoding="utf-8")) + version = data.get("version") branches = data.get("plugin_owned_branches") - if data.get("version") != self.OWNERSHIP_VERSION or not isinstance(branches, list): + filesystems = data.get("plugin_owned_filesystems", []) + if version not in (1, self.OWNERSHIP_VERSION) or not isinstance(branches, list): + raise ValueError("invalid ownership metadata") + if version == self.OWNERSHIP_VERSION and not isinstance(filesystems, list): + raise ValueError("invalid ownership metadata") + owned_branches = {self._validate_runtime(branch) for branch in branches} + if len(owned_branches) != len(branches): raise ValueError("invalid ownership metadata") - owned = {self._validate_runtime(branch) for branch in branches} - if len(owned) != len(branches): + owned_filesystems = { + self._validate_filesystem_path(filesystem) for filesystem in filesystems + } + if len(owned_filesystems) != len(filesystems): raise ValueError("invalid ownership metadata") - return owned + return owned_branches, owned_filesystems except (OSError, json.JSONDecodeError, TypeError, ValueError) as error: raise RuntimeError(f"Could not trust Flatpak ownership metadata: {error}") from error - def _write_owned_branches(self, branches: Set[str]) -> None: - if not branches: + def _write_ownership(self, branches: Set[str], filesystems: Set[str]) -> None: + if not branches and not filesystems: self.ownership_path.unlink(missing_ok=True) return self._write_file( @@ -173,136 +155,192 @@ class FlatpakService(BaseService): { "version": self.OWNERSHIP_VERSION, "plugin_owned_branches": sorted(branches), + "plugin_owned_filesystems": sorted(filesystems), }, indent=2, ) + "\n", ) - def get_extension_status(self): + def _configured_lossless_scaling_directory(self) -> Path: + default = self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling" + if not self.config_file_path.exists(): + return default try: - available = self.check_flatpak_available() - installed = self._installed_extension_branches() if available else set() - return self._success_response( - dict, - "Flatpak runtime extension status retrieved" if available else "Flatpak is not available", - available=available, - extension_id=self.EXTENSION_ID, - supported_branches=list(self.SUPPORTED_RUNTIMES), - installed_branches=sorted(installed), + content = self.config_file_path.read_text(encoding="utf-8") + match = re.search( + r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"', + content, ) + if not match: + return default + configured_dll = json.loads('"' + match.group(1) + '"') + if not configured_dll: + return default + if configured_dll.startswith("~/"): + return self.user_home / configured_dll[2:] + configured_path = Path(configured_dll) + return configured_path.parent if configured_path.is_absolute() else default + except Exception: + return default + + def _filesystem_grant_paths(self) -> Tuple[str, ...]: + paths = ( + str(self.config_dir), + str(self._configured_lossless_scaling_directory()), + ) + return tuple(dict.fromkeys(paths)) + + @staticmethod + def _parse_filesystems(output: str) -> Dict[str, str]: + entries: Dict[str, str] = {} + for raw_line in output.splitlines() if isinstance(output, str) else []: + line = raw_line.strip() + if not line.startswith("filesystems="): + continue + for raw_entry in line.partition("=")[2].split(";"): + entry = raw_entry.strip() + if not entry: + continue + for mode in ("ro", "rw", "create"): + suffix = f":{mode}" + if entry.endswith(suffix): + entries[entry[:-len(suffix)]] = mode + break + else: + entries[entry] = "rw" + return entries + + def _global_filesystems(self) -> Dict[str, str]: + result = self._run_flatpak_command( + ["override", "--user", "--show"], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or "Could not read Flatpak global overrides") + return self._parse_filesystems(result.stdout) + + def _status(self, message: Optional[str] = None): + available = self.check_flatpak_available() + owned_branches, owned_filesystems = self._read_ownership() + installed = self._installed_extension_branches() if available else set() + filesystems = self._global_filesystems() if available else {} + grant_paths = self._filesystem_grant_paths() + filesystem_grants = [ + { + "path": path, + "present": path in filesystems, + "read_only": filesystems.get(path) == "ro", + } + for path in grant_paths + ] + ready = available and all(branch in installed for branch in self.SUPPORTED_RUNTIMES) and all( + grant["present"] for grant in filesystem_grants + ) + return self._success_response( + dict, + message or ("Flatpak support is ready" if ready else "Flatpak support needs setup"), + available=available, + ready=ready, + extension_id=self.EXTENSION_ID, + supported_branches=list(self.SUPPORTED_RUNTIMES), + installed_branches=sorted(installed), + filesystem_grants=filesystem_grants, + missing_filesystem_grants=[ + grant["path"] for grant in filesystem_grants if not grant["present"] + ], + plugin_owned_branches=sorted(owned_branches), + plugin_owned_filesystems=sorted(owned_filesystems), + ) + + def get_extension_status(self): + try: + if not self.check_flatpak_available(): + return self._status("Flatpak is not available") + return self._status() except Exception as error: return self._error_response( dict, str(error), available=False, + ready=False, extension_id=self.EXTENSION_ID, supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=[], + filesystem_grants=[], + missing_filesystem_grants=list(self._filesystem_grant_paths()), ) get_flatpak_support_status = get_extension_status - def _resolve_runtime(self, app_id: str): - self._validate_app_id(app_id) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") + def _install_branch(self, branch: str) -> None: + bundle = self._bundled_extension_path(branch) + if not bundle.is_file(): + raise FileNotFoundError( + f"Bundled Flatpak extension not found at {bundle}; reinstall the plugin" + ) result = self._run_flatpak_command( - ["info", "--show-runtime", app_id], + ["install", "--user", "--noninteractive", "--or-update", str(bundle)], capture_output=True, text=True, ) if result.returncode != 0: - raise OSError(result.stderr.strip() or f"Could not inspect Flatpak app {app_id}") - runtime = result.stdout.strip().splitlines()[0] if result.stdout.strip() else "" - parts = runtime.split("/") - if len(parts) != 3: - raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}") - if parts[0] == "org.freedesktop.Platform": - branch = self._validate_runtime(parts[2]) - elif parts[0] in self.DERIVED_RUNTIME_IDS: - metadata_result = self._run_flatpak_command( - ["info", "--show-metadata", runtime], - capture_output=True, - text=True, + raise OSError(result.stderr.strip() or "Flatpak installation failed") + if branch not in self._installed_extension_branches("user"): + raise RuntimeError( + f"Flatpak install completed but {self._extension_ref(branch)} was not visible afterwards" ) - if metadata_result.returncode != 0: - raise OSError( - metadata_result.stderr.strip() - or f"Could not inspect Flatpak runtime {runtime}" - ) - branch = self.runtime_branch_from_metadata(metadata_result.stdout) - else: - raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}") - return runtime, branch - def resolve_app_support(self, app_id: str): + def _add_filesystem_grant(self, path: str) -> None: + result = self._run_flatpak_command( + ["override", "--user", f"--filesystem={path}:ro"], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or f"Could not grant Flatpak access to {path}") + current = self._global_filesystems() + if current.get(path) != "ro": + raise RuntimeError(f"Flatpak did not confirm the read-only grant for {path}") + + def ensure_plugin_support(self): + """Make the shared layer and exact read-only grants available once.""" try: - app_id = self._validate_app_id(app_id) - runtime, branch = self._resolve_runtime(app_id) - installed = self._installed_extension_branches() - ready = branch in installed - return self._success_response( - dict, - f"lsfg-vk support is ready for {app_id}" if ready - else f"lsfg-vk runtime extension {branch} is required for {app_id}", - flatpak_app_id=app_id, - runtime=runtime, - runtime_branch=branch, - support_status="ready" if ready else "needs-runtime", - extension_installed=ready, - installed_branches=sorted(installed), - ) - except ValueError as error: - return self._success_response( - dict, - str(error), - flatpak_app_id=app_id, - runtime=None, - runtime_branch=None, - support_status="unsupported", - extension_installed=False, - installed_branches=[], - error=str(error), - ) + with self._lock: + if not self.check_flatpak_available(): + return self._status("Flatpak is not available") + owned_branches, owned_filesystems = self._read_ownership() + installed = self._installed_extension_branches() + for branch in self.SUPPORTED_RUNTIMES: + if branch in installed: + continue + self._install_branch(branch) + installed.add(branch) + owned_branches.add(branch) + self._write_ownership(owned_branches, owned_filesystems) + + current_filesystems = self._global_filesystems() + for path in self._filesystem_grant_paths(): + if path in current_filesystems: + continue + self._add_filesystem_grant(path) + owned_filesystems.add(path) + self._write_ownership(owned_branches, owned_filesystems) + + return self._status("Flatpak support is ready") except Exception as error: return self._error_response( dict, str(error), - flatpak_app_id=app_id, - runtime=None, - runtime_branch=None, - support_status="error", - extension_installed=False, + available=self.check_flatpak_available(), + ready=False, + extension_id=self.EXTENSION_ID, + supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=[], + filesystem_grants=[], + missing_filesystem_grants=list(self._filesystem_grant_paths()), ) - def install_extension(self, branch: str): - try: - branch = self._validate_runtime(branch) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - with self._lock: - if branch in self._installed_extension_branches(): - return self._extension_result(branch, True, False, "already installed") - bundle = self._bundled_extension_path(branch) - if not bundle.is_file(): - raise FileNotFoundError(f"Bundled Flatpak extension not found at {bundle}; reinstall the plugin") - result = self._run_flatpak_command( - ["install", "--user", "--noninteractive", "--or-update", str(bundle)], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Flatpak installation failed") - if branch not in self._installed_extension_branches("user"): - raise RuntimeError(f"Flatpak install completed but {self._extension_ref(branch)} was not visible afterwards") - owned = self._owned_branches() - owned.add(branch) - self._write_owned_branches(owned) - return self._extension_result(branch, True, False, "installed") - except Exception as error: - return self._error_response(dict, str(error), runtime_branch=branch, installed=False, enabled=False) - def _remove_extension(self, branch: str) -> bool: if branch not in self._installed_extension_branches("user"): return False @@ -314,111 +352,89 @@ class FlatpakService(BaseService): if result.returncode != 0: raise OSError(result.stderr.strip() or "Flatpak uninstall failed") if branch in self._installed_extension_branches("user"): - raise RuntimeError(f"Flatpak uninstall completed but {self._extension_ref(branch)} is still installed") - return True - - def _extension_result(self, branch: str, installed: bool, removed: bool, verb: str): - return self._success_response( - dict, - f"lsfg-vk {branch} runtime extension {verb}", - runtime_branch=branch, - installed=installed, - enabled=installed, - removed=removed, - ) - - def uninstall_extension(self, branch: str): - try: - branch = self._validate_runtime(branch) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - with self._lock: - owned = self._owned_branches() - if branch not in owned: - installed = branch in self._installed_extension_branches() - return self._extension_result(branch, installed, False, "preserved (not plugin-owned)") - removed = self._remove_extension(branch) - owned.remove(branch) - self._write_owned_branches(owned) - installed = branch in self._installed_extension_branches() - return self._extension_result(branch, installed, removed, "uninstalled") - except Exception as error: - return self._error_response( - dict, - str(error), - runtime_branch=branch, - removed=False, - installed=False, - enabled=False, + raise RuntimeError( + f"Flatpak uninstall completed but {self._extension_ref(branch)} is still installed" ) + return True - def ensure_extension(self, branch: str): - try: - branch = self._validate_runtime(branch) - if branch in self._installed_extension_branches(): - return self._extension_result(branch, True, False, "is ready") - except Exception as error: - return self._error_response(dict, str(error), runtime_branch=branch, support_status="error") - return self.install_extension(branch) - - def ensure_app_support(self, app_id: str): - resolved = self.resolve_app_support(app_id) - if not resolved.get("success") or resolved.get("support_status") != "needs-runtime": - return resolved - result = self.ensure_extension(resolved["runtime_branch"]) - if not result.get("success"): - return self._error_response( - dict, - result.get("error") or "Could not install the required Flatpak runtime extension", - flatpak_app_id=app_id, - runtime=resolved.get("runtime"), - runtime_branch=resolved.get("runtime_branch"), - support_status="error", - extension_installed=False, + def _remove_filesystem_grant(self, path: str) -> bool: + current = self._global_filesystems() + if path not in current: + return False + if current[path] != "ro": + raise RuntimeError( + f"Refusing to remove Flatpak grant for {path}; its permissions changed externally" ) - return self.resolve_app_support(app_id) - - def set_extension_enabled(self, branch: str, enabled: bool): - if type(enabled) is not bool: - return self._error_response(dict, "enabled must be a boolean", runtime_branch=branch, installed=False, enabled=False) - return self.install_extension(branch) if enabled else self.uninstall_extension(branch) + result = self._run_flatpak_command( + ["override", "--user", f"--nofilesystem={path}"], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or f"Could not remove Flatpak access to {path}") + if path in self._global_filesystems(): + raise RuntimeError(f"Flatpak filesystem grant for {path} is still present") + return True def remove_plugin_owned_extensions(self): + """Remove only positively owned user branches and exact filesystem grants.""" try: with self._lock: - owned = self._owned_branches() - if not owned: + owned_branches, owned_filesystems = self._read_ownership() + if not owned_branches and not owned_filesystems: return self._success_response( dict, - "No plugin-owned Flatpak extensions to remove", + "No plugin-owned Flatpak state to remove", removed_branches=[], preserved_branches=[], + removed_filesystem_grants=[], + preserved_filesystem_grants=[], ownership_uncertain=False, ) if not self.check_flatpak_available(): - raise RuntimeError("Flatpak is not available; plugin-owned extension metadata was preserved") - removed, failures = [], [] - for branch in sorted(owned): + raise RuntimeError( + "Flatpak is not available; plugin-owned state metadata was preserved" + ) + + removed_branches, preserved_branches = [], [] + remaining_branches = set(owned_branches) + for branch in sorted(owned_branches): try: self._remove_extension(branch) - removed.append(branch) + removed_branches.append(branch) + remaining_branches.discard(branch) + except Exception as error: + preserved_branches.append(f"{branch}: {error}") + + removed_filesystems, preserved_filesystems = [], [] + remaining_filesystems = set(owned_filesystems) + for path in sorted(owned_filesystems): + try: + self._remove_filesystem_grant(path) + removed_filesystems.append(path) + remaining_filesystems.discard(path) except Exception as error: - failures.append(f"{branch}: {error}") - remaining = owned - set(removed) - self._write_owned_branches(remaining) + preserved_filesystems.append(f"{path}: {error}") + + self._write_ownership(remaining_branches, remaining_filesystems) + failures = [*preserved_branches, *preserved_filesystems] if failures: return self._error_response( dict, "; ".join(failures), - removed_branches=removed, - preserved_branches=sorted(remaining), + removed_branches=removed_branches, + preserved_branches=preserved_branches, + removed_filesystem_grants=removed_filesystems, + preserved_filesystem_grants=preserved_filesystems, ownership_uncertain=False, ) return self._success_response( dict, - "Plugin-owned Flatpak extensions removed", - removed_branches=removed, + "Plugin-owned Flatpak state removed", + removed_branches=removed_branches, preserved_branches=[], + removed_filesystem_grants=removed_filesystems, + preserved_filesystem_grants=[], ownership_uncertain=False, ) except Exception as error: @@ -427,5 +443,7 @@ class FlatpakService(BaseService): str(error), removed_branches=[], preserved_branches=[], + removed_filesystem_grants=[], + preserved_filesystem_grants=[], ownership_uncertain=True, ) diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py index 13df7a4..282ed96 100644 --- a/py_modules/lsfg_vk/plugin.py +++ b/py_modules/lsfg_vk/plugin.py @@ -24,7 +24,14 @@ class Plugin: self.wrapper_service = WrapperService() async def install_lsfg_vk(self): - return self.installation_service.install() + result = self.installation_service.install() + if not result.get("success"): + return result + flatpak = self.flatpak_service.ensure_plugin_support() + result["flatpak_support"] = flatpak + if not flatpak.get("success"): + decky.logger.warning(f"Flatpak support setup was not completed: {flatpak.get('error')}") + return result async def check_lsfg_vk_installed(self): return self.installation_service.check_installation() @@ -36,21 +43,7 @@ class Plugin: return self.configuration_service.get_game_configs() async def get_installed_games(self): - result = self.steam_service.get_installed_games() - if not result.get("success"): - return result - cache: Dict[str, Dict[str, Any]] = {} - for game in result.get("games", []): - transport = game.get("transport", {}) - if transport.get("kind") != "flatpak": - continue - app_id = transport.get("flatpakAppId") - if not app_id: - continue - if app_id not in cache: - cache[app_id] = self.flatpak_service.resolve_app_support(app_id) - game["flatpakSupport"] = cache[app_id] - return result + return self.steam_service.get_installed_games() async def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]): return self.configuration_service.update_game_config(appid, game_name, config) @@ -150,19 +143,21 @@ class Plugin: async def get_flatpak_support_status(self): return self.flatpak_service.get_flatpak_support_status() - async def ensure_flatpak_support(self, flatpak_app_id: str): - return self.flatpak_service.ensure_app_support(flatpak_app_id) - - async def repair_flatpak_support(self, flatpak_app_id: str): - return self.flatpak_service.ensure_app_support(flatpak_app_id) + async def ensure_flatpak_support(self): + return self.flatpak_service.ensure_plugin_support() - async def set_flatpak_extension_enabled(self, version: str, enabled: bool): - return self.flatpak_service.set_extension_enabled(version, enabled) + async def repair_flatpak_support(self): + return self.flatpak_service.ensure_plugin_support() async def _main(self): repair = self.wrapper_service.repair() if not repair.get("success"): decky.logger.error(f"Could not repair lsfg workaround wrapper: {repair.get('error')}") + installation = self.installation_service.check_installation() + if installation.get("installed"): + flatpak = self.flatpak_service.ensure_plugin_support() + if not flatpak.get("success"): + decky.logger.warning(f"Could not ensure Flatpak support: {flatpak.get('error')}") decky.logger.info("decky-lsfg-vk plugin loaded") async def _unload(self): diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py index a952fcb..a7c0083 100644 --- a/py_modules/lsfg_vk/steam_service.py +++ b/py_modules/lsfg_vk/steam_service.py @@ -10,8 +10,11 @@ from .constants import ( WRAPPER_FILENAME, ) -_FLATPAK_APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$") -_WRAPPER_TOKEN = f"~/{WRAPPER_FILENAME}" + +_DIRECT_FLATPAK_TARGETS = { + ("/usr/bin/flatpak",), + ("~/.lsfg", "/usr/bin/flatpak"), +} def _split_command(value: Optional[str]) -> Optional[list[str]]: @@ -23,34 +26,14 @@ def _split_command(value: Optional[str]) -> Optional[list[str]]: return None -def _is_managed_wrapper(value: str) -> bool: - if value in {_WRAPPER_TOKEN, f"$HOME/{WRAPPER_FILENAME}"}: - return True - path = Path(value) - return path.is_absolute() and path.name == WRAPPER_FILENAME - +def classify_shortcut_transport(executable: Optional[str], _launch_options: Optional[str] = None) -> Dict[str, object]: + """Recognize only the direct Flatpak executable Target. -def classify_shortcut_transport(executable: Optional[str], launch_options: Optional[str] = None) -> Dict[str, object]: + Launch scripts and wrapper commands remain host targets. Their launch + options are intentionally opaque to the plugin. + """ executable_tokens = _split_command(executable) - option_tokens = _split_command(launch_options) - if executable_tokens is None or option_tokens is None or not executable_tokens: - return {"kind": "host"} - direct_flatpak = executable_tokens[0] in {"flatpak", "/usr/bin/flatpak"} - managed_wrapper = len(executable_tokens) == 1 and _is_managed_wrapper(executable_tokens[0]) - if not direct_flatpak and not managed_wrapper: - return {"kind": "host"} - arguments = [*executable_tokens[1:], *option_tokens] - if not arguments or arguments[0] != "run": - return {"kind": "host"} - for argument in arguments[1:]: - if argument == "--" or argument.startswith("-"): - continue - return ( - {"kind": "flatpak", "flatpakAppId": argument} - if _FLATPAK_APP_ID.fullmatch(argument) - else {"kind": "host"} - ) - return {"kind": "host"} + return {"kind": "flatpak"} if tuple(executable_tokens or ()) in _DIRECT_FLATPAK_TARGETS else {"kind": "host"} def _first_string(values: Dict[str, object], *keys: str) -> Optional[str]: diff --git a/py_modules/lsfg_vk/wrapper_service.py b/py_modules/lsfg_vk/wrapper_service.py index 980f7ae..807a1c3 100644 --- a/py_modules/lsfg_vk/wrapper_service.py +++ b/py_modules/lsfg_vk/wrapper_service.py @@ -6,7 +6,6 @@ import json import re import shlex import threading -from pathlib import Path from typing import Any, Dict, Optional, Tuple from .base_service import BaseService @@ -43,6 +42,7 @@ class WrapperService(BaseService): "__GLX_VENDOR_LIBRARY_NAME", "GALLIUM_DRIVER", "DXVK_FRAME_RATE", + "LSFGVK_FLATPAK", ) def __init__(self, logger=None): @@ -92,44 +92,28 @@ class WrapperService(BaseService): if not isinstance(raw, dict): raise ValueError("Workaround transport must be an object") kind = raw.get("kind") - if kind == "host": - return {"kind": "host"} - if kind == "flatpak": - app_id = raw.get("flatpakAppId") - if ( - not isinstance(app_id, str) - or not re.fullmatch( - r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$", - app_id, - ) - ): - raise ValueError("Flatpak transport requires a valid application ID") - return {"kind": "flatpak", "flatpakAppId": app_id} + if kind in ("host", "flatpak"): + # Flatpak app IDs are deliberately not part of plugin state. The + # only special target is the direct /usr/bin/flatpak shortcut. + return {"kind": kind} raise ValueError("Workaround transport must be host or flatpak") @classmethod def _validate_entry(cls, raw: Any) -> Dict[str, Any]: if not isinstance(raw, dict): raise ValueError("Workaround AppID entry must be an object") + transport = cls._validate_transport(raw.get("transport")) entry = { "state": cls._validate_state(raw.get("state")), "command_token_added": raw.get("command_token_added", False), - # Version 1 entries had no transport field. They are preserved as - # host entries until the shortcut is explicitly repaired with the - # backend's classified transport. - "transport": cls._validate_transport(raw.get("transport")), } if type(entry["command_token_added"]) is not bool: raise ValueError("command_token_added must be a boolean") - if entry["transport"]["kind"] == "flatpak" and "shortcut_exe" in raw and raw["shortcut_exe"] is not None: - shortcut_exe = raw["shortcut_exe"] - if ( - not isinstance(shortcut_exe, str) - or not shortcut_exe.startswith("/") - or "\x00" in shortcut_exe - or not shortcut_exe.strip() - ): - raise ValueError("shortcut_exe must be an absolute executable path") + if transport["kind"] == "flatpak": + shortcut_exe = raw.get("shortcut_exe") + if shortcut_exe != "/usr/bin/flatpak": + raise ValueError("Flatpak workaround state must save /usr/bin/flatpak") + entry["transport"] = transport entry["shortcut_exe"] = shortcut_exe return entry @@ -160,11 +144,11 @@ class WrapperService(BaseService): if self.sidecar_path.is_symlink() or not self.sidecar_path.is_file(): raise RuntimeError("Workaround state path is not a regular file") try: - raw = json.loads(self.sidecar_path.read_text(encoding="utf-8")) + content = self.sidecar_path.read_text(encoding="utf-8") + raw = json.loads(content) except (OSError, json.JSONDecodeError) as error: raise RuntimeError(f"Could not read workaround state: {error}") from error - document = self._validate_document(raw) - return document, True, self.sidecar_path.read_text(encoding="utf-8") + return self._validate_document(raw), True, content def _wrapper_marker(self) -> bool: if self.wrapper_path.is_symlink() or not self.wrapper_path.exists(): @@ -190,20 +174,19 @@ class WrapperService(BaseService): def _shell(value: str) -> str: return shlex.quote(value) - @staticmethod - def _direct_flatpak_tokens(value: str) -> Optional[list[str]]: - """Parse the supported full executable form: /usr/bin/flatpak run APP.""" - try: - tokens = shlex.split(value, posix=True) - except ValueError: - return None - if len(tokens) >= 3 and Path(tokens[0]).name == "flatpak" and tokens[1] == "run": - return tokens - return None - - @classmethod - def _state_lines(cls, state: Dict[str, Any], shortcut_exe: Optional[str]) -> list[str]: - lines = [" unset " + " ".join(cls.MANAGED_ENV_KEYS)] + def _state_lines(self, state: Dict[str, Any], shortcut_exe: Optional[str]) -> list[str]: + lines = [ + " unset " + " ".join(self.MANAGED_ENV_KEYS), + ' SteamAppId="$appid"', + " export SteamAppId", + f" LSFGVK_CONFIG={self._shell(str(self.config_file_path))}", + " export LSFGVK_CONFIG", + ] + if shortcut_exe: + lines.extend([ + " LSFGVK_FLATPAK=1", + " export LSFGVK_FLATPAK", + ]) if state["disableGamescopeWsi"]: lines.extend([" ENABLE_GAMESCOPE_WSI=0", " export ENABLE_GAMESCOPE_WSI"]) if state["disableHdr"]: @@ -235,68 +218,9 @@ class WrapperService(BaseService): " fi", " export DXVK_CONFIG", ]) - lines.append(f" shortcut_exe={cls._shell(shortcut_exe or '')}") + lines.append(f" shortcut_exe={self._shell(shortcut_exe or '')}") return lines - def _dll_directory(self) -> Path: - if self.config_file_path.exists(): - try: - content = self.config_file_path.read_text(encoding="utf-8") - match = re.search( - r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"', - content, - ) - if match: - configured_dll = json.loads('"' + match.group(1) + '"') - if configured_dll: - return Path(configured_dll).parent - except Exception: - pass - return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling" - - def _flatpak_args(self, state: Dict[str, Any]) -> list[str]: - config_dir = str(self.config_dir) - config_file = str(self.config_file_path) - dll_dir = str(self._dll_directory()) - args = [ - self._shell(f"--filesystem={config_dir}:rw"), - self._shell(f"--filesystem={dll_dir}:ro"), - self._shell(f"--env=LSFGVK_CONFIG={config_file}"), - '"--env=LSFGVK_FLATPAK=1"', - '"--env=SteamAppId=$appid"', - '"--unset-env=DISABLE_LSFGVK" "--unset-env=DISABLE_LSFG"', - '"--unset-env=DISABLE_GAMESCOPE_WSI"', - '"--unset-env=ENABLE_GAMESCOPE_WSI"' if not state["disableGamescopeWsi"] else - '"--env=ENABLE_GAMESCOPE_WSI=0"', - '"--unset-env=DXVK_HDR"' if not state["disableHdr"] else - '"--env=DXVK_HDR=0"', - '"--unset-env=SteamDeck"' if not state["disableSteamdeckMode"] else - '"--env=SteamDeck=0"', - '"--unset-env=DISABLE_VKBASALT" "--unset-env=ENABLE_VKBASALT"', - ] - if state["disableVkbasalt"]: - args.append('"--env=DISABLE_VKBASALT=1"') - args.extend([ - '"--unset-env=MESA_LOADER_DRIVER_OVERRIDE" "--unset-env=__GLX_VENDOR_LIBRARY_NAME" "--unset-env=GALLIUM_DRIVER"', - ]) - if state["enableZink"]: - args.extend([ - '"--env=__GLX_VENDOR_LIBRARY_NAME=mesa"', - '"--env=MESA_LOADER_DRIVER_OVERRIDE=zink"', - '"--env=GALLIUM_DRIVER=zink"', - ]) - args.extend([ - '"--unset-env=DXVK_FRAME_RATE"', - ]) - static_args = " ".join(args) - return [ - ' if [ -n "${DXVK_CONFIG+x}" ]; then', - f' set -- "$flatpak_command" {static_args} "--env=DXVK_CONFIG=$DXVK_CONFIG" "$@"', - " else", - f' set -- "$flatpak_command" {static_args} "$@"', - " fi", - ] - def _render_wrapper(self, document: Dict[str, Any]) -> str: lines = [ "#!/bin/sh", @@ -332,55 +256,9 @@ class WrapperService(BaseService): "esac", "", 'if [ -n "$shortcut_exe" ]; then', - ]) - # The arguments are emitted per branch below so the values are static and - # the wrapper never needs a JSON parser or another helper executable. - lines.append(' case "$appid" in') - for appid in sorted(document["apps"], key=lambda value: int(value)): - entry = document["apps"][appid] - transport = entry.get("transport", {"kind": "host"}) - if transport.get("kind") != "flatpak": - continue - shortcut_exe = entry.get("shortcut_exe", "") - direct_flatpak_tokens = self._direct_flatpak_tokens(shortcut_exe) - if direct_flatpak_tokens is None and Path(shortcut_exe).name != "flatpak": - raise ValueError( - f"Flatpak target {appid} does not use a direct flatpak executable" - ) - lines.append(f" {appid})") - lines.extend([ - *( - [ - f" shortcut_exe={self._shell(direct_flatpak_tokens[0])}", - " set -- " - + " ".join(self._shell(token) for token in direct_flatpak_tokens[1:]) - + ' "$@"', - ] - if direct_flatpak_tokens - else [] - ), - ' if [ "${1-}" != "run" ]; then', - ' echo "lsfg-vk: Flatpak shortcut must use direct flatpak run transport" >&2', - " exit 64", - " fi", - ' flatpak_command="$1"', - " shift", - " flatpak_target=", - ' for flatpak_arg in "$@"; do', - ' case "$flatpak_arg" in', - ' -*) ;;', - ' *) flatpak_target="$flatpak_arg"; break ;;', - " esac", - " done", - f' if [ "$flatpak_target" != {self._shell(transport["flatpakAppId"])} ]; then', - ' echo "lsfg-vk: Flatpak shortcut application ID changed externally" >&2', - " exit 64", - " fi", - ]) - lines.extend(self._flatpak_args(entry["state"])) - lines.append(" ;;") - lines.extend([ - " esac", + ' if [ "${1-}" = "$shortcut_exe" ]; then', + " shift", + " fi", ' exec "$shortcut_exe" "$@"', "fi", 'exec "$@"', @@ -396,7 +274,11 @@ class WrapperService(BaseService): old_sidecar_exists = self.sidecar_path.exists() old_sidecar = self.sidecar_path.read_text(encoding="utf-8") if old_sidecar_exists else None old_wrapper_exists = self.wrapper_path.exists() or self.wrapper_path.is_symlink() - old_wrapper = self.wrapper_path.read_text(encoding="utf-8") if old_wrapper_exists and not self.wrapper_path.is_symlink() else None + old_wrapper = ( + self.wrapper_path.read_text(encoding="utf-8") + if old_wrapper_exists and not self.wrapper_path.is_symlink() + else None + ) try: self._write_document(document) self._write_file(self.wrapper_path, self._render_wrapper(document), 0o755) @@ -465,24 +347,22 @@ class WrapperService(BaseService): document, _, _ = self._read_document() previous_entry = document["apps"].get(normalized) selected_transport = self._validate_transport( - transport - if transport is not None - else ( - previous_entry.get("transport") - if previous_entry - else None + transport if transport is not None else ( + previous_entry.get("transport") if previous_entry else None ) ) entry: Dict[str, Any] = { "state": validated_state, "command_token_added": bool(command_token_added), - "transport": selected_transport, } if selected_transport["kind"] == "flatpak": - if shortcut_exe is not None: - entry = self._validate_entry({**entry, "shortcut_exe": shortcut_exe}) - elif previous_entry and "shortcut_exe" in previous_entry: - entry["shortcut_exe"] = previous_entry["shortcut_exe"] + selected_shortcut = shortcut_exe or ( + previous_entry.get("shortcut_exe") if previous_entry else None + ) + if selected_shortcut != "/usr/bin/flatpak": + raise ValueError("Flatpak workaround state must save /usr/bin/flatpak") + entry["transport"] = selected_transport + entry["shortcut_exe"] = selected_shortcut document["apps"][normalized] = entry self._write_pair(document) return self._response(document, normalized) diff --git a/tests/test_flatpak_service.py b/tests/test_flatpak_service.py index d5baf61..6e7067f 100644 --- a/tests/test_flatpak_service.py +++ b/tests/test_flatpak_service.py @@ -24,17 +24,16 @@ class FlatpakServiceTests(unittest.TestCase): self.service = FlatpakService() self.service.user_home = self.home self.service.config_dir = self.home / ".config/lsfg-vk" + self.service.config_dir.mkdir(parents=True) self.service.config_file_path = self.service.config_dir / "conf.toml" self.service.check_flatpak_available = Mock(return_value=True) self.service._run_flatpak_command = Mock(side_effect=self._run_flatpak_command) - self.runtime_ref = "org.freedesktop.Platform/x86_64/24.08" - self.runtime_metadata = "" self.user_branches = set() self.system_branches = set() - self.install_branch = "24.08" - self.bundle = self.home / "lsfg-vk-24.08.flatpak" - self.bundle.write_bytes(b"bundle") - self.service._bundled_extension_path = Mock(return_value=self.bundle) + self.filesystems = {} + self.service._bundled_extension_path = Mock( + side_effect=lambda branch: self._bundle(branch) + ) def tearDown(self): self.tempdir.cleanup() @@ -47,252 +46,172 @@ class FlatpakServiceTests(unittest.TestCase): def _extension_line(branch): return f"org.freedesktop.Platform.VulkanLayer.lsfgvk\tx86_64\t{branch}\n" + def _bundle(self, branch): + path = self.home / f"lsfg-vk-{branch}.flatpak" + path.write_bytes(b"bundle") + return path + + def _override_output(self): + if not self.filesystems: + return "" + values = ";".join( + f"{path}:{mode}" for path, mode in sorted(self.filesystems.items()) + ) + return f"[Context]\nfilesystems={values};\n" + def _run_flatpak_command(self, args, **_kwargs): - if args[0] == "info" and args[1] == "--show-runtime": - return self._result(self.runtime_ref + "\n") - if args[0] == "info" and args[1] == "--show-metadata": - return self._result(self.runtime_metadata) if args[0] == "list": branches = self.user_branches if "--user" in args else self.system_branches return self._result("".join(self._extension_line(branch) for branch in sorted(branches))) if args[0] == "install": - self.user_branches.add(self.install_branch) + bundle = Path(args[-1]) + self.user_branches.add(bundle.stem.removeprefix("lsfg-vk-")) return self._result() if args[0] == "uninstall": self.user_branches.discard(args[-1].rsplit("/", 1)[-1]) return self._result() - raise AssertionError(f"Unexpected Flatpak command: {args}") - - def test_runtime_branch_mapping_is_strict_and_branch_specific(self): - self.assertEqual( - FlatpakService.runtime_branch_from_ref( - "org.freedesktop.Platform/x86_64/24.08" - ), - "24.08", - ) - self.assertEqual( - FlatpakService.runtime_branch_from_ref( - "org.freedesktop.Platform//25.08" - ), - "25.08", - ) - with self.assertRaises(ValueError): - FlatpakService.runtime_branch_from_ref("org.gnome.Sdk/x86_64/46") - with self.assertRaises(ValueError): - FlatpakService.runtime_branch_from_ref( - "org.freedesktop.Platform/x86_64/26.08" + if args[0] == "override" and "--show" in args: + return self._result(self._override_output()) + if args[0] == "override": + filesystem = next( + (arg.removeprefix("--filesystem=") for arg in args if arg.startswith("--filesystem=")), + None, ) - - def test_runtime_branch_mapping_reads_documented_gl_metadata(self): - metadata = """ -[Extension org.freedesktop.Platform.GL] -versions=25.08;25.08-extra;1.4 -version=1.4 -""" - self.assertEqual(FlatpakService.runtime_branch_from_metadata(metadata), "25.08") - with self.assertRaises(ValueError): - FlatpakService.runtime_branch_from_metadata( - "[Extension org.freedesktop.Platform.GL]\nversions=26.08;26.08-extra;1.4\n" + if filesystem is not None: + path, _, mode = filesystem.rpartition(":") + self.filesystems[path] = mode + return self._result() + filesystem = next( + (arg.removeprefix("--nofilesystem=") for arg in args if arg.startswith("--nofilesystem=")), + None, ) + if filesystem is not None: + self.filesystems.pop(filesystem, None) + return self._result() + raise AssertionError(f"Unexpected Flatpak command: {args}") - def test_resolve_reads_required_runtime_instead_of_any_installed_branch(self): - self.user_branches = {"23.08"} - - response = self.service.resolve_app_support("com.example.Game") + def _mutating_commands(self): + return [ + call.args[0] + for call in self.service._run_flatpak_command.call_args_list + if call.args[0][0] in {"install", "uninstall", "override"} + and "--show" not in call.args[0] + ] - self.assertTrue(response["success"]) - self.assertEqual(response["runtime_branch"], "24.08") - self.assertEqual(response["support_status"], "needs-runtime") - self.assertFalse(response["extension_installed"]) - self.assertEqual( - self.service._run_flatpak_command.call_args_list[0].args[0], - ["info", "--show-runtime", "com.example.Game"], - ) - self.assertEqual( - self.service._run_flatpak_command.call_args_list[1].args[0], - ["list", "--user", "--runtime", "--columns=application,arch,branch"], - ) - self.assertEqual( - self.service._run_flatpak_command.call_args_list[2].args[0], - ["list", "--system", "--runtime", "--columns=application,arch,branch"], + def test_ensure_installs_all_bundled_branches_and_exact_read_only_grants(self): + configured_dll = self.home / "Games" / "Lossless Scaling" / "lsfg-vk.dll" + self.service.config_file_path.write_text( + f'version = 2\n[global]\ndll = {json.dumps(str(configured_dll))}\n', + encoding="utf-8", ) - def test_resolve_maps_kde_and_gnome_runtimes_from_gl_metadata(self): - metadata = "[Extension org.freedesktop.Platform.GL]\nversions=25.08;25.08-extra;1.4\n" - for runtime in ("org.kde.Platform/x86_64/6.10", "org.gnome.Platform/x86_64/49"): - with self.subTest(runtime=runtime): - self.service._run_flatpak_command.reset_mock() - self.runtime_ref = runtime - self.runtime_metadata = metadata - response = self.service.resolve_app_support("com.example.Game") - self.assertEqual(response["runtime_branch"], "25.08") - self.assertEqual(response["support_status"], "needs-runtime") - self.assertEqual( - self.service._run_flatpak_command.call_args_list[1].args[0], - ["info", "--show-metadata", runtime], - ) - - def test_system_extension_is_ready_without_installing_a_user_copy(self): - self.system_branches = {"24.08"} - - response = self.service.ensure_app_support("com.example.Game") + response = self.service.ensure_plugin_support() self.assertTrue(response["success"]) - self.assertEqual(response["support_status"], "ready") - self.assertEqual( - [call.args[0][0] for call in self.service._run_flatpak_command.call_args_list], - ["info", "list", "list"], - ) - self.assertFalse(any(call.args[0][0] == "install" for call in self.service._run_flatpak_command.call_args_list)) - - def test_install_records_only_a_new_user_owned_branch(self): - response = self.service.install_extension("24.08") + self.assertTrue(response["ready"]) + self.assertEqual(response["installed_branches"], ["23.08", "24.08", "25.08"]) + expected_grants = {str(self.service.config_dir), str(configured_dll.parent)} + self.assertEqual(set(self.filesystems), expected_grants) + self.assertEqual(set(self.filesystems.values()), {"ro"}) + metadata = json.loads(self.service.ownership_path.read_text(encoding="utf-8")) + self.assertEqual(metadata["version"], 2) + self.assertEqual(metadata["plugin_owned_branches"], ["23.08", "24.08", "25.08"]) + self.assertEqual(set(metadata["plugin_owned_filesystems"]), expected_grants) + installs = [command for command in self._mutating_commands() if command[0] == "install"] + grants = [command for command in self._mutating_commands() if "--filesystem=" in " ".join(command)] + self.assertEqual(len(installs), 3) + self.assertEqual(len(grants), 2) + self.assertNotIn("--filesystem=/home", " ".join(" ".join(command) for command in grants)) + self.assertNotIn("--filesystem=--", " ".join(" ".join(command) for command in grants)) + + def test_repeated_setup_is_a_no_op_when_state_is_correct(self): + self.assertTrue(self.service.ensure_plugin_support()["success"]) + self.service._run_flatpak_command.reset_mock() + + response = self.service.ensure_plugin_support() self.assertTrue(response["success"]) - self.assertTrue(response["enabled"]) - self.assertTrue(response["installed"]) - install_args = self.service._run_flatpak_command.call_args_list[2].args[0] - self.assertEqual(install_args[:4], ["install", "--user", "--noninteractive", "--or-update"]) - self.assertEqual( - json.loads(self.service.ownership_path.read_text(encoding="utf-8")), - {"version": 1, "plugin_owned_branches": ["24.08"]}, - ) - - def test_preexisting_branch_is_not_claimed_or_removed(self): - self.user_branches = {"24.08"} - - install_response = self.service.install_extension("24.08") - cleanup_response = self.service.remove_plugin_owned_extensions() + self.assertTrue(response["ready"]) + self.assertFalse(self._mutating_commands()) - self.assertTrue(install_response["success"]) - self.assertTrue(install_response["enabled"]) - self.assertTrue(install_response["installed"]) - self.assertFalse(self.service.ownership_path.exists()) - self.assertTrue(cleanup_response["success"]) - self.assertEqual(self.service._run_flatpak_command.call_count, 2) - - def test_extension_toggle_preserves_preexisting_branch(self): - self.user_branches = {"24.08"} - - enable_response = self.service.set_extension_enabled("24.08", True) - disable_response = self.service.set_extension_enabled("24.08", False) - - self.assertTrue(enable_response["success"]) - self.assertTrue(enable_response["enabled"]) - self.assertTrue(enable_response["installed"]) - self.assertTrue(disable_response["success"]) - self.assertTrue(disable_response["enabled"]) - self.assertTrue(disable_response["installed"]) - self.assertFalse(disable_response["removed"]) - self.assertEqual(self.user_branches, {"24.08"}) - self.assertEqual( - [call.args[0][0] for call in self.service._run_flatpak_command.call_args_list], - ["list", "list", "list", "list"], - ) - - def test_extension_toggle_removes_owned_user_branch_but_preserves_system_branch(self): - self.service.ownership_path.parent.mkdir(parents=True, exist_ok=True) - self.service.ownership_path.write_text( - json.dumps({"version": 1, "plugin_owned_branches": ["24.08"]}), - encoding="utf-8", - ) - self.user_branches = {"24.08"} + def test_preexisting_branches_and_grants_are_not_claimed_or_removed(self): + self.user_branches = {"23.08"} self.system_branches = {"24.08"} + preexisting_config = str(self.service.config_dir) + self.filesystems[preexisting_config] = "ro" - disable_response = self.service.set_extension_enabled("24.08", False) - repeat_response = self.service.set_extension_enabled("24.08", False) + setup = self.service.ensure_plugin_support() + cleanup = self.service.remove_plugin_owned_extensions() - self.assertTrue(disable_response["success"]) - self.assertTrue(disable_response["enabled"]) - self.assertTrue(disable_response["removed"]) - self.assertTrue(repeat_response["success"]) - self.assertTrue(repeat_response["enabled"]) - self.assertTrue(repeat_response["installed"]) - self.assertEqual(self.user_branches, set()) + self.assertTrue(setup["success"]) + self.assertTrue(cleanup["success"]) + self.assertEqual(self.user_branches, {"23.08"}) self.assertEqual(self.system_branches, {"24.08"}) + self.assertEqual(self.filesystems, {preexisting_config: "ro"}) + self.assertFalse(self.service.ownership_path.exists()) + repeat = self.service.remove_plugin_owned_extensions() + self.assertTrue(repeat["success"]) + + def test_uninstall_removes_plugin_owned_state_and_preserves_unrelated_override(self): + unrelated = str(self.home / "Games" / "Other") + self.filesystems[unrelated] = "ro" + + self.assertTrue(self.service.ensure_plugin_support()["success"]) + cleanup = self.service.remove_plugin_owned_extensions() + + self.assertTrue(cleanup["success"]) + self.assertEqual(cleanup["removed_branches"], ["23.08", "24.08", "25.08"]) + self.assertEqual(set(cleanup["removed_filesystem_grants"]), { + str(self.service.config_dir), + str(self.home / ".local/share/Steam/steamapps/common/Lossless Scaling"), + }) + self.assertEqual(self.user_branches, set()) + self.assertEqual(self.filesystems, {unrelated: "ro"}) self.assertFalse(self.service.ownership_path.exists()) - uninstall_commands = [ - call.args[0] - for call in self.service._run_flatpak_command.call_args_list - if call.args[0][0] == "uninstall" - ] - self.assertEqual(len(uninstall_commands), 1) - def test_corrupt_ownership_metadata_fails_closed(self): - self.service.ownership_path.parent.mkdir(parents=True, exist_ok=True) - self.service.ownership_path.write_text("{not-json", encoding="utf-8") + def test_cleanup_is_idempotent_after_owned_state_is_gone(self): + self.assertTrue(self.service.ensure_plugin_support()["success"]) + self.assertTrue(self.service.remove_plugin_owned_extensions()["success"]) + self.service._run_flatpak_command.reset_mock() response = self.service.remove_plugin_owned_extensions() - self.assertFalse(response["success"]) - self.assertTrue(response["ownership_uncertain"]) + self.assertTrue(response["success"]) + self.assertFalse(response["ownership_uncertain"]) self.assertEqual(self.service._run_flatpak_command.call_count, 0) - def test_dangling_ownership_symlink_fails_closed(self): - self.service.ownership_path.parent.mkdir(parents=True, exist_ok=True) - self.service.ownership_path.symlink_to(self.home / "missing-metadata") + def test_changed_owned_grant_is_preserved_and_metadata_is_retained(self): + self.assertTrue(self.service.ensure_plugin_support()["success"]) + config_path = str(self.service.config_dir) + self.filesystems[config_path] = "rw" response = self.service.remove_plugin_owned_extensions() self.assertFalse(response["success"]) - self.assertTrue(response["ownership_uncertain"]) - self.assertEqual(self.service._run_flatpak_command.call_count, 0) - - def test_ensure_app_support_installs_only_the_app_runtime_branch(self): - response = self.service.ensure_app_support("com.example.Game") + self.assertEqual(self.filesystems[config_path], "rw") + self.assertTrue(self.service.ownership_path.exists()) + remaining = json.loads(self.service.ownership_path.read_text(encoding="utf-8")) + self.assertIn(config_path, remaining["plugin_owned_filesystems"]) - self.assertTrue(response["success"]) - self.assertEqual(response["support_status"], "ready") - self.assertEqual(response["runtime_branch"], "24.08") - install_args = next( - call.args[0] - for call in self.service._run_flatpak_command.call_args_list - if call.args[0][0] == "install" - ) - self.assertEqual(install_args[0], "install") - self.assertIn("--user", install_args) - self.assertNotIn("23.08", install_args) - self.assertEqual( - json.loads(self.service.ownership_path.read_text(encoding="utf-8")), - {"version": 1, "plugin_owned_branches": ["24.08"]}, - ) + def test_corrupt_ownership_metadata_fails_closed_without_flatpak_commands(self): + self.service.ownership_path.write_text("{not-json", encoding="utf-8") - def test_two_shortcuts_using_one_flatpak_share_one_extension_branch(self): - first = self.service.ensure_app_support("net.pcsx2.PCSX2") - second = self.service.ensure_app_support("net.pcsx2.PCSX2.Dev") + response = self.service.remove_plugin_owned_extensions() - self.assertEqual(first["support_status"], "ready") - self.assertEqual(second["support_status"], "ready") - install_commands = [ - call.args[0] - for call in self.service._run_flatpak_command.call_args_list - if call.args[0][0] == "install" - ] - self.assertEqual(len(install_commands), 1) - self.assertEqual( - json.loads(self.service.ownership_path.read_text(encoding="utf-8")), - {"version": 1, "plugin_owned_branches": ["24.08"]}, - ) + self.assertFalse(response["success"]) + self.assertTrue(response["ownership_uncertain"]) + self.assertEqual(self.service._run_flatpak_command.call_count, 0) - def test_cleanup_removes_all_owned_branches_without_reusing_stale_metadata(self): - self.service.ownership_path.parent.mkdir(parents=True, exist_ok=True) - self.service.ownership_path.write_text( - json.dumps({"version": 1, "plugin_owned_branches": ["23.08", "24.08"]}), - encoding="utf-8", + def test_global_override_parser_handles_permissions_and_unrelated_entries(self): + parsed = FlatpakService._parse_filesystems( + "[Context]\nfilesystems=/home/deck/config:ro;/home/deck/other:rw;/tmp/create:create;\n" ) - self.user_branches = {"23.08", "24.08"} - - response = self.service.remove_plugin_owned_extensions() - self.assertTrue(response["success"]) - self.assertEqual(response["removed_branches"], ["23.08", "24.08"]) - self.assertFalse(self.service.ownership_path.exists()) - uninstall_commands = [ - call.args[0] - for call in self.service._run_flatpak_command.call_args_list - if call.args[0][0] == "uninstall" - ] - self.assertEqual(len(uninstall_commands), 2) + self.assertEqual(parsed, { + "/home/deck/config": "ro", + "/home/deck/other": "rw", + "/tmp/create": "create", + }) if __name__ == "__main__": diff --git a/tests/test_plugin_flatpak_lifecycle.py b/tests/test_plugin_flatpak_lifecycle.py new file mode 100644 index 0000000..25f09b6 --- /dev/null +++ b/tests/test_plugin_flatpak_lifecycle.py @@ -0,0 +1,62 @@ +import asyncio +import sys +import types +import unittest +from unittest.mock import Mock + + +sys.modules.setdefault( + "decky", + types.SimpleNamespace( + DECKY_HOME="/decky", + DECKY_USER_HOME="/home/deck", + logger=Mock(), + ), +) +sys.modules.setdefault("tomllib", types.SimpleNamespace(loads=Mock())) +sys.path.insert(0, "py_modules") + +from lsfg_vk.plugin import Plugin + + +class PluginFlatpakLifecycleTests(unittest.TestCase): + def test_install_and_reload_ensure_global_flatpak_support(self): + plugin = Plugin.__new__(Plugin) + plugin.installation_service = Mock() + plugin.installation_service.install.return_value = {"success": True} + plugin.installation_service.check_installation.return_value = {"installed": True} + plugin.flatpak_service = Mock() + plugin.flatpak_service.ensure_plugin_support.return_value = {"success": True} + plugin.wrapper_service = Mock() + plugin.wrapper_service.repair.return_value = {"success": True} + + installed = asyncio.run(plugin.install_lsfg_vk()) + asyncio.run(plugin._main()) + + self.assertTrue(installed["success"]) + self.assertEqual(plugin.flatpak_service.ensure_plugin_support.call_count, 2) + plugin.wrapper_service.repair.assert_called_once_with() + + def test_unload_does_not_remove_persistent_flatpak_setup(self): + plugin = Plugin.__new__(Plugin) + plugin.installation_service = Mock() + plugin.flatpak_service = Mock() + + asyncio.run(plugin._unload()) + + plugin.flatpak_service.remove_plugin_owned_extensions.assert_not_called() + + def test_uninstall_is_the_flatpak_cleanup_boundary(self): + plugin = Plugin.__new__(Plugin) + plugin.installation_service = Mock() + plugin.flatpak_service = Mock() + plugin.flatpak_service.remove_plugin_owned_extensions.return_value = {"success": True} + + asyncio.run(plugin._uninstall()) + + plugin.installation_service.cleanup_on_uninstall.assert_called_once_with() + plugin.flatpak_service.remove_plugin_owned_extensions.assert_called_once_with() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_plugin_migration.py b/tests/test_plugin_migration.py index a392f85..100f677 100644 --- a/tests/test_plugin_migration.py +++ b/tests/test_plugin_migration.py @@ -21,7 +21,10 @@ class PluginMigrationTests(unittest.TestCase): sys.modules["tomllib"] = types.SimpleNamespace(loads=Mock()) try: sys.path.insert(0, "py_modules") + import lsfg_vk.plugin as plugin_module from lsfg_vk.plugin import Plugin + previous_plugin_decky = plugin_module.decky + plugin_module.decky = decky plugin = Plugin.__new__(Plugin) plugin.installation_service = Mock() @@ -35,6 +38,8 @@ class PluginMigrationTests(unittest.TestCase): plugin.installation_service.install.assert_not_called() plugin.flatpak_service.migrate_v2.assert_not_called() finally: + if "plugin_module" in locals(): + plugin_module.decky = previous_plugin_decky sys.path.remove("py_modules") if previous_decky is None: sys.modules.pop("decky", None) diff --git a/tests/test_steam_service.py b/tests/test_steam_service.py index 636c87f..95bd2a2 100644 --- a/tests/test_steam_service.py +++ b/tests/test_steam_service.py @@ -1,3 +1,4 @@ +import tempfile import sys import types import unittest @@ -15,102 +16,79 @@ from lsfg_vk.steam_service import SteamService, classify_shortcut_transport class SteamTransportTests(unittest.TestCase): - def test_only_direct_canonical_flatpak_forms_are_classified(self): + def test_only_a_direct_usr_bin_flatpak_target_is_special(self): self.assertEqual( - classify_shortcut_transport( - "/usr/bin/flatpak", - "run com.example.PCSX2 --fullscreen", - ), - {"kind": "flatpak", "flatpakAppId": "com.example.PCSX2"}, - ) - self.assertEqual( - classify_shortcut_transport( - "flatpak", - "run com.example.PCSX2 --fullscreen", - ), - {"kind": "flatpak", "flatpakAppId": "com.example.PCSX2"}, - ) - self.assertEqual( - classify_shortcut_transport( - "/usr/bin/flatpak run com.example.PCSX2", - "--fullscreen", - ), - {"kind": "flatpak", "flatpakAppId": "com.example.PCSX2"}, - ) - self.assertEqual( - classify_shortcut_transport( - "/usr/bin/bash", - "~/launch-game.sh --fullscreen", - ), - {"kind": "host"}, - ) - self.assertEqual( - classify_shortcut_transport( - "/usr/bin/flatpak", - "--user run com.example.PCSX2", - ), - {"kind": "host"}, - ) - self.assertEqual( - classify_shortcut_transport( - "/usr/bin/flatpak", - "run bash ~/launch-game.sh", - ), - {"kind": "host"}, - ) - self.assertEqual( - classify_shortcut_transport( - "~/.lsfg", - "run --branch=stable --arch=x86_64 com.example.PCSX2", - ), - {"kind": "flatpak", "flatpakAppId": "com.example.PCSX2"}, + classify_shortcut_transport("/usr/bin/flatpak", "run com.example.Game --fullscreen"), + {"kind": "flatpak"}, ) + for executable, options in ( + ("flatpak", "run com.example.Game"), + ("/usr/bin/flatpak run com.example.Game", "--fullscreen"), + ("~/.lsfg", "run com.example.Game"), + ("/usr/bin/bash", "~/launch-game.sh --fullscreen"), + ): + with self.subTest(executable=executable, options=options): + self.assertEqual(classify_shortcut_transport(executable, options), {"kind": "host"}) + self.assertEqual( - classify_shortcut_transport( - "/home/deck/.lsfg", - "run com.example.PCSX2", - ), - {"kind": "flatpak", "flatpakAppId": "com.example.PCSX2"}, + classify_shortcut_transport("/usr/bin/flatpak", "--user run com.example.Game"), + {"kind": "flatpak"}, ) self.assertEqual( - classify_shortcut_transport("~/.lsfg", "--profile high"), - {"kind": "host"}, + classify_shortcut_transport('"~/.lsfg" "/usr/bin/flatpak"', "run com.example.Game"), + {"kind": "flatpak"}, ) + self.assertEqual(classify_shortcut_transport("~/.lsfg", "run com.example.Game"), {"kind": "host"}) - def test_shortcut_data_preserves_transport_inputs(self): + def test_direct_flatpak_shortcut_keeps_arguments_without_an_app_id(self): game = SteamService._shortcut_game( { "appid": 123456, - "AppName": "PCSX2 shortcut", + "AppName": "Flatpak shortcut", "Exe": "/usr/bin/flatpak", "LaunchOptions": "run net.pcsx2.PCSX2 --fullscreen", "StartDir": "/home/deck/Games", } ) - self.assertEqual(game["appid"], "123456") - self.assertEqual(game["transport"], { - "kind": "flatpak", - "flatpakAppId": "net.pcsx2.PCSX2", - }) + self.assertEqual(game["transport"], {"kind": "flatpak"}) self.assertEqual(game["executable"], "/usr/bin/flatpak") self.assertEqual(game["arguments"], "run net.pcsx2.PCSX2 --fullscreen") self.assertEqual(game["startDir"], "/home/deck/Games") - def test_wrapped_flatpak_shortcut_remains_a_flatpak_target(self): + def test_shell_targets_are_opaque_host_targets(self): game = SteamService._shortcut_game( { - "appid": 987654, - "AppName": "Wrapped Flatpak", - "Exe": "~/.lsfg", - "LaunchOptions": "run --branch=stable --arch=x86_64 com.example.Game", + "appid": 654321, + "AppName": "EmuDeck-style launcher", + "Exe": "/usr/bin/bash", + "LaunchOptions": "~/Emulators/launch-game.sh --game-id foo", } ) - self.assertEqual(game["transport"], { - "kind": "flatpak", - "flatpakAppId": "com.example.Game", - }) + self.assertEqual(game["transport"], {"kind": "host"}) + + def test_normal_steam_manifest_always_uses_host_transport(self): + with tempfile.TemporaryDirectory() as directory: + home = Path(directory) + steamapps = home / ".local/share/Steam/steamapps" + steamapps.mkdir(parents=True) + (steamapps / "appmanifest_123.acf").write_text( + '"AppState" {\n\t"name" "Steam game"\n}\n', + encoding="utf-8", + ) + service = SteamService() + service.user_home = home + + response = service.get_installed_games() + + self.assertTrue(response["success"]) + self.assertEqual(response["games"], [{ + "appid": "123", + "name": "Steam game", + "nonSteam": False, + "transport": {"kind": "host"}, + }]) if __name__ == "__main__": diff --git a/tests/test_wrapper_service.py b/tests/test_wrapper_service.py index 0632d93..52b2aef 100644 --- a/tests/test_wrapper_service.py +++ b/tests/test_wrapper_service.py @@ -1,3 +1,4 @@ +import json import os import subprocess import sys @@ -27,6 +28,7 @@ class WrapperServiceTests(unittest.TestCase): self.service.local_bin_dir = self.home / ".local/bin" self.service.config_dir = self.home / ".config/lsfg-vk" self.service.sidecar_path = self.service.config_dir / "workarounds.json" + self.service.config_file_path = self.service.config_dir / "conf.toml" self.service.wrapper_path = self.service.local_bin_dir / "lsfg" def tearDown(self): @@ -51,6 +53,7 @@ class WrapperServiceTests(unittest.TestCase): def test_writes_owned_dispatcher_and_validates_shell(self): response = self.service.set("123", self._state(dxvkFrameRate=60, enableZink=True)) + self.assertTrue(response["success"]) self.assertEqual(response["wrapper_path"], "~/.lsfg") self.assertTrue(response["wrapper_owned"]) @@ -58,8 +61,10 @@ class WrapperServiceTests(unittest.TestCase): self.assertEqual(subprocess.run(["/bin/sh", "-n", str(self.service.wrapper_path)]).returncode, 0) self.assertIn(self.service.MARKER, self.service.wrapper_path.read_text(encoding="utf-8")) self.assertEqual(self.service.get("123")["state"], self._state(dxvkFrameRate=60, enableZink=True)) + document = json.loads(self.service.sidecar_path.read_text(encoding="utf-8")) + self.assertNotIn("transport", document["apps"]["123"]) - def test_dispatch_clears_managed_values_preserves_other_environment_and_appends_config(self): + def test_dispatch_exports_required_workaround_context_and_clears_disable_flags(self): self.service.set( "123", self._state(dxvkFrameRate=30, disableSteamdeckMode=True, disableVkbasalt=True, enableZink=True), @@ -72,11 +77,15 @@ class WrapperServiceTests(unittest.TestCase): "DXVK_FRAME_RATE": "5", "ENABLE_GAMESCOPE_WSI": "1", "DISABLE_VKBASALT": "0", + "DISABLE_LSFGVK": "1", + "DISABLE_LSFG": "1", "MESA_LOADER_DRIVER_OVERRIDE": "llvmpipe", "MANGOHUD": "1", }, ) values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line) + self.assertEqual(values["SteamAppId"], "123") + self.assertEqual(values["LSFGVK_CONFIG"], str(self.service.config_file_path)) self.assertEqual(values["ENABLE_GAMESCOPE_WSI"], "0") self.assertEqual(values["DXVK_HDR"], "0") self.assertEqual(values["SteamDeck"], "0") @@ -87,7 +96,8 @@ class WrapperServiceTests(unittest.TestCase): self.assertEqual(values["DXVK_CONFIG"], "dxgi.syncInterval = 0; dxvk.maxFrameRate = 30") self.assertEqual(values["MANGOHUD"], "1") self.assertNotIn("DXVK_FRAME_RATE", values) - self.assertNotIn("ENABLE_VKBASALT", values) + self.assertNotIn("DISABLE_LSFGVK", values) + self.assertNotIn("DISABLE_LSFG", values) def test_appid_fallback_and_unmatched_passthrough(self): self.service.set("123", self._state(disableGamescopeWsi=False, disableHdr=False)) @@ -100,8 +110,9 @@ class WrapperServiceTests(unittest.TestCase): check=True, ) fallback_values = dict(line.split("=", 1) for line in fallback.stdout.splitlines() if "=" in line) + self.assertEqual(fallback_values["SteamAppId"], "456") self.assertEqual(fallback_values["SteamDeck"], "0") - self.assertEqual(fallback_values["SteamGameId"], "456") + self.assertEqual(fallback_values["LSFGVK_CONFIG"], str(self.service.config_file_path)) passthrough = subprocess.run( [str(self.service.wrapper_path), "/usr/bin/env"], @@ -114,130 +125,35 @@ class WrapperServiceTests(unittest.TestCase): self.assertEqual(passthrough_values["KEEP"], "yes") self.assertEqual(passthrough_values["DXVK_HDR"], "1") - def test_flatpak_shortcut_receives_env_arguments_and_original_target(self): - fake_flatpak = self.home / ".local/bin/flatpak" - fake_flatpak.parent.mkdir(parents=True, exist_ok=True) - fake_flatpak.write_text( - "#!/bin/sh\n" - "printf 'ARG:%s\\n' \"$@\"\n", - encoding="utf-8", - ) - fake_flatpak.chmod(0o755) - self.service.set( - "123", - self._state(dxvkFrameRate=20, enableZink=True), - str(fake_flatpak), - False, - {"kind": "flatpak", "flatpakAppId": "com.example.Game"}, - ) - result = self._run(123, "run", "com.example.Game", "--windowed", env={"DXVK_CONFIG": "foo=1"}) - args = result.stdout.splitlines() - self.assertEqual(args[0], "ARG:run") - self.assertIn("ARG:--filesystem=" + str(self.service.config_dir) + ":rw", args) - self.assertIn("ARG:--filesystem=" + str(self.home / ".local/share/Steam/steamapps/common/Lossless Scaling") + ":ro", args) - self.assertIn("ARG:--env=LSFGVK_CONFIG=" + str(self.service.config_file_path), args) - self.assertIn("ARG:--env=LSFGVK_FLATPAK=1", args) - self.assertIn("ARG:--env=SteamAppId=123", args) - self.assertIn("ARG:--env=ENABLE_GAMESCOPE_WSI=0", args) - self.assertIn("ARG:--env=DXVK_HDR=0", args) - self.assertIn("ARG:--env=__GLX_VENDOR_LIBRARY_NAME=mesa", args) - self.assertIn("ARG:--env=MESA_LOADER_DRIVER_OVERRIDE=zink", args) - self.assertIn("ARG:--env=GALLIUM_DRIVER=zink", args) - self.assertIn("ARG:--env=DXVK_CONFIG=foo=1; dxvk.maxFrameRate = 20", args) - self.assertIn("ARG:com.example.Game", args) - self.assertIn("ARG:--windowed", args) - - def test_flatpak_full_executable_form_is_preserved(self): - fake_flatpak = self.home / ".local/bin/flatpak" - fake_flatpak.parent.mkdir(parents=True, exist_ok=True) - fake_flatpak.write_text( - "#!/bin/sh\n" - "printf 'ARG:%s\\n' \"$@\"\n", - encoding="utf-8", - ) - fake_flatpak.chmod(0o755) + def test_direct_flatpak_state_has_only_original_target_transport(self): response = self.service.set( "123", - self._state(), - f"{fake_flatpak} run com.example.Game", - False, - {"kind": "flatpak", "flatpakAppId": "com.example.Game"}, - ) - self.assertTrue(response["success"]) - result = self._run(123, "--windowed") - args = result.stdout.splitlines() - self.assertEqual(args[0], "ARG:run") - self.assertIn("ARG:com.example.Game", args) - self.assertIn("ARG:--windowed", args) - - def test_host_transport_does_not_store_shortcut_target(self): - self.service.set( - "123", - self._state(), + self._state(enableZink=True), "/usr/bin/flatpak", False, - {"kind": "flatpak", "flatpakAppId": "com.example.Game"}, - ) - response = self.service.set( - "123", - self._state(), - "/usr/bin/ignored", - False, - {"kind": "host"}, - ) - self.assertTrue(response["success"]) - self.assertIsNone(response["shortcut_exe"]) - self.assertIsNone(self.service.get("123")["shortcut_exe"]) - - def test_flatpak_transport_rejects_non_run_invocation(self): - fake_flatpak = self.home / ".local/bin/flatpak" - fake_flatpak.parent.mkdir(parents=True, exist_ok=True) - fake_flatpak.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") - fake_flatpak.chmod(0o755) - response = self.service.set( - "123", - self._state(), - str(fake_flatpak), - False, - {"kind": "flatpak", "flatpakAppId": "com.example.Game"}, + {"kind": "flatpak"}, ) - self.assertTrue(response["success"]) - result = subprocess.run( - [str(self.service.wrapper_path), "bash", "launch-game.sh"], - env={"PATH": "/usr/bin:/bin", "SteamAppId": "123"}, - capture_output=True, - text=True, - ) - self.assertEqual(result.returncode, 64) - self.assertIn("direct flatpak run", result.stderr) - def test_flatpak_transport_rejects_external_app_id_change(self): - fake_flatpak = self.home / ".local/bin/flatpak" - fake_flatpak.parent.mkdir(parents=True, exist_ok=True) - fake_flatpak.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") - fake_flatpak.chmod(0o755) - response = self.service.set( - "123", - self._state(), - str(fake_flatpak), - False, - {"kind": "flatpak", "flatpakAppId": "com.example.Game"}, - ) self.assertTrue(response["success"]) - result = subprocess.run( - [str(self.service.wrapper_path), "run", "com.other.Game"], - env={"PATH": "/usr/bin:/bin", "SteamAppId": "123"}, - capture_output=True, - text=True, - ) - self.assertEqual(result.returncode, 64) - self.assertIn("application ID changed externally", result.stderr) + self.assertEqual(response["transport"], {"kind": "flatpak"}) + self.assertEqual(response["shortcut_exe"], "/usr/bin/flatpak") + document = json.loads(self.service.sidecar_path.read_text(encoding="utf-8")) + self.assertEqual(document["apps"]["123"]["transport"], {"kind": "flatpak"}) + wrapper = self.service.wrapper_path.read_text(encoding="utf-8") + self.assertIn("LSFGVK_FLATPAK=1", wrapper) + self.assertIn("SteamAppId=\"$appid\"", wrapper) + self.assertIn("export LSFGVK_CONFIG", wrapper) + self.assertNotIn("--filesystem=", wrapper) + self.assertIn('exec "$shortcut_exe" "$@"', wrapper) + self.assertEqual(subprocess.run(["/bin/sh", "-n", str(self.service.wrapper_path)]).returncode, 0) def test_invalid_state_and_foreign_wrapper_fail_closed(self): invalid = self.service.set("0", self.service.default_state()) self.assertFalse(invalid["success"]) invalid = self.service.set("123", {**self.service.default_state(), "dxvkFrameRate": 61}) self.assertFalse(invalid["success"]) + invalid = self.service.set("123", self.service.default_state(), "/tmp/flatpak", False, {"kind": "flatpak"}) + self.assertFalse(invalid["success"]) self.service.local_bin_dir.mkdir(parents=True, exist_ok=True) self.service.wrapper_path.write_text("#!/bin/sh\necho foreign\n", encoding="utf-8") @@ -249,6 +165,7 @@ class WrapperServiceTests(unittest.TestCase): def test_remove_keeps_a_safe_owned_passthrough_wrapper(self): self.service.set("123", self.service.default_state()) response = self.service.remove("123") + self.assertTrue(response["success"]) self.assertIsNone(self.service.get("123")["state"]) self.assertTrue(self.service.wrapper_path.exists()) |