Version Framey sample settings without losing old data

.github/workflows/decky-runtime.yml

@@ -1,32 +0,0 @@
-name: Decky backend runtime
-
-on:
-  push:
-    branches: [main]
-  pull_request:
-
-permissions:
-  contents: read
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}
-  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
-
-jobs:
-  decky-runtime:
-    strategy:
-      fail-fast: false
-      matrix:
-        os: [ubuntu-latest, ubuntu-24.04-arm]
-    runs-on: ${{ matrix.os }}
-    timeout-minutes: 10
-    steps:
-      - uses: actions/checkout@v4
-      - name: Actual Decky 3.3.0 discovery and safe backend probes
-        run: python3 scripts/framey-test-decky-runtime.py --output "$RUNNER_TEMP/decky-runtime.json"
-      - uses: actions/upload-artifact@v4
-        if: always()
-        with:
-          name: decky-runtime-${{ matrix.os }}
-          path: ${{ runner.temp }}/decky-runtime.json
-          if-no-files-found: warn

docs/testing.md

@@ -549,41 +549,3 @@ Fixtures cover duplicate instance identity, unknown states, GUID matching,
 localised/denied channel data and 6 GHz ambiguity. No tests contact a real Frame.
 A Windows CI runner/VM cannot prove physical Steam dongle behaviour or audio RF
 interference. See [sources and limits](windows-wireless-doctor.md).
-
-### Actual Decky backend shim gate
-
-Fake-Frame tests stub plugin execution, so they cannot prove Decky's Python
-calling convention. `Decky backend runtime` CI separately runs the published,
-SHA-256-pinned **Decky 3.3.0** executable on native amd64 and ARM64 Linux.
-The loader discovers production-managed official archives and calls their public
-methods over its actual websocket API. No test imports or instantiates `Plugin`.
-
-```sh
-python3 scripts/framey-test-decky-runtime.py --output /tmp/decky-runtime.json
-# Offline loader cache (still checked against its pinned native digest):
-python3 scripts/framey-test-decky-runtime.py --loader /path/to/PluginLoader --output /tmp/decky-runtime.json
-```
-
-Run on Linux with Docker. Image build and the pinned loader download need network;
-plugin execution has no network, published ports, host home/audio/D-Bus/device
-mounts, capabilities or writable repository. Each case has its own unprivileged
-container, private tmpfs home and temporary files. Nothing contacts the headset
-or its shared lock. Failed/timed-out containers and the private image tag are
-removed by the runner.
-
-Every `compat_shims.append` in the managed installer must have a case in
-`tests/decky-runtime/cases.json`. Add each affected pinned official archive and
-a safe public-method probe when introducing a shim. Ordinary unit tests reject
-uncovered shim names. The runtime job verifies the managed receipt, loader
-plugin discovery, and a successful backend call; a discovery listing alone
-could conceal a crashed backend process.
-
-Current cases cover both Music Control entries and Shotty. Music Control uses
-the private silent MPRIS fixture to check discovery, metadata, play/pause,
-next/previous, seek and volume, with independent service-state readbacks.
-Shotty checks backend import, rescuer state and empty screenshot aggregation
-inside its disposable home. A fourth negative-control case deliberately
-reintroduces #335's incorrect bound-method transformation and must reproduce
-the real loader's legacy backend-call failure. The gate fails if that regression goes
-undetected. These checks verify backend behavior in a container; they do not
-upgrade any plugin's Frame, actual media playback or worn-VR compatibility claim.

examples/framey-settings/README.md

@@ -14,6 +14,12 @@ action calls the backend, so it reads the persisted setting rather than a
 browser cache. A missing, damaged, or old non-boolean value reads as the
 caller's default;
 the next save atomically replaces the sample's settings file.
+The current file is `{"schema": 1, "enabled": true|false}`. An unversioned
+`{"enabled": true|false}` file from the first sample release still reads, and
+the next save writes schema 1. A file with another explicit schema returns a
+visible read or save error and remains untouched, so downgrading the sample
+cannot erase settings written by a newer version. Add a deliberate migration
+for each future schema instead of changing this guard silently.
 The panel also guards its own read and write calls synchronously. If two
 controller events arrive before React redraws a disabled control, the second
 event is ignored; after the request settles, a new change can be saved.

examples/framey-settings/main.py

@@ -5,26 +5,41 @@ from pathlib import Path
 import tempfile
 import decky
 
