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

"""Official Decky store transport and owned plugin transaction."""
import sandbox  # noqa: F401
import hashlib
import io
import json
import os
from pathlib import Path
import secrets
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
import zipfile

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'ui'))
import frame_decky_store


def archive(flags=(), extra=None):
    stream = io.BytesIO()
    with zipfile.ZipFile(stream, 'w') as bundle:
        bundle.writestr('Example/plugin.json', json.dumps({'name': 'Example', 'flags': list(flags)}))
        bundle.writestr('Example/dist/index.js', 'export default {};' + ' ' * 1200)
        if extra:
            bundle.writestr(*extra)
    data = stream.getvalue()
    return data, hashlib.sha256(data).hexdigest()


class ArchiveTests(unittest.TestCase):
    def test_catalog_selects_valid_visible_latest_versions(self):
        records = [{'id': 42, 'name': 'Example', 'visible': True, 'versions':
                    [{'name': '1.0', 'hash': 'a' * 64}], 'tags': []},
                   {'id': 43, 'name': 'Hidden', 'visible': False, 'versions':
                    [{'name': '1.0', 'hash': 'b' * 64}]},
                   {'id': 44, 'name': 'Unsafe', 'visible': True, 'versions':
                    [{'name': '1.0; touch /tmp/bad', 'hash': 'c' * 64}]}]
        with patch.object(frame_decky_store.urllib.request, 'urlopen', return_value=io.BytesIO(json.dumps(records).encode())):
            self.assertEqual([p['id'] for p in frame_decky_store.catalog()], [42])

    def test_accepts_hashed_non_root_archive(self):
        data, digest = archive()
        self.assertEqual(frame_decky_store.archive_info(data, digest, 'Example'), 'Example')
        with self.assertRaisesRegex(ValueError, 'SHA-256'):
            frame_decky_store.archive_info(data, '0' * 64, 'Example')

    def test_rejects_root_x86_and_escape(self):
        for data in (archive(flags=['root'])[0],
                     archive(extra=('Example/bin/helper', b'\x7fELF' + b'\x00' * 14 + (62).to_bytes(2, 'little')))[0],
                     archive(extra=('../outside', b'oops'))[0]):
            with self.subTest(data=hashlib.sha256(data).hexdigest()[:8]):
                with self.assertRaises(ValueError):
                    frame_decky_store.archive_info(data, hashlib.sha256(data).hexdigest(), 'Example')


@unittest.skipIf(os.name == 'nt', 'remote plugin transaction runs on Linux')
class StoreTransaction(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        home = Path(self.tmp.name)
        unit = home / '.config/systemd/user/frame-control-decky.service'
        unit.parent.mkdir(parents=True)
        unit.write_text('[Service]\n')
        fake = home / 'systemctl'
        fake.write_text('#!/bin/sh\nif [ "$2" = is-active ]; then exit 0; fi\nexit 0\n')
        fake.chmod(0o700)
        self.env = {**os.environ, 'HOME': self.tmp.name,
                    'PATH': self.tmp.name + os.pathsep + os.environ['PATH']}
        self.folder = home / '.local/share/frame-control/decky/plugins/Example'
        self.folder.parent.mkdir(parents=True)

    def apply(self, action, data=None):
        token = secrets.token_hex(12)
        source = Path('/tmp/frame-control-decky-plugin-' + token)
        if data is not None:
            source.write_bytes(data)
        digest = hashlib.sha256(data or b'').hexdigest()
        try:
            return subprocess.run([sys.executable, '-', action, token, '42', digest, 'Example', '1.0'],
                                  input=frame_decky_store.APPLY, text=True, capture_output=True,
                                  env=self.env, timeout=15)
        finally:
            source.unlink(missing_ok=True)

    def test_install_update_remove_only_owned_folder(self):
        first, _ = archive()
        result = self.apply('install', first)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads((self.folder / '.frame-control-store.json').read_text())['storeId'], 42)
        second, _ = archive(extra=('Example/data.txt', 'new'))
        result = self.apply('update', second)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual((self.folder / 'data.txt').read_text(), 'new')
        result = self.apply('remove')
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(self.folder.exists())
        self.folder.mkdir()
        (self.folder / 'plugin.json').write_text('{}')
        result = self.apply('remove')
        self.assertNotEqual(result.returncode, 0)
        self.assertTrue(self.folder.exists())
        external = Path(self.tmp.name) / 'other-plugin'
        external.mkdir()
        (external / '.frame-control-store.json').write_text('{"storeId":42}')
        for item in self.folder.iterdir(): item.unlink()
        self.folder.rmdir()
        self.folder.symlink_to(external, target_is_directory=True)
        result = self.apply('remove')
        self.assertNotEqual(result.returncode, 0)
        self.assertTrue(external.exists())


class StoreHost(unittest.TestCase):
    def test_owned_plugin_can_be_removed_without_store_network(self):
        state = {'managed': True, 'running': True, 'plugins':
                 [{'storeId': 42, 'folder': 'Example', 'name': 'Example', 'version': '1.0'}]}
        def ssh(command, **kwargs):
            return '{}'
        with patch.object(frame_decky_store.frame_decky, 'status', return_value=state), \
             patch.object(frame_decky_store, 'catalog') as network:
            frame_decky_store.change(ssh, 'remove', 42)
            network.assert_not_called()


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