mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-11-11 12:47:32 +01:00
fix(enforcer): idle when Steam is absent instead of crash-looping
A total gaming block on 2026-07-04 purged the Steam client (packages and
all data dirs, per total_block_purge_log.json) and it was never
reinstalled. The enforcer kept trying to drive it anyway: every start
called ensure_steam_debug_port() -> _launch_steam_with_debug(), waited
45s for a CDP port no process would ever open, raised, and exited 1 into
Restart=always. That ran to 1050+ restarts.
Each attempt also left a zombie behind: _run_as_user() launches
fire-and-forget via Popen and never reaps, so a launch that dies on exec
lingers as a dead process still named "steam". The focus-mode daemon
scans /proc for process names and read that zombie as a live Steam,
entering GAMING mode and killing the user's browsers every few seconds.
- steam_is_installed(): probe the real /usr/bin/steam. Deliberately not
shutil.which("steam") - a launcher wrapper on $PATH keeps `which`
truthy long after the package is gone, which is precisely how a
deleted Steam went on looking installed.
- SteamUnavailableError (a RuntimeError, so existing handlers still
catch it) raised on both "not installed" and CDP timeout, and caught
at the _enforce_hide_games boundary: no library to hide is not fatal,
the store block and install guard still work.
- Guard _enforce_loop_iteration and game_install._ensure_steam_running,
whose 15s sleep was the source of the observed flapping interval.
- do_enforce falls through to the idle loop rather than returning:
returning exits the process, which under Restart=always is the same
crash loop by another name.
- _reap_spawned() retires exited launches so no zombie named "steam"
outlives them.
Verified live: NRestarts 1050 -> 0, service active/running for 60s+ with
zero errors and zero steam zombies, logging "Steam is not installed -
nothing to enforce."
Also removed the orphaned /usr/local/bin/steam wrapper, a leftover of
the purge that pointed at the deleted binary. Its body was:
#!/bin/sh
exec /usr/bin/steam -cef-disable-gpu -cef-disable-gpu-compositing "$@"
Restore those flags if a Steam reinstall brings back GPU trouble.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013rN2usTS3pYLJqeVg2ZpgR
This commit is contained in:
parent
e9f60e8ee8
commit
d54df3a1ad
@ -37,7 +37,11 @@ from steam_backlog_enforcer.game_install import (
|
||||
uninstall_game,
|
||||
uninstall_other_games,
|
||||
)
|
||||
from steam_backlog_enforcer.library_hider import hide_other_games
|
||||
from steam_backlog_enforcer.library_hider import (
|
||||
SteamUnavailableError,
|
||||
hide_other_games,
|
||||
steam_is_installed,
|
||||
)
|
||||
from steam_backlog_enforcer.steam_api import SteamAPIClient
|
||||
from steam_backlog_enforcer.store_blocker import block_store
|
||||
|
||||
@ -232,14 +236,26 @@ def _enforce_hide_games(config: Config, state: State) -> None:
|
||||
state: Current enforcer state.
|
||||
"""
|
||||
owned_ids = get_all_owned_app_ids(config)
|
||||
if owned_ids:
|
||||
hidden = hide_other_games(owned_ids, state.current_app_id)
|
||||
if hidden > 0:
|
||||
_echo(f" Library: hid {hidden} games (only assigned game visible)")
|
||||
else:
|
||||
_echo(" Library: games already hidden")
|
||||
else:
|
||||
if not owned_ids:
|
||||
_echo(" Library hiding: skipped (no owned game list — run 'scan' first)")
|
||||
return
|
||||
|
||||
# An unreachable Steam is not fatal: with no client there is no library to
|
||||
# hide, and everything else the enforcer does (store block, install guard)
|
||||
# still works. Letting this escape used to exit(1) into Restart=always,
|
||||
# which spun the service through ~1000 restarts against a Steam that had
|
||||
# been uninstalled - each attempt leaving a dead process named "steam"
|
||||
# behind that /proc scanners misread as a live Steam.
|
||||
try:
|
||||
hidden = hide_other_games(owned_ids, state.current_app_id)
|
||||
except SteamUnavailableError as exc:
|
||||
_echo(f" Library hiding: skipped ({exc})")
|
||||
return
|
||||
|
||||
if hidden > 0:
|
||||
_echo(f" Library: hid {hidden} games (only assigned game visible)")
|
||||
else:
|
||||
_echo(" Library: games already hidden")
|
||||
|
||||
|
||||
def _enforce_loop_iteration(config: Config, state: State) -> None:
|
||||
@ -259,6 +275,14 @@ def _enforce_loop_iteration(config: Config, state: State) -> None:
|
||||
if total_block_needs_cleanup():
|
||||
end_total_block_cleanup()
|
||||
|
||||
# With no Steam client there is no library, no installs and no game
|
||||
# processes, so every branch below is a no-op at best - and at worst a
|
||||
# 3s-interval error loop trying to write manifests into a steamapps
|
||||
# directory that a total block deleted. The total-block tick above still
|
||||
# runs: keeping Steam uninstalled is exactly what it is for.
|
||||
if not steam_is_installed():
|
||||
return
|
||||
|
||||
if state.current_app_id is None:
|
||||
return
|
||||
|
||||
@ -315,6 +339,13 @@ def do_enforce(config: Config, state: State) -> None:
|
||||
elif state.current_app_id is None:
|
||||
_echo("No game assigned. Run 'scan' first.")
|
||||
return
|
||||
elif not steam_is_installed():
|
||||
# Fall through to the idle loop rather than returning: returning exits
|
||||
# the process, and under Restart=always that is just the crash loop
|
||||
# again by another name. Staying alive also means a later Steam
|
||||
# reinstall is picked up without needing a restart.
|
||||
_echo("Steam is not installed — nothing to enforce.")
|
||||
_echo(" (Reinstall Steam to resume backlog enforcement.)")
|
||||
else:
|
||||
_echo(f"Enforcing: {state.current_game_name} (AppID={state.current_app_id})")
|
||||
_enforce_setup(config, state)
|
||||
|
||||
@ -14,6 +14,7 @@ import sys
|
||||
import time
|
||||
|
||||
from steam_backlog_enforcer._whitelist import get_approved_exception_ids
|
||||
from steam_backlog_enforcer.library_hider import steam_is_installed
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -138,7 +139,15 @@ def is_game_installed(app_id: int) -> bool:
|
||||
|
||||
|
||||
def _ensure_steam_running() -> None:
|
||||
"""Start the Steam client if it is not already running."""
|
||||
"""Start the Steam client if it is not already running.
|
||||
|
||||
Does nothing if Steam is not installed - there is no client to start, and
|
||||
trying anyway only sleeps 15s waiting on a process that died on exec.
|
||||
"""
|
||||
if not steam_is_installed():
|
||||
logger.info("Steam is not installed — skipping client start.")
|
||||
return
|
||||
|
||||
# Check if any steam process is running (main client, not just helpers).
|
||||
try:
|
||||
result = subprocess.run(
|
||||
|
||||
@ -20,6 +20,7 @@ import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
import pwd
|
||||
import shutil
|
||||
import subprocess
|
||||
@ -34,6 +35,33 @@ _CDP_PORT = 8080
|
||||
_CDP_TIMEOUT = 120
|
||||
_STEAM_STARTUP_WAIT = 45
|
||||
|
||||
# Real Steam client binary, as shipped by the distro's `steam` package.
|
||||
#
|
||||
# Deliberately NOT probed with shutil.which("steam"): a launcher wrapper on
|
||||
# $PATH (e.g. /usr/local/bin/steam adding -cef-* flags) keeps `which` truthy
|
||||
# long after the package itself is gone - as happens when a total block
|
||||
# uninstalls Steam. Checking the real binary is what actually answers
|
||||
# "can we launch Steam at all?".
|
||||
_STEAM_BINARY = "/usr/bin/steam"
|
||||
|
||||
# Handles for fire-and-forget launches, kept only so they can be reaped.
|
||||
_SPAWNED: list[subprocess.Popen[bytes]] = []
|
||||
|
||||
|
||||
class SteamUnavailableError(RuntimeError):
|
||||
"""Raised when Steam cannot be driven over CDP.
|
||||
|
||||
Covers both "Steam is not installed" and "Steam is installed but never
|
||||
opened its debug port". Callers are expected to degrade gracefully rather
|
||||
than abort: an unreachable Steam means there is no library to hide, which
|
||||
is not a fatal condition for the enforcer.
|
||||
"""
|
||||
|
||||
|
||||
def steam_is_installed() -> bool:
|
||||
"""Return True if the real Steam client binary is present."""
|
||||
return Path(_STEAM_BINARY).exists()
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────
|
||||
# CDP (Chrome DevTools Protocol) helpers
|
||||
@ -194,11 +222,22 @@ def ensure_steam_debug_port() -> None:
|
||||
|
||||
If Steam is running without the port, it is restarted.
|
||||
If Steam is not running, it is launched.
|
||||
|
||||
Raises:
|
||||
SteamUnavailableError: If Steam is not installed, or is installed but
|
||||
never opens its debug port.
|
||||
"""
|
||||
if _steam_has_debug_port():
|
||||
logger.debug("Steam CDP port already available.")
|
||||
return
|
||||
|
||||
# Bail out before the ~45s launch-and-wait: with no binary to exec there
|
||||
# is nothing to wait for, and retrying every pass only burns time and
|
||||
# leaves dead processes behind.
|
||||
if not steam_is_installed():
|
||||
msg = f"Steam is not installed ({_STEAM_BINARY} does not exist)"
|
||||
raise SteamUnavailableError(msg)
|
||||
|
||||
logger.info("Steam CDP port not available — (re)starting Steam...")
|
||||
if _is_steam_running():
|
||||
_shutdown_steam()
|
||||
@ -207,12 +246,12 @@ def ensure_steam_debug_port() -> None:
|
||||
|
||||
if not _wait_for_cdp_ready():
|
||||
msg = "Timed out waiting for Steam CDP port to become ready"
|
||||
raise RuntimeError(msg)
|
||||
raise SteamUnavailableError(msg)
|
||||
logger.info("Steam CDP port ready.")
|
||||
|
||||
if not _wait_for_collections_ready():
|
||||
msg = "Timed out waiting for Steam collections to initialise"
|
||||
raise RuntimeError(msg)
|
||||
raise SteamUnavailableError(msg)
|
||||
logger.info("Steam collection store ready.")
|
||||
|
||||
|
||||
@ -355,8 +394,21 @@ def restart_steam() -> None:
|
||||
logger.info("Steam restarted with CDP port ready.")
|
||||
|
||||
|
||||
def _reap_spawned() -> None:
|
||||
"""Clear out previously launched processes that have since exited.
|
||||
|
||||
Launches here are fire-and-forget: Steam is meant to outlive the call, so
|
||||
it is never waited on. That leaves any launch which dies immediately - a
|
||||
missing binary, a broken wrapper - sitting as a zombie that still carries
|
||||
the name ``steam``, which anything scanning /proc reads as "Steam is
|
||||
running". Polling the old handles reaps them and retires the name.
|
||||
"""
|
||||
_SPAWNED[:] = [proc for proc in _SPAWNED if proc.poll() is None]
|
||||
|
||||
|
||||
def _run_as_user(cmd: list[str], user: str | None) -> None:
|
||||
"""Run a command, dropping to *user* if currently root."""
|
||||
_reap_spawned()
|
||||
if os.geteuid() == 0 and user and user != "root":
|
||||
try:
|
||||
pw = pwd.getpwnam(user)
|
||||
@ -380,8 +432,10 @@ def _run_as_user(cmd: list[str], user: str | None) -> None:
|
||||
else:
|
||||
full_cmd = cmd
|
||||
|
||||
subprocess.Popen(
|
||||
full_cmd,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
_SPAWNED.append(
|
||||
subprocess.Popen(
|
||||
full_cmd,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
)
|
||||
|
||||
@ -16,6 +16,7 @@ from steam_backlog_enforcer._enforce_loop import (
|
||||
get_all_owned_app_ids,
|
||||
)
|
||||
from steam_backlog_enforcer.config import Config, State
|
||||
from steam_backlog_enforcer.library_hider import SteamUnavailableError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
@ -371,3 +372,23 @@ class TestEnforceHideGames:
|
||||
):
|
||||
_enforce_hide_games(Config(), state)
|
||||
assert any("skipped" in str(c) for c in mock_echo.call_args_list)
|
||||
|
||||
def test_unreachable_steam_is_not_fatal(self) -> None:
|
||||
"""An unreachable Steam must degrade, never propagate.
|
||||
|
||||
Regression guard: this exception used to escape all the way out of
|
||||
do_enforce, exiting the process into Restart=always - which spun the
|
||||
service through ~1000 restarts against an uninstalled Steam.
|
||||
"""
|
||||
state = State(current_app_id=1)
|
||||
with (
|
||||
patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]),
|
||||
patch(
|
||||
f"{PKG}.hide_other_games",
|
||||
side_effect=SteamUnavailableError("Steam is not installed"),
|
||||
),
|
||||
patch(f"{PKG}._echo") as mock_echo,
|
||||
):
|
||||
_enforce_hide_games(Config(), state)
|
||||
|
||||
assert any("skipped" in str(c) for c in mock_echo.call_args_list)
|
||||
|
||||
@ -2,20 +2,37 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from steam_backlog_enforcer._enforce_loop import (
|
||||
_enforce_loop_iteration,
|
||||
do_enforce,
|
||||
)
|
||||
from steam_backlog_enforcer.config import Config, State
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
PKG = "steam_backlog_enforcer._enforce_loop"
|
||||
|
||||
|
||||
class TestEnforceLoopIteration:
|
||||
"""Tests for _enforce_loop_iteration."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _steam_present(self) -> Iterator[None]:
|
||||
"""Pretend Steam is installed for every test in this class.
|
||||
|
||||
The iteration short-circuits when it is not, which on a test machine
|
||||
without Steam would silently make every assertion below vacuous. The
|
||||
absent case is covered explicitly by test_skips_when_steam_absent.
|
||||
"""
|
||||
with patch(f"{PKG}.steam_is_installed", return_value=True):
|
||||
yield
|
||||
|
||||
def test_kills_unauthorized(self) -> None:
|
||||
config = Config(
|
||||
kill_unauthorized_games=True,
|
||||
@ -71,6 +88,31 @@ class TestEnforceLoopIteration:
|
||||
):
|
||||
_enforce_loop_iteration(config, state)
|
||||
|
||||
def test_skips_when_steam_absent(self) -> None:
|
||||
"""With Steam uninstalled the iteration must do nothing.
|
||||
|
||||
Regression guard: this path used to try to write an appmanifest into
|
||||
a steamapps directory a total block had deleted, erroring every 3s.
|
||||
"""
|
||||
config = Config(
|
||||
kill_unauthorized_games=True,
|
||||
uninstall_other_games=True,
|
||||
)
|
||||
state = State(current_app_id=1, current_game_name="G")
|
||||
with (
|
||||
patch(f"{PKG}.steam_is_installed", return_value=False),
|
||||
patch(f"{PKG}.enforce_allowed_game") as mock_enforce,
|
||||
patch(f"{PKG}._guard_installed_games") as mock_guard,
|
||||
patch(f"{PKG}.is_game_installed") as mock_installed,
|
||||
patch(f"{PKG}.install_game") as mock_install,
|
||||
):
|
||||
_enforce_loop_iteration(config, state)
|
||||
|
||||
mock_enforce.assert_not_called()
|
||||
mock_guard.assert_not_called()
|
||||
mock_installed.assert_not_called()
|
||||
mock_install.assert_not_called()
|
||||
|
||||
def test_reinstalls_missing(self) -> None:
|
||||
config = Config(
|
||||
kill_unauthorized_games=False,
|
||||
@ -104,11 +146,48 @@ class TestEnforceLoopIteration:
|
||||
class TestDoEnforce:
|
||||
"""Tests for do_enforce."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _steam_present(self) -> Iterator[None]:
|
||||
"""Pretend Steam is installed for every test in this class.
|
||||
|
||||
Without it do_enforce takes the "not installed" branch and never
|
||||
reaches the normal enforcement path these tests are about. The
|
||||
absent case is covered by test_steam_absent_idles_without_setup,
|
||||
which opts back out.
|
||||
"""
|
||||
with patch(f"{PKG}.steam_is_installed", return_value=True):
|
||||
yield
|
||||
|
||||
def test_no_game(self) -> None:
|
||||
with patch(f"{PKG}._echo") as mock_echo:
|
||||
do_enforce(Config(), State())
|
||||
assert any("No game" in str(c) for c in mock_echo.call_args_list)
|
||||
|
||||
def test_steam_absent_idles_without_setup(self) -> None:
|
||||
"""With Steam gone, say so and keep looping - never exit.
|
||||
|
||||
Returning here would end the process, and under Restart=always that
|
||||
is the crash loop again by another name. Staying alive also lets a
|
||||
later reinstall be picked up without a restart.
|
||||
"""
|
||||
state = State(current_app_id=1, current_game_name="G")
|
||||
with (
|
||||
patch(f"{PKG}.is_total_block_active", return_value=False),
|
||||
patch(f"{PKG}.steam_is_installed", return_value=False),
|
||||
patch(f"{PKG}._enforce_setup") as mock_setup,
|
||||
patch(f"{PKG}._echo") as mock_echo,
|
||||
patch.object(State, "load", return_value=state),
|
||||
patch(
|
||||
f"{PKG}._enforce_loop_iteration",
|
||||
side_effect=KeyboardInterrupt,
|
||||
),
|
||||
patch(f"{PKG}.time.sleep"),
|
||||
):
|
||||
do_enforce(Config(), state)
|
||||
|
||||
mock_setup.assert_not_called()
|
||||
assert any("not installed" in str(c) for c in mock_echo.call_args_list)
|
||||
|
||||
def test_keyboard_interrupt(self) -> None:
|
||||
state = State(current_app_id=1, current_game_name="G")
|
||||
config = Config()
|
||||
|
||||
@ -19,6 +19,7 @@ from steam_backlog_enforcer.game_install import (
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@ -183,6 +184,33 @@ class TestIsGameInstalled:
|
||||
class TestEnsureSteamRunning:
|
||||
"""Tests for _ensure_steam_running."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _steam_present(self) -> Iterator[None]:
|
||||
"""Pretend Steam is installed for every test in this class.
|
||||
|
||||
Without it the function returns before any launch, which on a test
|
||||
machine without Steam would make the launch assertions vacuous. The
|
||||
absent case is covered by test_skips_when_steam_absent.
|
||||
"""
|
||||
with patch(f"{PKG}.steam_is_installed", return_value=True):
|
||||
yield
|
||||
|
||||
def test_skips_when_steam_absent(self) -> None:
|
||||
"""With Steam uninstalled, do not try to launch a missing client.
|
||||
|
||||
Regression guard: this path used to exec a dead launcher wrapper and
|
||||
then sleep 15s, leaving a zombie named "steam" behind each time.
|
||||
"""
|
||||
with (
|
||||
patch(f"{PKG}.steam_is_installed", return_value=False),
|
||||
patch(f"{PKG}.subprocess.Popen") as mock_popen,
|
||||
patch(f"{PKG}.time.sleep") as mock_sleep,
|
||||
):
|
||||
_ensure_steam_running()
|
||||
|
||||
mock_popen.assert_not_called()
|
||||
mock_sleep.assert_not_called()
|
||||
|
||||
def test_already_running(self) -> None:
|
||||
mock_result = MagicMock(returncode=0)
|
||||
with patch(
|
||||
|
||||
@ -4,11 +4,13 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from steam_backlog_enforcer.library_hider import (
|
||||
SteamUnavailableError,
|
||||
_cdp_result_value,
|
||||
_evaluate_js,
|
||||
_evaluate_js_async,
|
||||
@ -22,6 +24,9 @@ from steam_backlog_enforcer.library_hider import (
|
||||
ensure_steam_debug_port,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
|
||||
class TestGetSharedJsWsUrl:
|
||||
"""Tests for _get_shared_js_ws_url."""
|
||||
@ -318,6 +323,44 @@ class TestLaunchSteamWithDebug:
|
||||
class TestEnsureSteamDebugPort:
|
||||
"""Tests for ensure_steam_debug_port."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _steam_present(self) -> Iterator[None]:
|
||||
"""Pretend Steam is installed for every test in this class.
|
||||
|
||||
Without it the function raises before reaching any launch logic,
|
||||
which on a test machine without Steam would make the cases below
|
||||
vacuous. The absent case has its own test.
|
||||
"""
|
||||
with patch(
|
||||
"steam_backlog_enforcer.library_hider.steam_is_installed",
|
||||
return_value=True,
|
||||
):
|
||||
yield
|
||||
|
||||
def test_raises_when_steam_absent(self) -> None:
|
||||
"""An uninstalled Steam must fail fast, before any launch attempt.
|
||||
|
||||
Regression guard: without this the caller spent ~45s waiting on a
|
||||
CDP port that no process was ever going to open.
|
||||
"""
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer.library_hider._steam_has_debug_port",
|
||||
return_value=False,
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer.library_hider.steam_is_installed",
|
||||
return_value=False,
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer.library_hider._launch_steam_with_debug",
|
||||
) as mock_launch,
|
||||
pytest.raises(SteamUnavailableError, match="not installed"),
|
||||
):
|
||||
ensure_steam_debug_port()
|
||||
|
||||
mock_launch.assert_not_called()
|
||||
|
||||
def test_already_available(self) -> None:
|
||||
with patch(
|
||||
"steam_backlog_enforcer.library_hider._steam_has_debug_port",
|
||||
|
||||
@ -7,18 +7,89 @@ import tempfile
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from steam_backlog_enforcer.library_hider import (
|
||||
_SPAWNED,
|
||||
_reap_spawned,
|
||||
_run_as_user,
|
||||
hide_other_games,
|
||||
restart_steam,
|
||||
steam_is_installed,
|
||||
unhide_all_games,
|
||||
)
|
||||
|
||||
PKG = "steam_backlog_enforcer.library_hider"
|
||||
|
||||
|
||||
class TestSteamIsInstalled:
|
||||
"""Tests for steam_is_installed."""
|
||||
|
||||
def test_true_when_binary_exists(self) -> None:
|
||||
with patch(f"{PKG}.Path") as mock_path:
|
||||
mock_path.return_value.exists.return_value = True
|
||||
assert steam_is_installed() is True
|
||||
|
||||
def test_false_when_binary_missing(self) -> None:
|
||||
with patch(f"{PKG}.Path") as mock_path:
|
||||
mock_path.return_value.exists.return_value = False
|
||||
assert steam_is_installed() is False
|
||||
|
||||
def test_checks_real_binary_not_path_lookup(self) -> None:
|
||||
"""Must probe the real binary, never a $PATH lookup.
|
||||
|
||||
A launcher wrapper on $PATH keeps `which steam` truthy long after the
|
||||
package is uninstalled - which is exactly how a dead Steam went on
|
||||
looking installed and got launched ~1000 times.
|
||||
"""
|
||||
with patch(f"{PKG}.Path") as mock_path:
|
||||
mock_path.return_value.exists.return_value = False
|
||||
steam_is_installed()
|
||||
mock_path.assert_called_once_with("/usr/bin/steam")
|
||||
|
||||
|
||||
class TestReapSpawned:
|
||||
"""Tests for _reap_spawned."""
|
||||
|
||||
def test_drops_exited_processes(self) -> None:
|
||||
"""An exited launch must be reaped so its name stops showing in /proc.
|
||||
|
||||
This is the zombie that focus-mode read as a live Steam.
|
||||
"""
|
||||
dead = MagicMock()
|
||||
dead.poll.return_value = 1
|
||||
_SPAWNED[:] = [dead]
|
||||
try:
|
||||
_reap_spawned()
|
||||
assert _SPAWNED == []
|
||||
finally:
|
||||
_SPAWNED.clear()
|
||||
|
||||
def test_keeps_running_processes(self) -> None:
|
||||
"""A Steam that is still alive must not be dropped from tracking."""
|
||||
alive = MagicMock()
|
||||
alive.poll.return_value = None
|
||||
_SPAWNED[:] = [alive]
|
||||
try:
|
||||
_reap_spawned()
|
||||
assert [alive] == _SPAWNED
|
||||
finally:
|
||||
_SPAWNED.clear()
|
||||
|
||||
|
||||
class TestRunAsUser:
|
||||
"""Tests for _run_as_user."""
|
||||
|
||||
def test_tracks_spawned_process_for_reaping(self) -> None:
|
||||
"""Every launch must be tracked, or it can never be reaped."""
|
||||
_SPAWNED.clear()
|
||||
with (
|
||||
patch(f"{PKG}.os.geteuid", return_value=1000),
|
||||
patch(f"{PKG}.subprocess.Popen") as mock_popen,
|
||||
):
|
||||
_run_as_user(["steam"], "alice")
|
||||
try:
|
||||
assert [mock_popen.return_value] == _SPAWNED
|
||||
finally:
|
||||
_SPAWNED.clear()
|
||||
|
||||
def test_non_root_runs_directly(self) -> None:
|
||||
with (
|
||||
patch(f"{PKG}.os.geteuid", return_value=1000),
|
||||
|
||||
Loading…
Reference in New Issue
Block a user