+SETTINGS_SCHEMA = 1
+
 
 def settings_file():
     # Decky provides this per-plugin path through its official Python module.
     return Path(decky.DECKY_PLUGIN_SETTINGS_DIR) / 'settings.json'
 
 
+def read_settings():
+    try:
+        stored = json.loads(settings_file().read_text(encoding='utf-8'))
+    except (FileNotFoundError, json.JSONDecodeError, UnicodeDecodeError):
+        return {}
+    if not isinstance(stored, dict):
+        return {}
+    # The first sample release wrote {"enabled": bool} without a schema.
+    # Keep reading it; the next save writes the current format.
+    schema = stored.get('schema', SETTINGS_SCHEMA)
+    if type(schema) is not int or schema != SETTINGS_SCHEMA:
+        raise ValueError('Settings were written by an unsupported sample version')
+    return stored
+
+
 class Plugin:
     async def get_setting(self, key: str, default=False):
         if key != 'enabled':
             raise ValueError('Unknown setting')
-        try:
-            stored = json.loads(settings_file().read_text(encoding='utf-8'))
-        except (FileNotFoundError, json.JSONDecodeError, UnicodeDecodeError):
-            return default
-        value = stored.get(key) if isinstance(stored, dict) else None
+        value = read_settings().get(key)
         return value if type(value) is bool else default
 
     async def set_setting(self, key: str, value: bool):
         if key != 'enabled' or type(value) is not bool:
             raise ValueError('Expected an enabled boolean')
+        # Refuse to erase settings from a newer plugin after a downgrade.
+        read_settings()
         target = settings_file()
         target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
         temporary = None
@@ -32,7 +47,7 @@ class Plugin:
             with tempfile.NamedTemporaryFile('w', encoding='utf-8', dir=target.parent,
                                              prefix='.settings-', delete=False) as output:
                 temporary = Path(output.name)
-                json.dump({'enabled': value}, output)
+                json.dump({'schema': SETTINGS_SCHEMA, 'enabled': value}, output)
                 output.write('\n')
             temporary.chmod(0o600)
             os.replace(temporary, target)

scripts/framey-test-decky-runtime.py

@@ -1,114 +0,0 @@
-#!/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()

tests/decky-runtime/Containerfile

