feat: block-gaming purges Steam/Proton filesystem remnants + helper packages
Some checks failed
pre-commit / pre-commit (push) Has been cancelled
Tests / test (3.10) (push) Has been cancelled
Tests / test (3.11) (push) Has been cancelled
Tests / test (3.12) (push) Has been cancelled

Total block now also deletes a curated allowlist of Steam/Proton paths
(~/.steam, ~/steam, ~/.local/share/Steam, ~/.steampath, ~/.steampid,
~/.config/steamtinkerlaunch, ~/.config/CSDSteamBuild) and uninstalls known
Proton helper packages (protondb-tags-git, protonhax-git, protontricks-git,
protonup-ng-git, protonup-qt), logging every removal to a purge log.

A curated allowlist rather than a recursive "anything named steam" sweep,
since the latter would also catch unrelated files (AUR build checkouts,
archives, other apps' own save data) that merely share the substring.

Live-verified against the real filesystem: all 7 paths and 5 packages were
actually removed, steam-backlog-enforcer's own directory/config untouched.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YFdVut4N4kMbJTUEEmfuQU
This commit is contained in:
Krzysztof Rudnicki 2026-07-04 17:25:08 +02:00
parent 7554b58ab7
commit 95bd0b93bd
3 changed files with 457 additions and 5 deletions

View File

@ -1,7 +1,8 @@
"""Total gaming block: no in-app command to lift it early.
Uninstalls Steam, kills all game/launcher processes, and blocks all
Steam + game-website domains for a fixed number of days.
Uninstalls Steam, kills all game/launcher processes, blocks all Steam +
game-website domains, purges known Steam/Proton filesystem remnants, and
uninstalls Proton helper packages, for a fixed number of days.
Tamper-resistance is provided by guard-lib (~/guard-lib): the lock file's
``until`` timestamp is protected by a bind-mounted, chattr-immutable
@ -153,6 +154,37 @@ IPTABLES_CHAIN = "STEAM_TOTAL_BLOCK"
# mistake it for a socket bind-all-interfaces address (it never is one).
_NULL_ROUTE_IP = ".".join(["0"] * 4)
_STEAM_PURGE_LOG_FILE = CONFIG_DIR / "total_block_purge_log.json"
# Fixed allowlist of known Steam/Proton filesystem remnants to delete - NOT
# a recursive "anything whose name contains 'steam'" sweep, which would also
# catch unrelated files (AUR build checkouts, archives, other apps' own save
# data) that merely share the substring. Confirmed present on the reference
# machine: ~/.steam (symlink farm), ~/steam (secondary/portable install),
# ~/.local/share/Steam (the real install - steamapps, userdata, screenshots,
# and compatibilitytools.d's GE-Proton builds all live under here).
_STEAM_REMNANT_PATHS: tuple[Path, ...] = (
Path.home() / ".steam",
Path.home() / "steam",
Path.home() / ".local" / "share" / "Steam",
Path.home() / ".steampath",
Path.home() / ".steampid",
Path.home() / ".config" / "steamtinkerlaunch",
Path.home() / ".config" / "CSDSteamBuild",
)
# Proton-management helper packages (not Steam itself, not launched by
# STEAM_CLIENT_PROCESS_NAMES) - installed via pacman/AUR, so pacman -R
# is sufficient; no separate process-kill needed as these are short-lived
# CLI/GUI tools, not background daemons.
_PROTON_HELPER_PACKAGES: tuple[str, ...] = (
"protondb-tags-git",
"protonhax-git",
"protontricks-git",
"protonup-ng-git",
"protonup-qt",
)
@dataclass
class TotalBlockStatus:
@ -334,10 +366,10 @@ def _kill_steam_and_launchers() -> list[tuple[int, str]]:
# ──────────────────────────────────────────────────────────────
def _is_steam_installed() -> bool:
"""Return True if the ``steam`` pacman package is currently installed."""
def _is_package_installed(package: str) -> bool:
"""Return True if *package* is currently installed via pacman."""
result = subprocess.run(
[_PACMAN, "-Qi", _STEAM_PACKAGE],
[_PACMAN, "-Qi", package],
capture_output=True,
timeout=10,
check=False,
@ -345,6 +377,11 @@ def _is_steam_installed() -> bool:
return result.returncode == 0
def _is_steam_installed() -> bool:
"""Return True if the ``steam`` pacman package is currently installed."""
return _is_package_installed(_STEAM_PACKAGE)
def _uninstall_steam_package() -> bool:
"""Remove the ``steam`` pacman package.
@ -353,6 +390,90 @@ def _uninstall_steam_package() -> bool:
return _uninstall_package(_STEAM_PACKAGE)
def _uninstall_proton_helpers() -> list[str]:
"""Uninstall known Proton-management helper packages that are present.
Returns the subset of :data:`_PROTON_HELPER_PACKAGES` that were actually
installed and successfully removed (for logging), not the full fixed
list.
"""
removed: list[str] = []
for package in _PROTON_HELPER_PACKAGES:
if not _is_package_installed(package):
continue
if _uninstall_package(package):
removed.append(package)
else:
logger.warning("Total block: failed to uninstall proton helper %s", package)
return removed
def _remove_steam_remnants() -> list[str]:
"""Delete the curated Steam/Proton filesystem remnants that exist.
Symlinks are unlinked directly rather than following them into
:func:`shutil.rmtree`, since e.g. ``~/.steampath`` -> a file inside
``~/.steam``, which this same pass may already have removed.
"""
removed: list[str] = []
for path in _STEAM_REMNANT_PATHS:
if not (path.is_symlink() or path.exists()):
continue
try:
if path.is_symlink() or path.is_file():
path.unlink()
else:
shutil.rmtree(path)
except OSError:
logger.exception("Failed to remove steam remnant %s", path)
continue
removed.append(str(path))
return removed
def _log_steam_purge(removed_paths: list[str], removed_packages: list[str]) -> None:
"""Append a timestamped record of a Steam/Proton purge to disk.
A no-op when nothing was actually removed, so the log only grows on
ticks that did real work (this runs every enforce tick while the block
is active - see :func:`enforce_total_block_tick`).
"""
if not removed_paths and not removed_packages:
return
try:
existing = json.loads(_STEAM_PURGE_LOG_FILE.read_text(encoding="utf-8"))
if not isinstance(existing, list):
existing = []
except (OSError, json.JSONDecodeError, ValueError):
existing = []
existing.append(
{
"timestamp": datetime.now(timezone.utc).isoformat(),
"removed_paths": removed_paths,
"removed_packages": removed_packages,
}
)
_atomic_write(_STEAM_PURGE_LOG_FILE, json.dumps(existing, indent=2) + "\n")
def _purge_steam_and_proton() -> None:
"""Remove curated Steam/Proton filesystem remnants and helper packages.
Best-effort and re-run every enforce tick, same as the Steam-package
reappearance check: a stat() per fixed path and a ``pacman -Qi`` per
helper package are cheap even when there is nothing left to do.
"""
removed_paths = _remove_steam_remnants()
removed_packages = _uninstall_proton_helpers()
if removed_paths or removed_packages:
logger.info(
"Total block: purged steam path(s) %s, proton package(s) %s",
removed_paths,
removed_packages,
)
_log_steam_purge(removed_paths, removed_packages)
# ──────────────────────────────────────────────────────────────
# Domain blocking (hosts + iptables) - separate from store_blocker's own
# BLOCKED_DOMAINS/STEAM_ENFORCER state, so ending the total block never
@ -631,6 +752,8 @@ def start_total_block(days: int) -> bool:
if not _uninstall_steam_package():
logger.warning("Total block: failed to uninstall steam (will retry each tick)")
_purge_steam_and_proton()
# iptables MUST be applied before hosts: it resolves real upstream IPs,
# and once the hosts block is written, local resolution for these same
# domains collapses to 0.0.0.0 (see _apply_total_block_iptables).
@ -655,6 +778,8 @@ def enforce_total_block_tick() -> None:
logger.warning("Steam reappeared during total block - removing again")
_uninstall_steam_package()
_purge_steam_and_proton()
_apply_total_block_iptables()
_apply_total_block_hosts()

View File

@ -9,6 +9,8 @@ to temporary directories. This stops tests from accidentally:
- Modifying /etc/hosts via the store blocker
- Corrupting the HLTB cache on disk
- Launching real Steam or calling real subprocess commands
- Deleting real ~/.steam, ~/.local/share/Steam, etc. via the total-block
Steam/Proton remnant purge
"""
from __future__ import annotations
@ -91,6 +93,25 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]:
"steam_backlog_enforcer._total_block._IPTABLES_IP_CACHE_FILE",
fake_config / "total_block_ip_cache.json",
),
patch(
"steam_backlog_enforcer._total_block._STEAM_PURGE_LOG_FILE",
fake_config / "total_block_purge_log.json",
),
# Steam/Proton remnant paths (real ~/.steam, ~/.local/share/Steam,
# etc. - _remove_steam_remnants() deletes these, so tests must never
# see the real ones)
patch(
"steam_backlog_enforcer._total_block._STEAM_REMNANT_PATHS",
(
tmp_path / "fake_home" / ".steam",
tmp_path / "fake_home" / "steam",
tmp_path / "fake_home" / ".local" / "share" / "Steam",
tmp_path / "fake_home" / ".steampath",
tmp_path / "fake_home" / ".steampid",
tmp_path / "fake_home" / ".config" / "steamtinkerlaunch",
tmp_path / "fake_home" / ".config" / "CSDSteamBuild",
),
),
# Whitelist exception files (_whitelist module-level constants)
patch(
"steam_backlog_enforcer._whitelist.PENDING_EXCEPTIONS_FILE",

View File

@ -26,15 +26,20 @@ from steam_backlog_enforcer._total_block import (
_apply_total_block_hosts,
_apply_total_block_iptables,
_iptables_chain_intact,
_is_package_installed,
_is_steam_installed,
_kill_and_uninstall_launchers,
_kill_steam_and_launchers,
_load_cached_ips,
_log_steam_purge,
_pacman_owner,
_purge_steam_and_proton,
_remove_steam_remnants,
_remove_total_block_hosts,
_remove_total_block_iptables,
_save_cached_ips,
_uninstall_package,
_uninstall_proton_helpers,
_uninstall_steam_package,
end_total_block_cleanup,
enforce_total_block_tick,
@ -53,28 +58,46 @@ PKG = "steam_backlog_enforcer._total_block"
@dataclass
class _Paths:
"""Tests for paths."""
lock_file: Path
ip_cache_file: Path
hosts_file: Path
purge_log_file: Path
remnant_paths: tuple[Path, ...]
@pytest.fixture(autouse=True)
def paths(tmp_path: Path) -> Iterator[_Paths]:
"""Redirect the module's path constants to tmp_path for every test."""
remnant_paths = (
tmp_path / "home" / ".steam",
tmp_path / "home" / "steam",
tmp_path / "home" / ".local" / "share" / "Steam",
tmp_path / "home" / ".steampath",
tmp_path / "home" / ".steampid",
tmp_path / "home" / ".config" / "steamtinkerlaunch",
tmp_path / "home" / ".config" / "CSDSteamBuild",
)
paths = _Paths(
lock_file=tmp_path / "total_block_lock.json",
ip_cache_file=tmp_path / "total_block_ip_cache.json",
hosts_file=tmp_path / "hosts",
purge_log_file=tmp_path / "total_block_purge_log.json",
remnant_paths=remnant_paths,
)
with (
patch(f"{PKG}.TOTAL_BLOCK_LOCK_FILE", paths.lock_file),
patch(f"{PKG}._IPTABLES_IP_CACHE_FILE", paths.ip_cache_file),
patch(f"{PKG}.HOSTS_FILE", paths.hosts_file),
patch(f"{PKG}._STEAM_PURGE_LOG_FILE", paths.purge_log_file),
patch(f"{PKG}._STEAM_REMNANT_PATHS", paths.remnant_paths),
):
yield paths
def _write_lock(paths: _Paths, started_at: float, until: float, days: int = 1) -> None:
"""Test that write lock."""
paths.lock_file.parent.mkdir(parents=True, exist_ok=True)
paths.lock_file.write_text(
json.dumps({"started_at": started_at, "until": until, "days": days}),
@ -91,33 +114,42 @@ _NOW = datetime.now(timezone.utc).timestamp()
class TestIsTotalBlockActive:
"""Tests for is total block active."""
def test_no_lock_file(self) -> None:
"""Test that no lock file."""
assert is_total_block_active() is False
def test_active_lock(self, paths: _Paths) -> None:
"""Test that active lock."""
_write_lock(paths, _NOW, _NOW + 3600)
assert is_total_block_active() is True
def test_expired_lock(self, paths: _Paths) -> None:
"""Test that expired lock."""
_write_lock(paths, _NOW - 3600, _NOW - 1)
assert is_total_block_active() is False
def test_malformed_json(self, paths: _Paths) -> None:
"""Test that malformed json."""
paths.lock_file.parent.mkdir(parents=True, exist_ok=True)
paths.lock_file.write_text("not json", encoding="utf-8")
assert is_total_block_active() is False
def test_non_dict_json(self, paths: _Paths) -> None:
"""Test that non dict json."""
paths.lock_file.parent.mkdir(parents=True, exist_ok=True)
paths.lock_file.write_text("[1, 2, 3]", encoding="utf-8")
assert is_total_block_active() is False
def test_missing_until_key(self, paths: _Paths) -> None:
"""Test that missing until key."""
paths.lock_file.parent.mkdir(parents=True, exist_ok=True)
paths.lock_file.write_text(json.dumps({"days": 1}), encoding="utf-8")
assert is_total_block_active() is False
def test_non_numeric_until(self, paths: _Paths) -> None:
"""Test that non numeric until."""
paths.lock_file.parent.mkdir(parents=True, exist_ok=True)
paths.lock_file.write_text(
json.dumps({"until": "not-a-number"}), encoding="utf-8"
@ -126,20 +158,28 @@ class TestIsTotalBlockActive:
class TestTotalBlockNeedsCleanup:
"""Tests for total block needs cleanup."""
def test_no_lock_file(self) -> None:
"""Test that no lock file."""
assert total_block_needs_cleanup() is False
def test_active_lock_no_cleanup_needed(self, paths: _Paths) -> None:
"""Test that active lock no cleanup needed."""
_write_lock(paths, _NOW, _NOW + 3600)
assert total_block_needs_cleanup() is False
def test_expired_lock_needs_cleanup(self, paths: _Paths) -> None:
"""Test that expired lock needs cleanup."""
_write_lock(paths, _NOW - 3600, _NOW - 1)
assert total_block_needs_cleanup() is True
class TestGetTotalBlockStatus:
"""Tests for get total block status."""
def test_no_lock(self) -> None:
"""Test that no lock."""
status = get_total_block_status()
assert status.active is False
assert status.started_at is None
@ -148,6 +188,7 @@ class TestGetTotalBlockStatus:
assert status.days_remaining == 0.0
def test_active_lock(self, paths: _Paths) -> None:
"""Test that active lock."""
_write_lock(paths, _NOW, _NOW + 86400, days=1)
status = get_total_block_status()
assert status.active is True
@ -157,18 +198,21 @@ class TestGetTotalBlockStatus:
assert status.until is not None
def test_expired_lock(self, paths: _Paths) -> None:
"""Test that expired lock."""
_write_lock(paths, _NOW - 7200, _NOW - 3600, days=1)
status = get_total_block_status()
assert status.active is False
assert status.days_remaining == 0.0
def test_malformed_json_returns_inactive(self, paths: _Paths) -> None:
"""Test that malformed json returns inactive."""
paths.lock_file.parent.mkdir(parents=True, exist_ok=True)
paths.lock_file.write_text("garbage", encoding="utf-8")
status = get_total_block_status()
assert status.active is False
def test_non_int_days_defaults_to_zero(self, paths: _Paths) -> None:
"""Test that non int days defaults to zero."""
paths.lock_file.parent.mkdir(parents=True, exist_ok=True)
paths.lock_file.write_text(
json.dumps({"started_at": _NOW, "until": _NOW + 3600, "days": "one"}),
@ -184,7 +228,10 @@ class TestGetTotalBlockStatus:
class TestKillSteamAndLaunchers:
"""Tests for kill steam and launchers."""
def test_combines_steam_and_launcher_kills(self) -> None:
"""Test that combines steam and launcher kills."""
with (
patch(f"{PKG}.kill_processes_by_name", return_value=[(1, "steam")]),
patch(
@ -198,7 +245,10 @@ class TestKillSteamAndLaunchers:
class TestPacmanOwner:
"""Tests for pacman owner."""
def test_owned_path_returns_package_name(self) -> None:
"""Test that owned path returns package name."""
result = MagicMock(
returncode=0,
stdout="/usr/bin/prismlauncher is owned by prismlauncher-git 11.0.0-1\n",
@ -207,24 +257,30 @@ class TestPacmanOwner:
assert _pacman_owner("/usr/bin/prismlauncher") == "prismlauncher-git"
def test_unowned_path_returns_none(self) -> None:
"""Test that unowned path returns none."""
result = MagicMock(returncode=1, stdout="")
with patch(f"{PKG}.subprocess.run", return_value=result):
assert _pacman_owner("/opt/foo/bar") is None
def test_unexpected_output_format_returns_none(self) -> None:
"""Test that unexpected output format returns none."""
result = MagicMock(returncode=0, stdout="something unexpected\n")
with patch(f"{PKG}.subprocess.run", return_value=result):
assert _pacman_owner("/usr/bin/x") is None
class TestUninstallPackage:
"""Tests for uninstall package."""
def test_success(self) -> None:
"""Test that success."""
with patch(
f"{PKG}.subprocess.run", return_value=MagicMock(returncode=0, stderr="")
):
assert _uninstall_package("foo") is True
def test_already_absent_treated_as_success(self) -> None:
"""Test that already absent treated as success."""
with patch(
f"{PKG}.subprocess.run",
return_value=MagicMock(returncode=1, stderr="error: target not found: foo"),
@ -232,6 +288,7 @@ class TestUninstallPackage:
assert _uninstall_package("foo") is True
def test_real_failure_returns_false(self) -> None:
"""Test that real failure returns false."""
with patch(
f"{PKG}.subprocess.run",
return_value=MagicMock(returncode=1, stderr="some other error"),
@ -239,12 +296,16 @@ class TestUninstallPackage:
assert _uninstall_package("foo") is False
def test_subprocess_error_returns_false(self) -> None:
"""Test that subprocess error returns false."""
with patch(f"{PKG}.subprocess.run", side_effect=OSError):
assert _uninstall_package("foo") is False
class TestKillAndUninstallLaunchers:
"""Tests for kill and uninstall launchers."""
def test_no_launchers_running(self) -> None:
"""Test that no launchers running."""
with (
patch(f"{PKG}.get_pids_by_process_names", return_value={}),
patch(f"{PKG}.kill_processes_by_name", return_value=[]),
@ -252,6 +313,7 @@ class TestKillAndUninstallLaunchers:
assert _kill_and_uninstall_launchers() == []
def test_kills_and_uninstalls_owned_package(self) -> None:
"""Test that kills and uninstalls owned package."""
with (
patch(
f"{PKG}.get_pids_by_process_names",
@ -271,6 +333,7 @@ class TestKillAndUninstallLaunchers:
mock_uninstall.assert_called_once_with("prismlauncher-git")
def test_exe_path_unreadable_skips_uninstall(self) -> None:
"""Test that exe path unreadable skips uninstall."""
with (
patch(
f"{PKG}.get_pids_by_process_names",
@ -291,6 +354,7 @@ class TestKillAndUninstallLaunchers:
mock_uninstall.assert_not_called()
def test_unowned_package_not_uninstalled(self) -> None:
"""Test that unowned package not uninstalled."""
with (
patch(f"{PKG}.get_pids_by_process_names", return_value={123: "custom"}),
patch(f"{PKG}.Path") as mock_path_cls,
@ -303,6 +367,7 @@ class TestKillAndUninstallLaunchers:
mock_uninstall.assert_not_called()
def test_uninstall_failure_is_logged_not_raised(self) -> None:
"""Test that uninstall failure is logged not raised."""
with (
patch(
f"{PKG}.get_pids_by_process_names",
@ -326,23 +391,31 @@ class TestKillAndUninstallLaunchers:
class TestIsSteamInstalled:
"""Tests for is steam installed."""
def test_installed(self) -> None:
"""Test that installed."""
with patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=0)):
assert _is_steam_installed() is True
def test_not_installed(self) -> None:
"""Test that not installed."""
with patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=1)):
assert _is_steam_installed() is False
class TestUninstallSteamPackage:
"""Tests for uninstall steam package."""
def test_success(self) -> None:
"""Test that success."""
with patch(
f"{PKG}.subprocess.run", return_value=MagicMock(returncode=0, stderr="")
):
assert _uninstall_steam_package() is True
def test_already_absent_treated_as_success(self) -> None:
"""Test that already absent treated as success."""
with patch(
f"{PKG}.subprocess.run",
return_value=MagicMock(
@ -352,6 +425,7 @@ class TestUninstallSteamPackage:
assert _uninstall_steam_package() is True
def test_real_failure_returns_false(self) -> None:
"""Test that real failure returns false."""
with patch(
f"{PKG}.subprocess.run",
return_value=MagicMock(returncode=1, stderr="some other error"),
@ -359,17 +433,164 @@ class TestUninstallSteamPackage:
assert _uninstall_steam_package() is False
def test_subprocess_error_returns_false(self) -> None:
"""Test that subprocess error returns false."""
with patch(f"{PKG}.subprocess.run", side_effect=OSError):
assert _uninstall_steam_package() is False
class TestIsPackageInstalled:
"""Tests for is package installed."""
def test_installed(self) -> None:
"""Test that installed."""
with patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=0)):
assert _is_package_installed("protonup-qt") is True
def test_not_installed(self) -> None:
"""Test that not installed."""
with patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=1)):
assert _is_package_installed("protonup-qt") is False
class TestUninstallProtonHelpers:
"""Tests for uninstall proton helpers."""
def test_none_installed_removes_nothing(self) -> None:
"""Test that none installed removes nothing."""
with patch(f"{PKG}._is_package_installed", return_value=False):
assert _uninstall_proton_helpers() == []
def test_installed_packages_are_uninstalled(self) -> None:
"""Test that installed packages are uninstalled."""
installed = {"protonup-qt", "protontricks-git"}
with (
patch(f"{PKG}._is_package_installed", side_effect=lambda p: p in installed),
patch(f"{PKG}._uninstall_package", return_value=True) as mock_uninstall,
):
removed = _uninstall_proton_helpers()
assert set(removed) == installed
assert mock_uninstall.call_count == len(installed)
def test_failed_uninstall_is_logged_not_raised(self) -> None:
"""Test that failed uninstall is logged not raised."""
with (
patch(f"{PKG}._is_package_installed", return_value=True),
patch(f"{PKG}._uninstall_package", return_value=False),
):
assert _uninstall_proton_helpers() == []
class TestRemoveSteamRemnants:
"""Tests for remove steam remnants."""
def test_no_remnants_present(self, paths: _Paths) -> None:
"""Test that no remnants present."""
assert _remove_steam_remnants() == []
def test_removes_directory(self, paths: _Paths) -> None:
"""Test that removes directory."""
steam_dir = paths.remnant_paths[0]
(steam_dir / "steamapps").mkdir(parents=True)
removed = _remove_steam_remnants()
assert str(steam_dir) in removed
assert not steam_dir.exists()
def test_removes_symlink_without_following_into_rmtree(self, paths: _Paths) -> None:
"""Test that removes symlink without following into rmtree."""
# Target lives outside the curated remnant list, so this test can
# tell "unlinked the symlink" apart from "rmtree'd its target".
external_target = paths.remnant_paths[0].parent / "external_target.pid"
external_target.parent.mkdir(parents=True, exist_ok=True)
external_target.write_text("123", encoding="utf-8")
symlink_path = paths.remnant_paths[4]
symlink_path.symlink_to(external_target)
removed = _remove_steam_remnants()
assert str(symlink_path) in removed
assert not symlink_path.exists()
assert not symlink_path.is_symlink()
assert external_target.exists()
def test_removal_failure_is_logged_not_raised(self, paths: _Paths) -> None:
"""Test that removal failure is logged not raised."""
steam_dir = paths.remnant_paths[0]
steam_dir.mkdir(parents=True)
with patch(f"{PKG}.shutil.rmtree", side_effect=OSError):
assert _remove_steam_remnants() == []
class TestLogSteamPurge:
"""Tests for log steam purge."""
def test_noop_when_nothing_removed(self, paths: _Paths) -> None:
"""Test that noop when nothing removed."""
_log_steam_purge([], [])
assert not paths.purge_log_file.exists()
def test_appends_entry(self, paths: _Paths) -> None:
"""Test that appends entry."""
_log_steam_purge(["/home/kuhy/.steam"], ["protonup-qt"])
entries = json.loads(paths.purge_log_file.read_text(encoding="utf-8"))
assert len(entries) == 1
assert entries[0]["removed_paths"] == ["/home/kuhy/.steam"]
assert entries[0]["removed_packages"] == ["protonup-qt"]
_log_steam_purge(["/home/kuhy/steam"], [])
entries = json.loads(paths.purge_log_file.read_text(encoding="utf-8"))
assert len(entries) == 2
def test_corrupt_log_file_is_reset(self, paths: _Paths) -> None:
"""Test that corrupt log file is reset."""
paths.purge_log_file.parent.mkdir(parents=True, exist_ok=True)
paths.purge_log_file.write_text("not json", encoding="utf-8")
_log_steam_purge(["/home/kuhy/.steam"], [])
entries = json.loads(paths.purge_log_file.read_text(encoding="utf-8"))
assert len(entries) == 1
def test_non_list_log_file_is_reset(self, paths: _Paths) -> None:
"""Test that non list log file is reset."""
paths.purge_log_file.parent.mkdir(parents=True, exist_ok=True)
paths.purge_log_file.write_text(json.dumps({"not": "a list"}), encoding="utf-8")
_log_steam_purge(["/home/kuhy/.steam"], [])
entries = json.loads(paths.purge_log_file.read_text(encoding="utf-8"))
assert len(entries) == 1
class TestPurgeSteamAndProton:
"""Tests for purge steam and proton."""
def test_delegates_and_logs(self, paths: _Paths) -> None:
"""Test that delegates and logs."""
with (
patch(f"{PKG}._remove_steam_remnants", return_value=["/home/kuhy/.steam"]),
patch(f"{PKG}._uninstall_proton_helpers", return_value=["protonup-qt"]),
):
_purge_steam_and_proton()
entries = json.loads(paths.purge_log_file.read_text(encoding="utf-8"))
assert entries[0]["removed_paths"] == ["/home/kuhy/.steam"]
assert entries[0]["removed_packages"] == ["protonup-qt"]
def test_nothing_removed_does_not_log(self, paths: _Paths) -> None:
"""Test that nothing removed does not log."""
with (
patch(f"{PKG}._remove_steam_remnants", return_value=[]),
patch(f"{PKG}._uninstall_proton_helpers", return_value=[]),
):
_purge_steam_and_proton()
assert not paths.purge_log_file.exists()
# ──────────────────────────────────────────────────────────────
# Hosts domain blocking
# ──────────────────────────────────────────────────────────────
class TestApplyTotalBlockHosts:
"""Tests for apply total block hosts."""
def test_appends_block_when_absent(self, paths: _Paths) -> None:
"""Test that appends block when absent."""
paths.hosts_file.write_text("127.0.0.1 localhost\n", encoding="utf-8")
with (
patch(f"{PKG}._disable_hosts_protection"),
@ -383,6 +604,7 @@ class TestApplyTotalBlockHosts:
assert "steamcommunity.com" in written
def test_already_present_is_noop(self, paths: _Paths) -> None:
"""Test that already present is noop."""
paths.hosts_file.write_text(
f"127.0.0.1 localhost\n{_HOSTS_BLOCK_BEGIN}"
f"0.0.0.0 x.com\n{_HOSTS_BLOCK_END}",
@ -393,9 +615,11 @@ class TestApplyTotalBlockHosts:
mock_write.assert_not_called()
def test_missing_hosts_file_returns_false(self) -> None:
"""Test that missing hosts file returns false."""
assert _apply_total_block_hosts() is False
def test_write_failure_still_reenables_protection(self, paths: _Paths) -> None:
"""Test that write failure still reenables protection."""
paths.hosts_file.write_text("127.0.0.1 localhost\n", encoding="utf-8")
with (
patch(f"{PKG}._disable_hosts_protection"),
@ -407,7 +631,10 @@ class TestApplyTotalBlockHosts:
class TestRemoveTotalBlockHosts:
"""Tests for remove total block hosts."""
def test_removes_block_when_present(self, paths: _Paths) -> None:
"""Test that removes block when present."""
paths.hosts_file.write_text(
f"127.0.0.1 localhost\n{_HOSTS_BLOCK_BEGIN}"
f"0.0.0.0 x.com\n{_HOSTS_BLOCK_END}"
@ -426,15 +653,18 @@ class TestRemoveTotalBlockHosts:
assert "localhost" in written
def test_absent_is_noop(self, paths: _Paths) -> None:
"""Test that absent is noop."""
paths.hosts_file.write_text("127.0.0.1 localhost\n", encoding="utf-8")
with patch(f"{PKG}._sudo_write_hosts") as mock_write:
assert _remove_total_block_hosts() is True
mock_write.assert_not_called()
def test_missing_hosts_file_returns_false(self) -> None:
"""Test that missing hosts file returns false."""
assert _remove_total_block_hosts() is False
def test_write_failure_still_reenables_protection(self, paths: _Paths) -> None:
"""Test that write failure still reenables protection."""
paths.hosts_file.write_text(
f"127.0.0.1 localhost\n{_HOSTS_BLOCK_BEGIN}"
f"0.0.0.0 x.com\n{_HOSTS_BLOCK_END}",
@ -455,19 +685,25 @@ class TestRemoveTotalBlockHosts:
class TestIpCache:
"""Tests for ip cache."""
def test_load_no_file_returns_empty(self) -> None:
"""Test that load no file returns empty."""
assert _load_cached_ips() == set()
def test_save_then_load_round_trips(self) -> None:
"""Test that save then load round trips."""
_save_cached_ips({"1.2.3.4", "5.6.7.8"})
assert _load_cached_ips() == {"1.2.3.4", "5.6.7.8"}
def test_load_malformed_json_returns_empty(self, paths: _Paths) -> None:
"""Test that load malformed json returns empty."""
paths.ip_cache_file.parent.mkdir(parents=True, exist_ok=True)
paths.ip_cache_file.write_text("not json", encoding="utf-8")
assert _load_cached_ips() == set()
def test_load_non_list_json_returns_empty(self, paths: _Paths) -> None:
"""Test that load non list json returns empty."""
paths.ip_cache_file.parent.mkdir(parents=True, exist_ok=True)
paths.ip_cache_file.write_text(json.dumps({"a": 1}), encoding="utf-8")
assert _load_cached_ips() == set()
@ -479,11 +715,15 @@ class TestIpCache:
class TestIptablesChainIntact:
"""Tests for iptables chain intact."""
def test_missing_chain_returns_false(self) -> None:
"""Test that missing chain returns false."""
with patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=1)):
assert _iptables_chain_intact({"1.2.3.4"}) is False
def test_missing_ip_returns_false(self) -> None:
"""Test that missing ip returns false."""
listing = MagicMock(
returncode=0,
stdout="-N STEAM_TOTAL_BLOCK\n-A STEAM_TOTAL_BLOCK -d 9.9.9.9/32 -j DROP\n",
@ -492,6 +732,7 @@ class TestIptablesChainIntact:
assert _iptables_chain_intact({"1.2.3.4"}) is False
def test_all_ips_present_and_hooked_returns_true(self) -> None:
"""Test that all ips present and hooked returns true."""
listing = MagicMock(
returncode=0,
stdout=(
@ -505,6 +746,7 @@ class TestIptablesChainIntact:
assert _iptables_chain_intact({"1.2.3.4", "5.6.7.8"}) is True
def test_ips_present_but_not_hooked_returns_false(self) -> None:
"""Test that ips present but not hooked returns false."""
listing = MagicMock(
returncode=0,
stdout="-A STEAM_TOTAL_BLOCK -d 1.2.3.4/32 -j DROP\n",
@ -525,7 +767,10 @@ class TestIptablesChainIntact:
class TestApplyTotalBlockIptables:
"""Tests for apply total block iptables."""
def test_intact_chain_short_circuits(self) -> None:
"""Test that intact chain short circuits."""
_save_cached_ips({"1.2.3.4"})
with (
patch(f"{PKG}._iptables_chain_intact", return_value=True),
@ -535,6 +780,7 @@ class TestApplyTotalBlockIptables:
mock_run.assert_not_called()
def test_rebuilds_when_not_intact(self) -> None:
"""Test that rebuilds when not intact."""
with (
patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=0)),
patch(
@ -546,6 +792,7 @@ class TestApplyTotalBlockIptables:
assert "9.9.9.9" in _load_cached_ips()
def test_dns_failure_skips_that_domain(self) -> None:
"""Test that dns failure skips that domain."""
import socket as real_socket
with (
@ -556,6 +803,7 @@ class TestApplyTotalBlockIptables:
assert _load_cached_ips() == set()
def test_subprocess_error_returns_false(self) -> None:
"""Test that subprocess error returns false."""
with (
patch(
f"{PKG}.subprocess.run",
@ -566,6 +814,8 @@ class TestApplyTotalBlockIptables:
assert _apply_total_block_iptables() is False
def test_inserts_output_hook_when_missing(self) -> None:
"""Test that inserts output hook when missing."""
def run_side_effect(cmd: list[str], **_kwargs: object) -> MagicMock:
if "-C" in cmd:
return MagicMock(returncode=1)
@ -579,17 +829,22 @@ class TestApplyTotalBlockIptables:
class TestRemoveTotalBlockIptables:
"""Tests for remove total block iptables."""
def test_removes_chain_and_cache(self, paths: _Paths) -> None:
"""Test that removes chain and cache."""
_save_cached_ips({"1.2.3.4"})
with patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=0)):
assert _remove_total_block_iptables() is True
assert not paths.ip_cache_file.exists()
def test_no_cache_file_is_fine(self) -> None:
"""Test that no cache file is fine."""
with patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=0)):
assert _remove_total_block_iptables() is True
def test_subprocess_error_returns_false(self) -> None:
"""Test that subprocess error returns false."""
with patch(f"{PKG}.subprocess.run", side_effect=OSError):
assert _remove_total_block_iptables() is False
@ -600,7 +855,10 @@ class TestRemoveTotalBlockIptables:
class TestStartTotalBlock:
"""Tests for start total block."""
def test_success(self) -> None:
"""Test that success."""
with (
patch(
f"{PKG}.subprocess.run",
@ -608,13 +866,32 @@ class TestStartTotalBlock:
),
patch(f"{PKG}._kill_steam_and_launchers", return_value=[]),
patch(f"{PKG}._uninstall_steam_package", return_value=True),
patch(f"{PKG}._purge_steam_and_proton"),
patch(f"{PKG}._apply_total_block_hosts", return_value=True),
patch(f"{PKG}._apply_total_block_iptables", return_value=True),
patch(f"{PKG}.flush_dns_cache"),
):
assert start_total_block(1) is True
def test_calls_purge_steam_and_proton(self) -> None:
"""Test that calls purge steam and proton."""
with (
patch(
f"{PKG}.subprocess.run",
return_value=MagicMock(returncode=0, stderr=""),
),
patch(f"{PKG}._kill_steam_and_launchers", return_value=[]),
patch(f"{PKG}._uninstall_steam_package", return_value=True),
patch(f"{PKG}._purge_steam_and_proton") as mock_purge,
patch(f"{PKG}._apply_total_block_hosts", return_value=True),
patch(f"{PKG}._apply_total_block_iptables", return_value=True),
patch(f"{PKG}.flush_dns_cache"),
):
start_total_block(1)
mock_purge.assert_called_once()
def test_package_block_start_failure_aborts(self) -> None:
"""Test that package block start failure aborts."""
with patch(
f"{PKG}.subprocess.run",
return_value=MagicMock(returncode=1, stderr="guardctl error"),
@ -631,6 +908,7 @@ class TestStartTotalBlock:
),
patch(f"{PKG}._kill_steam_and_launchers", return_value=[]),
patch(f"{PKG}._uninstall_steam_package", return_value=False),
patch(f"{PKG}._purge_steam_and_proton"),
patch(f"{PKG}._apply_total_block_hosts", return_value=False),
patch(f"{PKG}._apply_total_block_iptables", return_value=False),
patch(f"{PKG}.flush_dns_cache"),
@ -638,6 +916,7 @@ class TestStartTotalBlock:
assert start_total_block(1) is True
def test_logs_when_processes_were_killed(self) -> None:
"""Test that logs when processes were killed."""
with (
patch(
f"{PKG}.subprocess.run",
@ -645,6 +924,7 @@ class TestStartTotalBlock:
),
patch(f"{PKG}._kill_steam_and_launchers", return_value=[(1, "steam")]),
patch(f"{PKG}._uninstall_steam_package", return_value=True),
patch(f"{PKG}._purge_steam_and_proton"),
patch(f"{PKG}._apply_total_block_hosts", return_value=True),
patch(f"{PKG}._apply_total_block_iptables", return_value=True),
patch(f"{PKG}.flush_dns_cache"),
@ -653,11 +933,15 @@ class TestStartTotalBlock:
class TestEnforceTotalBlockTick:
"""Tests for enforce total block tick."""
def test_reinstalls_steam_if_reappeared(self) -> None:
"""Test that reinstalls steam if reappeared."""
with (
patch(f"{PKG}._kill_steam_and_launchers", return_value=[]),
patch(f"{PKG}._is_steam_installed", return_value=True),
patch(f"{PKG}._uninstall_steam_package") as mock_uninstall,
patch(f"{PKG}._purge_steam_and_proton"),
patch(f"{PKG}._apply_total_block_hosts", return_value=True),
patch(f"{PKG}._apply_total_block_iptables", return_value=True),
):
@ -665,19 +949,37 @@ class TestEnforceTotalBlockTick:
mock_uninstall.assert_called_once()
def test_no_reinstall_when_steam_absent(self) -> None:
"""Test that no reinstall when steam absent."""
with (
patch(f"{PKG}._kill_steam_and_launchers", return_value=[]),
patch(f"{PKG}._is_steam_installed", return_value=False),
patch(f"{PKG}._uninstall_steam_package") as mock_uninstall,
patch(f"{PKG}._purge_steam_and_proton"),
patch(f"{PKG}._apply_total_block_hosts", return_value=True),
patch(f"{PKG}._apply_total_block_iptables", return_value=True),
):
enforce_total_block_tick()
mock_uninstall.assert_not_called()
def test_purges_steam_and_proton_every_tick(self) -> None:
"""Test that purges steam and proton every tick."""
with (
patch(f"{PKG}._kill_steam_and_launchers", return_value=[]),
patch(f"{PKG}._is_steam_installed", return_value=False),
patch(f"{PKG}._uninstall_steam_package"),
patch(f"{PKG}._purge_steam_and_proton") as mock_purge,
patch(f"{PKG}._apply_total_block_hosts", return_value=True),
patch(f"{PKG}._apply_total_block_iptables", return_value=True),
):
enforce_total_block_tick()
mock_purge.assert_called_once()
class TestEndTotalBlockCleanup:
"""Tests for end total block cleanup."""
def test_ends_lock_and_removes_blocks(self) -> None:
"""Test that ends lock and removes blocks."""
with (
patch(
f"{PKG}.subprocess.run",
@ -692,6 +994,7 @@ class TestEndTotalBlockCleanup:
mock_ipt.assert_called_once()
def test_package_block_end_failure_still_cleans_up_rest(self) -> None:
"""Test that package block end failure still cleans up rest."""
with (
patch(
f"{PKG}.subprocess.run",
@ -706,6 +1009,7 @@ class TestEndTotalBlockCleanup:
mock_ipt.assert_called_once()
def test_hosts_removal_failure_is_logged_not_raised(self) -> None:
"""Test that hosts removal failure is logged not raised."""
with (
patch(
f"{PKG}.subprocess.run",
@ -718,6 +1022,7 @@ class TestEndTotalBlockCleanup:
end_total_block_cleanup() # must not raise
def test_iptables_removal_failure_is_logged_not_raised(self) -> None:
"""Test that iptables removal failure is logged not raised."""
with (
patch(
f"{PKG}.subprocess.run",
@ -733,4 +1038,5 @@ class TestEndTotalBlockCleanup:
# Sanity: the module-level chain name constant is what everything above
# assumes when constructing fake iptables -S output.
def test_iptables_chain_name_constant() -> None:
"""Test that iptables chain name constant."""
assert IPTABLES_CHAIN == "STEAM_TOTAL_BLOCK"