saphid--frame-control/tests/test_frame_android_library.py

"""Offline artwork, Steam API and launcher supervision regressions."""
import importlib.util
import json
import os
from pathlib import Path
import signal
import struct
import subprocess
import sys
import tempfile
import time
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'ui'))
import frame_android as android
import frame_artwork as art

spec = importlib.util.spec_from_file_location('steam_shortcuts', ROOT / 'frame/android/steam_shortcuts.py')
shortcuts = importlib.util.module_from_spec(spec)
spec.loader.exec_module(shortcuts)


@unittest.skipIf(os.name == 'nt', 'POSIX launcher')
class LauncherTests(unittest.TestCase):
    def exercise(self, terminate, sig=signal.SIGTERM, blocked=None):
        with tempfile.TemporaryDirectory() as tmp:
            d = Path(tmp)
            app = d / 'Applications/Android/org.test.app'
            app.mkdir(parents=True)
            (app / 'launch.sh').write_bytes((ROOT / 'frame/android/lepton-app.sh').read_bytes())
            (app / 'app.apk').touch()
            (app / 'instance.id').write_text('2800000001')
            bin_dir = d / 'bin'
            bin_dir.mkdir()
            lepton = d / '.local/share/Steam/steamapps/common/Lepton/lepton'
            lepton.parent.mkdir(parents=True)
            def script(path, body):
                path.write_text('#!' + sys.executable + '\n' + body)
                path.chmod(0o755)
            script(lepton, 'import os,time\nfrom pathlib import Path\n'
                   'Path(os.environ["HOME"],"started").write_text(str(os.getpid()))\n'
                   + ('time.sleep(30)\n' if terminate else 'raise SystemExit(23)\n'))
            script(bin_dir / 'setsid', 'import os,sys\nos.setsid()\nos.execv(sys.argv[2],sys.argv[2:])\n')
            script(bin_dir / 'flock', 'import os\nraise SystemExit(1 if os.environ.get("TEST_LOCKED") else 0)\n')  # lock semantics belong to Linux; no flock on macOS
            script(bin_dir / 'podman', 'import os,sys\nfrom pathlib import Path\n'
                   'p=Path(os.environ["HOME"],"podman-calls")\n'
                   'with p.open("a") as f: f.write(" ".join(sys.argv[1:])+"\\n")\n'
                   'if sys.argv[1:2]==["inspect"] and os.environ.get("TEST_RUNNING"): print("true")\n')
            env = {**os.environ, 'HOME': str(d), 'PATH': str(bin_dir) + os.pathsep + os.environ['PATH']}
            if blocked:
                env['TEST_' + blocked] = '1'
            saved = d / '.local/share/Steam/steamapps/compatdata/2800000001/internal/save'
            saved.parent.mkdir(parents=True)
            saved.write_text('saved game')
            proc = subprocess.Popen(['bash', str(app / 'launch.sh')], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            try:
                if blocked:
                    proc.communicate(timeout=5)
                    self.assertEqual(proc.returncode, 1)
                    self.assertFalse((d / 'started').exists())
                    calls = (d / 'podman-calls').read_text() if (d / 'podman-calls').exists() else ''
                    self.assertNotIn('stop ', calls)
                    return
                deadline = time.monotonic() + 5
                while not (d / 'started').exists() and proc.poll() is None and time.monotonic() < deadline:
                    time.sleep(.02)
                self.assertTrue((d / 'started').exists(), 'launcher did not start Lepton')
                if terminate:
                    self.assertIsNone(proc.poll(), 'Steam-tracked wrapper exited during the session')
                    proc.send_signal(sig)
                _, err = proc.communicate(timeout=5)
                calls = (d / 'podman-calls').read_text() if (d / 'podman-calls').exists() else ''
                self.assertIn('stop -t 5 lepton-steamlaunch-2800000001', calls, err.decode())
                self.assertEqual(proc.returncode, 128 + sig if terminate else 23)
                self.assertEqual(saved.read_text(), 'saved game')
                self.assertTrue((app / 'app.apk').exists())
            finally:
                if proc.poll() is None:
                    proc.kill()
                    proc.communicate()
                if (d / 'started').exists():
                    try:
                        os.kill(int((d / 'started').read_text()), signal.SIGKILL)
                    except ProcessLookupError:
                        pass

    def test_steam_stop_cleans_container(self):
        self.exercise(True)

    def test_hangup_and_interrupt_cleanup(self):
        for sig in (signal.SIGHUP, signal.SIGINT):
            with self.subTest(sig=sig):
                self.exercise(True, sig)

    def test_duplicate_launch_leaves_existing_session_alone(self):
        for blocked in ('LOCKED', 'RUNNING'):
            with self.subTest(blocked=blocked):
                self.exercise(False, blocked=blocked)

    def test_normal_exit_cleans_container_and_keeps_exit_code(self):
        self.exercise(False)


FIXTURES = ROOT / 'tests/fixtures/library'


class ArtworkTests(unittest.TestCase):
    def test_icon_roundtrip_and_transparency(self):
        w, h, pixels = art.decode((FIXTURES / 'icon.png').read_bytes())
        self.assertEqual((w, h), (2, 2))
        self.assertEqual(pixels, bytes([255, 0, 0, 255, 0, 255, 0, 255,
                                       0, 0, 255, 128, 0, 0, 0, 0]))
        background = bytearray([10, 20, 30, 255] * 4)
        art.stamp(background, 2, (w, h, pixels), 0, 0, 2)
        self.assertEqual(background[:8], pixels[:8])
        self.assertEqual(background[8:12], bytes([4, 9, 142, 255]))
        self.assertEqual(background[12:], bytes([10, 20, 30, 255]))

    def test_all_slots_have_expected_dimensions(self):
        images = art.prepare('Open Saber Plus', (FIXTURES / 'icon.png').read_bytes())
        self.assertEqual(set(images), set(art.SLOTS))
        for slot, (ext, data) in images.items():
            self.assertEqual(ext, 'png')
            self.assertEqual(struct.unpack_from('>II', data, 16), art.SLOTS[slot])
        # Transparent logo, opaque capsules; corrupt/missing icon gets a monogram.
        _, _, logo = art.decode(images['logo'][1])
        self.assertIn(0, logo[3::4])
        self.assertIn(255, logo[3::4])
        self.assertEqual(set(art.fallback('日本語', b'bad icon')), set(art.SLOTS))

    def test_supplied_slots_and_url(self):
        import io
        data = (FIXTURES / 'icon.png').read_bytes()
        response = io.BytesIO(data)
        response.geturl = lambda: 'https://example.org/icon.png'
        with patch.object(art.urllib.request, 'urlopen', return_value=response) as fetch:
            images = art.prepare('Game', artwork={'hero': data, 'icon': 'https://example.org/icon.png'})
        self.assertEqual(images['hero'], ('png', data))
        self.assertEqual(images['icon'], ('png', data))
        self.assertEqual(fetch.call_count, 1)
        self.assertEqual(fetch.call_args.kwargs['timeout'], 20)
        self.assertEqual(fetch.call_args.args[0].get_header('User-agent'), 'FrameControl/1.0')

    def test_supplied_jpeg(self):
        data = (FIXTURES / 'icon.jpg').read_bytes()
        self.assertEqual(art.image_type(data), 'jpg')
        self.assertEqual(art.prepare('Game', artwork={'wide': data})['wide'], ('jpg', data))
        with self.assertRaises(ValueError):
            art.image_type(data[:30])

    def test_bad_artwork_and_expansion_limits(self):
        import zlib
        for value in ({'bad': b'bad'}, {'hero': 'file:///etc/passwd'}, {'hero': b'bad'}, ['hero']):
            with self.subTest(value=value), self.assertRaises(ValueError):
                art.prepare('Game', artwork=value)
        data = bytearray((FIXTURES / 'icon.png').read_bytes())
        data[45] ^= 1
        with self.assertRaisesRegex(ValueError, 'checksum'):
            art.decode(bytes(data))
        bomb = art.PNG + art.chunk(b'IHDR', struct.pack('>IIBBBBB', 1, 1, 8, 6, 0, 0, 0)) + \
            art.chunk(b'IDAT', zlib.compress(b'\0' * 1_000_000)) + art.chunk(b'IEND', b'')
        with self.assertRaisesRegex(ValueError, 'pixels'):
            art.decode(bomb)
        huge = art.PNG + art.chunk(b'IHDR', struct.pack('>IIBBBBB', 10000, 10000, 8, 6, 0, 0, 0)) + art.chunk(b'IEND', b'')
        with self.assertRaisesRegex(ValueError, 'dimensions'):
            art.decode(huge)

    def test_palette_and_filters(self):
        import zlib
        header = art.chunk(b'IHDR', struct.pack('>IIBBBBB', 2, 1, 1, 3, 0, 0, 0))
        data = art.PNG + header + art.chunk(b'PLTE', b'\xff\0\0\0\xff\0') + art.chunk(b'tRNS', b'\xff\x80') + \
            art.chunk(b'IDAT', zlib.compress(b'\0\x40')) + art.chunk(b'IEND', b'')
        self.assertEqual(art.decode(data)[2], bytes([255, 0, 0, 255, 0, 255, 0, 128]))
        for method in range(5):
            # Two identical RGBA rows: exercise each predictor with known filtered bytes.
            first = bytes([10, 20, 30, 255] * 2)
            filtered = bytearray()
            for x, value in enumerate(first):
                a, b, c = first[x-4] if x >= 4 else 0, first[x], first[x-4] if x >= 4 else 0
                predictor = (0, a, b, (a+b)//2, b)[method]
                filtered.append((value - predictor) & 255)
            data = art.PNG + art.chunk(b'IHDR', struct.pack('>IIBBBBB', 2, 2, 8, 6, 0, 0, 0)) + \
                art.chunk(b'IDAT', zlib.compress(b'\0' + first + bytes([method]) + filtered)) + art.chunk(b'IEND', b'')
            self.assertEqual(art.decode(data)[2], first * 2)

    def test_godot_project_icon(self):
        import io
        import zipfile
        data = (FIXTURES / 'icon.png').read_bytes()
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, 'w') as archive:
            archive.writestr('assets/icon.png', data)
        with zipfile.ZipFile(buffer) as archive:
            self.assertEqual(android.frame_apk._icon_png(archive, set(archive.namelist()), []), data)


class InstallTests(unittest.TestCase):
    def setUp(self):
        self.responses = json.loads((FIXTURES / 'steam-responses.json').read_text())
        self.info = {'package': 'org.test.vr', 'label': 'VR', 'version': '1', 'icon_png': None,
                     'vr': True, 'launchable': True, 'repairable': False, 'abis': [], 'min_sdk': 24}
        self.images = {slot: ('png', b'PNG ' + slot.encode()) for slot in art.SLOTS}
        self.existing = {'package': 'org.test.vr', 'instance': 2800000001,
                         'shortcut': 3346865537, 'label': 'Old name'}

    def install(self, existing, tool=None):
        def shortcut(*args, **kwargs):
            if args[0] == 'add':
                return '3346865537'
            if args[0] == 'list':
                return json.dumps(self.responses['shortcuts'])
            return json.dumps(self.responses['configure'])
        with patch.object(android.frame_artwork, 'prepare', return_value=self.images), \
                patch.object(android, 'read_meta', return_value=existing), \
                patch.object(android, '_copy') as copy, \
                patch.object(android, 'ssh', return_value=self.responses['home']) as ssh, \
                patch.object(android, 'shortcut_tool', side_effect=tool or shortcut) as api, \
                patch.object(android, '_write_meta') as meta:
            result = android._install('game.apk', self.info, self.info['package'], False, 'New name', 'test')
        return result, ssh, api, meta

    def test_existing_shortcut_refreshes_name_vr_and_every_slot(self):
        result, ssh, api, meta = self.install(self.existing)
        calls = [c.args for c in api.call_args_list]
        self.assertNotIn('add', [c[0] for c in calls])
        configure = next(c for c in calls if c[0] == 'configure')
        self.assertEqual(configure[1:3], ('3346865537', 'New name'))
        self.assertEqual(configure[6], '1')
        self.assertEqual(set(json.loads(configure[7])), set(art.SLOTS))
        self.assertTrue(configure[5].endswith('/org.test.vr/icon.png'))
        self.assertEqual(result['shortcut'], self.existing['shortcut'])
        self.assertEqual(result['label'], 'New name')
        self.assertEqual(result['library_warnings'], [])
        self.assertEqual(len([c for c in ssh.call_args_list if isinstance(c.kwargs.get('input'), bytes)]), 5)
        meta.assert_called_once()

    def test_first_install_adds_shortcut(self):
        _, _, api, _ = self.install(None)
        self.assertEqual([c.args[0] for c in api.call_args_list], ['add', 'configure'])

    def test_artwork_forwarded_through_patch(self):
        artwork = {'hero': b'provided'}
        with patch.object(android, 'apk_info', return_value={**self.info, 'repairable': True}), \
                patch.object(android, 'xr_compat_files', return_value={}), \
                patch.object(android, 'patch', return_value={'patched': ['launcher']}), \
                patch.object(android, '_install', return_value={}) as install:
            android.install('x.apk', artwork=artwork)
        self.assertIs(install.call_args.args[-1], artwork)

    def test_failed_new_install_removes_shortcut(self):
        def tool(*args):
            if args[0] == 'add':
                return '3346865537'
            if args[0] == 'configure':
                raise android.FrameError('write failed')
            return '{}'
        with patch.object(android.frame_artwork, 'prepare', return_value=self.images), \
                patch.object(android, 'read_meta', return_value=None), \
                patch.object(android, '_copy'), patch.object(android, 'ssh', return_value='/home/steamos') as ssh, \
                patch.object(android, 'shortcut_tool', side_effect=tool) as api:
            with self.assertRaisesRegex(android.FrameError, 'write failed'):
                android._install('x.apk', self.info, 'org.test.vr', False, None, None)
        self.assertIn(('remove', '3346865537'), [c.args for c in api.call_args_list])
        self.assertTrue(any(c.args[0] == 'rm -rf Applications/Android/org.test.vr' for c in ssh.call_args_list))

    def test_remove_keeps_data_when_requested_and_surfaces_api_failure(self):
        with patch.object(android, '_meta_or_fail', return_value=dict(self.existing)), \
                patch.object(android, 'stop'), \
                patch.object(android, 'shortcut_tool', return_value='{"warnings": []}') as api, \
                patch.object(android, 'ssh') as ssh:
            android.remove('org.test.vr', keep_data=True)
            api.assert_called_once_with('remove', '3346865537')
            ssh.assert_called_once_with('rm -rf Applications/Android/org.test.vr')
            api.side_effect = android.FrameError('CDP unavailable')
            ssh.reset_mock()
            with self.assertRaises(android.FrameError):
                android.remove('org.test.vr')
            ssh.assert_not_called()

    def test_stop_requests_steam_and_has_container_fallback(self):
        with patch.object(android, '_meta_or_fail', return_value=self.existing), \
                patch.object(android, 'shortcut_tool', side_effect=android.FrameError('offline')) as api, \
                patch.object(android, 'ssh') as ssh:
            android.stop('org.test.vr')
        api.assert_called_once_with('stop', '3346865537')
        self.assertIn('podman stop -t 5 lepton-steamlaunch-2800000001', ssh.call_args.args[0])


class SteamAPITests(unittest.TestCase):
    def test_artwork_api_enums_and_safe_serialization(self):
        with patch.object(shortcuts, 'evaluate', return_value={'warnings': []}) as evaluate:
            shortcuts.configure(42, 'A "name"\n', '/path', '/start', '/icon', True,
                                {slot: str(FIXTURES / 'icon.png') for slot in art.SLOTS})
        js = evaluate.call_args.args[0]
        self.assertIn('SetShortcutIsVR(id, true)', js)
        self.assertIn('SetShortcutName(id, "A \\"name\\"\\n")', js)
        self.assertIn('SetCustomArtworkForApp(id, data, ext, type)', js)
        self.assertEqual(shortcuts.ASSETS, {'grid': 0, 'hero': 1, 'logo': 2, 'wide': 3, 'icon': 4})
        self.assertIn('NewUnsavedCollection(name, undefined, [app])', js)

    def test_remove_clears_every_slot_before_shortcut(self):
        with patch.object(shortcuts, 'evaluate', return_value={}) as evaluate:
            shortcuts.remove(42)
        js = evaluate.call_args.args[0]
        self.assertIn('[0, 1, 2, 3, 4]', js)
        self.assertLess(js.index('ClearCustomArtworkForApp(id, type)'), js.index('RemoveShortcut(id)'))
        self.assertIn('const wanted = []', js)

    def test_stop_uses_exact_64_bit_game_id_string(self):
        with patch.object(sys, 'argv', ['steam_shortcuts.py', 'stop', '3346865537']), \
                patch.object(shortcuts, 'evaluate') as evaluate, patch('builtins.print'):
            shortcuts.main()
        self.assertEqual(evaluate.call_args.args[0], 'SteamClient.Apps.TerminateApp("14374678025558032384", false)')


class SteamContextTests(unittest.TestCase):
    @unittest.skipUnless(__import__('shutil').which('node'), 'optional V8 fixture check requires node')
    def test_collection_lifecycle_and_native_artwork_calls(self):
        steam = {'apps': [], 'shortcuts': [{'appid': 42, 'name': 'Before'}], 'compat_tools': {},
                 'collections': [{'name': 'Android', 'apps': [999]}]}
        def evaluate(expression):
            nonlocal steam
            proc = subprocess.run(['node', str(ROOT / 'tests/fakeframe/rootfs/usr/local/lib/fakeframe/cef_shim.js')],
                                  input=json.dumps({'id': 1, 'expression': expression, 'awaitPromise': True,
                                                    'steam': steam}) + '\n',
                                  text=True, capture_output=True, timeout=10, check=True)
            reply = json.loads(proc.stdout)
            self.assertNotIn('exceptionDetails', reply['result'])
            steam = reply['steam']
            return reply['result']['result'].get('value')
        with patch.object(shortcuts, 'evaluate', side_effect=evaluate):
            result = shortcuts.configure(42, 'Game', '/exe', '/dir', '/icon', True,
                                         {slot: str(FIXTURES / 'icon.png') for slot in art.SLOTS})
            self.assertEqual(result['warnings'], [])
            self.assertTrue(steam['shortcuts'][0]['vr'])
            self.assertEqual(set(steam['shortcuts'][0]['artwork']), {'0', '1', '2', '3', '4'})
            self.assertEqual(steam['collections'], [{'name': 'Android', 'apps': [999, 42]},
                                                   {'name': 'Android VR', 'apps': [42]}])
            shortcuts.configure(42, 'Renamed', '/exe', '/dir', '/icon', False, {})
            self.assertEqual(steam['shortcuts'][0]['name'], 'Renamed')
            self.assertEqual(steam['collections'][1]['apps'], [])
            shortcuts.remove(42)
            self.assertEqual(steam['shortcuts'], [])
            self.assertEqual(steam['collections'][0]['apps'], [999])


if __name__ == '__main__':
    unittest.main()