From 3d2f1384df78bfc21ea55f808c697759ab353795 Mon Sep 17 00:00:00 2001 From: Adrian Bonpin Date: Mon, 13 Jul 2026 04:59:17 +0800 Subject: [PATCH] fix(plugin): extract _collect_screenshots pure function, add fallback test --- plugins/decky-vault/main.py | 88 ++++++++------- plugins/decky-vault/tests/test_screenshots.py | 104 ++++++++++-------- 2 files changed, 106 insertions(+), 86 deletions(-) diff --git a/plugins/decky-vault/main.py b/plugins/decky-vault/main.py index 8efce87..c51ccd4 100644 --- a/plugins/decky-vault/main.py +++ b/plugins/decky-vault/main.py @@ -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)} diff --git a/plugins/decky-vault/tests/test_screenshots.py b/plugins/decky-vault/tests/test_screenshots.py index ae8ffdf..da0cdba 100644 --- a/plugins/decky-vault/tests/test_screenshots.py +++ b/plugins/decky-vault/tests/test_screenshots.py @@ -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 \ No newline at end of file + 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) \ No newline at end of file