diff --git a/steam_backlog_enforcer/_total_block.py b/steam_backlog_enforcer/_total_block.py index 1dc577d..e85b7e4 100644 --- a/steam_backlog_enforcer/_total_block.py +++ b/steam_backlog_enforcer/_total_block.py @@ -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() diff --git a/steam_backlog_enforcer/tests/conftest.py b/steam_backlog_enforcer/tests/conftest.py index 1faca7c..e66974c 100644 --- a/steam_backlog_enforcer/tests/conftest.py +++ b/steam_backlog_enforcer/tests/conftest.py @@ -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", diff --git a/steam_backlog_enforcer/tests/test_total_block.py b/steam_backlog_enforcer/tests/test_total_block.py index 65a5f15..e85121a 100644 --- a/steam_backlog_enforcer/tests/test_total_block.py +++ b/steam_backlog_enforcer/tests/test_total_block.py @@ -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"