@@ -1,12 +0,0 @@
-# Native amd64 or ARM64; no emulation, host session bus, audio, devices or home.
-FROM debian:trixie-slim
-RUN apt-get update && apt-get install -y --no-install-recommends \
-      python3 python3-dbus python3-gi python3-websockets dbus-bin dbus-daemon \
-      ca-certificates zlib1g \
-    && rm -rf /var/lib/apt/lists/* \
-    && useradd -m -u 1100 -s /bin/sh tester \
-    && mkdir -p /run/user/1100 && chown tester:tester /run/user/1100
-ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 XDG_RUNTIME_DIR=/run/user/1100
-USER tester
-WORKDIR /repo
-ENTRYPOINT ["python3", "/repo/tests/decky-runtime/check.py"]

tests/decky-runtime/cases.json

@@ -1,5 +0,0 @@
-[
-  {"id": 12, "shim": "music-control-system-dbus-env", "probe": "mpris"},
-  {"id": 111, "shim": "music-control-system-dbus-env", "probe": "mpris"},
-  {"id": 53, "shim": "shotty-click-app-dir", "probe": "screenshots"}
-]

tests/decky-runtime/check.py

@@ -1,199 +0,0 @@
-"""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 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')
-    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())
-    assert receipt['compatShims'] == [case['shim']], receipt
-    # Exercise the production manifest's API convention, never import Plugin ourselves.
-    manifest = json.loads((plugin/'plugin.json').read_text())
-    assert 'api_version' not in manifest, 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 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()

tests/test_framey_decky_runtime.py

@@ -1,36 +0,0 @@
-"""Managed backend shims must stay covered by the actual-Decky CI gate."""
-import sandbox  # noqa: F401
-import importlib.util
-from pathlib import Path
-import unittest
-from unittest.mock import patch
-
-ROOT = Path(__file__).resolve().parents[1]
-spec = importlib.util.spec_from_file_location('decky_runtime_gate', ROOT/'scripts/framey-test-decky-runtime.py')
-gate = importlib.util.module_from_spec(spec)
-spec.loader.exec_module(gate)
-
-
-class RuntimeCoverage(unittest.TestCase):
-    def test_current_shims_have_unique_official_archive_probes(self):
-        from framey_official import PLUGINS
-        cases = gate.check_coverage()
-        official = {record['id'] for record in PLUGINS}
-        self.assertEqual(len(cases), len({case['id'] for case in cases}))
-        for case in cases:
-            self.assertIn(case['id'], official)
-            self.assertIn(case['probe'], ('mpris', 'screenshots'))
-
-    def test_new_shim_without_runtime_case_fails_coverage(self):
-        import frame_decky_store
-        with patch.object(frame_decky_store, 'APPLY', frame_decky_store.APPLY+
-                          '\ncompat_shims.append("uncovered-new-shim")\n'):
-            with self.assertRaisesRegex(ValueError, 'Every managed shim'):
-                gate.check_coverage()
-
-    def test_dynamic_shim_names_cannot_evade_inventory(self):
-        import frame_decky_store
-        with patch.object(frame_decky_store, 'APPLY', frame_decky_store.APPLY+
-                          '\ncompat_shims.append(variable_name)\n'):
-            with self.assertRaisesRegex(ValueError, 'literal name'):
-                gate.check_coverage()

tests/test_framey_sample.py

@@ -1,6 +1,7 @@
 """The author example stores settings as a normal Decky user plugin."""
 import asyncio
 import importlib.util
+import json
 from pathlib import Path
 import subprocess
 import sys
@@ -36,7 +37,8 @@ class SamplePlugin(unittest.TestCase):
             self.assertTrue(asyncio.run(first.set_setting('enabled', True)))
             second = backend(folder)
             self.assertTrue(asyncio.run(second.get_setting('enabled', False)))
-            self.assertTrue((Path(folder) / 'settings.json').is_file())
+            self.assertEqual(json.loads((Path(folder) / 'settings.json').read_text()),
+                             {'schema': 1, 'enabled': True})
 
     def test_backend_rejects_unknown_keys_and_non_boolean_values(self):
         with tempfile.TemporaryDirectory() as folder:
@@ -57,6 +59,43 @@ class SamplePlugin(unittest.TestCase):
             self.assertTrue(asyncio.run(plugin.set_setting('enabled', True)))
             self.assertTrue(asyncio.run(backend(folder).get_setting('enabled', False)))
 
+    def test_legacy_boolean_survives_restart_and_migrates_on_save(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path = Path(folder) / 'settings.json'
+            path.write_text('{"enabled": true}\n')
+            self.assertTrue(asyncio.run(backend(folder).get_setting('enabled', False)))
+            self.assertEqual(json.loads(path.read_text()), {'enabled': True},
+                             'Reading must not rewrite a user setting')
+            self.assertTrue(asyncio.run(backend(folder).set_setting('enabled', False)))
+            self.assertEqual(json.loads(path.read_text()), {'schema': 1, 'enabled': False})
+            self.assertFalse(asyncio.run(backend(folder).get_setting('enabled', True)))
+
+    def test_newer_schema_is_preserved_on_read_and_save(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path = Path(folder) / 'settings.json'
+            original = '{"schema": 2, "enabled": true, "future": "keep"}\n'
+            path.write_text(original)
+            plugin = backend(folder)
+            with self.assertRaisesRegex(ValueError, 'unsupported sample version'):
+                asyncio.run(plugin.get_setting('enabled', False))
+            with self.assertRaisesRegex(ValueError, 'unsupported sample version'):
+                asyncio.run(plugin.set_setting('enabled', False))
+            self.assertEqual(path.read_text(), original)
+            self.assertEqual(list(Path(folder).glob('.settings-*')), [])
+
+    def test_failed_atomic_replace_keeps_previous_setting(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path = Path(folder) / 'settings.json'
+            plugin = backend(folder)
+            self.assertTrue(asyncio.run(plugin.set_setting('enabled', True)))
+            previous = path.read_bytes()
+            with patch('os.replace', side_effect=OSError('storage unavailable')):
+                with self.assertRaisesRegex(OSError, 'storage unavailable'):
+                    asyncio.run(plugin.set_setting('enabled', False))
+            self.assertEqual(path.read_bytes(), previous)
+            self.assertEqual(list(Path(folder).glob('.settings-*')), [])
+            self.assertTrue(asyncio.run(backend(folder).get_setting('enabled', False)))
+
 
 if __name__ == '__main__':
     unittest.main()