Fix Riot session aborts, CDN downloads and hidden client recovery

tests/test_desktop_run.py

@@ -0,0 +1,36 @@
+"""Verify menu launch diagnostics without starting Darling or a client."""
+import os
+from pathlib import Path
+import shutil
+import stat
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class DesktopRunTests(unittest.TestCase):
+    def test_private_log_and_arguments(self):
+        with tempfile.TemporaryDirectory(prefix='desktop runner ') as folder:
+            root = Path(folder)
+            (root / 'scripts').mkdir()
+            shutil.copy2(ROOT / 'scripts/desktop-run.sh', root / 'scripts/desktop-run.sh')
+            run = root / 'scripts/run.sh'
+            run.write_text('#!/bin/sh\nprintf "stdout:%s\\n" "$1"\nprintf "stderr\\n" >&2\nexit 7\n')
+            run.chmod(0o755)
+            (root / 'build').mkdir()
+            log = root / 'build/desktop-launch.log'
+            log.write_text('previous\n')
+            log.chmod(0o644)
+            result = subprocess.run(['sh', str(root / 'scripts/desktop-run.sh'), 'two words'],
+                                    capture_output=True, text=True)
+            self.assertEqual(result.returncode, 7)
+            self.assertEqual(result.stdout, '')
+            self.assertEqual(result.stderr, '')
+            self.assertEqual(stat.S_IMODE(log.stat().st_mode), 0o600)
+            self.assertEqual(log.read_text(), 'previous\nstdout:two words\nstderr\n')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_hidden_riot_recovery.py

@@ -0,0 +1,116 @@
+"""Exercise desktop recovery with fake /proc entries and mocked commands only."""
+import importlib.util
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location('recovery', ROOT / 'tools/recover-hidden-riot-client.py')
+recovery = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(recovery)
+
+
+class RecoveryTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.proc = self.root / 'proc'
+        self.proc.mkdir()
+        self.home = self.root / 'home'
+        self.home_patch = patch.object(Path, 'home', return_value=self.home)
+        self.home_patch.start()
+        self.addCleanup(self.home_patch.stop)
+        self.env_patch = patch.dict(os.environ, {'DPREFIX': str(self.home / '.darling')})
+        self.env_patch.start()
+        self.addCleanup(self.env_patch.stop)
+
+    def process(self, pid, executable, prefix=None):
+        folder = self.proc / str(pid)
+        folder.mkdir()
+        (folder / 'cmdline').write_bytes(executable.encode() + b'\0')
+        (folder / 'environ').write_bytes(b'' if prefix is None else b'DPREFIX=' + os.fsencode(prefix) + b'\0')
+        return folder
+
+    def riot(self):
+        return self.process(123, '/Users/Shared/Riot Games/Riot Client.app/Contents/MacOS/RiotClientServices')
+
+    def run_recovery(self, windows=None, error=None, raw=None):
+        def command(args, **kwargs):
+            if args[0] == 'hyprctl':
+                if error:
+                    raise error
+                return subprocess.CompletedProcess(args, 0, stdout=raw if raw is not None else json.dumps(windows or []))
+            return subprocess.CompletedProcess(args, 0)
+        with patch.object(recovery.subprocess, 'run', side_effect=command) as run:
+            result = recovery.recover(self.proc)
+        return result, [call.args[0] for call in run.call_args_list]
+
+    def test_windowless_client_is_recovered(self):
+        self.riot()
+        result, calls = self.run_recovery()
+        self.assertTrue(result)
+        self.assertEqual(calls, [['hyprctl', 'clients', '-j'], ['darling', 'shutdown']])
+
+    def test_visible_client_is_preserved(self):
+        self.riot()
+        for identity in ('Riot Client', 'League of Legends'):
+            result, calls = self.run_recovery([{'class': identity, 'mapped': True, 'hidden': False}])
+            self.assertFalse(result)
+            self.assertEqual(len(calls), 1)
+
+    def test_game_is_preserved_even_without_windows(self):
+        self.riot()
+        self.process(124, '/Applications/LeagueofLegends.app/Contents/MacOS/LeagueofLegends')
+        self.assertEqual(self.run_recovery(), (False, []))
+
+    def test_no_riot_is_preserved(self):
+        self.process(124, '/usr/bin/editor')
+        self.assertEqual(self.run_recovery(), (False, []))
+
+    def test_foreign_owned_process_is_preserved(self):
+        self.riot()
+        with patch.object(recovery.os, 'getuid', return_value=os.getuid() + 1):
+            self.assertEqual(self.run_recovery(), (False, []))
+
+    def test_custom_prefix_is_preserved(self):
+        self.riot()
+        with patch.dict(os.environ, {'DPREFIX': str(self.root / 'other-prefix')}):
+            self.assertEqual(self.run_recovery(), (False, []))
+
+    def test_riot_in_another_prefix_is_preserved(self):
+        self.process(123, '/Applications/Riot Client.app/Contents/MacOS/RiotClientServices', self.root / 'other-prefix')
+        self.assertEqual(self.run_recovery(), (False, []))
+
+    def test_missing_or_failed_compositor_is_preserved(self):
+        self.riot()
+        for error in (FileNotFoundError(), subprocess.TimeoutExpired('hyprctl', 5),
+                      subprocess.CalledProcessError(1, 'hyprctl')):
+            result, calls = self.run_recovery(error=error)
+            self.assertFalse(result)
+            self.assertEqual(len(calls), 1)
+
+    def test_malformed_compositor_response_is_preserved(self):
+        self.riot()
+        for raw in ('invalid', '{}', '[null]'):
+            result, calls = self.run_recovery(raw=raw)
+            self.assertFalse(result)
+            self.assertEqual(len(calls), 1)
+
+    def test_unreadable_environment_is_preserved(self):
+        self.riot()
+        original = Path.read_bytes
+        def read(path):
+            if path.name == 'environ':
+                raise PermissionError()
+            return original(path)
+        with patch.object(Path, 'read_bytes', read):
+            self.assertEqual(self.run_recovery(), (False, []))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_urlsession_runner.py

