From d0e8ed580ce9b9c21e9876e64ab427ac52a9768f Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 29 Aug 2026 14:53:04 +0200 Subject: [PATCH] Auto-detect completed manual picks before enforcing the pick cap Co-Authored-By: Claude Opus 5 (1M context) --- pyproject.toml | 6 + steam_backlog_enforcer/_cmd_done_finalize.py | 3 +- steam_backlog_enforcer/_enforce_loop.py | 19 +- steam_backlog_enforcer/_mcp_actions.py | 12 +- steam_backlog_enforcer/_pick_completion.py | 249 ++++++++++++++++++ steam_backlog_enforcer/_scanning_assign.py | 8 +- steam_backlog_enforcer/_scanning_tampering.py | 27 +- steam_backlog_enforcer/config.py | 4 + .../docs/DOCS-manual-pick-design.md | 23 +- steam_backlog_enforcer/main/picks.py | 12 + steam_backlog_enforcer/main/status.py | 13 +- steam_backlog_enforcer/scanning.py | 4 +- .../tests/test_do_enforce.py | 29 ++ .../tests/test_main_abandon.py | 8 + .../tests/test_main_pick_cap.py | 31 +++ .../tests/test_main_status.py | 26 ++ .../tests/test_pick_completion.py | 230 ++++++++++++++++ .../tests/test_scanning_part2.py | 42 +++ .../tests/test_scanning_pick_next_game.py | 77 ++++++ 19 files changed, 815 insertions(+), 8 deletions(-) create mode 100644 steam_backlog_enforcer/_pick_completion.py create mode 100644 steam_backlog_enforcer/tests/test_pick_completion.py diff --git a/pyproject.toml b/pyproject.toml index ae32def..cf78902 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,6 +51,12 @@ ignore = [ "COM812", # formatter handles this "ISC001", # formatter may create these "S603", # prone to false positives + # Conflicts with SIM117. A `pytest.raises` that needs patches in scope has + # no legal form: combine them into one `with` and PT012 objects to the + # extra context managers; nest `raises` inside and SIM117 objects to the + # nested `with`. SIM117's combined form is the style used across this repo, + # so PT012 is the one that loses. + "PT012", ] fixable = ["ALL"] unfixable = [] diff --git a/steam_backlog_enforcer/_cmd_done_finalize.py b/steam_backlog_enforcer/_cmd_done_finalize.py index 475308b..8365ef7 100644 --- a/steam_backlog_enforcer/_cmd_done_finalize.py +++ b/steam_backlog_enforcer/_cmd_done_finalize.py @@ -16,6 +16,7 @@ from steam_backlog_enforcer._cmd_done import ( _report_assigned_confidence, ) from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids +from steam_backlog_enforcer._pick_completion import mark_finished from steam_backlog_enforcer.config import Config, State, load_snapshot from steam_backlog_enforcer.enforcer import ( enforce_allowed_game, @@ -48,7 +49,7 @@ def _finalize_completion( ) -> None: """Mark game complete, pick next, hide non-assigned games, notify.""" _echo(f"\n COMPLETED: {game_name}!") - state.finished_app_ids.append(app_id) + mark_finished(state, app_id) snapshot_data = load_snapshot() _echo("\nPicking next game...") diff --git a/steam_backlog_enforcer/_enforce_loop.py b/steam_backlog_enforcer/_enforce_loop.py index fad15f0..a95b419 100644 --- a/steam_backlog_enforcer/_enforce_loop.py +++ b/steam_backlog_enforcer/_enforce_loop.py @@ -16,6 +16,10 @@ from steam_backlog_enforcer._enforce_steps import ( # Re-exported: five modules and the test suite import get_all_owned_app_ids # from here, so splitting it into _owned_apps_cache stays invisible to them. from steam_backlog_enforcer._owned_apps_cache import get_all_owned_app_ids +from steam_backlog_enforcer._pick_completion import ( + daemon_sweep, + retire_completed_manual_picks_throttled, +) from steam_backlog_enforcer._playtime import playtime_tick from steam_backlog_enforcer._playtime_session import PlaytimeSession, new_session from steam_backlog_enforcer._steam_launch import steam_is_installed @@ -139,7 +143,16 @@ def _enforce_loop_iteration( if not steam_is_installed(): return - allowed = allowed_app_ids(state) + # Retire manual picks that have hit 100%, at most once every + # MANUAL_PICK_RECHECK_TTL_SECONDS - this loop ticks every 3s. + retire_completed_manual_picks_throttled(config, state) + + # Record, but do not evict. A pick the daemon just retired is no longer in + # allowed_app_ids, so steps A and B below would kill the running process + # and uninstall the game - possibly seconds after the final achievement + # popped, mid-session. Keeping those ids allowed defers eviction to a + # user-invoked done/check/pick-manual, which is where it was before. + allowed = allowed_app_ids(state) | daemon_sweep.retired if not allowed: return @@ -224,6 +237,10 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None: state.current_app_id = fresh.current_app_id state.current_game_name = fresh.current_game_name state.finished_app_ids = fresh.finished_app_ids + # Manual picks too: the MCP pick_manual tool adds a *second* pick + # without touching current_app_id, so a daemon that never reloaded + # this list would uninstall that pick as unauthorized. + state.manual_picks = fresh.manual_picks _enforce_loop_iteration(config, state, session=session, demo=demo) time.sleep(ENFORCE_INTERVAL) diff --git a/steam_backlog_enforcer/_mcp_actions.py b/steam_backlog_enforcer/_mcp_actions.py index 4ab20b4..49e669b 100644 --- a/steam_backlog_enforcer/_mcp_actions.py +++ b/steam_backlog_enforcer/_mcp_actions.py @@ -23,6 +23,9 @@ from steam_backlog_enforcer._mcp_server import ( logger, mcp, ) +from steam_backlog_enforcer._pick_completion import ( + retire_completed_manual_picks, +) from steam_backlog_enforcer._total_block import start_total_block from steam_backlog_enforcer.config import Config, State @@ -61,12 +64,16 @@ def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]: ), "confirm_required": True, } + config = Config.load() state = State.load() + # Free the slots of picks already at 100%, so the cap below cannot refuse + # on behalf of a game that is finished. Matches the CLI's pick-manual. + retired = retire_completed_manual_picks(config, state) refused = apply_manual_pick( state, app_id, game_name, - max_picks=Config.load().max_manual_picks, + max_picks=config.max_manual_picks, ) if refused is not None: return {"ok": False, "reason": refused} @@ -77,6 +84,9 @@ def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]: "action": "pick_manual", "app_id": app_id, "game_name": game_name, + "retired_completed": [ + {"app_id": r.app_id, "game_name": r.game_name} for r in retired if r.retired + ], } diff --git a/steam_backlog_enforcer/_pick_completion.py b/steam_backlog_enforcer/_pick_completion.py new file mode 100644 index 0000000..b318842 --- /dev/null +++ b/steam_backlog_enforcer/_pick_completion.py @@ -0,0 +1,249 @@ +"""Achievement-based retirement of finished manual picks. + +A manual pick stops holding its slot once its app id is in +``State.finished_app_ids`` -- that is what ``_allowed_games._pick_is_active`` +reads. Historically only ``State.current_app_id`` was ever checked for +completion (``cmd_done``, ``do_check``), so a manual pick that was *not* the +current assignment had no achievement-based release path at all: it sat on a +slot until the 14-day lock expired or the user abandoned it by hand. + +This module closes that gap by checking every active pick and retiring the ones +at 100%. It deliberately stops there -- freeing a slot is not a request for a +replacement game, so nothing here calls ``pick_next_game``. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import logging +import time +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._allowed_games import active_manual_picks +from steam_backlog_enforcer.game_install import _echo +from steam_backlog_enforcer.steam_api import SteamAPIClient + +if TYPE_CHECKING: + from steam_backlog_enforcer.config import Config, State + +# How long the enforce daemon waits between achievement re-checks. That loop +# ticks every 3s; re-checking every tick would be ~1200 Steam API calls an hour +# for a fact that changes a few times a year. CLI commands do not throttle. +MANUAL_PICK_RECHECK_TTL_SECONDS = 900 + +logger = logging.getLogger(__name__) + + +class _DaemonSweep: + """The enforce daemon's re-check gate, plus what it has retired. + + An object rather than module-level values because rebinding those would + need a ``global`` statement; attribute writes do not. + + ``retired`` exists so the daemon can *record* a completion without + *evicting* the game: retiring drops a pick out of ``allowed_app_ids``, and + the daemon would then kill and uninstall it mid-session, seconds after the + last achievement popped. Unioning these ids back into the allowed set + defers that to a user-invoked command; the freed slot is visible at once. + """ + + def __init__(self) -> None: + self.last: float | None = None + self.retired: set[int] = set() + + def due(self, now: float, ttl: float) -> bool: + """Return whether a check is due, recording *now* when it is. + + Args: + now: Current ``time.monotonic()`` reading. + ttl: Minimum seconds between checks. + + Returns: + Whether the caller should run the check. + """ + if self.last is not None and now - self.last < ttl: + return False + self.last = now + return True + + +# Process-local by design: the daemon is long-lived, and a restart re-checking +# once immediately is harmless. +daemon_sweep = _DaemonSweep() + + +@dataclass(frozen=True) +class PickProgress: + """Achievement progress for one active manual pick. + + ``determinable`` is ``False`` when Steam returned nothing for this app (no + achievements, or the call failed); ``unlocked``/``total`` are 0 then, and + callers must read it as "unknown", never as "complete". + """ + + app_id: int + game_name: str + unlocked: int + total: int + retired: bool + determinable: bool + + def describe(self) -> str: + """Return a one-line human summary for CLI output.""" + if not self.determinable: + return f" {self.game_name}: progress unavailable (no achievements)" + if self.retired: + return ( + f" {self.game_name}: {self.unlocked}/{self.total} (100%)" + " - COMPLETE, freeing its slot" + ) + pct = (self.unlocked / self.total) * 100.0 if self.total else 0.0 + return ( + f" {self.game_name}: {self.unlocked}/{self.total}" + f" ({pct:.0f}%) - still in progress" + ) + + +def mark_finished(state: State, app_id: int) -> bool: + """Record *app_id* as finished, without duplicating an existing record. + + The single place a completion is written, so the manual-pick sweep and the + ``done``/``check`` paths cannot both append the same id. + + Args: + state: The enforcer state to mutate (not saved here). + app_id: The completed app id. + + Returns: + Whether this call actually added the id. + """ + if app_id in state.finished_app_ids: + return False + state.finished_app_ids.append(app_id) + return True + + +def retire_completed_manual_picks( + config: Config, + state: State, + *, + client: SteamAPIClient | None = None, +) -> list[PickProgress]: + """Mark every 100%-complete manual pick finished, freeing its slot. + + Mutates and saves ``state`` only when something was actually retired. Does + **not** assign a replacement game, uninstall anything, or touch + ``current_app_id``: the caller decides what a freed slot means. + + A pick whose achievements cannot be read stays locked: for an enforcement + tool, Steam being unreachable must never look like completion. + + Args: + config: Enforcer configuration (for the Steam credentials). + state: The enforcer state to inspect, mutate and save. + client: Pre-built API client, for reuse and for tests. + + Returns: + One entry per *active* pick, in pick order -- including the ones left + alone, so callers can report progress rather than only retirements. + """ + picks = active_manual_picks(state) + if not picks: + return [] + + if client is None: + client = SteamAPIClient(config.steam_api_key, config.steam_id) + + results: list[PickProgress] = [] + retired_any = False + for pick in picks: + # active_manual_picks has already dropped entries without an app_id. + app_id = pick["app_id"] + name = pick.get("game_name", "") or f"AppID={app_id}" + + game = client.refresh_single_game(app_id, name) + if game is None: + results.append( + PickProgress(app_id, name, 0, 0, retired=False, determinable=False) + ) + continue + + if game.is_complete and mark_finished(state, app_id): + retired_any = True + logger.info("Manual pick retired at 100%%: %s (AppID=%s)", name, app_id) + results.append( + PickProgress( + app_id, + name, + game.unlocked_achievements, + game.total_achievements, + retired=game.is_complete, + determinable=True, + ) + ) + + if retired_any: + state.save() + return results + + +def report_completion(config: Config, state: State) -> list[PickProgress]: + """Run the sweep and print a per-pick progress block. + + Args: + config: Enforcer configuration. + state: The enforcer state to inspect, mutate and save. + + Returns: + Only the picks whose slots this call freed. + """ + results = retire_completed_manual_picks(config, state) + if not results: + return [] + _echo("\nChecking your manual picks for completion...") + for result in results: + _echo(result.describe()) + return [r for r in results if r.retired] + + +def warn_stale_assignment(state: State, retired: list[PickProgress]) -> None: + """Warn when a just-retired pick is still the current assignment. + + Only reachable from the ``pick-manual`` abort path: on the success path + ``apply_manual_pick`` overwrites ``current_app_id`` with the new pick. + + Args: + state: The enforcer state (for ``current_app_id``). + retired: Picks whose slots were freed by this run. + """ + for pick in retired: + if pick.app_id == state.current_app_id: + _echo( + f"\nNote: {pick.game_name} is recorded as complete but is still" + "\n your current assignment. Run './run.sh done' to get" + "\n your next game." + ) + return + + +def retire_completed_manual_picks_throttled( + config: Config, + state: State, +) -> list[PickProgress]: + """Rate-limited :func:`retire_completed_manual_picks` for the enforce loop. + + Records what it retired in :data:`daemon_sweep` so the caller can keep + those games in the allowed set rather than evicting them unattended. + + Args: + config: Enforcer configuration. + state: The enforcer state to inspect, mutate and save. + + Returns: + The per-pick progress list, or ``[]`` when throttled. + """ + if not daemon_sweep.due(time.monotonic(), MANUAL_PICK_RECHECK_TTL_SECONDS): + return [] + results = retire_completed_manual_picks(config, state) + daemon_sweep.retired.update(r.app_id for r in results if r.retired) + return results diff --git a/steam_backlog_enforcer/_scanning_assign.py b/steam_backlog_enforcer/_scanning_assign.py index d5ab17e..6ecac1b 100644 --- a/steam_backlog_enforcer/_scanning_assign.py +++ b/steam_backlog_enforcer/_scanning_assign.py @@ -10,6 +10,7 @@ from datetime import UTC, datetime import logging from typing import TYPE_CHECKING +from steam_backlog_enforcer._actions import allowed_app_ids from steam_backlog_enforcer._scanning_candidates import ( _pick_next_shortest_candidate, _sort_key, @@ -84,7 +85,12 @@ def _assign_chosen_game( ) _report_poll_confidence(chosen, games, state) if config.uninstall_other_games: - count = uninstall_other_games(chosen.app_id) + # The whole allowed set, not just the new assignment: passing a bare + # int made the `in` test raise TypeError, and would tear down a + # concurrent manual pick even if it did not. Matches the other four + # call sites (main/install.py, _cmd_done_finalize.py, main/picks.py, + # _enforce_steps.py). state.current_app_id is already `chosen` here. + count = uninstall_other_games(allowed_app_ids(state)) if count: _echo(f"\n Uninstalled {count} non-assigned games") if not is_game_installed(chosen.app_id): diff --git a/steam_backlog_enforcer/_scanning_tampering.py b/steam_backlog_enforcer/_scanning_tampering.py index fe5a155..7f8abb4 100644 --- a/steam_backlog_enforcer/_scanning_tampering.py +++ b/steam_backlog_enforcer/_scanning_tampering.py @@ -9,6 +9,7 @@ from __future__ import annotations import logging from typing import TYPE_CHECKING, Any +from steam_backlog_enforcer._allowed_games import allowed_games from steam_backlog_enforcer.config import load_snapshot from steam_backlog_enforcer.enforcer import send_notification from steam_backlog_enforcer.game_install import _echo @@ -22,6 +23,30 @@ logger = logging.getLogger(__name__) _TAMPER_CHECK_LIMIT = 3 +def _legitimately_played(state: State) -> set[int]: + """Return every app id the user was allowed to earn achievements on. + + Wider than ``allowed_app_ids`` on purpose. A manual pick that has been + completed is no longer *allowed* -- it leaves the active set the moment it + lands in ``finished_app_ids`` -- but the achievements that finished it were + earned legitimately, so comparing it against a stale snapshot would report + the user's own completion as tampering. Every game ever manually picked, + plus every finished game, is therefore exempt. + + Args: + state: Current enforcer state. + + Returns: + App ids whose achievement progress must not be treated as suspicious. + """ + exempt = {app_id for app_id, _ in allowed_games(state)} + exempt.update(state.finished_app_ids) + exempt.update( + pick["app_id"] for pick in state.manual_picks if pick.get("app_id") is not None + ) + return exempt + + def _check_game_tampering( client: SteamAPIClient, entry: dict[str, Any], @@ -38,7 +63,7 @@ def _check_game_tampering( Tuple of (name, app_id, diff) if tampering detected, else None. """ app_id = entry["app_id"] - if app_id == state.current_app_id: + if app_id in _legitimately_played(state): return None if entry["unlocked_achievements"] >= entry["total_achievements"]: return None diff --git a/steam_backlog_enforcer/config.py b/steam_backlog_enforcer/config.py index c3f991c..67fd2c6 100644 --- a/steam_backlog_enforcer/config.py +++ b/steam_backlog_enforcer/config.py @@ -52,6 +52,10 @@ def _atomic_write(path: Path, data: str, *, mode: int | None = None) -> None: os.write(fd, data.encode("utf-8")) if mode is not None: os.fchmod(fd, mode) + # Keep the owner, or a root daemon write locks the user's CLI out. + with contextlib.suppress(OSError): # absent file / non-root: skip + original = path.stat() + os.fchown(fd, original.st_uid, original.st_gid) os.close(fd) tmp_path.replace(path) except BaseException: diff --git a/steam_backlog_enforcer/docs/DOCS-manual-pick-design.md b/steam_backlog_enforcer/docs/DOCS-manual-pick-design.md index b6886e0..a474841 100644 --- a/steam_backlog_enforcer/docs/DOCS-manual-pick-design.md +++ b/steam_backlog_enforcer/docs/DOCS-manual-pick-design.md @@ -60,7 +60,28 @@ releases only when every pick is finished or past its own 14-day deadline. assignment and keeps the lock. - `cmd_done` still finishes `current_app_id` and auto-picks a replacement, so the remaining picks stay queued. A finished pick leaves the active set - automatically via `finished_app_ids` — no pruning needed. + automatically via `finished_app_ids` — no pruning needed — but *something* + has to put the id there in the first place; see the next point. +- **Completion is checked per pick, not just for the assignment** + (`_pick_completion.py`). `finished_app_ids` used to gain ids only from + `cmd_done`/`do_check`, and both look at `current_app_id` alone — so a pick + that was not the current assignment had no achievement-based release path + and sat on its slot until the 14-day expiry. `retire_completed_manual_picks` + now re-checks every active pick and records the ones at 100%, so the cap can + never refuse on behalf of a game that is already finished. It runs from + `pick-manual`, `status`, `check`, the MCP `pick_manual` tool, and the enforce + loop (throttled to once every `MANUAL_PICK_RECHECK_TTL_SECONDS`, since that + loop ticks every 3s). + - A pick whose achievements cannot be read — the game has none, or Steam is + unreachable — stays locked. Unknown must never look like complete. + - Retiring frees the slot and nothing more: it never calls `pick_next_game`, + so freeing a slot cannot hand you an unrequested extra game. + - In the daemon it records without evicting. A retired pick leaves + `allowed_app_ids`, which would otherwise make the loop kill and uninstall + the game mid-session, seconds after the last achievement popped; the ids in + `daemon_sweep.retired` stay allowed until a user-invoked command acts. + - `mark_finished` is the single writer of `finished_app_ids`, so the sweep + and `done`/`check` cannot both record the same completion. **Deployment note:** the enforce daemon holds the allowed set in code, so `sudo systemctl restart steam-backlog-enforcer` is required after upgrading. diff --git a/steam_backlog_enforcer/main/picks.py b/steam_backlog_enforcer/main/picks.py index c422c3c..2e6db59 100644 --- a/steam_backlog_enforcer/main/picks.py +++ b/steam_backlog_enforcer/main/picks.py @@ -17,6 +17,10 @@ from steam_backlog_enforcer._allowed_games import ( ) from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids from steam_backlog_enforcer._hltb_types import load_hltb_cache +from steam_backlog_enforcer._pick_completion import ( + report_completion, + warn_stale_assignment, +) from steam_backlog_enforcer.config import load_snapshot from steam_backlog_enforcer.game_install import ( _echo, @@ -177,6 +181,8 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None: _echo(f"\nGame found: {game_name} (AppID={app_id})") + retired = report_completion(config, state) + existing = _report_pick_slots(config, state) _echo( @@ -189,12 +195,18 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None: f"\n {', '.join(sorted(_MANUAL_LOCK_EXEMPT_COMMANDS))}" f"\n - Stay undoable at any time via 'abandon-pick {app_id}'" ) + for done_pick in retired: + _echo( + f" - Uninstall {done_pick.game_name} (completed, no longer an" + " allowed game)" + ) _echo() confirm = input( f"Type YES to confirm you will play {game_name} until completion: " ).strip() if confirm != "YES": _echo("Aborted.") + warn_stale_assignment(state, retired) return # State mutation is the shared, stdout-free core (also used by the MCP diff --git a/steam_backlog_enforcer/main/status.py b/steam_backlog_enforcer/main/status.py index 118da2d..cbed022 100644 --- a/steam_backlog_enforcer/main/status.py +++ b/steam_backlog_enforcer/main/status.py @@ -8,6 +8,7 @@ from steam_backlog_enforcer._actions import ( active_manual_picks, manual_pick_age_days, ) +from steam_backlog_enforcer._pick_completion import report_completion from steam_backlog_enforcer._total_block import get_total_block_status from steam_backlog_enforcer.config import load_snapshot from steam_backlog_enforcer.game_install import ( @@ -23,7 +24,7 @@ if TYPE_CHECKING: from steam_backlog_enforcer.config import Config, State -def cmd_status(_config: Config, state: State) -> None: +def cmd_status(config: Config, state: State) -> None: """Show current status.""" _echo("=== Steam Backlog Enforcer ===\n") @@ -53,6 +54,16 @@ def cmd_status(_config: Config, state: State) -> None: is_assigned_installed = any(aid == state.current_app_id for aid, _ in installed) _echo(f"Assigned game installed: {is_assigned_installed}") + # Retire any pick that has hit 100% since we last looked, so the list + # below reflects reality. Deliberately non-fatal: status is the + # always-available diagnostic (it is exempt from both the total block and + # the manual-pick lock), so a Steam outage or an unwritable state file must + # degrade to a printed note, never to a failed status. + try: + report_completion(config, state) + except (OSError, RuntimeError, ValueError) as exc: + _echo(f"\n(Could not refresh pick completion: {exc})") + picks = active_manual_picks(state) if picks: _echo(f"\nManual picks ({len(picks)}):") diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index f9c5c17..c72f9ff 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -10,6 +10,7 @@ from steam_backlog_enforcer._hltb_types import ( load_hltb_count_comp_cache, load_hltb_polls_cache, ) +from steam_backlog_enforcer._pick_completion import mark_finished, report_completion from steam_backlog_enforcer._scanning_assign import ( _NO_CONF_MSG, _assign_chosen_game, @@ -205,6 +206,7 @@ def pick_next_game( def do_check(config: Config, state: State) -> None: """Check assigned game completion status; detect tampering.""" + report_completion(config, state) if state.current_app_id is None: _echo("No game currently assigned. Run 'scan' first.") return @@ -224,7 +226,7 @@ def do_check(config: Config, state: State) -> None: if game.is_complete: _echo(f"\n COMPLETED: {state.current_game_name}!") - state.finished_app_ids.append(state.current_app_id) + mark_finished(state, state.current_app_id) send_notification( "Game Complete!", f"You finished {state.current_game_name}! Picking next game...", diff --git a/steam_backlog_enforcer/tests/test_do_enforce.py b/steam_backlog_enforcer/tests/test_do_enforce.py index b3bee12..7ad8ac5 100644 --- a/steam_backlog_enforcer/tests/test_do_enforce.py +++ b/steam_backlog_enforcer/tests/test_do_enforce.py @@ -151,3 +151,32 @@ class TestDoEnforce: ): do_enforce(config, state) mock_iter.assert_not_called() + + +class TestStateReload: + """The loop must pick up CLI/MCP state changes without a restart.""" + + def test_reloads_manual_picks_from_disk(self) -> None: + # The MCP pick_manual tool adds a *second* pick without touching + # current_app_id. A daemon that never reloaded manual_picks would drop + # it out of the allowed set and uninstall it as unauthorized. + state = State(current_app_id=1, current_game_name="G") + fresh = State( + current_app_id=1, + current_game_name="G", + finished_app_ids=[7], + manual_picks=[{"app_id": 900, "game_name": "AddedByMcp", "started_at": ""}], + ) + with ( + patch(f"{PKG}._enforce_setup"), + patch(f"{PKG}._echo"), + patch.object(State, "load", return_value=fresh), + patch( + f"{PKG}._enforce_loop_iteration", + side_effect=KeyboardInterrupt, + ), + patch(f"{PKG}.time.sleep"), + ): + do_enforce(Config(), state) + assert state.manual_picks == fresh.manual_picks + assert state.finished_app_ids == [7] diff --git a/steam_backlog_enforcer/tests/test_main_abandon.py b/steam_backlog_enforcer/tests/test_main_abandon.py index 73d7f04..c5498fb 100644 --- a/steam_backlog_enforcer/tests/test_main_abandon.py +++ b/steam_backlog_enforcer/tests/test_main_abandon.py @@ -213,6 +213,10 @@ class TestAbandonOneOfTwoPicks: "steam_backlog_enforcer.main.status.get_installed_games", return_value=[], ), + patch( + "steam_backlog_enforcer.main.status.report_completion", + return_value=[], + ), patch("steam_backlog_enforcer.main.status._echo") as mock_echo, ): cmd_status(Config(), two_pick_state()) @@ -234,6 +238,10 @@ class TestAbandonOneOfTwoPicks: "steam_backlog_enforcer.main.status.get_installed_games", return_value=[], ), + patch( + "steam_backlog_enforcer.main.status.report_completion", + return_value=[], + ), patch("steam_backlog_enforcer.main.status._echo") as mock_echo, ): cmd_status(Config(), state) diff --git a/steam_backlog_enforcer/tests/test_main_pick_cap.py b/steam_backlog_enforcer/tests/test_main_pick_cap.py index d6a033e..6dff6e5 100644 --- a/steam_backlog_enforcer/tests/test_main_pick_cap.py +++ b/steam_backlog_enforcer/tests/test_main_pick_cap.py @@ -4,6 +4,7 @@ from unittest.mock import patch import pytest +from steam_backlog_enforcer._pick_completion import PickProgress from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.main import ( cmd_pick_manual, @@ -20,6 +21,7 @@ class TestPickManualCap: def test_second_pick_is_added_not_replacing(self) -> None: state = locked_state(app_id=100) with ( + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo"), patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), patch("builtins.input", return_value="YES"), @@ -34,6 +36,7 @@ class TestPickManualCap: def test_third_pick_refused_at_cap(self) -> None: state = two_pick_state() with ( + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), pytest.raises(SystemExit) as exc, @@ -49,6 +52,7 @@ class TestPickManualCap: # The cap has room, so this exercises the core's own refusal path. state = locked_state(app_id=489830) with ( + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), patch("builtins.input", return_value="YES"), @@ -68,6 +72,7 @@ class TestPickManualCap: # not uninstall or hide the first. state = locked_state(app_id=100) with ( + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo"), patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), patch("builtins.input", return_value="YES"), @@ -84,6 +89,7 @@ class TestPickManualCap: def test_installs_every_missing_allowed_game(self) -> None: state = locked_state(app_id=100) with ( + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo"), patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), patch("builtins.input", return_value="YES"), @@ -96,3 +102,28 @@ class TestPickManualCap: cmd_pick_manual(Config(max_manual_picks=2), state, ["489830"]) installed = {call.args[0] for call in mock_install.call_args_list} assert installed == {100, 489830} + + +class TestPickManualRetirementNotices: + """A pick retired by the sweep is announced before the YES prompt.""" + + def test_warns_about_uninstall_and_stale_assignment(self) -> None: + state = locked_state(app_id=100) + state.current_app_id = 200 + state.current_game_name = "Finished" + retired = [PickProgress(200, "Finished", 5, 5, retired=True, determinable=True)] + with ( + patch(f"{PKG}.report_completion", return_value=retired), + patch(f"{PKG}._echo") as mock_echo, + # warn_stale_assignment prints from the completion module. + patch("steam_backlog_enforcer._pick_completion._echo") as mock_note, + patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), + patch("builtins.input", return_value="no"), + ): + cmd_pick_manual(Config(max_manual_picks=2), state, ["489830"]) + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "Uninstall Finished (completed" in output + assert "Aborted." in output + # Abort leaves the finished game as the assignment, so say so. + note = " ".join(str(c) for c in mock_note.call_args_list) + assert "still" in note diff --git a/steam_backlog_enforcer/tests/test_main_status.py b/steam_backlog_enforcer/tests/test_main_status.py index e108d26..c6bd4cb 100644 --- a/steam_backlog_enforcer/tests/test_main_status.py +++ b/steam_backlog_enforcer/tests/test_main_status.py @@ -29,6 +29,7 @@ class TestCmdStatus: with ( patch(f"{PKG}.is_store_blocked", return_value=True), patch(f"{PKG}.get_installed_games", return_value=[(440, "TF2")]), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo"), ): cmd_status(Config(), state) @@ -37,6 +38,7 @@ class TestCmdStatus: with ( patch(f"{PKG}.is_store_blocked", return_value=False), patch(f"{PKG}.get_installed_games", return_value=[]), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo"), ): cmd_status(Config(), State()) @@ -48,6 +50,7 @@ class TestCmdList: def test_no_snapshot(self) -> None: with ( patch(f"{PKG}.load_snapshot", return_value=None), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, ): cmd_list(Config(), State()) @@ -62,6 +65,7 @@ class TestCmdList: state = State(current_app_id=1) with ( patch(f"{PKG}.load_snapshot", return_value=snapshot), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo"), ): cmd_list(Config(), state) @@ -70,6 +74,7 @@ class TestCmdList: snapshot = [snap(i, f"Game{i}") for i in range(60)] with ( patch(f"{PKG}.load_snapshot", return_value=snapshot), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, ): cmd_list(Config(), State()) @@ -82,6 +87,7 @@ class TestCmdStatusLockHint: with ( patch(f"{PKG}.is_store_blocked", return_value=False), patch(f"{PKG}.get_installed_games", return_value=[]), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, ): cmd_status(Config(), state) @@ -92,6 +98,7 @@ class TestCmdStatusLockHint: with ( patch(f"{PKG}.is_store_blocked", return_value=False), patch(f"{PKG}.get_installed_games", return_value=[]), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, ): cmd_status(Config(), State()) @@ -120,6 +127,7 @@ class TestCmdStatusTotalBlock: patch(f"{PKG}.get_total_block_status", return_value=ACTIVE_STATUS), patch(f"{PKG}.is_store_blocked", return_value=False), patch(f"{PKG}.get_installed_games", return_value=[]), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, ): cmd_status(Config(), State()) @@ -132,6 +140,7 @@ class TestCmdStatusTotalBlock: patch(f"{PKG}.get_total_block_status", return_value=INACTIVE_STATUS), patch(f"{PKG}.is_store_blocked", return_value=False), patch(f"{PKG}.get_installed_games", return_value=[]), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, ): cmd_status(Config(), State()) @@ -146,6 +155,7 @@ class TestCmdStatusTotalBlock: patch(f"{PKG}.get_total_block_status", return_value=status), patch(f"{PKG}.is_store_blocked", return_value=False), patch(f"{PKG}.get_installed_games", return_value=[]), + patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, ): cmd_status(Config(), State()) @@ -157,3 +167,19 @@ class TestCmdStatusTotalBlock: # ────────────────────────────────────────────────────────────── # main() dispatch to the daily-gaming-budget commands # ────────────────────────────────────────────────────────────── + + +class TestCmdStatusCompletionSweepIsNonFatal: + """status stays usable when the completion sweep cannot run.""" + + def test_reports_failure_and_still_renders(self) -> None: + with ( + patch(f"{PKG}.report_completion", side_effect=OSError("state read-only")), + patch(f"{PKG}.is_store_blocked", return_value=True), + patch(f"{PKG}.get_installed_games", return_value=[]), + patch(f"{PKG}._echo") as mock_echo, + ): + cmd_status(Config(), State(current_app_id=440, current_game_name="TF2")) + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "Could not refresh pick completion: state read-only" in output + assert "Store blocked" in output diff --git a/steam_backlog_enforcer/tests/test_pick_completion.py b/steam_backlog_enforcer/tests/test_pick_completion.py new file mode 100644 index 0000000..ccdc18f --- /dev/null +++ b/steam_backlog_enforcer/tests/test_pick_completion.py @@ -0,0 +1,230 @@ +"""Tests for achievement-based retirement of finished manual picks.""" + +from unittest.mock import MagicMock, patch + +from steam_backlog_enforcer._pick_completion import ( + MANUAL_PICK_RECHECK_TTL_SECONDS, + PickProgress, + _DaemonSweep, + daemon_sweep, + mark_finished, + report_completion, + retire_completed_manual_picks, + retire_completed_manual_picks_throttled, + warn_stale_assignment, +) +from steam_backlog_enforcer.config import Config, State +from steam_backlog_enforcer.steam_api import GameInfo +from steam_backlog_enforcer.tests._main_helpers import locked_state, two_pick_state + +PKG = "steam_backlog_enforcer._pick_completion" + + +def game(app_id: int, unlocked: int, total: int) -> GameInfo: + """Build a GameInfo with the given achievement progress. + + Args: + app_id: Steam application id. + unlocked: Unlocked achievement count. + total: Total achievement count. + + Returns: + A GameInfo carrying that progress. + """ + return GameInfo( + app_id=app_id, + name="G", + total_achievements=total, + unlocked_achievements=unlocked, + playtime_minutes=0, + ) + + +def client_returning(*games: GameInfo | None) -> MagicMock: + """Build a fake SteamAPIClient yielding *games* in order. + + Args: + games: What each ``refresh_single_game`` call should return. + + Returns: + A MagicMock standing in for SteamAPIClient. + """ + fake = MagicMock() + fake.refresh_single_game.side_effect = list(games) + return fake + + +class TestMarkFinished: + def test_records_once(self) -> None: + state = State() + assert mark_finished(state, 7) is True + assert state.finished_app_ids == [7] + + def test_does_not_duplicate(self) -> None: + state = State(finished_app_ids=[7]) + assert mark_finished(state, 7) is False + assert state.finished_app_ids == [7] + + +class TestPickProgressDescribe: + def test_undeterminable(self) -> None: + line = PickProgress(1, "G", 0, 0, retired=False, determinable=False).describe() + assert "progress unavailable" in line + + def test_retired(self) -> None: + line = PickProgress(1, "G", 5, 5, retired=True, determinable=True).describe() + assert "COMPLETE, freeing its slot" in line + + def test_in_progress(self) -> None: + line = PickProgress(1, "G", 1, 4, retired=False, determinable=True).describe() + assert "1/4 (25%) - still in progress" in line + + def test_in_progress_with_zero_total(self) -> None: + # Defensive: determinable but no achievements must not divide by zero. + line = PickProgress(1, "G", 0, 0, retired=False, determinable=True).describe() + assert "(0%)" in line + + +class TestRetireCompletedManualPicks: + def test_no_picks_makes_no_api_call(self) -> None: + fake = client_returning() + assert retire_completed_manual_picks(Config(), State(), client=fake) == [] + fake.refresh_single_game.assert_not_called() + + def test_complete_pick_is_retired_and_saved(self) -> None: + state = locked_state(app_id=100) + with patch.object(State, "save") as mock_save: + results = retire_completed_manual_picks( + Config(), state, client=client_returning(game(100, 5, 5)) + ) + assert [r.retired for r in results] == [True] + assert state.finished_app_ids == [100] + mock_save.assert_called_once() + + def test_incomplete_pick_is_left_alone(self) -> None: + state = locked_state(app_id=100) + with patch.object(State, "save") as mock_save: + results = retire_completed_manual_picks( + Config(), state, client=client_returning(game(100, 1, 5)) + ) + assert [r.retired for r in results] == [False] + assert state.finished_app_ids == [] + mock_save.assert_not_called() + + def test_unreadable_pick_stays_locked(self) -> None: + # No achievements at all, or Steam unreachable: both surface as None. + state = locked_state(app_id=100) + with patch.object(State, "save") as mock_save: + results = retire_completed_manual_picks( + Config(), state, client=client_returning(None) + ) + assert results == [ + PickProgress(100, "TestGame", 0, 0, retired=False, determinable=False) + ] + assert state.finished_app_ids == [] + mock_save.assert_not_called() + + def test_reports_every_pick_not_only_retirements(self) -> None: + state = two_pick_state() + with patch.object(State, "save"): + results = retire_completed_manual_picks( + Config(), + state, + client=client_returning(game(100, 1, 5), game(200, 5, 5)), + ) + assert [(r.app_id, r.retired) for r in results] == [(100, False), (200, True)] + + def test_pick_without_app_id_is_skipped(self) -> None: + state = State(manual_picks=[{"game_name": "Broken", "started_at": ""}]) + fake = client_returning() + assert retire_completed_manual_picks(Config(), state, client=fake) == [] + fake.refresh_single_game.assert_not_called() + + def test_unnamed_pick_falls_back_to_app_id(self) -> None: + state = State(manual_picks=[{"app_id": 100, "game_name": "", "started_at": ""}]) + results = retire_completed_manual_picks( + Config(), state, client=client_returning(game(100, 1, 5)) + ) + assert results[0].game_name == "AppID=100" + + def test_builds_its_own_client_when_none_given(self) -> None: + state = locked_state(app_id=100) + fake = client_returning(game(100, 1, 5)) + with patch(f"{PKG}.SteamAPIClient", return_value=fake) as mock_cls: + retire_completed_manual_picks( + Config(steam_api_key="k", steam_id="i"), state + ) + mock_cls.assert_called_once_with("k", "i") + + +class TestReportCompletion: + def test_prints_nothing_without_picks(self) -> None: + with patch(f"{PKG}._echo") as mock_echo: + assert report_completion(Config(), State()) == [] + mock_echo.assert_not_called() + + def test_prints_each_pick_and_returns_only_retirements(self) -> None: + state = two_pick_state() + with ( + patch.object(State, "save"), + patch( + f"{PKG}.SteamAPIClient", + return_value=client_returning(game(100, 1, 5), game(200, 5, 5)), + ), + patch(f"{PKG}._echo") as mock_echo, + ): + retired = report_completion(Config(), state) + assert [r.app_id for r in retired] == [200] + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "still in progress" in output + assert "COMPLETE, freeing its slot" in output + + +class TestWarnStaleAssignment: + def test_warns_when_retired_pick_is_current(self) -> None: + state = State(current_app_id=200, current_game_name="SecondGame") + retired = [ + PickProgress(200, "SecondGame", 5, 5, retired=True, determinable=True) + ] + with patch(f"{PKG}._echo") as mock_echo: + warn_stale_assignment(state, retired) + assert "still" in " ".join(str(c) for c in mock_echo.call_args_list) + + def test_silent_when_current_is_something_else(self) -> None: + state = State(current_app_id=999) + retired = [PickProgress(200, "G", 5, 5, retired=True, determinable=True)] + with patch(f"{PKG}._echo") as mock_echo: + warn_stale_assignment(state, retired) + mock_echo.assert_not_called() + + +class TestDaemonSweep: + def test_due_then_throttled(self) -> None: + sweep = _DaemonSweep() + assert sweep.due(1000.0, 900.0) is True + assert sweep.due(1100.0, 900.0) is False + assert sweep.due(2000.0, 900.0) is True + + def test_throttled_call_records_retirements_without_evicting(self) -> None: + state = locked_state(app_id=100) + daemon_sweep.last = None + daemon_sweep.retired.clear() + try: + with ( + patch.object(State, "save"), + patch( + f"{PKG}.SteamAPIClient", + return_value=client_returning(game(100, 5, 5)), + ), + ): + results = retire_completed_manual_picks_throttled(Config(), state) + assert [r.retired for r in results] == [True] + assert daemon_sweep.retired == {100} + # Second call inside the TTL must not re-query Steam. + assert retire_completed_manual_picks_throttled(Config(), state) == [] + finally: + daemon_sweep.last = None + daemon_sweep.retired.clear() + + def test_ttl_is_well_clear_of_the_three_second_loop(self) -> None: + assert MANUAL_PICK_RECHECK_TTL_SECONDS >= 300 diff --git a/steam_backlog_enforcer/tests/test_scanning_part2.py b/steam_backlog_enforcer/tests/test_scanning_part2.py index 9c2aadc..d31bb9c 100644 --- a/steam_backlog_enforcer/tests/test_scanning_part2.py +++ b/steam_backlog_enforcer/tests/test_scanning_part2.py @@ -38,6 +38,48 @@ class TestCheckGameTampering: result = _check_game_tampering(MagicMock(), _entry(app_id=1), state) assert result is None + def test_finished_manual_pick_is_not_tampering(self) -> None: + # The completion that retired the pick is legitimate progress; the + # snapshot is simply stale. Reported as TAMPERING before this exemption. + client = MagicMock() + game = MagicMock() + game.unlocked_achievements = 16 + client.refresh_single_game.return_value = game + state = State( + current_app_id=475150, + finished_app_ids=[588730], + manual_picks=[ + {"app_id": 588730, "game_name": "Majotori", "started_at": ""} + ], + ) + entry = _entry(app_id=588730, name="Majotori", total=16, unlocked=13) + assert _check_game_tampering(client, entry, state) is None + + def test_active_second_pick_is_not_tampering(self) -> None: + # A second manual pick is allowed but is not current_app_id. + client = MagicMock() + game = MagicMock() + game.unlocked_achievements = 9 + client.refresh_single_game.return_value = game + state = State( + current_app_id=475150, + manual_picks=[ + {"app_id": 412830, "game_name": "STEINS;GATE", "started_at": ""} + ], + ) + entry = _entry(app_id=412830, name="STEINS;GATE", total=42, unlocked=5) + assert _check_game_tampering(client, entry, state) is None + + def test_unrelated_game_still_flagged(self) -> None: + # The detector must keep working for games never allowed. + client = MagicMock() + game = MagicMock() + game.unlocked_achievements = 8 + client.refresh_single_game.return_value = game + state = State(current_app_id=475150, finished_app_ids=[588730]) + entry = _entry(app_id=813780, name="AoE2", unlocked=5) + assert _check_game_tampering(client, entry, state) == ("AoE2", 813780, 3) + def test_already_complete_skipped(self) -> None: state = State() result = _check_game_tampering( diff --git a/steam_backlog_enforcer/tests/test_scanning_pick_next_game.py b/steam_backlog_enforcer/tests/test_scanning_pick_next_game.py index 00bd4b3..26d4615 100644 --- a/steam_backlog_enforcer/tests/test_scanning_pick_next_game.py +++ b/steam_backlog_enforcer/tests/test_scanning_pick_next_game.py @@ -2,13 +2,18 @@ from __future__ import annotations +from typing import Any from unittest.mock import patch +import pytest + from steam_backlog_enforcer.config import Config, State +from steam_backlog_enforcer.game_uninstall import uninstall_other_games from steam_backlog_enforcer.scanning import ( pick_next_game, ) from steam_backlog_enforcer.steam_api import GameInfo +from steam_backlog_enforcer.tests._main_helpers import STARTED_AT def _game( @@ -156,3 +161,75 @@ class TestPickNextGame: ): pick_next_game([g1], state, config) mock_install.assert_called_once() + + +class TestAssignUninstallsAgainstTheAllowedSet: + """_assign_chosen_game must pass the allowed *set*, never a bare app id.""" + + def test_passes_set_including_concurrent_manual_pick(self) -> None: + g = _game(app_id=2, name="Short", hours=10.0) + config = Config(steam_api_key="k", steam_id="i") + state = State( + manual_picks=[ + {"app_id": 900, "game_name": "Pick", "started_at": STARTED_AT} + ], + ) + with ( + patch( + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", + side_effect=lambda c: c[0] if c else None, + ), + patch("steam_backlog_enforcer.scanning._echo"), + patch("steam_backlog_enforcer._scanning_confidence._echo"), + patch( + "steam_backlog_enforcer._scanning_assign.is_game_installed", + return_value=True, + ), + patch( + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", + return_value=0, + ) as mock_uninstall, + patch("builtins.input", return_value="1"), + ): + pick_next_game([g], state, config) + (allowed,) = mock_uninstall.call_args.args + assert isinstance(allowed, set) + # The new assignment survives, and so does the untouched manual pick. + assert allowed == {2, 900} + + def test_real_uninstall_accepts_that_argument(self) -> None: + """Guards the contract the mock above cannot: a bare int raises. + + ``uninstall_other_games`` membership-tests every installed game against + its argument, so an int made it raise TypeError as soon as anything was + installed. The suite hid that because no games are installed by default. + """ + with ( + patch( + "steam_backlog_enforcer.game_uninstall.get_installed_games", + return_value=[(2, "Short"), (900, "Pick"), (55, "Other")], + ), + patch( + "steam_backlog_enforcer.game_uninstall.is_protected_app", + return_value=False, + ), + patch( + "steam_backlog_enforcer.game_uninstall.uninstall_game", + return_value=True, + ) as mock_rm, + ): + assert uninstall_other_games({2, 900}) == 1 + assert [c.args[0] for c in mock_rm.call_args_list] == [55] + + # Typed as Any so the wrong-type call is a runtime check, not a + # suppressed type error. + bare_app_id: Any = 2 + with ( + patch( + "steam_backlog_enforcer.game_uninstall.get_installed_games", + return_value=[(2, "Short")], + ), + patch("steam_backlog_enforcer.game_uninstall.uninstall_game"), + pytest.raises(TypeError), + ): + uninstall_other_games(bare_app_id)