adrianbonpin/deckyvault
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(plugin): extract _collect_screenshots pure function, add fallback test

Browse Source
This commit is contained in:
adrianbonpin
2026-07-13 04:59:17 +08:00
parent 0018673747
commit 3d2f1384df
2 changed files with 106 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/decky-vault/main.py
+48 -40
View File
@@ -23,6 +23,52 @@ except ImportError:
decky = None
def _collect_screenshots(home: str, app_id: int | None = None, limit: int = 50) -> list[dict]:
"""Pure helper: collect screenshots from Desktop and Game Mode paths under `home`.
Returns list of dicts sorted by mtime desc, limited to `limit` entries.
When `app_id` is given, keeps Desktop exports + that app's userdata shots."""
import glob
base = os.path.join(home, "Pictures", "Screenshots")
userdata = os.path.join(home, ".local", "share", "Steam", "userdata")
patterns = [
os.path.join(base, "*.jpg"),
os.path.join(base, "*.png"),
os.path.join(base, "Steam Client", "*.jpg"),
os.path.join(base, "Steam Client", "*.png"),
os.path.join(userdata, "*", "760", "remote", "*", "screenshots", "*.jpg"),
os.path.join(userdata, "*", "760", "remote", "*", "screenshots", "*.png"),
]
seen, files = set(), []
for pat in patterns:
for f in glob.glob(pat):
if not os.path.isfile(f) or f in seen:
continue
if os.path.basename(f) == "most_recent.jpg":
continue
seen.add(f)
f_app_id = None
parts = f.split(os.sep)
if "760" in parts:
idx = parts.index("760")
if idx + 2 < len(parts):
try:
f_app_id = int(parts[idx + 2])
except ValueError:
pass
try:
files.append({
"path": f, "name": os.path.basename(f),
"mtime": os.path.getmtime(f), "size": os.path.getsize(f),
"appId": f_app_id,
})
except OSError:
continue
if app_id is not None:
files = [x for x in files if x["appId"] is None or x["appId"] == app_id]
files.sort(key=lambda x: x["mtime"], reverse=True)
return files[:limit]
def parse_mangohud_log(log_content: str) -> dict:
"""Parse a MangoHud log file's content and return FPS stats.
@@ -645,48 +691,10 @@ exec mangohud "$@"
"""RPC: List recent Steam screenshots from both Game Mode (userdata/760/remote)
and Desktop Mode (~/Pictures/Screenshots). Returns newest first.
When app_id is given, keeps all Desktop exports + that app's userdata shots."""
import glob
try:
home = os.path.expanduser("~")
base = os.path.join(home, "Pictures", "Screenshots")
userdata = os.path.join(home, ".local", "share", "Steam", "userdata")
patterns = [
os.path.join(base, "*.jpg"),
os.path.join(base, "*.png"),
os.path.join(base, "Steam Client", "*.jpg"),
os.path.join(base, "Steam Client", "*.png"),
os.path.join(userdata, "*", "760", "remote", "*", "screenshots", "*.jpg"),
os.path.join(userdata, "*", "760", "remote", "*", "screenshots", "*.png"),
]
seen, files = set(), []
for pat in patterns:
for f in glob.glob(pat):
if not os.path.isfile(f) or f in seen:
continue
if os.path.basename(f) == "most_recent.jpg":
continue
seen.add(f)
f_app_id = None
parts = f.split(os.sep)
if "760" in parts:
idx = parts.index("760")
if idx >= 2:
try:
f_app_id = int(parts[idx + 2])
except (ValueError, IndexError):
pass
try:
files.append({
"path": f, "name": os.path.basename(f),
"mtime": os.path.getmtime(f), "size": os.path.getsize(f),
"appId": f_app_id,
})
except OSError:
continue
if app_id is not None:
files = [x for x in files if x["appId"] is None or x["appId"] == app_id]
files.sort(key=lambda x: x["mtime"], reverse=True)
return {"screenshots": files[:limit]}
files = _collect_screenshots(home, app_id=app_id, limit=limit)
return {"screenshots": files}
except Exception as e:
return {"screenshots": [], "error": str(e)}
plugins/decky-vault/tests/test_screenshots.py
+57 -45
View File
@@ -1,9 +1,14 @@
"""Tests for screenshot discovery across Steam Game Mode + Desktop paths (Bug 3)."""
import os
import shutil
import sys
import tempfile
import time
import types
import pytest
from main import _collect_screenshots
def _touch(path, mtime_age=10, content=b"\xff\xd8\xff\xe0"):
os.makedirs(os.path.dirname(path), exist_ok=True)
@@ -13,55 +18,13 @@ def _touch(path, mtime_age=10, content=b"\xff\xd8\xff\xe0"):
os.utime(path, (t, t))
def _discover(home, app_id=None):
"""Mirror of Plugin.list_screenshot discovery against a fake home dir."""
import glob
base = os.path.join(home, "Pictures", "Screenshots")
userdata = os.path.join(home, ".local", "share", "Steam", "userdata")
patterns = [
os.path.join(base, "*.jpg"),
os.path.join(base, "*.png"),
os.path.join(base, "Steam Client", "*.jpg"),
os.path.join(base, "Steam Client", "*.png"),
os.path.join(userdata, "*", "760", "remote", "*", "screenshots", "*.jpg"),
os.path.join(userdata, "*", "760", "remote", "*", "screenshots", "*.png"),
]
seen, files = set(), []
for pat in patterns:
for f in glob.glob(pat):
if not os.path.isfile(f) or f in seen:
continue
if os.path.basename(f) == "most_recent.jpg":
continue
seen.add(f)
parts = f.split(os.sep)
f_app_id = None
if "760" in parts:
idx = parts.index("760")
if idx >= 2:
try:
f_app_id = int(parts[idx + 2])
except (ValueError, IndexError):
pass
try:
files.append({"path": f, "name": os.path.basename(f),
"mtime": os.path.getmtime(f), "size": os.path.getsize(f),
"appId": f_app_id})
except OSError:
continue
files.sort(key=lambda x: x["mtime"], reverse=True)
if app_id is not None:
files = [x for x in files if x["appId"] is None or x["appId"] == app_id]
return files
def test_discovers_all_three_locations():
with tempfile.TemporaryDirectory() as home:
_touch(os.path.join(home, "Pictures", "Screenshots", "desktop.jpg"), mtime_age=30)
_touch(os.path.join(home, "Pictures", "Screenshots", "Steam Client", "sc.jpg"), mtime_age=20)
_touch(os.path.join(home, ".local", "share", "Steam", "userdata", "111", "760",
"remote", "2531310", "screenshots", "game.jpg"), mtime_age=10)
found = _discover(home)
found = _collect_screenshots(home)
names = [f["name"] for f in found]
assert set(names) == {"desktop.jpg", "sc.jpg", "game.jpg"}
assert found[0]["name"] == "game.jpg"
@@ -74,7 +37,7 @@ def test_app_id_filter_keeps_desktop_plus_app():
"remote", "2531310", "screenshots", "want.jpg"), mtime_age=10)
_touch(os.path.join(home, ".local", "share", "Steam", "userdata", "111", "760",
"remote", "9999", "screenshots", "other.jpg"), mtime_age=5)
found = _discover(home, app_id=2531310)
found = _collect_screenshots(home, app_id=2531310)
names = [f["name"] for f in found]
assert "want.jpg" in names
assert "desktop.jpg" in names
@@ -85,7 +48,56 @@ def test_most_recent_duplicate_skipped():
with tempfile.TemporaryDirectory() as home:
_touch(os.path.join(home, "Pictures", "Screenshots", "most_recent.jpg"), mtime_age=1)
_touch(os.path.join(home, "Pictures", "Screenshots", "2026-01-01.jpg"), mtime_age=2)
found = _discover(home)
found = _collect_screenshots(home)
names = [f["name"] for f in found]
assert "most_recent.jpg" not in names
assert "2026-01-01.jpg" in names
@pytest.mark.asyncio
async def test_read_screenshot_fallback_no_pillow():
"""When Pillow is absent, >1MB files return preview-unavailable, small files return data URL."""
# Mock the decky module so main.py imports cleanly even when the real
# decky package (only present on the Deck) is unavailable.
sys.modules.pop("main", None)
if "decky" not in sys.modules:
mock_decky = types.ModuleType("decky")
mock_decky.logger = type(sys)("Logger")
mock_decky.logger.info = lambda *a, **kw: None
mock_decky.DECKY_PLUGIN_NAME = "test"
mock_decky.DECKY_PLUGIN_SETTINGS_DIR = "/tmp/decky-test"
sys.modules["decky"] = mock_decky
from main import Plugin
plugin = Plugin()
test_dir = f"/tmp/deckyvault_test_screenshot_{os.getpid()}"
try:
os.makedirs(test_dir, exist_ok=True)
small_path = os.path.join(test_dir, "small.jpg")
with open(small_path, "wb") as f:
f.write(b"\xff\xd8\xff\xe0" * 100) # ~400 bytes
large_path = os.path.join(test_dir, "large.jpg")
with open(large_path, "wb") as f:
f.write(b"\xff\xd8\xff\xe0" * 300000) # ~1.2MB
# Mock PIL ImportError by temporarily removing PIL from sys.modules
had_pil = "PIL" in sys.modules
if had_pil:
pil_mod = sys.modules.pop("PIL")
try:
r_small = await plugin.read_screenshot(small_path, max_width=320)
assert r_small["dataUrl"] != "", f"Expected data URL, got: {r_small}"
assert "error" not in r_small or not r_small["error"]
r_large = await plugin.read_screenshot(large_path, max_width=320)
assert r_large["dataUrl"] == "", f"Expected empty dataUrl, got: {r_large}"
assert "too large" in r_large.get("error", "")
finally:
if had_pil:
sys.modules["PIL"] = pil_mod
finally:
shutil.rmtree(test_dir, ignore_errors=True)
sys.modules.pop("main", None)