diff --git a/steam_backlog_enforcer/_achievement_skip.py b/steam_backlog_enforcer/_achievement_skip.py index 6ff031f..da96697 100644 --- a/steam_backlog_enforcer/_achievement_skip.py +++ b/steam_backlog_enforcer/_achievement_skip.py @@ -11,6 +11,10 @@ achievements at all. A game is skipped only while all three hold: So a game that later gains achievements without gaining the flag is still picked up, at the latest that long after. + +The same cache doubles as the *always allowed* list (:func:`zero_achievement_ids`): +an owned game Steam answered "no achievements" for is never uninstalled, hidden +or killed as unauthorized. Only the daily budget still applies to it. """ from __future__ import annotations @@ -136,3 +140,64 @@ def record_scan( _atomic_write(_cache_file(), json.dumps(payload, indent=2) + "\n") except OSError: logger.warning("Could not save %s", _cache_file(), exc_info=True) + + +# (file stamps, ids) of the last derivation; the derivation reads three files. +_memo: dict[str, Any] = {} + + +def _stamp(path: Path) -> tuple[str, int, int]: + """``(path, mtime_ns, size)``; zeros when absent, so a new file changes it.""" + try: + st = path.stat() + except OSError: + return (str(path), 0, 0) + return (str(path), st.st_mtime_ns, st.st_size) + + +def _derive_zero_ids() -> frozenset[int]: + """Cache keys for the configured account minus anything in the snapshot. + + Fails closed: a missing or corrupt file, another account, an unknown + account or an unreadable snapshot all give the empty set. + """ + try: + steam_id = json.loads(config.CONFIG_FILE.read_text(encoding="utf-8"))[ + "steam_id" + ] + except OSError, ValueError, KeyError, TypeError: + return frozenset() + if not isinstance(steam_id, str) or not steam_id: + return frozenset() + confirmed = set(_load_checked(steam_id)) + if not config.SNAPSHOT_FILE.exists(): + return frozenset(confirmed) + try: + # Only games with achievements are ever snapshotted, so one that is in + # it contradicts the cache and is enforced. + in_snapshot = {int(entry["app_id"]) for entry in load_snapshot() or []} + except OSError, ValueError, KeyError, TypeError: + return frozenset() + return frozenset(confirmed - in_snapshot) + + +def zero_achievement_ids() -> frozenset[int]: + """Owned games Steam confirmed have zero achievements: always allowed. + + Presence in the cache is enough (no recheck age): only a scan that finds + achievements removes an entry, so the game is never dropped mid-session + for want of a rescan. Memoized on the files' stamps because the enforce + loop asks once per installed or running game every few seconds. + + Returns: + The allowed app ids; empty whenever the status is unknown. + """ + key = tuple( + _stamp(path) + for path in (_cache_file(), config.CONFIG_FILE, config.SNAPSHOT_FILE) + ) + if _memo.get("key") != key: + _memo["key"] = key + _memo["ids"] = _derive_zero_ids() + ids: frozenset[int] = _memo["ids"] + return ids diff --git a/steam_backlog_enforcer/_scanning_tampering.py b/steam_backlog_enforcer/_scanning_tampering.py index 382a282..8e8e827 100644 --- a/steam_backlog_enforcer/_scanning_tampering.py +++ b/steam_backlog_enforcer/_scanning_tampering.py @@ -13,6 +13,7 @@ from itertools import islice import logging from typing import TYPE_CHECKING, Any +from steam_backlog_enforcer._achievement_skip import zero_achievement_ids from steam_backlog_enforcer._allowed_games import allowed_games from steam_backlog_enforcer._assignment_progress import iso_to_epoch from steam_backlog_enforcer._progress import current_progress @@ -56,6 +57,7 @@ def _legitimately_played(state: State) -> set[int]: exempt = {app_id for app_id, _ in allowed_games(state)} exempt.update(state.finished_app_ids) exempt.update(get_approved_exception_ids()) + exempt.update(zero_achievement_ids()) exempt.update( pick["app_id"] for pick in state.manual_picks if pick.get("app_id") is not None ) diff --git a/steam_backlog_enforcer/game_uninstall.py b/steam_backlog_enforcer/game_uninstall.py index 4a5c5dd..7a42b37 100644 --- a/steam_backlog_enforcer/game_uninstall.py +++ b/steam_backlog_enforcer/game_uninstall.py @@ -13,6 +13,7 @@ import re import shutil from typing import TYPE_CHECKING +from steam_backlog_enforcer._achievement_skip import zero_achievement_ids from steam_backlog_enforcer._game_names import _is_protected_name from steam_backlog_enforcer._progress import tracked from steam_backlog_enforcer._protected_apps import PROTECTED_APP_IDS @@ -173,7 +174,10 @@ def is_protected_app(app_id: int) -> bool: """Return True if *app_id* must never be uninstalled. Combines the hardcoded Steam infrastructure set with any app IDs that - have been approved via the time-locked exception mechanism. + have been approved via the exception mechanism, and with owned games Steam + confirmed have zero achievements (nothing to complete, so always allowed; + the gaming-time budget still applies to them). Unknown status is not + protected. This one test also gates the unauthorized-game kill. Args: app_id: Steam application ID to check. @@ -181,4 +185,8 @@ def is_protected_app(app_id: int) -> bool: Returns: True if the app should be left alone by the enforcer. """ - return app_id in PROTECTED_APP_IDS or app_id in get_approved_exception_ids() + return ( + app_id in PROTECTED_APP_IDS + or app_id in get_approved_exception_ids() + or app_id in zero_achievement_ids() + ) diff --git a/steam_backlog_enforcer/library_hider.py b/steam_backlog_enforcer/library_hider.py index 83fba4d..fa9a2ef 100644 --- a/steam_backlog_enforcer/library_hider.py +++ b/steam_backlog_enforcer/library_hider.py @@ -20,6 +20,7 @@ import json import logging from typing import TYPE_CHECKING +from steam_backlog_enforcer._achievement_skip import zero_achievement_ids from steam_backlog_enforcer._actions import allowed_app_ids from steam_backlog_enforcer._cdp import _cdp_result_value, _evaluate_js from steam_backlog_enforcer._echo import _echo @@ -86,12 +87,15 @@ def hide_other_games( Approved whitelist exceptions (``add-exception``) are always treated as allowed, so a permanently whitelisted game such as a co-op title stays visible next to the assignment instead of being hidden with the rest. + So are owned games with zero achievements (see ``zero_achievement_ids``). Returns the total number of games hidden across all passes. """ ensure_steam_debug_port() - allowed_app_ids = allowed_app_ids | get_approved_exception_ids() + allowed_app_ids = ( + allowed_app_ids | get_approved_exception_ids() | zero_achievement_ids() + ) allowed_json = json.dumps(sorted(allowed_app_ids)) extra_ids = sorted(aid for aid in owned_app_ids if aid not in allowed_app_ids) extra_json = json.dumps(extra_ids) diff --git a/steam_backlog_enforcer/tests/_zero_ach.py b/steam_backlog_enforcer/tests/_zero_ach.py new file mode 100644 index 0000000..c627ebb --- /dev/null +++ b/steam_backlog_enforcer/tests/_zero_ach.py @@ -0,0 +1,55 @@ +"""Shared fixtures for the zero-achievement "always allowed" tests.""" + +from __future__ import annotations + +import json +import re +from typing import TYPE_CHECKING + +from steam_backlog_enforcer import _achievement_skip as skip +from steam_backlog_enforcer import config + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + +SIMS3 = 47890 +ASSIGNED = 1091500 +WITH_ACH = 570 +STEAM_ID = "76561198009649542" +NOW = 1_000_000_000.0 + + +def write(path: Path, payload: object) -> None: + path.write_text(json.dumps(payload), encoding="utf-8") + + +def confirm_zero(*app_ids: int, steam_id: str = STEAM_ID) -> None: + """Make the real cache say Steam answered "no achievements" for these.""" + write(config.CONFIG_FILE, {"steam_id": STEAM_ID}) + write( + skip._cache_file(), + {"steam_id": steam_id, "checked": {str(a): NOW for a in app_ids}}, + ) + + +def write_snapshot(*app_ids: int) -> None: + write( + config.SNAPSHOT_FILE, [{"app_id": a, "total_achievements": 5} for a in app_ids] + ) + + +def recorder(removed: list[int]) -> Callable[..., bool]: + """An ``uninstall_game`` stand-in that notes which app was removed.""" + + def record(app_id: int, _name: str = "") -> bool: + removed.append(app_id) + return True + + return record + + +def js_list(pattern: str, js: str) -> list[int]: + match = re.search(pattern, js) + assert match is not None + return list(json.loads(match.group(1))) diff --git a/steam_backlog_enforcer/tests/test_zero_achievement_allowed.py b/steam_backlog_enforcer/tests/test_zero_achievement_allowed.py new file mode 100644 index 0000000..37c5a75 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_zero_achievement_allowed.py @@ -0,0 +1,153 @@ +"""Owned games with zero achievements are always allowed, except by the budget.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +from leisure_watch.procscan import qualifying_pids +import pytest + +from steam_backlog_enforcer import config +from steam_backlog_enforcer._enforce_guards import guard_installed_games +from steam_backlog_enforcer._playtime_cutoff import _kill_set +from steam_backlog_enforcer._scanning_tampering import _legitimately_played +from steam_backlog_enforcer.config import State +from steam_backlog_enforcer.enforcer import enforce_allowed_game +from steam_backlog_enforcer.game_uninstall import ( + is_protected_app, + uninstall_other_games, +) +from steam_backlog_enforcer.library_hider import hide_other_games +from steam_backlog_enforcer.tests._zero_ach import ( + ASSIGNED, + SIMS3, + WITH_ACH, + confirm_zero, + js_list, + recorder, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + +@pytest.fixture(autouse=True) +def _no_exceptions() -> Iterator[None]: + """No approved exceptions, so only the zero-achievement rule can protect.""" + with patch( + "steam_backlog_enforcer.game_uninstall.get_approved_exception_ids", + return_value=frozenset(), + ): + yield + + +class TestUninstall: + def test_zero_achievement_game_is_kept(self) -> None: + confirm_zero(SIMS3) + removed: list[int] = [] + with ( + patch( + "steam_backlog_enforcer.game_uninstall.get_installed_games", + return_value=[(SIMS3, "The Sims 3"), (WITH_ACH, "Other")], + ), + patch( + "steam_backlog_enforcer.game_uninstall.uninstall_game", + side_effect=recorder(removed), + ), + ): + uninstall_other_games({ASSIGNED}) + assert removed == [WITH_ACH] + + def test_guard_keeps_it_and_removes_game_with_achievements(self) -> None: + confirm_zero(SIMS3) + removed: list[int] = [] + with ( + patch( + "steam_backlog_enforcer._enforce_guards.get_installed_games", + return_value=[(SIMS3, "The Sims 3"), (WITH_ACH, "Other")], + ), + patch( + "steam_backlog_enforcer._enforce_guards.uninstall_game", + side_effect=recorder(removed), + ), + patch("steam_backlog_enforcer._enforce_guards.send_notification"), + ): + assert guard_installed_games({ASSIGNED}) == 1 + assert removed == [WITH_ACH] + + +class TestHiding: + def test_zero_achievement_game_stays_visible(self) -> None: + confirm_zero(SIMS3) + pkg = "steam_backlog_enforcer.library_hider" + captured: dict[str, str] = {} + + def fake_eval(js: str) -> dict[str, object]: + captured["js"] = js + return {} + + with ( + patch(f"{pkg}.ensure_steam_debug_port"), + patch(f"{pkg}._evaluate_js", side_effect=fake_eval), + patch(f"{pkg}._cdp_result_value", return_value='{"totalHidden": 0}'), + patch(f"{pkg}.get_approved_exception_ids", return_value=frozenset()), + ): + hide_other_games([SIMS3, ASSIGNED, WITH_ACH], {ASSIGNED}) + visible = js_list(r"new Set\((\[.*?\])\)", captured["js"]) + to_hide = js_list(r"const extraIds = (\[.*?\]);", captured["js"]) + assert SIMS3 in visible + assert to_hide == [WITH_ACH] + + +class TestTampering: + def test_zero_achievement_game_is_exempt(self) -> None: + confirm_zero(SIMS3) + with patch( + "steam_backlog_enforcer._scanning_tampering.get_approved_exception_ids", + return_value=frozenset(), + ): + exempt = _legitimately_played(State(current_app_id=ASSIGNED)) + assert SIMS3 in exempt + + +class TestUnauthorizedKill: + def test_not_killed_but_other_games_are(self) -> None: + confirm_zero(SIMS3) + with ( + patch( + "steam_backlog_enforcer.enforcer.get_running_steam_game_pids", + return_value={100: SIMS3, 200: WITH_ACH}, + ), + patch("steam_backlog_enforcer.enforcer.kill_process") as kill, + ): + assert enforce_allowed_game({ASSIGNED}) == [(200, WITH_ACH)] + kill.assert_called_once_with(200, WITH_ACH) + + +class TestBudgetStillApplies: + """The playtime budget never asks is_protected_app, so it still bites.""" + + def test_running_zero_game_is_counted_and_in_the_kill_set(self) -> None: + confirm_zero(SIMS3) + rules = MagicMock(count_launchers=False) + with ( + patch( + "leisure_watch.procscan._qualifying.get_running_steam_game_pids", + return_value={100: SIMS3}, + ), + patch( + "steam_backlog_enforcer.enforcer.get_running_steam_game_pids", + return_value={100: SIMS3}, + ), + patch( + "steam_backlog_enforcer._playtime_cutoff.steam_and_launcher_pids", + return_value=set(), + ), + patch("steam_backlog_enforcer._playtime_cutoff.config_mod") as cfg, + ): + cfg.CONFIG_FILE = config.CONFIG_FILE + assert is_protected_app(SIMS3) + assert enforce_allowed_game({ASSIGNED}) == [] + assert qualifying_pids(rules, config.CONFIG_FILE) == {100: f"app:{SIMS3}"} + assert 100 in _kill_set(rules) diff --git a/steam_backlog_enforcer/tests/test_zero_achievement_fail_closed.py b/steam_backlog_enforcer/tests/test_zero_achievement_fail_closed.py new file mode 100644 index 0000000..eaae8bf --- /dev/null +++ b/steam_backlog_enforcer/tests/test_zero_achievement_fail_closed.py @@ -0,0 +1,132 @@ +"""Zero-achievement "always allowed": unknown status stays enforced.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from steam_backlog_enforcer import _achievement_skip as skip +from steam_backlog_enforcer import config +from steam_backlog_enforcer.enforcer import enforce_allowed_game +from steam_backlog_enforcer.game_uninstall import ( + is_protected_app, + uninstall_other_games, +) +from steam_backlog_enforcer.tests._zero_ach import ( + ASSIGNED, + NOW, + SIMS3, + STEAM_ID, + WITH_ACH, + confirm_zero, + recorder, + write, + write_snapshot, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + +@pytest.fixture(autouse=True) +def _no_exceptions() -> Iterator[None]: + """No approved exceptions, so only the zero-achievement rule can protect.""" + with patch( + "steam_backlog_enforcer.game_uninstall.get_approved_exception_ids", + return_value=frozenset(), + ): + yield + + +class TestFailClosed: + def test_no_cache_file(self) -> None: + assert not is_protected_app(SIMS3) + + def test_corrupt_cache_file(self) -> None: + write(config.CONFIG_FILE, {"steam_id": STEAM_ID}) + skip._cache_file().write_text("{nope", encoding="utf-8") + assert not is_protected_app(SIMS3) + + def test_cache_of_another_account(self) -> None: + confirm_zero(SIMS3, steam_id="someone-else") + assert not is_protected_app(SIMS3) + + def test_unknown_account(self) -> None: + confirm_zero(SIMS3) + config.CONFIG_FILE.unlink() + assert not is_protected_app(SIMS3) + + def test_account_id_of_the_wrong_type(self) -> None: + confirm_zero(SIMS3) + write(config.CONFIG_FILE, {"steam_id": 12345}) + assert not is_protected_app(SIMS3) + + def test_unreadable_snapshot(self) -> None: + confirm_zero(SIMS3) + config.SNAPSHOT_FILE.write_text("{nope", encoding="utf-8") + assert not is_protected_app(SIMS3) + + def test_answer_is_reused_until_a_file_changes(self) -> None: + confirm_zero(SIMS3) + with patch.object(skip, "_derive_zero_ids", wraps=skip._derive_zero_ids) as d: + assert is_protected_app(SIMS3) + assert is_protected_app(SIMS3) + assert d.call_count <= 1 + + def test_unconfirmed_game_is_enforced(self) -> None: + confirm_zero(SIMS3) + assert not is_protected_app(WITH_ACH) + + def test_game_that_gained_achievements_is_enforced(self) -> None: + confirm_zero(SIMS3) + write_snapshot(SIMS3) + assert not is_protected_app(SIMS3) + + def test_unknown_status_game_is_uninstalled(self) -> None: + removed: list[int] = [] + with ( + patch( + "steam_backlog_enforcer.game_uninstall.get_installed_games", + return_value=[(SIMS3, "The Sims 3")], + ), + patch( + "steam_backlog_enforcer.game_uninstall.uninstall_game", + side_effect=recorder(removed), + ), + ): + uninstall_other_games({ASSIGNED}) + assert removed == [SIMS3] + + def test_unknown_status_game_is_killed(self) -> None: + with ( + patch( + "steam_backlog_enforcer.enforcer.get_running_steam_game_pids", + return_value={100: SIMS3}, + ), + patch("steam_backlog_enforcer.enforcer.kill_process"), + ): + assert enforce_allowed_game({ASSIGNED}) == [(100, SIMS3)] + + +class TestRecordedByScan: + def test_a_scan_that_finds_no_achievements_allows_the_game(self) -> None: + write(config.CONFIG_FILE, {"steam_id": STEAM_ID}) + game = {"appid": SIMS3, "has_community_visible_stats": False} + skip.record_scan(STEAM_ID, [game], set(), NOW) + assert is_protected_app(SIMS3) + + def test_a_later_scan_that_finds_achievements_enforces_it_again(self) -> None: + write(config.CONFIG_FILE, {"steam_id": STEAM_ID}) + game = {"appid": SIMS3, "has_community_visible_stats": False} + skip.record_scan(STEAM_ID, [game], set(), NOW) + assert is_protected_app(SIMS3) + skip.record_scan(STEAM_ID, [game], {SIMS3}, NOW + 1) + assert not is_protected_app(SIMS3) + + def test_old_confirmation_still_counts(self) -> None: + """Presence is enough: only a scan seeing achievements revokes it.""" + confirm_zero(SIMS3) + write(skip._cache_file(), {"steam_id": STEAM_ID, "checked": {str(SIMS3): 1.0}}) + assert is_protected_app(SIMS3)