@@ -0,0 +1,55 @@
+"""Check fixture traffic and private-prefix cleanup with a fake darling executable."""
+import json
+import os
+from pathlib import Path
+import stat
+import subprocess
+import sys
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class URLSessionRunnerTests(unittest.TestCase):
+    def test_fixture_and_shutdown_use_only_a_private_prefix(self):
+        with tempfile.TemporaryDirectory() as folder:
+            root = Path(folder)
+            executable = root / 'darling'
+            trace = root / 'trace.jsonl'
+            executable.write_text(f'#!{sys.executable}\n' + f'TRACE = {str(trace)!r}\n' + '''
+import json, os, sys, urllib.request
+with open(TRACE, 'a') as stream:
+    stream.write(json.dumps({'args':sys.argv[1:], 'prefix':os.environ['DPREFIX']}) + '\\n')
+if sys.argv[1] == 'shutdown':
+    raise SystemExit(0)
+assert urllib.request.urlopen(sys.argv[2]).read() == b'darling-session-ok'
+try:
+    urllib.request.urlopen(sys.argv[3], timeout=2)
+except OSError:
+    pass
+else:
+    raise AssertionError('reserved error endpoint unexpectedly accepted a connection')
+print('LIFECYCLE_PASS: fixture')
+print('URLSESSION_PASS: fixture')
+''')
+            executable.chmod(0o755)
+            probe = root / 'probe'
+            probe.write_text('fake probe; never executed')
+            log = root / 'probe.log'
+            result = subprocess.run([sys.executable, str(ROOT / 'tools/riot-urlsession-check.py'),
+                                     '--probe', str(probe), '--log', str(log)],
+                                    env=dict(os.environ, PATH=str(root) + os.pathsep + os.environ['PATH']),
+                                    capture_output=True, text=True, timeout=15)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            calls = [json.loads(line) for line in trace.read_text().splitlines()]
+            self.assertEqual(len(calls), 2)
+            self.assertEqual(calls[1]['args'], ['shutdown'])
+            self.assertEqual(calls[0]['prefix'], calls[1]['prefix'])
+            self.assertNotEqual(Path(calls[0]['prefix']), Path.home() / '.darling')
+            self.assertFalse(Path(calls[0]['prefix']).exists())
+            self.assertEqual(stat.S_IMODE(log.stat().st_mode), 0o600)
+
+
+if __name__ == '__main__':
+    unittest.main()