From 04bd0837f13d8a7b0343c1ebe2e70065fb53607b Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 19 Oct 2026 17:01:21 +0200 Subject: [PATCH] leisure kill: wsg-grabber only as the program run; see through (deleted) exes A path naming the repo (git -C ~/src/wsg-grabber, pytest, an editor) no longer counts as the reviewer: only the exe itself, or a Python interpreter's script / -m module, and still never `scrape`. An updated browser or player (exe link ' (deleted)') is still recognised, for the sweep and the browser focus exemption. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Dsh4Znwh2KnC7q6vbGpX1N --- steam_backlog_enforcer/_leisure_procs.py | 73 +++++++++++++----- .../tests/test_leisure_procs_identity.py | 74 +++++++++++++++++++ 2 files changed, 130 insertions(+), 17 deletions(-) create mode 100644 steam_backlog_enforcer/tests/test_leisure_procs_identity.py diff --git a/steam_backlog_enforcer/_leisure_procs.py b/steam_backlog_enforcer/_leisure_procs.py index 3d28079..e590d3e 100644 --- a/steam_backlog_enforcer/_leisure_procs.py +++ b/steam_backlog_enforcer/_leisure_procs.py @@ -13,6 +13,7 @@ from __future__ import annotations import os from pathlib import Path +import re from typing import Final from leisure_watch.facts import DESKTOP_UID @@ -42,7 +43,13 @@ BROWSER_FAMILIES: Final = ( "mullvad-browser", ) MEDIA_PLAYERS: Final = ("mpv", "vlc", "totem", "freetube") -_WSG_NAMES: Final = frozenset({"wsg_grabber", "wsg-grabber"}) +_WSG: Final = "wsg_grabber" +# An interpreter by exe basename (python, python3, python3.14), and its +# options that take a separate value, so the script after them is found. +_PYTHON: Final = re.compile(r"python[0-9.]*") +_OPTS_WITH_VALUE: Final = frozenset({"-W", "-X", "--check-hash-based-pycs"}) +# An updated binary's exe link (the package manager replaced the file). +_DELETED: Final = " (deleted)" # Headless `wsg-grabber scrape` never counts as leisure (Q7): never swept. _WSG_HEADLESS: Final = "scrape" # A process whose argv names one of these -- or any ancestor's does -- is a @@ -77,12 +84,20 @@ def _read(pid: int, name: str) -> str | None: return None +def exe_path(pid: int) -> str: + """The pid's binary, without an update's `` (deleted)``; ``""`` if unreadable.""" + try: + return str((_PROC / str(pid) / "exe").readlink()).removesuffix(_DELETED) + except OSError: + return "" + + def exe_name(pid: int) -> str: """Lower-case basename of the pid's binary (``comm`` if exe is unreadable).""" - try: - return (_PROC / str(pid) / "exe").readlink().name.lower() - except OSError: - return (_read(pid, "comm") or "").strip().lower() + path = exe_path(pid) + if path: + return Path(path).name.lower() + return (_read(pid, "comm") or "").strip().lower() def argv(pid: int) -> tuple[str, ...]: @@ -129,14 +144,33 @@ def is_browser(pid: int) -> bool: return _family(exe_name(pid), BROWSER_FAMILIES) -def _arg_names(pid: int) -> set[str]: - # "/x/leetcode_guard_tray.py" -> leetcode_guard_tray; "-m screen_locker._web" +def _stem(arg: str) -> str: + # "/x/leetcode_guard_tray.py" -> leetcode_guard_tray; "screen_locker._web" # -> screen_locker; "book-guard" -> book_guard. - names = set() - for arg in argv(pid): - stem = Path(arg).name.lower().removesuffix(".py").partition(".")[0] - names.add(stem.replace("-", "_")) - return names + stem = Path(arg).name.lower().removesuffix(".py").partition(".")[0] + return stem.replace("-", "_") + + +def _arg_names(pid: int) -> set[str]: + return {_stem(arg) for arg in argv(pid)} + + +def script_of(pid: int) -> tuple[str, tuple[str, ...]] | None: + """What a Python interpreter runs: (script or ``-m`` module stem, its args). + + ``None`` for any other binary, ``-c`` code or a bare interpreter. + """ + if not _PYTHON.fullmatch(exe_name(pid)): + return None + args = argv(pid)[1:] + i = 0 + while i < len(args) and args[i].startswith("-"): + if args[i] == "-c": + return None + if args[i] == "-m": + return (_stem(args[i + 1]), args[i + 2 :]) if args[i + 1 :] else None + i += 2 if args[i] in _OPTS_WITH_VALUE else 1 + return (_stem(args[i]), args[i + 1 :]) if i < len(args) else None def _ancestry(pid: int) -> list[int]: @@ -173,11 +207,16 @@ def is_protected(pid: int, chain: set[int]) -> bool: def _wsg_reviewer(pid: int) -> bool: - args = argv(pid) - for i, arg in enumerate(args): - if Path(arg).name in _WSG_NAMES: - return args[i + 1 : i + 2] != (_WSG_HEADLESS,) - return False + # Only the program actually run: a path naming the repo (git -C, pytest, + # an editor) is not the reviewer. + if _stem(exe_name(pid)) == _WSG: + rest = argv(pid)[1:] + else: + script = script_of(pid) + if script is None or script[0] != _WSG: + return False + rest = script[1] + return rest[:1] != (_WSG_HEADLESS,) def sweep_targets() -> dict[int, str]: diff --git a/steam_backlog_enforcer/tests/test_leisure_procs_identity.py b/steam_backlog_enforcer/tests/test_leisure_procs_identity.py new file mode 100644 index 0000000..b211ec6 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_leisure_procs_identity.py @@ -0,0 +1,74 @@ +"""What a process really runs: wsg-grabber as the program, updated binaries.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from steam_backlog_enforcer import _leisure_procs as procs +from steam_backlog_enforcer.tests._isolate_leisure import leisure_proc +from steam_backlog_enforcer.tests._leisure_fake_proc import FakeProc, add_proc + +if TYPE_CHECKING: + from pathlib import Path + +_PY = "/usr/bin/python3.14" +_WSG = "/home/kuhy/.local/bin/wsg-grabber" + + +@pytest.fixture +def proc(tmp_path: Path) -> Path: + return leisure_proc(tmp_path) + + +class TestWsgIsTheProgram: + @pytest.mark.parametrize( + ("exe", "args"), + [ + ("/usr/bin/git", ("git", "-C", "/home/kuhy/src/wsg-grabber", "status")), + (_PY, ("python3", "-m", "pytest", "/home/kuhy/src/wsg-grabber")), + ("/usr/bin/vim", ("vim", _WSG)), + (_PY, ("python3", _WSG, "scrape")), + (_PY, ("python3", "-u", "-m", "wsg_grabber", "scrape")), + ("/usr/bin/wsg-grabber", ("wsg-grabber", "scrape")), + (_PY, ("python3", "-c", "import wsg_grabber")), + (_PY, ("python3", "-m")), + (_PY, ("python3", "-X", "dev")), + (_PY, ("python3",)), + ], + ) + def test_not_a_reviewer(self, proc: Path, exe: str, args: tuple[str, ...]) -> None: + add_proc(proc, FakeProc(10, exe=exe, argv=args)) + assert procs.sweep_targets() == {} + + @pytest.mark.parametrize( + ("exe", "args"), + [ + (_PY, ("python3", _WSG)), + (_PY, ("python3", _WSG, "review")), + (_PY, ("python3", "-X", "dev", "-m", "wsg_grabber.cli")), + ("/usr/bin/python", ("python", "/x/wsg_grabber.py")), + ("/usr/bin/wsg_grabber", ("wsg_grabber", "review")), + ], + ) + def test_a_reviewer(self, proc: Path, exe: str, args: tuple[str, ...]) -> None: + add_proc(proc, FakeProc(10, exe=exe, argv=args)) + assert procs.sweep_targets() == {10: "sweep:wsg-grabber"} + + def test_script_of_a_non_interpreter_is_none(self, proc: Path) -> None: + add_proc(proc, FakeProc(10, exe="/usr/bin/git", argv=("git", "x.py"))) + assert procs.script_of(10) is None + + +class TestDeletedExe: + def test_an_updated_browser_is_still_a_browser(self, proc: Path) -> None: + add_proc(proc, FakeProc(10, exe="/opt/helium-browser-bin/helium (deleted)")) + assert procs.exe_name(10) == "helium" + assert procs.is_browser(10) + assert procs.sweep_targets() == {10: "sweep:browser:helium"} + + def test_an_updated_player_is_still_swept(self, proc: Path) -> None: + add_proc(proc, FakeProc(10, exe="/usr/bin/mpv (deleted)")) + assert procs.sweep_targets() == {10: "sweep:media:mpv"} + assert procs.exe_path(10) == "/usr/bin/mpv"