import sys import tempfile import types import unittest from pathlib import Path from unittest.mock import Mock sys.modules.setdefault( "decky", types.SimpleNamespace(DECKY_USER_HOME="/home/deck", logger=Mock()), ) sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "py_modules")) from lsfg_vk.configuration import ConfigurationService from lsfg_vk.flatpak_service import FlatpakService from lsfg_vk.flatpak_profile_service import FlatpakProfileService class FakeFlatpakService: def __init__(self, home: Path): self.user_home = home self.config_dir = home / ".config/lsfg-vk" self.config_file_path = self.config_dir / "conf.toml" self.state = {"version": 2, "plugin_owned_branches": [], "prepared_apps": {}} self.commands = [] self.running = "" self.start_times = {} def _read_state(self): return self.state def _write_state(self, state): self.state = state def _override_path(self, app_id): return self.user_home / ".local/share/flatpak/overrides" / app_id def _reset_app_override(self, app_id): self._override_path(app_id).unlink(missing_ok=True) def _process_start_time(self, pid): return self.start_times.get(pid) @staticmethod def _write_file(path, content, mode=0o644): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") path.chmod(mode) def prepare_app(self, app_id): apps = self.state["prepared_apps"] if app_id not in apps: self._reset_app_override(app_id) apps[app_id] = {} self._write_file( self._override_path(app_id), "[Context]\nfilesystems=/config:ro;/dll:ro;\n" "unset-environment=DISABLE_LSFGVK;DISABLE_LSFG;\n" "[Environment]\nLSFGVK_CONFIG=/config/conf.toml\nLSFGVK_FLATPAK=1\n", ) return { "success": True, "owned": True, "prepared": True, "runtime": "org.freedesktop.Platform/x86_64/24.08", "runtime_branch": "24.08", } def remove_app_override(self, app_id): if app_id not in self.state["prepared_apps"]: return {"success": True, "prepared": False, "owned": False} self._reset_app_override(app_id) self.state["prepared_apps"].pop(app_id) return {"success": True, "prepared": False, "owned": False} def get_flatpak_apps(self): app_id = "org.example.Game" return { "success": True, "apps": [{ "app_id": app_id, "app_name": "Example Game", "runtime": "org.freedesktop.Platform/x86_64/24.08", "runtime_branch": "24.08", "runtime_ready": True, "prepared": app_id in self.state["prepared_apps"], "owned": app_id in self.state["prepared_apps"], "error": None, }], } def _run_flatpak_command(self, args, **_kwargs): self.commands.append(args) if args[:3] == ["override", "--user", "--show"]: path = self._override_path(args[3]) return types.SimpleNamespace( returncode=0, stdout=path.read_text(encoding="utf-8") if path.exists() else "", stderr="", ) if args[:3] == ["override", "--user", "--reset"]: self._reset_app_override(args[3]) return types.SimpleNamespace(returncode=0, stdout="", stderr="") if args[0] == "override": app_id = args[-1] path = self._override_path(app_id) content = path.read_text(encoding="utf-8") if path.exists() else "" env = [item.removeprefix("--env=") for item in args if item.startswith("--env=")] unset = [item.removeprefix("--unset-env=") for item in args if item.startswith("--unset-env=")] if unset: content += "\n[Context]\nunset-environment=" + ";".join(unset) + ";\n" if env: content += "\n[Environment]\n" + "\n".join(env) + "\n" self._write_file(path, content) return types.SimpleNamespace(returncode=0, stdout="", stderr="") if args[0] == "ps": return types.SimpleNamespace(returncode=0, stdout=self.running, stderr="") raise AssertionError(args) class FlatpakProfileServiceTests(unittest.TestCase): def setUp(self): self.tempdir = tempfile.TemporaryDirectory() self.home = Path(self.tempdir.name) / "home" / "deck" self.home.mkdir(parents=True) self.flatpak = FakeFlatpakService(self.home) self.runtime = Mock() self.configuration = ConfigurationService(runtime_service=self.runtime) self.configuration.user_home = self.home self.configuration.config_dir = self.flatpak.config_dir self.configuration.config_file_path = self.flatpak.config_file_path self.service = FlatpakProfileService(self.flatpak, self.configuration) self.app_id = "org.example.Game" def tearDown(self): self.tempdir.cleanup() def test_enable_creates_selector_profile_and_default_workarounds(self): result = self.service.enable_app(self.app_id) config = self.configuration.get_flatpak_config(self.app_id) content = self.flatpak._override_path(self.app_id).read_text(encoding="utf-8") self.assertTrue(result["success"]) self.assertTrue(config["exists"]) self.assertEqual(config["profile"], "flatpak:org.example.Game") self.assertEqual(config["config"]["active_in"], []) self.assertIn("LSFGVK_PROFILE=flatpak:org.example.Game", content) self.assertIn("ENABLE_GAMESCOPE_WSI=0", content) self.assertIn("DXVK_HDR=0", content) def test_workaround_update_rebuilds_from_clean_override(self): self.flatpak._write_file( self.flatpak._override_path(self.app_id), "[Environment]\nDXVK_CONFIG=dxgi.syncInterval = 0\nKEEP=yes\n", ) self.assertTrue(self.service.enable_app(self.app_id)["success"]) state = self.service.default_state() state.update({"dxvkFrameRate": 30, "disableHdr": False, "enableZink": True}) result = self.service.set_workaround_state(self.app_id, state) command = next( args for args in reversed(self.flatpak.commands) if args[0] == "override" and any(item.startswith("--env=LSFGVK_PROFILE=") for item in args) ) self.assertTrue(result["success"]) self.assertIn("--env=DXVK_CONFIG=dxvk.maxFrameRate = 30", command) self.assertNotIn("dxgi.syncInterval", " ".join(command)) self.assertNotIn("--env=DXVK_HDR=0", command) self.assertIn("--env=MESA_LOADER_DRIVER_OVERRIDE=zink", command) def test_config_update_returns_without_relisting_flatpaks(self): self.assertTrue(self.service.enable_app(self.app_id)["success"]) self.service.get_app = Mock(side_effect=AssertionError("config updates must not relist Flatpaks")) result = self.service.update_config(self.app_id, {"multiplier": 4}) self.assertTrue(result["success"]) self.assertEqual(result["app_id"], self.app_id) self.assertEqual(result["config"]["multiplier"], 4) self.service.get_app.assert_not_called() def test_remove_resets_override_and_profile(self): self.flatpak._write_file(self.flatpak._override_path(self.app_id), "[Environment]\nKEEP=yes\n") self.assertTrue(self.service.enable_app(self.app_id)["success"]) removed = self.service.remove_app(self.app_id) self.assertTrue(removed["success"]) self.assertFalse(self.flatpak._override_path(self.app_id).exists()) self.assertFalse(self.configuration.get_flatpak_config(self.app_id)["exists"]) def test_external_override_change_is_replaced_on_update(self): self.assertTrue(self.service.enable_app(self.app_id)["success"]) path = self.flatpak._override_path(self.app_id) path.write_text(path.read_text(encoding="utf-8") + "EXTERNAL=yes\n", encoding="utf-8") result = self.service.set_workaround_state(self.app_id, self.service.default_state()) self.assertTrue(result["success"]) self.assertNotIn("EXTERNAL=yes", path.read_text(encoding="utf-8")) def test_running_detection_uses_owned_selector_state(self): self.assertTrue(self.service.enable_app(self.app_id)["success"]) self.flatpak.running = "org.example.Game\ttrue\t1234\norg.other.App\ttrue\t9999\n" self.flatpak.start_times["1234"] = 200 result = self.service.get_running_apps() self.assertTrue(result["success"]) self.assertEqual(result["apps"], [{"app_id": self.app_id, "active": True, "pid": "1234", "start_time": 200}]) def test_process_start_time_parser_handles_parentheses_in_command_name(self): fields = ["S"] + ["0"] * 18 + ["4242"] stat = "1234 (retro)arch) " + " ".join(fields) self.assertEqual(FlatpakService._parse_process_start_time(stat), 4242) if __name__ == "__main__": unittest.main()