saphid--frame-control/scripts/framey-test-decky-runtime.py

#!/usr/bin/env python3
"""Run every managed backend shim in native, isolated, published Decky 3.3.0.

Downloads a digest-pinned loader, or accepts --loader for an offline cached copy.
Only image build/download use network. All official plugin execution is in
network/device/session-isolated disposable Docker containers. Never uses Frame.
"""
import argparse
import ast
import hashlib
import json
from pathlib import Path
import platform
import shutil
import subprocess
import sys
import tempfile
import urllib.request
import uuid

ROOT = Path(__file__).resolve().parents[1]
RELEASE = 'v3.3.0'
LOADERS = {
    'x86_64': ('PluginLoader', 'e80d03e641d1766a33a817cb27ab8d2e0a4633685e5d27e4de948a6687526181', 62),
    'aarch64': ('PluginLoader-arm64', '8ef5dcf36cbbbb6e97201a6baed5d79c4cc44ec27789fc9af1c5b6ae8f0b25b6', 183),
}


def check_coverage():
    """A new production shim must declare an actual-runtime probe before CI runs."""
    sys.path.insert(0, str(ROOT/'ui'))
    import frame_decky_store
    tree = ast.parse(frame_decky_store.APPLY)
    shims = set()
    for node in ast.walk(tree):
        if (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
                and isinstance(node.func.value, ast.Name)
                and node.func.value.id == 'compat_shims' and node.func.attr == 'append'):
            if len(node.args) != 1 or not isinstance(node.args[0], ast.Constant) or not isinstance(node.args[0].value, str):
                raise ValueError('Managed shims must declare a literal name for runtime test coverage')
            shims.add(node.args[0].value)
    cases = json.loads((ROOT/'tests/decky-runtime/cases.json').read_text())
    if shims != {case['shim'] for case in cases}:
        raise ValueError('Every managed shim needs an actual Decky runtime case')
    return cases


def loader_bytes(path):
    arch = platform.machine()
    if arch not in LOADERS:
        raise ValueError('Run on native Linux x86_64 or aarch64')
    asset, digest, machine = LOADERS[arch]
    if path:
        data = path.read_bytes()
    else:
        url = f'https://github.com/SteamDeckHomebrew/decky-loader/releases/download/{RELEASE}/{asset}'
        request = urllib.request.Request(url, headers={'User-Agent': 'Frame-Control-runtime-tests'})
        with urllib.request.urlopen(request, timeout=60) as response:
            data = response.read(80*1024*1024+1)
    if hashlib.sha256(data).hexdigest() != digest:
        raise ValueError('Decky 3.3.0 loader digest mismatch')
    if data[:4] != b'\x7fELF' or int.from_bytes(data[18:20], 'little') != machine:
        raise ValueError('Loader architecture does not match the native runner')
    return data, digest


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--loader', type=Path, help='Optional cached, digest-verified native Decky 3.3.0 binary')
    parser.add_argument('--output', type=Path, required=True)
    args = parser.parse_args()
    if sys.platform != 'linux' or not shutil.which('docker'):
        parser.error('Linux with Docker is required')
    cases = check_coverage()
    data, digest = loader_bytes(args.loader)
    tag = 'framey-decky-runtime:'+uuid.uuid4().hex[:12]
    report = {'decky': RELEASE, 'architecture': platform.machine(), 'loaderSha256': digest, 'cases': []}
    try:
        subprocess.run(['docker', 'build', '-q', '-t', tag, '-f', str(ROOT/'tests/decky-runtime/Containerfile'),
                        str(ROOT/'tests/decky-runtime')], check=True, timeout=300)
        with tempfile.TemporaryDirectory(prefix='framey-decky-runtime-') as temp:
            loader = Path(temp)/'PluginLoader'
            loader.write_bytes(data)
            loader.chmod(0o644)  # Private container user copies/chmods its own executable.
            probes = [(case['id'], False) for case in cases] + [(111, True)]
            for plugin_id, negative in probes:
                name = 'framey-decky-'+uuid.uuid4().hex[:12]
                command = ['docker', 'run', '--rm', '--name', name, '--network', 'none', '--read-only',
                    '--cap-drop', 'ALL', '--security-opt', 'no-new-privileges', '--pids-limit', '128',
                    '--tmpfs', '/tmp:rw,exec,mode=1777',
                    '--tmpfs', '/home/tester:rw,exec,uid=1100,gid=1100,mode=700',
                    '--tmpfs', '/run/user/1100:rw,uid=1100,gid=1100,mode=700',
                    '--mount', f'type=bind,source={ROOT},target=/repo,readonly',
                    '--mount', f'type=bind,source={loader},target=/input/PluginLoader,readonly',
                    tag, str(plugin_id)]
                if negative:
                    command.append('--reproduce-bound-method-regression')
                try:
                    result = subprocess.run(command, capture_output=True, text=True, timeout=120)
                    if result.returncode:
                        raise RuntimeError(f'Decky runtime case {plugin_id} failed:\n{result.stdout}\n{result.stderr}')
                    observation = json.loads(result.stdout)
                    report['cases'].append(observation)
                    print(f"PASS {plugin_id}: {observation['result']}", flush=True)
                finally:
                    subprocess.run(['docker', 'rm', '-f', name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    finally:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(json.dumps(report, indent=2)+'\n')
        subprocess.run(['docker', 'image', 'rm', tag], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


if __name__ == '__main__':
    main()