mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-27 16:01:24 +01:00
230 lines
11 KiB
Python
230 lines
11 KiB
Python
"""Container-only real Decky backend discovery and smoke tests; no host services."""
|
|
import argparse
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import zipfile
|
|
import io
|
|
|
|
import websockets
|
|
from websockets.exceptions import WebSocketException
|
|
|
|
ROOT = Path('/repo')
|
|
sys.path[:0] = [str(ROOT/'ui'), str(ROOT/'tests')]
|
|
import frame_decky_store
|
|
from framey_official import PLUGINS, archive
|
|
|
|
|
|
async def request(route, args):
|
|
token = urllib.request.urlopen('http://127.0.0.1:1337/auth/token', timeout=2).read().decode()
|
|
async with websockets.connect('ws://127.0.0.1:1337/ws?auth='+token,
|
|
open_timeout=3, close_timeout=2) as ws:
|
|
await ws.send(json.dumps({'type': 0, 'id': 1, 'route': route, 'args': args}))
|
|
while True:
|
|
response = json.loads(await asyncio.wait_for(ws.recv(), timeout=8))
|
|
if response.get('id') != 1:
|
|
continue
|
|
await ws.send(json.dumps({'type': 3, 'id': 1}))
|
|
assert response['type'] == 1, response
|
|
return response['result']
|
|
|
|
|
|
def rpc(route, *args):
|
|
return asyncio.run(request(route, list(args)))
|
|
|
|
|
|
def until_ready(read, description):
|
|
deadline = time.monotonic()+20
|
|
last = None
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
value = read()
|
|
if value:
|
|
return value
|
|
except (OSError, ValueError, KeyError, TimeoutError, WebSocketException) as error:
|
|
last = str(error)
|
|
time.sleep(.1)
|
|
raise RuntimeError(f'{description} not ready: {last}')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('id', type=int)
|
|
parser.add_argument('--reproduce-bound-method-regression', action='store_true')
|
|
parser.add_argument('--reproduce-timer-regression', action='store_true')
|
|
args = parser.parse_args()
|
|
assert os.getuid() == 1100 and ROOT.is_dir(), 'Run through framey-test-decky-runtime.py'
|
|
record = next(p for p in PLUGINS if p['id'] == args.id)
|
|
case = next(p for p in json.loads((ROOT/'tests/decky-runtime/cases.json').read_text()) if p['id'] == args.id)
|
|
data = archive(record)
|
|
assert hashlib.sha256(data).hexdigest() == record['sha256']
|
|
assert frame_decky_store.archive_info(data, record['sha256'], record['name']) == record['folder']
|
|
home = Path.home()
|
|
base = home/'.local/share/frame-control/decky'
|
|
(base/'plugins').mkdir(parents=True)
|
|
unit = home/'.config/systemd/user/frame-control-decky.service'
|
|
unit.parent.mkdir(parents=True)
|
|
unit.write_text('# Private fixture marker; no real service manager is contacted.\n')
|
|
tools = home/'bin'
|
|
tools.mkdir()
|
|
systemctl = tools/'systemctl'
|
|
systemctl.write_text('#!/bin/sh\ncase "$*" in *is-active*) exit 3;; *) exit 0;; esac\n')
|
|
systemctl.chmod(0o700)
|
|
env = dict(os.environ, PATH=str(tools)+':'+os.environ['PATH'])
|
|
Path('/tmp/frame-control-decky-plugin-runtime-test').write_bytes(data)
|
|
install = subprocess.run([sys.executable, '-', 'install', 'runtime-test', str(record['id']),
|
|
record['sha256'], record['folder'], record['version'], '-'], input=frame_decky_store.APPLY,
|
|
capture_output=True, text=True, env=env, timeout=15)
|
|
assert install.returncode == 0, install.stderr
|
|
plugin = base/'plugins'/record['folder']
|
|
receipt = json.loads((plugin/'.frame-control-store.json').read_text())
|
|
if not args.reproduce_timer_regression:
|
|
assert receipt['compatShims'] == [case['shim']], receipt
|
|
else:
|
|
assert record['id'] == 96
|
|
with zipfile.ZipFile(io.BytesIO(data)) as original:
|
|
(plugin/'main.py').write_bytes(original.read(record['folder']+'/main.py'))
|
|
# Exercise the production manifest's API convention, never import Plugin ourselves.
|
|
manifest = json.loads((plugin/'plugin.json').read_text())
|
|
assert manifest.get('api_version', 'legacy') == record['api_version'], manifest
|
|
if args.reproduce_bound_method_regression:
|
|
assert record['id'] == 111
|
|
source = (plugin/'main.py').read_text()
|
|
for old, new in [('self._get_dbus_env(self)', 'self._get_dbus_env()'),
|
|
('self._sp_player_dbus(self,', 'self._sp_player_dbus(')]:
|
|
assert old in source
|
|
source = source.replace(old, new)
|
|
(plugin/'main.py').write_text(source)
|
|
|
|
(home/'.local/share/Steam/steamui/images').mkdir(parents=True)
|
|
loader = base/'PluginLoader'
|
|
shutil.copy('/input/PluginLoader', loader)
|
|
loader.chmod(0o700)
|
|
env.update(UNPRIVILEGED_PATH=str(base), PRIVILEGED_PATH=str(base), CHOWN_PLUGIN_PATH='0',
|
|
LIVE_RELOAD='0', USE_CEF_CLOSE_WORKAROUND='0', KEEP_SYSTEMD_SERVICE='1',
|
|
DBUS_SESSION_BUS_ADDRESS='unix:path=/run/user/1100/bus')
|
|
processes = []
|
|
report = {'id': record['id'], 'name': record['name'], 'version': record['version'],
|
|
'sha256': record['sha256'], 'shim': case['shim'], 'calls': []}
|
|
with (home/'runtime.log').open('w+') as log:
|
|
try:
|
|
if case['probe'] == 'mpris':
|
|
bus = subprocess.Popen(['dbus-daemon', '--session', '--nofork',
|
|
'--address='+env['DBUS_SESSION_BUS_ADDRESS']], env=env, stdout=log, stderr=log,
|
|
start_new_session=True)
|
|
processes.append(bus)
|
|
until_ready(lambda: Path('/run/user/1100/bus').is_socket(), 'private session bus')
|
|
player = subprocess.Popen([sys.executable, str(ROOT/'tests/fixtures/framey-mpris/player.py')],
|
|
env=env, stdout=log, stderr=log, start_new_session=True)
|
|
processes.append(player)
|
|
until_ready(lambda: Path('/tmp/mpris-state.json').is_file(), 'silent MPRIS player')
|
|
backend = subprocess.Popen([str(loader)], env=env, stdout=log, stderr=log, start_new_session=True)
|
|
processes.append(backend)
|
|
discovered = until_ready(lambda: rpc('loader/get_plugins'), 'published Decky plugin discovery')
|
|
assert len(discovered) == 1 and discovered[0]['name'] == record['name'], discovered
|
|
assert discovered[0]['version'] == record['version'] and not discovered[0]['disabled'], discovered
|
|
report['discovered'] = discovered
|
|
|
|
def legacy(method, **kwargs):
|
|
response = rpc('loader/call_legacy_plugin_method', record['name'], method, kwargs)
|
|
assert response['success'] is True, (method, response)
|
|
report['calls'].append({'method': method, 'result': response['result']})
|
|
return response['result']
|
|
|
|
if case['probe'] == 'timer-lifecycle':
|
|
from timer_lifecycle import probe
|
|
|
|
def restart(settings=None):
|
|
nonlocal backend
|
|
os.killpg(backend.pid, signal.SIGTERM)
|
|
backend.wait(timeout=8)
|
|
if settings is not None:
|
|
path = base/'settings'/record['folder']/'settings.json'
|
|
assert path.exists(), path
|
|
path.write_text(json.dumps(settings))
|
|
backend = subprocess.Popen([str(loader)], env=env, stdout=log, stderr=log,
|
|
start_new_session=True)
|
|
processes.append(backend)
|
|
until_ready(lambda: rpc('loader/get_plugins'), 'restarted Decky discovery')
|
|
until_ready(lambda: rpc('loader/call_plugin_method', record['name'],
|
|
'settings_getSetting', 'subtle_mode', False), 'restarted instance RPC')
|
|
|
|
report['lifecycle'] = probe(record['name'], restart, home/'runtime.log',
|
|
args.reproduce_timer_regression)
|
|
report['result'] = ('upstream timer lifecycle regressions reproduced' if args.reproduce_timer_regression
|
|
else 'discovery, cancel, replace, completion and persistent restart passed')
|
|
elif args.reproduce_bound_method_regression:
|
|
response = rpc('loader/call_legacy_plugin_method', record['name'], 'sp_list_media_players', {})
|
|
assert response['success'] is False and 'self' in response['result'], response
|
|
report['expectedRegression'] = response
|
|
report['result'] = 'historical bad shim rejected by actual Decky legacy dispatch'
|
|
elif case['probe'] == 'mpris':
|
|
player_name = 'org.mpris.MediaPlayer2.FrameySilent'
|
|
assert legacy('sp_list_media_players') == player_name
|
|
legacy('sp_set_media_player', player=player_name)
|
|
assert legacy('sp_get_media_player') == player_name
|
|
assert legacy('sp_identity', orgPath=player_name) == 'Player' # Upstream truncation.
|
|
metadata = legacy('get_meta_data')
|
|
assert 'title|Silent Test Track' in metadata and 'length|180000000' in metadata, metadata
|
|
assert legacy('sp_track_status') == 'Paused'
|
|
assert legacy('sp_track_progress') == '12000000'
|
|
assert legacy('sp_can_seek') == 'true'
|
|
assert float(legacy('sp_get_volume')) == .4
|
|
assert legacy('sp_test_volume_control') == 'true'
|
|
assert json.loads(Path('/tmp/mpris-state.json').read_text())['Volume'] == .4
|
|
legacy('sp_play')
|
|
assert legacy('sp_track_status') == 'Playing'
|
|
legacy('sp_pause')
|
|
legacy('sp_next')
|
|
legacy('sp_previous')
|
|
legacy('sp_seek', amount=2000000)
|
|
assert legacy('sp_track_progress') == '14000000'
|
|
legacy('sp_set_position', position=45000000, trackid='/org/framey/TestTrack/1')
|
|
legacy('sp_set_volume', volume=.25)
|
|
state = json.loads(Path('/tmp/mpris-state.json').read_text())
|
|
assert state['PlaybackStatus'] == 'Paused' and state['Volume'] == .25, state
|
|
assert state['next'] == state['previous'] == 1 and state['Position'] == 45000000, state
|
|
report['independentState'] = state
|
|
report['result'] = 'discovery, metadata, transport, seek and volume passed'
|
|
else:
|
|
assert case['probe'] == 'screenshots'
|
|
assert legacy('was_rescued') is False
|
|
assert legacy('aggregate_all', allapps=[]) == 0
|
|
report['result'] = 'discovery, backend import and empty screenshot aggregation passed'
|
|
log.flush()
|
|
log.seek(0)
|
|
assert 'Starting Decky version v3.3.0' in log.read()
|
|
print(json.dumps(report))
|
|
except Exception:
|
|
log.flush()
|
|
log.seek(0)
|
|
print(log.read()[-12000:], file=sys.stderr)
|
|
raise
|
|
finally:
|
|
for process in reversed(processes):
|
|
if process.poll() is None:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
continue
|
|
try:
|
|
process.wait(timeout=4)
|
|
except subprocess.TimeoutExpired:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
process.wait(timeout=4)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|