feat(plugin): add Python backend skeleton with settings management
This commit is contained in:
@@ -0,0 +1,56 @@
|
|||||||
|
import asyncio
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
|
||||||
|
try:
|
||||||
|
import decky
|
||||||
|
except ImportError:
|
||||||
|
# Allow running tests without the decky module (tests mock the path)
|
||||||
|
decky = None
|
||||||
|
|
||||||
|
|
||||||
|
class Plugin:
|
||||||
|
async def _main(self):
|
||||||
|
if decky:
|
||||||
|
decky.logger.info(f"DeckyVault plugin loaded: {decky.DECKY_PLUGIN_NAME}")
|
||||||
|
self._settings_path = self._get_settings_path()
|
||||||
|
self._settings = self._read_settings()
|
||||||
|
|
||||||
|
async def _unload(self):
|
||||||
|
if decky:
|
||||||
|
decky.logger.info("DeckyVault plugin unloading")
|
||||||
|
|
||||||
|
async def _uninstall(self):
|
||||||
|
if decky:
|
||||||
|
decky.logger.info("DeckyVault plugin uninstalled")
|
||||||
|
|
||||||
|
def _get_settings_path(self):
|
||||||
|
if decky:
|
||||||
|
return os.path.join(decky.DECKY_PLUGIN_SETTINGS_DIR, "settings.json")
|
||||||
|
return os.path.join(os.path.expanduser("~"), ".deckyvault-test", "settings.json")
|
||||||
|
|
||||||
|
def _read_settings(self):
|
||||||
|
"""Read settings from JSON file. Returns empty dict if file missing."""
|
||||||
|
if os.path.exists(self._settings_path):
|
||||||
|
try:
|
||||||
|
with open(self._settings_path, 'r') as f:
|
||||||
|
return json.load(f)
|
||||||
|
except (json.JSONDecodeError, IOError):
|
||||||
|
return {}
|
||||||
|
return {}
|
||||||
|
|
||||||
|
def _write_settings(self, settings):
|
||||||
|
"""Write settings to JSON file, creating directory if needed."""
|
||||||
|
os.makedirs(os.path.dirname(self._settings_path), exist_ok=True)
|
||||||
|
with open(self._settings_path, 'w') as f:
|
||||||
|
json.dump(settings, f, indent=2)
|
||||||
|
|
||||||
|
async def get_settings(self) -> dict:
|
||||||
|
"""RPC: Return all plugin settings."""
|
||||||
|
return self._settings
|
||||||
|
|
||||||
|
async def set_setting(self, key: str, value) -> dict:
|
||||||
|
"""RPC: Set a single setting and persist. Returns updated settings."""
|
||||||
|
self._settings[key] = value
|
||||||
|
self._write_settings(self._settings)
|
||||||
|
return self._settings
|
||||||
@@ -0,0 +1,49 @@
|
|||||||
|
"""Tests for settings persistence in the Python backend."""
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
# We test the settings logic directly, not through the Plugin class,
|
||||||
|
# so we can run tests without the decky module.
|
||||||
|
|
||||||
|
def write_settings(settings_path, settings):
|
||||||
|
"""Write settings JSON to the given path."""
|
||||||
|
os.makedirs(os.path.dirname(settings_path), exist_ok=True)
|
||||||
|
with open(settings_path, 'w') as f:
|
||||||
|
json.dump(settings, f, indent=2)
|
||||||
|
|
||||||
|
def read_settings(settings_path):
|
||||||
|
"""Read settings JSON from the given path, return empty dict if missing."""
|
||||||
|
if os.path.exists(settings_path):
|
||||||
|
with open(settings_path, 'r') as f:
|
||||||
|
return json.load(f)
|
||||||
|
return {}
|
||||||
|
|
||||||
|
def test_read_settings_returns_empty_when_file_missing():
|
||||||
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
|
path = os.path.join(tmpdir, "settings.json")
|
||||||
|
assert read_settings(path) == {}
|
||||||
|
|
||||||
|
def test_write_then_read_settings():
|
||||||
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
|
path = os.path.join(tmpdir, "subdir", "settings.json")
|
||||||
|
write_settings(path, {"apiKey": "dv_test123", "exportPath": "/home/deck/Downloads"})
|
||||||
|
result = read_settings(path)
|
||||||
|
assert result["apiKey"] == "dv_test123"
|
||||||
|
assert result["exportPath"] == "/home/deck/Downloads"
|
||||||
|
|
||||||
|
def test_write_settings_creates_directory():
|
||||||
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
|
path = os.path.join(tmpdir, "newdir", "settings.json")
|
||||||
|
write_settings(path, {"key": "value"})
|
||||||
|
assert os.path.exists(path)
|
||||||
|
|
||||||
|
def test_read_settings_handles_corrupt_json():
|
||||||
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
|
path = os.path.join(tmpdir, "settings.json")
|
||||||
|
with open(path, 'w') as f:
|
||||||
|
f.write("{invalid json")
|
||||||
|
with pytest.raises(json.JSONDecodeError):
|
||||||
|
read_settings(path)
|
||||||
Reference in New Issue
Block a user