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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dsh4Znwh2KnC7q6vbGpX1N
This commit is contained in:
Krzysztof Rudnicki 2026-10-19 17:01:21 +02:00
parent c99dd12984
commit 04bd0837f1
2 changed files with 130 additions and 17 deletions

View File

@ -13,6 +13,7 @@ from __future__ import annotations
import os import os
from pathlib import Path from pathlib import Path
import re
from typing import Final from typing import Final
from leisure_watch.facts import DESKTOP_UID from leisure_watch.facts import DESKTOP_UID
@ -42,7 +43,13 @@ BROWSER_FAMILIES: Final = (
"mullvad-browser", "mullvad-browser",
) )
MEDIA_PLAYERS: Final = ("mpv", "vlc", "totem", "freetube") 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. # Headless `wsg-grabber scrape` never counts as leisure (Q7): never swept.
_WSG_HEADLESS: Final = "scrape" _WSG_HEADLESS: Final = "scrape"
# A process whose argv names one of these -- or any ancestor's does -- is a # 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 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: def exe_name(pid: int) -> str:
"""Lower-case basename of the pid's binary (``comm`` if exe is unreadable).""" """Lower-case basename of the pid's binary (``comm`` if exe is unreadable)."""
try: path = exe_path(pid)
return (_PROC / str(pid) / "exe").readlink().name.lower() if path:
except OSError: return Path(path).name.lower()
return (_read(pid, "comm") or "").strip().lower() return (_read(pid, "comm") or "").strip().lower()
def argv(pid: int) -> tuple[str, ...]: def argv(pid: int) -> tuple[str, ...]:
@ -129,14 +144,33 @@ def is_browser(pid: int) -> bool:
return _family(exe_name(pid), BROWSER_FAMILIES) return _family(exe_name(pid), BROWSER_FAMILIES)
def _arg_names(pid: int) -> set[str]: def _stem(arg: str) -> str:
# "/x/leetcode_guard_tray.py" -> leetcode_guard_tray; "-m screen_locker._web" # "/x/leetcode_guard_tray.py" -> leetcode_guard_tray; "screen_locker._web"
# -> screen_locker; "book-guard" -> book_guard. # -> screen_locker; "book-guard" -> book_guard.
names = set() stem = Path(arg).name.lower().removesuffix(".py").partition(".")[0]
for arg in argv(pid): return stem.replace("-", "_")
stem = Path(arg).name.lower().removesuffix(".py").partition(".")[0]
names.add(stem.replace("-", "_"))
return names 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]: 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: def _wsg_reviewer(pid: int) -> bool:
args = argv(pid) # Only the program actually run: a path naming the repo (git -C, pytest,
for i, arg in enumerate(args): # an editor) is not the reviewer.
if Path(arg).name in _WSG_NAMES: if _stem(exe_name(pid)) == _WSG:
return args[i + 1 : i + 2] != (_WSG_HEADLESS,) rest = argv(pid)[1:]
return False 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]: def sweep_targets() -> dict[int, str]:

View File

@ -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"