250 lines
14 KiB
Python
250 lines
14 KiB
Python
"""Installed desktop discovery: inert trees/files, no real applications launched."""
|
|
import copy
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
from common import BACKEND, Catalog, Spec
|
|
from test_xwayland import Sway
|
|
from session_v2.applications import DEFAULT, installed_specs, launch, launch_count, _gio
|
|
from session_v2.checkpoint import Store
|
|
from session_v2.errors import Failure, Deferred
|
|
from session_v2.executor import Executor
|
|
from session_v2.ipc import workspace, X11_CLASS_PREFIX
|
|
from session_v2.recovery import JOURNAL
|
|
from session_v2.schema import validate
|
|
from session_v2.snapshot import capture
|
|
|
|
|
|
class DesktopCatalogTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name)
|
|
self.Gio, self.GioUnix, self.GLib = _gio()
|
|
|
|
def desktop(self, identifier="org.test.Editor.desktop", extra="", executable="/usr/bin/true"):
|
|
path = self.root / identifier
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("[Desktop Entry]\nType=Application\nName=TEST_ONLY\nExec=" + executable + "\n" + extra)
|
|
path.chmod(0o600)
|
|
app = self.GioUnix.DesktopAppInfo.new_from_filename(str(path))
|
|
self.assertIsNotNone(app)
|
|
return app
|
|
|
|
def catalog(self, *apps):
|
|
return Catalog(desktop_apps=apps)
|
|
|
|
def test_explicit_catalog_is_hermetic(self):
|
|
with patch("session_v2.applications.installed_specs") as discovery:
|
|
self.assertEqual(Catalog(()).specs, {})
|
|
self.assertEqual(Catalog(DEFAULT).specs["firefox"].provider, "firefox-session-window-uuid")
|
|
discovery.assert_not_called()
|
|
|
|
def test_exact_desktop_and_declared_class_with_no_title_matching(self):
|
|
app = self.desktop(extra="StartupWMClass=TestEditor\n")
|
|
catalog = self.catalog(app)
|
|
spec = catalog.classify({"app_id": "org.test.Editor"})
|
|
self.assertEqual(spec.application_id, "org.test.Editor.desktop")
|
|
self.assertEqual(spec.strategy, "application-self-restore")
|
|
self.assertEqual(catalog.classify({"app_id": "TestEditor"}), spec)
|
|
self.assertEqual(catalog.classify({"app_id": None, "window": 12,
|
|
"window_properties": {"class": "TestEditor"}}), spec)
|
|
for identity in ("testeditor", "org.test.Unregistered"):
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_UNSUPPORTED"):
|
|
catalog.classify({"app_id": identity, "name": "TestEditor"})
|
|
with self.assertRaisesRegex(Failure, "SURFACE_IDENTITY_RESERVED"):
|
|
catalog.classify({"app_id": "x11-class:TestEditor"})
|
|
|
|
def test_menu_visibility_does_not_disable_installed_window_support(self):
|
|
hidden = self.desktop("hidden.desktop", "Hidden=true\n")
|
|
helper = self.desktop("helper.desktop", "NoDisplay=true\n")
|
|
terminal = self.desktop("terminal.desktop", "Terminal=true\n")
|
|
other = self.desktop("other.desktop", "OnlyShowIn=TEST_ONLY_UNKNOWN_DESKTOP;\n")
|
|
future_webapp = self.desktop("firefox.webapp-new.desktop")
|
|
# GIO itself may reject an absent Exec; a valid wrapper path with a
|
|
# missing runtime target must not be inferred as a unique app identity.
|
|
catalog = self.catalog(hidden, helper, terminal, other, future_webapp)
|
|
self.assertEqual(catalog.excluded, {"hidden.desktop": "hidden",
|
|
"terminal.desktop": "terminal-content",
|
|
"firefox.webapp-new.desktop": "provider-required"})
|
|
self.assertEqual(len(catalog.specs), len(DEFAULT) + 2)
|
|
self.assertEqual(catalog.classify({"app_id": "helper"}).desktop_entry, "helper.desktop")
|
|
self.assertEqual(catalog.classify({"app_id": "other"}).desktop_entry, "other.desktop")
|
|
|
|
def test_shared_aliases_refuse_but_exact_declared_ids_remain(self):
|
|
first = self.desktop("org.test.First.desktop", "StartupWMClass=SharedWindow\n")
|
|
second = self.desktop("org.test.Second.desktop", "StartupWMClass=SharedWindow\n")
|
|
catalog = self.catalog(first, second)
|
|
for identity in ("SharedWindow", "true"):
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_UNSUPPORTED"):
|
|
catalog.classify({"app_id": identity})
|
|
for wm_class in ("SharedWindow", "true"):
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_UNSUPPORTED"):
|
|
catalog.classify({"app_id": None, "window": 1, "window_properties": {"class": wm_class}})
|
|
self.assertEqual(catalog.classify({"app_id": "org.test.First"}).desktop_entry, "org.test.First.desktop")
|
|
|
|
def test_exact_desktop_identity_outranks_an_executable_alias(self):
|
|
canonical = self.desktop("true.desktop")
|
|
other = self.desktop("org.test.Other.desktop")
|
|
self.assertEqual(self.catalog(canonical, other).classify({"app_id": "true"}).desktop_entry, "true.desktop")
|
|
|
|
def test_native_desktop_id_outranks_other_launcher_wm_class(self):
|
|
canonical = self.desktop("org.test.Native.desktop")
|
|
helper = self.desktop("org.test.Helper.desktop", "NoDisplay=true\nStartupWMClass=org.test.Native\n")
|
|
catalog = self.catalog(canonical, helper)
|
|
self.assertEqual(catalog.classify({"app_id": "org.test.Native"}).desktop_entry, "org.test.Native.desktop")
|
|
self.assertEqual(catalog.classify({"app_id": None, "window": 1, "window_properties": {"class": "org.test.Native"}}).desktop_entry, "org.test.Helper.desktop")
|
|
|
|
def test_static_policy_cannot_be_replaced_by_desktop_metadata(self):
|
|
override = self.desktop("firefox.desktop", "StartupWMClass=OtherFirefox\n")
|
|
impostor = self.desktop("org.test.Impostor.desktop", "StartupWMClass=firefox\n")
|
|
catalog = self.catalog(override, impostor)
|
|
self.assertEqual(catalog.classify({"app_id": "firefox"}), next(s for s in DEFAULT if s.application_id == "firefox"))
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_UNSUPPORTED"):
|
|
catalog.classify({"app_id": "OtherFirefox"})
|
|
self.assertEqual(catalog.specs["whatsapp-web"].managed_by, "sway-autostart")
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_UNSUPPORTED"):
|
|
catalog.classify({"app_id": None, "window": 1, "window_properties": {"class": "firefox"}})
|
|
|
|
def test_launch_uses_pinned_desktop_file_without_snapshot_arguments(self):
|
|
app = self.desktop(executable='"/usr/bin/true" "%U"')
|
|
spec = self.catalog(app).classify({"app_id": "org.test.Editor"})
|
|
with patch("session_v2.applications.subprocess.Popen") as popen:
|
|
launch(spec, 1, {"TEST_ONLY": "1"})
|
|
popen.assert_called_once()
|
|
self.assertEqual(popen.call_args.args[0], ["/usr/bin/gio", "launch", app.get_filename()])
|
|
self.assertNotIn("shell", popen.call_args.kwargs)
|
|
self.assertEqual(popen.call_args.kwargs["env"], {"TEST_ONLY": "1"})
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_LAUNCH_FAILED"):
|
|
launch(spec, 2)
|
|
forged = {"application_id": spec.application_id, "desktop_entry": "/tmp/evil.desktop",
|
|
"strategy": spec.strategy, "managed_by": None, "identity_provider": None}
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_POLICY_MISMATCH"):
|
|
self.catalog(app).validate(forged)
|
|
|
|
def test_changed_or_writable_launcher_is_not_executed(self):
|
|
app = self.desktop()
|
|
spec = self.catalog(app).classify({"app_id": "org.test.Editor"})
|
|
path = Path(app.get_filename())
|
|
path.write_text(path.read_text() + "Comment=changed\n")
|
|
with patch("session_v2.applications.subprocess.Popen") as popen:
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_LAUNCH_FAILED"):
|
|
launch(spec, 1)
|
|
popen.assert_not_called()
|
|
path.chmod(0o666)
|
|
self.assertEqual(self.catalog(app).excluded[app.get_id()], "invalid-desktop-file")
|
|
|
|
def test_nonregular_launcher_is_refused_without_reading_it(self):
|
|
app = self.desktop()
|
|
spec = self.catalog(app).classify({"app_id": "org.test.Editor"})
|
|
path = Path(app.get_filename())
|
|
path.unlink()
|
|
os.mkfifo(path)
|
|
with patch("session_v2.applications.subprocess.Popen") as popen:
|
|
with self.assertRaisesRegex(Failure, "APPLICATION_LAUNCH_FAILED"):
|
|
launch(spec, 1)
|
|
popen.assert_not_called()
|
|
|
|
def test_flatpak_export_symlink_and_shared_wrapper(self):
|
|
target = self.desktop("export/org.test.Flatpak.desktop", executable="/usr/bin/env TEST_ONLY=1 /usr/bin/true")
|
|
alias = self.root / "org.test.Flatpak.desktop"
|
|
alias.symlink_to(target.get_filename())
|
|
app = self.GioUnix.DesktopAppInfo.new_from_filename(str(alias))
|
|
spec = self.catalog(app).classify({"app_id": "org.test.Flatpak"})
|
|
self.assertEqual(spec.desktop_signature[0], str(Path(target.get_filename()).resolve()))
|
|
self.assertNotIn("env", spec.app_ids)
|
|
with patch("session_v2.applications.subprocess.Popen") as popen:
|
|
launch(spec, 1)
|
|
popen.assert_called_once()
|
|
|
|
def test_xdg_user_override_hidden_and_system_flatpak_roots(self):
|
|
local, system, flatpak = (self.root / name for name in ("local", "system", "flatpak"))
|
|
for directory in (local, system, flatpak):
|
|
(directory / "applications").mkdir(parents=True)
|
|
def put(directory, identifier, extra=""):
|
|
(directory / "applications" / identifier).write_text("[Desktop Entry]\nType=Application\nName=TEST_ONLY\nExec=/usr/bin/true\n" + extra)
|
|
put(system, "org.test.Shadow.desktop", "StartupWMClass=SystemClass\n")
|
|
put(local, "org.test.Shadow.desktop", "StartupWMClass=LocalClass\n")
|
|
put(system, "org.test.Deleted.desktop")
|
|
put(local, "org.test.Deleted.desktop", "Hidden=true\n")
|
|
put(flatpak, "org.test.Flatpak.desktop")
|
|
program = "import sys,json;sys.path.insert(0,sys.argv[1]);from session_v2.applications import Catalog;c=Catalog();print(json.dumps({s.desktop_entry:s.app_ids for s in c.specs.values() if s.desktop_signature is not None}))"
|
|
env = {**os.environ, "XDG_DATA_HOME": str(local), "XDG_DATA_DIRS": str(system) + ":" + str(flatpak), "XDG_CURRENT_DESKTOP": "sway"}
|
|
data = json.loads(subprocess.check_output(["python3", "-B", "-c", program, str(BACKEND)], env=env, text=True, timeout=5))
|
|
self.assertIn("LocalClass", data["org.test.Shadow.desktop"])
|
|
self.assertNotIn("SystemClass", data["org.test.Shadow.desktop"])
|
|
self.assertNotIn("org.test.Deleted.desktop", data)
|
|
self.assertIn("org.test.Flatpak.desktop", data)
|
|
|
|
def source(self):
|
|
catalog = self.catalog(self.desktop())
|
|
sway = Sway()
|
|
sway.add("org.test.Editor", 7, focus=False)
|
|
sway.add("org.test.Editor", 8, focus=True)
|
|
data = capture(sway, catalog, None)
|
|
return catalog, sway, data
|
|
|
|
def test_multiwindow_capture_persists_layout_only_and_preserves_v1(self):
|
|
catalog, sway, data = self.source()
|
|
self.assertEqual(data["applications"][0]["expected_windows"], 2)
|
|
self.assertEqual(data["applications"][0]["strategy"], "application-self-restore")
|
|
self.assertTrue(all(slot["identity_requirement"] == "interchangeable" for slot in data["window_slots"]))
|
|
for forbidden in ("argv", "desktop_signature", str(self.root)):
|
|
self.assertNotIn(forbidden, json.dumps(data))
|
|
store = Store(self.root / "state")
|
|
store.checkpoint(data)
|
|
self.assertEqual(store.last(), data)
|
|
old = copy.deepcopy(data)
|
|
old["version"] = 1
|
|
old.pop("focus_workspace")
|
|
for slot in old["window_slots"]: slot["state"]["scratchpad"] = False
|
|
with self.assertRaisesRegex(Failure, "SCHEMA_STRATEGY"): validate(old)
|
|
schema = json.loads((BACKEND / "session_v2/schema-v2.json").read_text())
|
|
self.assertIn("application-self-restore", schema["$defs"]["application"]["properties"]["strategy"]["enum"])
|
|
schema = json.loads((BACKEND / "session_v2/schema-v1.json").read_text())
|
|
self.assertNotIn("application-self-restore", schema["$defs"]["application"]["properties"]["strategy"]["enum"])
|
|
|
|
def test_launch_count_is_once_for_absent_app_and_zero_for_partial_presence(self):
|
|
app = {"strategy": "application-self-restore", "expected_windows": 3}
|
|
self.assertEqual([launch_count(app, count) for count in (0, 1, 2, 3)], [1, 0, 0, 0])
|
|
with self.assertRaisesRegex(Failure, "UNEXPECTED_EXTRA_WINDOW"): launch_count(app, 4)
|
|
|
|
def test_prewarm_launches_once_and_requires_full_window_set(self):
|
|
catalog, sway, data = self.source()
|
|
workspace(sway.value, "1")["nodes"] = []
|
|
engine = Executor(sway, catalog, None, self.root / "runtime", "/forbidden", process_identity=lambda pid: 123)
|
|
def once(spec, count, env):
|
|
self.assertEqual(count, 1)
|
|
sway.add("org.test.Editor", 20, focus=False)
|
|
sway.add("org.test.Editor", 21, focus=True)
|
|
return []
|
|
with patch("session_v2.executor.launch", side_effect=once) as launcher, patch.object(engine, "_apply", return_value="admitted"):
|
|
self.assertEqual(engine.apply(data, True, timeout=.5), "admitted")
|
|
launcher.assert_called_once()
|
|
self.assertEqual(sway.commands, [])
|
|
self.assertFalse((self.root / "runtime" / JOURNAL).exists())
|
|
|
|
def test_partial_generic_restore_defers_without_layout_or_retry_launch(self):
|
|
catalog, sway, data = self.source()
|
|
workspace(sway.value, "1")["nodes"] = []
|
|
runtime = self.root / "runtime"
|
|
engine = Executor(sway, catalog, None, runtime, "/forbidden", process_identity=lambda pid: 123)
|
|
def only_one(spec, count, env):
|
|
sway.add("org.test.Editor", 20, focus=True)
|
|
return []
|
|
with patch("session_v2.executor.launch", side_effect=only_one) as launcher, patch.object(engine, "_apply") as apply:
|
|
for _ in range(2):
|
|
with self.assertRaisesRegex(Deferred, "APPLICATION_WINDOWS_INCOMPLETE"):
|
|
engine.apply(data, True, timeout=.08)
|
|
launcher.assert_called_once()
|
|
apply.assert_not_called()
|
|
self.assertEqual(sway.commands, [])
|
|
self.assertFalse((runtime / JOURNAL).exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|