import asyncio import logging import sys import tempfile import types import unittest from pathlib import Path from unittest import mock PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT / "py_modules")) sys.modules.setdefault("decky", types.SimpleNamespace(logger=logging.getLogger("decky-test"))) from lsfg_vk.base_service import BaseService from lsfg_vk.config_schema import ConfigurationManager from lsfg_vk.configuration import ConfigurationService LEGACY_CONFIG = '''\ version = 1 [global] current_profile = "decky-lsfg-vk" dll = "/games/Lossless Scaling/Lossless.dll" no_fp16 = true [[game]] exe = "decky-lsfg-vk" multiplier = 1 flow_scale = 0.9 performance_mode = true hdr_mode = true experimental_present_mode = "mailbox" ''' class ConfigurationMigrationTests(unittest.TestCase): def test_v1_configuration_is_converted_to_valid_v2_toml(self): migrated = ConfigurationManager.parse_toml_content_multi_profile(LEGACY_CONFIG) serialized = ConfigurationManager.generate_toml_content_multi_profile(migrated) self.assertIn("version = 2", serialized) self.assertIn("allow_fp16 = false", serialized) self.assertIn("[[profile]]", serialized) self.assertIn('name = "decky-lsfg-vk"', serialized) self.assertIn("multiplier = 2", serialized) self.assertNotIn("[[game]]", serialized) self.assertNotIn("no_fp16", serialized) self.assertNotIn("hdr_mode", serialized) self.assertNotIn("experimental_present_mode", serialized) def test_v2_normalization_is_idempotent(self): migrated = ConfigurationManager.parse_toml_content_multi_profile(LEGACY_CONFIG) first_write = ConfigurationManager.generate_toml_content_multi_profile(migrated) second_write = ConfigurationManager.generate_toml_content_multi_profile( ConfigurationManager.parse_toml_content_multi_profile(first_write) ) self.assertEqual(second_write, first_write) def test_launch_script_uses_v2_profile_override(self): service = ConfigurationService.__new__(ConfigurationService) service.local_share_dir = Path("/home/deck/.local/share/vulkan/implicit_layer.d") service.config_file_path = Path("/home/deck/.config/lsfg-vk/conf.toml") profile_data = { "current_profile": "decky-lsfg-vk", "profiles": {"decky-lsfg-vk": ConfigurationManager.get_defaults()}, "global_config": {"dll": "/games/Lossless Scaling/Lossless.dll", "allow_fp16": True}, } script = service._generate_script_content_for_profile(profile_data) self.assertIn("LSFGVK_CONFIG=/home/deck/.config/lsfg-vk/conf.toml", script) self.assertIn("LSFGVK_PROFILE=decky-lsfg-vk", script) self.assertNotIn("LSFG_PROCESS", script) def test_active_in_does_not_implicitly_override_the_selected_profile(self): config = ConfigurationManager.get_defaults() config["active_in"] = "Game.exe, GameThread" self.assertEqual( ConfigurationService._profile_selection_lines("decky-lsfg-vk", config), ["export LSFGVK_PROFILE=decky-lsfg-vk"], ) config["use_native_matching"] = True self.assertEqual( ConfigurationService._profile_selection_lines("decky-lsfg-vk", config), ["# lsfg-vk will select a matching profile from Active In."], ) self.assertEqual( ConfigurationManager.parse_script_content("export DECKY_LSFGVK_AUTO_PROFILE=1\n"), {"use_native_matching": True}, ) def test_cloned_profile_disables_native_matching_until_explicitly_enabled(self): config = ConfigurationManager.get_defaults() config["active_in"] = "Game.exe" config["use_native_matching"] = True profile_data = { "current_profile": "decky-lsfg-vk", "profiles": {"decky-lsfg-vk": config}, "global_config": {"dll": "", "allow_fp16": True}, } cloned = ConfigurationManager.create_profile(profile_data, "Other") self.assertEqual(cloned["profiles"]["Other"]["active_in"], "Game.exe") self.assertFalse(cloned["profiles"]["Other"]["use_native_matching"]) class InstallerInfrastructureTests(unittest.TestCase): def test_layer_archive_selection_has_no_legacy_arm_override(self): from lsfg_vk.constants import ( LAYER_ARCHIVE_AARCH64, LAYER_ARCHIVE_X86_64, get_layer_archive_filename, ) self.assertEqual(get_layer_archive_filename(False), LAYER_ARCHIVE_X86_64) self.assertEqual(get_layer_archive_filename(True), LAYER_ARCHIVE_AARCH64) self.assertNotEqual(LAYER_ARCHIVE_AARCH64, LAYER_ARCHIVE_X86_64) def test_install_migrates_v1_config_once_and_preserves_script_workarounds(self): from lsfg_vk.installation import InstallationService with tempfile.TemporaryDirectory() as directory: home = Path(directory) config_dir = home / ".config" / "lsfg-vk" config_dir.mkdir(parents=True) config_path = config_dir / "conf.toml" config_path.write_text(LEGACY_CONFIG, encoding="utf-8") script_path = home / "lsfg" script_path.write_text("#!/bin/bash\nexport DXVK_FRAME_RATE=45\nexec \"$@\"\n", encoding="utf-8") service = InstallationService.__new__(InstallationService) service.log = mock.Mock() service.user_home = home service.config_dir = config_dir service.config_file_path = config_path service.lsfg_script_path = script_path service.lsfg_launch_script_path = script_path with mock.patch("lsfg_vk.dll_detection.DllDetectionService") as detection_service: detection_service.return_value.check_lossless_scaling_dll.return_value = {"detected": False} profile_data = service._create_config_file() self.assertEqual( (config_dir / "conf.toml.v1.bak").read_text(encoding="utf-8"), LEGACY_CONFIG, ) migrated = config_path.read_text(encoding="utf-8") self.assertIn("version = 2", migrated) self.assertIn("allow_fp16 = false", migrated) self.assertEqual(profile_data["profiles"]["decky-lsfg-vk"]["dxvk_frame_rate"], 45) service._create_lsfg_launch_script(profile_data) launch_script = script_path.read_text(encoding="utf-8") self.assertIn("export DXVK_FRAME_RATE=45", launch_script) self.assertIn("export DISABLE_LSFGVK=1", launch_script) with mock.patch("lsfg_vk.dll_detection.DllDetectionService") as detection_service: detection_service.return_value.check_lossless_scaling_dll.return_value = {"detected": False} service._create_config_file() self.assertEqual(len(list(config_dir.glob("conf.toml.v1.bak*"))), 1) def test_unrecognized_configuration_is_backed_up_before_reset(self): from lsfg_vk.installation import InstallationService with tempfile.TemporaryDirectory() as directory: home = Path(directory) config_dir = home / ".config" / "lsfg-vk" config_dir.mkdir(parents=True) config_path = config_dir / "conf.toml" unsupported_config = "version = 3\n" config_path.write_text(unsupported_config, encoding="utf-8") service = InstallationService.__new__(InstallationService) service.log = mock.Mock() service.user_home = home service.config_dir = config_dir service.config_file_path = config_path service.lsfg_script_path = home / "lsfg" service.lsfg_launch_script_path = home / "lsfg" with mock.patch("lsfg_vk.dll_detection.DllDetectionService") as detection_service: detection_service.return_value.check_lossless_scaling_dll.return_value = {"detected": False} service._create_config_file() self.assertEqual( (config_dir / "conf.toml.unrecognized.bak").read_text(encoding="utf-8"), unsupported_config, ) self.assertIn("version = 2", config_path.read_text(encoding="utf-8")) self.assertEqual(service._config_recovery_backup, config_dir / "conf.toml.unrecognized.bak") def test_stale_dll_path_is_replaced_when_detection_finds_a_valid_file(self): from lsfg_vk.installation import InstallationService with tempfile.TemporaryDirectory() as directory: home = Path(directory) config_dir = home / ".config" / "lsfg-vk" config_dir.mkdir(parents=True) config_path = config_dir / "conf.toml" detected_dll = home / "Lossless.dll" detected_dll.write_bytes(b"dll") profile_data = { "current_profile": "decky-lsfg-vk", "profiles": {"decky-lsfg-vk": ConfigurationManager.get_defaults()}, "global_config": {"dll": str(home / "missing" / "Lossless.dll"), "allow_fp16": True}, } config_path.write_text( ConfigurationManager.generate_toml_content_multi_profile(profile_data), encoding="utf-8", ) service = InstallationService.__new__(InstallationService) service.log = mock.Mock() service.user_home = home service.config_dir = config_dir service.config_file_path = config_path service.lsfg_script_path = home / "lsfg" service.lsfg_launch_script_path = home / "lsfg" with mock.patch("lsfg_vk.dll_detection.DllDetectionService") as detection_service: detection_service.return_value.check_lossless_scaling_dll.return_value = { "detected": True, "path": str(detected_dll), } migrated = service._create_config_file() self.assertEqual(migrated["global_config"]["dll"], str(detected_dll)) self.assertIn(str(detected_dll), config_path.read_text(encoding="utf-8")) def test_invalid_layer_manifest_fails_without_copying_unmodified_json(self): from lsfg_vk.installation import InstallationService with tempfile.TemporaryDirectory() as directory: source = Path(directory) / "manifest.json" destination = Path(directory) / "installed.json" source.write_text('{"not_layer": true}', encoding="utf-8") service = InstallationService.__new__(InstallationService) service.log = mock.Mock() with self.assertRaises(OSError): service._copy_and_fix_json_file(source, destination) self.assertFalse(destination.exists()) def test_installation_snapshot_restores_previous_files(self): from lsfg_vk.installation import InstallationService with tempfile.TemporaryDirectory() as directory: home = Path(directory) service = InstallationService.__new__(InstallationService) service.log = mock.Mock() service.local_lib_dir = home / "lib" service.local_share_dir = home / "share" service.config_file_path = home / "config" / "conf.toml" service.lsfg_launch_script_path = home / "lsfg" service.legacy_lib_file = service.local_lib_dir / "liblsfg-vk.so" service.legacy_json_file = service.local_share_dir / "legacy.json" service.lib_file = service.local_lib_dir / "liblsfg-vk-layer.so" service.json_file = service.local_share_dir / "manifest.json" service.config_file_path.parent.mkdir(parents=True) service.local_lib_dir.mkdir(parents=True) service.local_share_dir.mkdir(parents=True) service.lib_file.write_bytes(b"old layer") service.json_file.write_text("old manifest", encoding="utf-8") service.config_file_path.write_text("old config", encoding="utf-8") service.lsfg_launch_script_path.write_text("old script", encoding="utf-8") with tempfile.TemporaryDirectory() as transaction_dir: snapshots = service._snapshot_installation_files(Path(transaction_dir)) service.lib_file.write_bytes(b"new layer") service.json_file.write_text("new manifest", encoding="utf-8") service.config_file_path.unlink() service.lsfg_launch_script_path.write_text("new script", encoding="utf-8") service._restore_installation_files(snapshots) self.assertEqual(service.lib_file.read_bytes(), b"old layer") self.assertEqual(service.json_file.read_text(encoding="utf-8"), "old manifest") self.assertEqual(service.config_file_path.read_text(encoding="utf-8"), "old config") self.assertEqual(service.lsfg_launch_script_path.read_text(encoding="utf-8"), "old script") def test_writes_replace_the_config_atomically(self): service = BaseService.__new__(BaseService) service.log = mock.Mock() with tempfile.TemporaryDirectory() as directory: target = Path(directory) / "conf.toml" target.write_text("old config", encoding="utf-8") with mock.patch("lsfg_vk.base_service.os.replace", wraps=__import__("os").replace) as replace: service._write_file(target, "version = 2\n", 0o644) self.assertEqual(target.read_text(encoding="utf-8"), "version = 2\n") replace.assert_called_once() class FlatpakMigrationTests(unittest.TestCase): def test_install_uses_the_bundled_flatpak_asset(self): from lsfg_vk.flatpak_service import FlatpakService with tempfile.TemporaryDirectory() as directory: bundle_path = Path(directory) / "org.freedesktop.Platform.VulkanLayer.lsfg_vk_23.08.flatpak" bundle_path.write_bytes(b"flatpak bundle") service = FlatpakService.__new__(FlatpakService) service.log = mock.Mock() service.check_flatpak_available = mock.Mock(return_value=True) service._get_bundled_extension_path = mock.Mock(return_value=bundle_path) service._run_flatpak_command = mock.Mock( return_value=types.SimpleNamespace(returncode=0, stdout="", stderr="") ) result = service.install_extension("23.08") self.assertTrue(result["success"]) service._run_flatpak_command.assert_called_once_with( [ "install", "--user", "--or-update", "--assumeyes", "--noninteractive", str(bundle_path), ], capture_output=True, text=True, ) def test_update_installed_extensions_only_updates_user_runtimes_and_migrates_apps(self): from lsfg_vk.flatpak_service import FlatpakService service = FlatpakService.__new__(FlatpakService) service.log = mock.Mock() service.check_flatpak_available = mock.Mock(return_value=True) service._get_installed_extension_versions = mock.Mock(return_value=["23.08", "25.08"]) bundle = mock.Mock() bundle.is_file.return_value = True service._get_bundled_extension_path = mock.Mock(return_value=bundle) service._get_installed_extension_commit = mock.Mock(side_effect=lambda version: f"old-{version}") service.install_extension = mock.Mock( side_effect=[{"success": True}, {"success": True}] ) service._migrate_legacy_app_overrides = mock.Mock( return_value={ "success": True, "migrated_apps": ["org.example.Game"], "failed_apps": [], } ) result = service.update_installed_extensions() self.assertTrue(result["success"]) self.assertEqual(result["updated_versions"], ["23.08", "25.08"]) self.assertEqual(result["migrated_apps"], ["org.example.Game"]) self.assertEqual( service.install_extension.call_args_list, [mock.call("23.08"), mock.call("25.08")], ) service._migrate_legacy_app_overrides.assert_called_once_with() def test_installed_runtime_scan_is_scoped_to_user_installation(self): from lsfg_vk.flatpak_service import FlatpakService service = FlatpakService.__new__(FlatpakService) service._run_flatpak_command = mock.Mock( return_value=types.SimpleNamespace( returncode=0, stdout=( "lsfg-vk\torg.freedesktop.Platform.VulkanLayer.lsfgvk\t" "2.0\t23.08\tx86_64\n" ), stderr="", ) ) self.assertEqual(service._get_installed_extension_versions(), ["23.08"]) service._run_flatpak_command.assert_called_once_with( ["list", "--user", "--runtime"], capture_output=True, text=True, check=True, ) def test_failed_runtime_update_leaves_legacy_overrides_untouched(self): from lsfg_vk.flatpak_service import FlatpakService service = FlatpakService.__new__(FlatpakService) service.log = mock.Mock() service.check_flatpak_available = mock.Mock(return_value=True) service._get_installed_extension_versions = mock.Mock(return_value=["24.08"]) bundle = mock.Mock() bundle.is_file.return_value = True service._get_bundled_extension_path = mock.Mock(return_value=bundle) service._get_installed_extension_commit = mock.Mock(return_value="old-24.08") service.install_extension = mock.Mock( return_value={"success": False, "error": "bundle unavailable"} ) service._migrate_legacy_app_overrides = mock.Mock() result = service.update_installed_extensions() self.assertFalse(result["success"]) self.assertEqual(result["failed_versions"][0]["version"], "24.08") service._migrate_legacy_app_overrides.assert_not_called() def test_failed_runtime_update_rolls_back_previously_updated_runtimes(self): from lsfg_vk.flatpak_service import FlatpakService service = FlatpakService.__new__(FlatpakService) service.log = mock.Mock() service.check_flatpak_available = mock.Mock(return_value=True) service._get_installed_extension_versions = mock.Mock(return_value=["23.08", "24.08"]) bundle = mock.Mock() bundle.is_file.return_value = True service._get_bundled_extension_path = mock.Mock(return_value=bundle) service._get_installed_extension_commit = mock.Mock( side_effect=lambda version: f"old-{version}" ) service.install_extension = mock.Mock( side_effect=[ {"success": True}, {"success": False, "error": "transaction failed"}, ] ) service._rollback_extension = mock.Mock(return_value=None) service._migrate_legacy_app_overrides = mock.Mock() result = service.update_installed_extensions() self.assertFalse(result["success"]) self.assertEqual(result["rolled_back_versions"], ["23.08"]) service._rollback_extension.assert_called_once_with("23.08", "old-23.08") service._migrate_legacy_app_overrides.assert_not_called() def test_flatpak_app_scan_is_scoped_to_user_installation(self): from lsfg_vk.flatpak_service import FlatpakService service = FlatpakService.__new__(FlatpakService) service._run_flatpak_command = mock.Mock( return_value=types.SimpleNamespace( returncode=0, stdout="Game\torg.example.Game\t1.0\tx86_64\n", stderr="", ) ) self.assertEqual(service._get_flatpak_app_ids(), ["org.example.Game"]) service._run_flatpak_command.assert_called_once_with( ["list", "--user", "--app"], capture_output=True, text=True, check=True, ) def test_native_install_runs_flatpak_migration_after_success(self): from lsfg_vk.plugin import Plugin plugin = Plugin.__new__(Plugin) plugin.installation_service = mock.Mock() plugin.flatpak_service = mock.Mock() plugin.installation_service.install.return_value = { "success": True, "message": "lsfg-vk v2 installed successfully", "error": None, } plugin.flatpak_service.update_installed_extensions.return_value = { "success": True, "message": "Updated one Flatpak runtime", "skipped": False, "updated_versions": ["23.08"], "failed_versions": [], "migrated_apps": [], "failed_apps": [], } result = asyncio.run(plugin.install_lsfg_vk()) self.assertTrue(result["success"]) self.assertIn("flatpak_update", result) plugin.flatpak_service.update_installed_extensions.assert_called_once_with() def test_legacy_override_detection_does_not_match_v2_environment(self): from lsfg_vk.flatpak_service import FlatpakService with tempfile.TemporaryDirectory() as directory: home = Path(directory) service = FlatpakService.__new__(FlatpakService) service.user_home = home service.config_dir = home / ".config/lsfg-vk" service.config_file_path = service.config_dir / "conf.toml" service.lsfg_launch_script_path = home / "lsfg" service.log = mock.Mock() self.assertTrue( service._has_legacy_app_override( f"LSFG_CONFIG={service.config_file_path}\n" ) ) self.assertFalse( service._has_legacy_app_override( f"LSFGVK_CONFIG={service.config_file_path}\n" ) ) def test_legacy_override_migration_only_touches_plugin_owned_apps(self): from lsfg_vk.flatpak_service import FlatpakService with tempfile.TemporaryDirectory() as directory: home = Path(directory) service = FlatpakService.__new__(FlatpakService) service.user_home = home service.config_dir = home / ".config/lsfg-vk" service.config_file_path = service.config_dir / "conf.toml" service.lsfg_launch_script_path = home / "lsfg" service.log = mock.Mock() service._get_flatpak_app_ids = mock.Mock( return_value=["org.example.Legacy", "org.example.V2", "org.example.Other"] ) service._get_app_override_output = mock.Mock( side_effect=[ f"LSFG_CONFIG={service.config_file_path}\n", f"LSFGVK_CONFIG={service.config_file_path}\n", "", ] ) service.set_app_override = mock.Mock(return_value={"success": True}) result = service._migrate_legacy_app_overrides() self.assertTrue(result["success"]) self.assertEqual(result["migrated_apps"], ["org.example.Legacy"]) service.set_app_override.assert_called_once_with("org.example.Legacy") def test_v2_override_setup_and_removal_clean_legacy_overrides(self): from lsfg_vk.flatpak_service import FlatpakService with tempfile.TemporaryDirectory() as directory: home = Path(directory) config_dir = home / ".config" / "lsfg-vk" config_dir.mkdir(parents=True) config_path = config_dir / "conf.toml" config_path.write_text( ConfigurationManager.generate_toml_content(ConfigurationManager.get_defaults()), encoding="utf-8", ) service = FlatpakService.__new__(FlatpakService) service.log = mock.Mock() service.user_home = home service.config_dir = config_dir service.config_file_path = config_path service.lsfg_launch_script_path = home / "lsfg" service.check_flatpak_available = mock.Mock(return_value=True) service._run_flatpak_command = mock.Mock( return_value=types.SimpleNamespace(returncode=0, stdout="", stderr="") ) app_id = "org.example.Game" result = service.set_app_override(app_id) self.assertTrue(result["success"]) setup_calls = [call.args[0] for call in service._run_flatpak_command.call_args_list] self.assertIn( ["override", "--user", f"--env=LSFGVK_CONFIG={config_path}", app_id], setup_calls, ) self.assertIn( ["override", "--user", "--unset-env=LSFG_CONFIG", app_id], setup_calls, ) self.assertIn( [ "override", "--user", f"--nofilesystem={home / '.local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll'}", app_id, ], setup_calls, ) self.assertIn( ["override", "--user", f"--nofilesystem={home / 'lsfg'}", app_id], setup_calls, ) service._run_flatpak_command.reset_mock() result = service.remove_app_override(app_id) self.assertTrue(result["success"]) removal_calls = [call.args[0] for call in service._run_flatpak_command.call_args_list] self.assertIn( ["override", "--user", "--unset-env=LSFG_CONFIG", app_id], removal_calls, ) self.assertIn( ["override", "--user", f"--nofilesystem={home / 'lsfg'}", app_id], removal_calls, ) if __name__ == "__main__": unittest.main()