diff --git a/steam_backlog_enforcer/_actions.py b/steam_backlog_enforcer/_actions.py index 54c2927..43668ec 100644 --- a/steam_backlog_enforcer/_actions.py +++ b/steam_backlog_enforcer/_actions.py @@ -23,6 +23,15 @@ if TYPE_CHECKING: # truth: both ``main.py`` and the MCP server import it from here. MANUAL_LOCK_DAYS = 14 +# Mistake-correction window: for this many days after a manual pick the user +# may still back out via ``abandon-pick``. Deliberately short — it exists to +# undo a wrong pick, not to serve as an escape hatch from the lock. +MANUAL_GRACE_DAYS = 4 + +# How long an abandoned pick stays out of the auto-assignment pool, so that +# ``scan`` does not immediately hand back the game the user just rejected. +ABANDON_COOLDOWN_DAYS = 30 + def is_manual_pick_locked(state: State) -> bool: """Return ``True`` if the manual-pick lock is currently in force. @@ -82,6 +91,76 @@ def apply_manual_pick(state: State, app_id: int, game_name: str) -> None: state.save() +def manual_pick_grace_remaining(state: State) -> float | None: + """Return days left in the manual-pick grace window, or ``None``. + + ``None`` means "no grace window applies": there is no manual pick, or its + timestamp is missing/malformed. A value <= 0 means the window has closed. + + Args: + state: The loaded enforcer state. + + Returns: + Fractional days remaining before the pick can no longer be abandoned, + or ``None`` when the question does not apply. + """ + if state.manual_pick_app_id is None or not state.manual_pick_started_at: + return None + try: + started = datetime.fromisoformat(state.manual_pick_started_at) + except ValueError: + return None + deadline = started + timedelta(days=MANUAL_GRACE_DAYS) + return (deadline - datetime.now(timezone.utc)).total_seconds() / 86400 + + +def can_abandon_manual_pick(state: State) -> bool: + """Return ``True`` if the current manual pick is still inside its grace window. + + Args: + state: The loaded enforcer state. + + Returns: + Whether ``abandon_manual_pick`` would be accepted right now. + """ + remaining = manual_pick_grace_remaining(state) + return remaining is not None and remaining > 0 + + +def abandon_manual_pick(state: State) -> bool: + """Drop the manual pick and persist ``state``, if still inside the grace window. + + Clears both the manual-pick lock and the game assignment, and puts the + abandoned app id on the existing skip cooldown so ``scan`` will not hand + the same game straight back. State-only: like ``apply_manual_pick`` this + deliberately performs no uninstall/hide cascade, so the MCP server can call + it without touching the filesystem. + + Args: + state: The enforcer state to mutate and save. + + Returns: + ``True`` if the pick was abandoned, ``False`` if the grace window has + closed (or there was no pick), in which case ``state`` is untouched. + """ + if not can_abandon_manual_pick(state): + return False + + app_id = state.manual_pick_app_id + if app_id is not None: + state.skip_for_days(app_id, ABANDON_COOLDOWN_DAYS) + + state.manual_pick_app_id = None + state.manual_pick_game_name = "" + state.manual_pick_started_at = "" + # The pick was also the assignment; clearing it lets 'scan' reassign. + if state.current_app_id == app_id: + state.current_app_id = None + state.current_game_name = "" + state.save() + return True + + def status_payload(state: State) -> dict[str, Any]: """Build the structured status snapshot that ``cmd_status`` renders as text. @@ -116,4 +195,5 @@ def status_payload(state: State) -> dict[str, Any]: "until": total_block.until.isoformat() if total_block.until else None, }, "manual_pick_locked": is_manual_pick_locked(state), + "manual_pick_grace_days_left": manual_pick_grace_remaining(state), } diff --git a/steam_backlog_enforcer/_cmd_done.py b/steam_backlog_enforcer/_cmd_done.py index 644dd98..af16a22 100644 --- a/steam_backlog_enforcer/_cmd_done.py +++ b/steam_backlog_enforcer/_cmd_done.py @@ -24,7 +24,7 @@ from steam_backlog_enforcer.hltb import ( load_hltb_polls_cache, save_hltb_cache, ) -from steam_backlog_enforcer.library_hider import hide_other_games +from steam_backlog_enforcer.library_hider import try_hide_other_games from steam_backlog_enforcer.scanning import pick_next_game from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient @@ -212,8 +212,10 @@ def _finalize_completion( owned_ids = get_all_owned_app_ids(config) if owned_ids: - hidden = hide_other_games(owned_ids, state.current_app_id) - if hidden > 0: + hidden, skipped = try_hide_other_games(owned_ids, state.current_app_id) + if skipped is not None: + _echo(f"\n Library hiding: skipped ({skipped})") + elif hidden > 0: _echo(f"\n Library: hid {hidden} games") if not is_game_installed(state.current_app_id): @@ -276,8 +278,10 @@ def _enforce_on_done(config: Config, state: State) -> None: # assigned game hidden and stale games visible. owned_ids = get_all_owned_app_ids(config) if owned_ids: - hidden = hide_other_games(owned_ids, state.current_app_id) - if hidden > 0: + hidden, skipped = try_hide_other_games(owned_ids, state.current_app_id) + if skipped is not None: + _echo(f" Library hiding: skipped ({skipped})") + elif hidden > 0: _echo(f" Library: hid {hidden} games") diff --git a/steam_backlog_enforcer/_enforce_loop.py b/steam_backlog_enforcer/_enforce_loop.py index da651bb..4ee5307 100644 --- a/steam_backlog_enforcer/_enforce_loop.py +++ b/steam_backlog_enforcer/_enforce_loop.py @@ -38,9 +38,8 @@ from steam_backlog_enforcer.game_install import ( uninstall_other_games, ) from steam_backlog_enforcer.library_hider import ( - SteamUnavailableError, - hide_other_games, steam_is_installed, + try_hide_other_games, ) from steam_backlog_enforcer.steam_api import SteamAPIClient from steam_backlog_enforcer.store_blocker import block_store @@ -246,10 +245,9 @@ def _enforce_hide_games(config: Config, state: State) -> None: # which spun the service through ~1000 restarts against a Steam that had # been uninstalled - each attempt leaving a dead process named "steam" # behind that /proc scanners misread as a live Steam. - try: - hidden = hide_other_games(owned_ids, state.current_app_id) - except SteamUnavailableError as exc: - _echo(f" Library hiding: skipped ({exc})") + hidden, skipped = try_hide_other_games(owned_ids, state.current_app_id) + if skipped is not None: + _echo(f" Library hiding: skipped ({skipped})") return if hidden > 0: diff --git a/steam_backlog_enforcer/_mcp.py b/steam_backlog_enforcer/_mcp.py index 739330d..c44bd6e 100644 --- a/steam_backlog_enforcer/_mcp.py +++ b/steam_backlog_enforcer/_mcp.py @@ -32,8 +32,13 @@ from typing import Any from mcp.server.fastmcp import FastMCP from steam_backlog_enforcer._actions import ( + ABANDON_COOLDOWN_DAYS, + MANUAL_GRACE_DAYS, MANUAL_LOCK_DAYS, + abandon_manual_pick, apply_manual_pick, + can_abandon_manual_pick, + manual_pick_grace_remaining, status_payload, ) from steam_backlog_enforcer._total_block import start_total_block @@ -209,6 +214,69 @@ def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]: } +@mcp.tool() +def abandon_pick(app_id: int, *, confirm: bool = False) -> dict[str, Any]: + """Undo a manual pick inside its grace window (gated write). + + With ``confirm=False`` (the default) this performs **no** mutation and + returns a preview. Only works within ``MANUAL_GRACE_DAYS`` days of the + pick. Like ``pick_manual``, this mutates **state only** — the CLI's + ``abandon-pick`` additionally uninstalls the abandoned game. + + Args: + app_id: The manually-picked app id to back out of. + confirm: Set ``True`` to actually abandon the pick; otherwise preview. + """ + state = State.load() + if state.manual_pick_app_id is None: + return {"ok": False, "reason": "No manual pick is active."} + if app_id != state.manual_pick_app_id: + return { + "ok": False, + "reason": ( + f"AppID={app_id} is not the active manual pick " + f"(AppID={state.manual_pick_app_id})." + ), + } + + remaining = manual_pick_grace_remaining(state) + if not can_abandon_manual_pick(state): + return { + "ok": False, + "reason": ( + f"The {MANUAL_GRACE_DAYS}-day grace period has expired; " + "the pick can no longer be abandoned." + ), + } + + game_name = state.manual_pick_game_name + if not confirm: + return { + "ok": True, + "preview": True, + "action": "abandon_pick", + "app_id": app_id, + "game_name": game_name, + "grace_days_left": remaining, + "effect": ( + "Releases the manual pick lock, clears the assignment, and " + f"keeps the game out of auto-assignment for " + f"{ABANDON_COOLDOWN_DAYS} days." + ), + "confirm_required": True, + } + + abandon_manual_pick(state) + logger.info("abandon_pick applied: %s (AppID=%s)", game_name, app_id) + return { + "ok": True, + "applied": True, + "action": "abandon_pick", + "app_id": app_id, + "game_name": game_name, + } + + @mcp.tool() def block_gaming(days: int, *, confirm: bool = False) -> dict[str, Any]: """Start a total gaming block for ``days`` days (gated, privileged write). diff --git a/steam_backlog_enforcer/docs/pick_game_manually_design.md b/steam_backlog_enforcer/docs/pick_game_manually_design.md index cd7fd8a..5aa9b58 100644 --- a/steam_backlog_enforcer/docs/pick_game_manually_design.md +++ b/steam_backlog_enforcer/docs/pick_game_manually_design.md @@ -17,3 +17,24 @@ Logic should be as follows: test the functionality with 489830 (The Elder Scrolls V: Skyrim Special Edition) as always first write full functionality confirm that it works alone and with the user and only AFTER that write tests and coverage and fix issues + +## Grace period (added 2026-07-19) + +A manual pick can be a mistake, and with no way out the user is stuck for the +full 2 weeks. There is now a short mistake-correction window: + +- `MANUAL_GRACE_DAYS = 4` — for 4 days after the pick, `abandon-pick ` + backs out of it. Outside that window the command refuses and exits 1. +- The app_id must be passed explicitly and must match the active pick, so an + abandon cannot be triggered by muscle memory. +- `abandon-pick` is in `_MANUAL_LOCK_EXEMPT_COMMANDS` — otherwise the + pre-dispatch lock check in `main()` would block the only way out. +- Abandoning clears the lock **and** the assignment, uninstalls the game, and + puts it on the existing `skipped_until` cooldown for + `ABANDON_COOLDOWN_DAYS = 30` so `scan` does not hand it straight back. +- `_actions.abandon_manual_pick` is state-only (no uninstall), matching the + `apply_manual_pick` rule, so the MCP `abandon_pick` tool can reuse it. Both + MCP tools stay gated behind `confirm=True`. + +The lock-active message advertises `abandon-pick` only while the window is +still open, and the `pick-manual` warning mentions it up front. diff --git a/steam_backlog_enforcer/library_hider.py b/steam_backlog_enforcer/library_hider.py index a4063de..d20f1a8 100644 --- a/steam_backlog_enforcer/library_hider.py +++ b/steam_backlog_enforcer/library_hider.py @@ -394,6 +394,35 @@ def hide_other_games( return count +def try_hide_other_games( + owned_app_ids: list[int], + allowed_app_id: int | None, +) -> tuple[int, str | None]: + """Hide other games, degrading gracefully when Steam cannot be driven. + + An unreachable Steam (not running, no debug port) or a deferred restart + (a game update is in flight) is never fatal: there is simply no library to + reconcile this pass. Every interactive command should skip and carry on + rather than abort with a traceback, so this wrapper turns the + :class:`SteamUnavailableError` family into a return value. + + Stdout-free on purpose — callers phrase the skip message themselves. + + Args: + owned_app_ids: All owned app ids, used to seed the first hide pass. + allowed_app_id: The one game that must stay visible, if any. + + Returns: + ``(hidden_count, skip_reason)``. ``skip_reason`` is ``None`` when the + pass ran; otherwise it explains why hiding was skipped and the count + is 0. + """ + try: + return hide_other_games(owned_app_ids, allowed_app_id), None + except SteamUnavailableError as exc: + return 0, str(exc) + + def unhide_all_games(owned_app_ids: list[int]) -> int: """Remove all games from the hidden collection. diff --git a/steam_backlog_enforcer/main.py b/steam_backlog_enforcer/main.py index 72438cb..a0e9d47 100644 --- a/steam_backlog_enforcer/main.py +++ b/steam_backlog_enforcer/main.py @@ -7,11 +7,20 @@ import logging import sys from typing import TYPE_CHECKING +from steam_backlog_enforcer._actions import ( + ABANDON_COOLDOWN_DAYS as _ABANDON_COOLDOWN_DAYS, +) +from steam_backlog_enforcer._actions import ( + MANUAL_GRACE_DAYS as _MANUAL_GRACE_DAYS, +) from steam_backlog_enforcer._actions import ( MANUAL_LOCK_DAYS as _MANUAL_LOCK_DAYS, ) from steam_backlog_enforcer._actions import ( + abandon_manual_pick, apply_manual_pick, + can_abandon_manual_pick, + manual_pick_grace_remaining, ) from steam_backlog_enforcer._actions import ( is_manual_pick_locked as _is_manual_pick_locked, @@ -46,11 +55,12 @@ from steam_backlog_enforcer.game_install import ( install_game, is_game_installed, is_protected_app, + uninstall_game, uninstall_other_games, ) from steam_backlog_enforcer.library_hider import ( - hide_other_games, restart_steam, + try_hide_other_games, unhide_all_games, ) from steam_backlog_enforcer.scanning import ( @@ -82,7 +92,7 @@ _MIN_CLI_ARGS = 2 # Principle: only what is needed to release the lock (done/check) or # that cannot change the game assignment (status, enforce, setup, serve). _MANUAL_LOCK_EXEMPT_COMMANDS = frozenset( - {"done", "check", "status", "enforce", "setup", "serve"} + {"done", "check", "status", "enforce", "setup", "serve", "abandon-pick"} ) # Commands that remain usable while a total gaming block is active. Far @@ -158,8 +168,24 @@ def _show_manual_pick_lock_message(state: State) -> None: "\nYou CANNOT use any other feature until you finish this game" "\n(100% achievements) or the 2-week deadline passes." "\n\nTo release the lock: finish the game, then run 'done' or 'check'." - f"\n\nAllowed commands: {', '.join(sorted(_MANUAL_LOCK_EXEMPT_COMMANDS))}" ) + + # Mistake-correction window: only advertised while it is still open, and + # once closed 'abandon-pick' is dropped from the allowed list rather than + # offered as a command that would only refuse. + grace_left = manual_pick_grace_remaining(state) + in_grace = grace_left is not None and grace_left > 0 + if in_grace: + _echo( + f"\nPicked this by mistake? You have {grace_left:.1f} day(s) of the" + f"\n{_MANUAL_GRACE_DAYS}-day grace period left:" + f"\n abandon-pick {state.manual_pick_app_id}" + ) + + usable = set(_MANUAL_LOCK_EXEMPT_COMMANDS) + if not in_grace: + usable.discard("abandon-pick") + _echo(f"\nAllowed commands: {', '.join(sorted(usable))}") _echo("=" * 60 + "\n") @@ -442,7 +468,10 @@ def cmd_hide(config: Config, state: State) -> None: return _echo(f"Hiding all games except {state.current_game_name}...") - hidden = hide_other_games(owned_ids, state.current_app_id) + hidden, skipped = try_hide_other_games(owned_ids, state.current_app_id) + if skipped is not None: + _echo(f"Library hiding: skipped ({skipped})") + return _echo(f"Hidden {hidden} games.") if hidden > 0: @@ -482,8 +511,10 @@ def cmd_pick(config: Config, state: State) -> None: if state.current_app_id is not None: owned_ids = get_all_owned_app_ids(config) if owned_ids: - hidden = hide_other_games(owned_ids, state.current_app_id) - if hidden > 0: + hidden, skipped = try_hide_other_games(owned_ids, state.current_app_id) + if skipped is not None: + _echo(f"\n Library hiding: skipped ({skipped})") + elif hidden > 0: _echo(f"\n Library: hid {hidden} games") @@ -548,8 +579,10 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None: f"\n - Override your current assignment" f"\n - Lock ALL other commands for {_MANUAL_LOCK_DAYS} DAYS or until" f"\n you reach 100% achievements" - f"\n - Only 'done', 'check', 'status', 'enforce', 'setup', 'serve'" - f"\n will remain usable during this period" + f"\n - Leave only these commands usable:" + f"\n {', '.join(sorted(_MANUAL_LOCK_EXEMPT_COMMANDS))}" + f"\n - Stay undoable via 'abandon-pick {app_id}' for the first" + f"\n {_MANUAL_GRACE_DAYS} days" ) _echo() confirm = input( @@ -582,11 +615,122 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None: owned_ids = get_all_owned_app_ids(config) if owned_ids: - hidden = hide_other_games(owned_ids, app_id) - if hidden > 0: + hidden, skipped = try_hide_other_games(owned_ids, app_id) + if skipped is not None: + _echo(f" Library hiding: skipped ({skipped})") + elif hidden > 0: _echo(f" Library: hid {hidden} games") +_ABANDON_PICK_USAGE = ( + "Usage: abandon-pick \n" + " app_id : the manually-picked game you want to back out of.\n\n" + f"Only works within {_MANUAL_GRACE_DAYS} days of the pick." +) + + +def _abandon_pick_target(state: State, args: list[str]) -> int | None: + """Validate the abandon-pick argument against the active manual pick. + + Requiring the explicit app_id (rather than defaulting to the current pick) + makes an accidental abandon impossible to trigger by muscle memory. + + Args: + state: Current enforcer state. + args: Remaining CLI args (first element should be the app_id). + + Returns: + The validated app id, or ``None`` if the input was unusable (a message + has already been printed in that case). + """ + if not args: + _echo(_ABANDON_PICK_USAGE) + return None + + try: + app_id = int(args[0]) + except ValueError: + _echo(f"Error: app_id must be a number, got '{args[0]}'.") + return None + + if state.manual_pick_app_id is None: + _echo("No manual pick is active — nothing to abandon.") + return None + + if app_id != state.manual_pick_app_id: + _echo( + f"Error: AppID={app_id} is not your manual pick.\n" + f"Current pick: {state.manual_pick_game_name}" + f" (AppID={state.manual_pick_app_id})." + ) + return None + + return app_id + + +def cmd_abandon_pick(_config: Config, state: State, args: list[str]) -> None: + """Back out of a manual pick while still inside the grace period. + + Args: + _config: Enforcer configuration (unused, kept for dispatch symmetry). + state: Current enforcer state. + args: Remaining CLI args (first element should be the app_id). + """ + app_id = _abandon_pick_target(state, args) + if app_id is None: + sys.exit(1) + + game_name = state.manual_pick_game_name + + if not can_abandon_manual_pick(state): + # ``remaining`` is negative once the window has closed, so its + # magnitude is how long ago that happened. + remaining = manual_pick_grace_remaining(state) + elapsed = ( + f"{abs(remaining):.1f} day(s) ago" + if remaining is not None + else "at an unknown time" + ) + _echo( + f"\nGrace period EXPIRED for {game_name} (AppID={app_id}).\n" + f"The {_MANUAL_GRACE_DAYS}-day window closed {elapsed}.\n" + "Finish the game (100% achievements) or wait out the lock." + ) + sys.exit(1) + + _echo(f"\nAbandoning manual pick: {game_name} (AppID={app_id})") + _echo( + f"\nThis will:" + f"\n - Release the manual pick lock" + f"\n - Uninstall {game_name}" + f"\n - Keep it out of auto-assignment for" + f" {_ABANDON_COOLDOWN_DAYS} days" + f"\n - Leave you with no assigned game (run 'scan' to get one)" + ) + _echo() + confirm = input(f"Type YES to abandon {game_name}: ").strip() + if confirm != "YES": + _echo("Aborted.") + return + + if not abandon_manual_pick(state): + # Should be unreachable: the grace check above already passed. + _echo("Error: the grace period closed before the pick could be abandoned.") + sys.exit(1) + + _echo(f"\nManual pick abandoned: {game_name}") + + if is_game_installed(app_id): + _echo(f" Uninstalling {game_name}...") + if uninstall_game(app_id, game_name): + _echo(" Uninstalled.") + else: + _echo(" Warning: could not uninstall — remove it from Steam manually.") + + _echo("\nNo game is assigned now. Run 'scan' to get a new assignment,") + _echo("or 'pick-manual ' to choose one yourself.\n") + + _BLOCK_GAMING_USAGE = ( "Usage: block-gaming \n" " days : whole number of days to block ALL gaming:\n" @@ -668,6 +812,9 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = { _EXTRA_COMMAND_DESCRIPTIONS: dict[str, str] = { "add-exception": "Request 24h-locked whitelist exception (use --reason)", "pick-manual": f"Pick a game by app_id, lock enforcer for {_MANUAL_LOCK_DAYS} days", + "abandon-pick": ( + f"Undo a manual pick within {_MANUAL_GRACE_DAYS} days (needs app_id)" + ), "block-gaming": "Block ALL gaming for days, no in-app undo", } @@ -717,6 +864,10 @@ def main() -> None: cmd_pick_manual(config, state, sys.argv[2:]) return + if command == "abandon-pick": + cmd_abandon_pick(config, state, sys.argv[2:]) + return + _, func = COMMANDS[command] func(config, state) diff --git a/steam_backlog_enforcer/tests/test_actions.py b/steam_backlog_enforcer/tests/test_actions.py index f276883..9c9280f 100644 --- a/steam_backlog_enforcer/tests/test_actions.py +++ b/steam_backlog_enforcer/tests/test_actions.py @@ -5,10 +5,15 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone from unittest.mock import patch +import pytest + from steam_backlog_enforcer import _actions from steam_backlog_enforcer._actions import ( + abandon_manual_pick, apply_manual_pick, + can_abandon_manual_pick, is_manual_pick_locked, + manual_pick_grace_remaining, status_payload, ) from steam_backlog_enforcer._total_block import TotalBlockStatus @@ -71,6 +76,115 @@ class TestApplyManualPick: assert State.load().current_app_id == 70 +class TestManualPickGraceRemaining: + def test_no_pick_returns_none(self) -> None: + assert manual_pick_grace_remaining(State()) is None + + def test_missing_timestamp_returns_none(self) -> None: + state = State(manual_pick_app_id=5, manual_pick_started_at="") + assert manual_pick_grace_remaining(state) is None + + def test_malformed_timestamp_returns_none(self) -> None: + state = State(manual_pick_app_id=5, manual_pick_started_at="not-a-date") + assert manual_pick_grace_remaining(state) is None + + def test_fresh_pick_has_almost_the_full_window(self) -> None: + state = State(manual_pick_app_id=5, manual_pick_started_at=_iso_days_ago(0)) + remaining = manual_pick_grace_remaining(state) + assert remaining is not None + assert remaining == pytest.approx(_actions.MANUAL_GRACE_DAYS, abs=0.01) + + def test_expired_window_is_negative(self) -> None: + state = State( + manual_pick_app_id=5, + manual_pick_started_at=_iso_days_ago(_actions.MANUAL_GRACE_DAYS + 1), + ) + remaining = manual_pick_grace_remaining(state) + assert remaining is not None + assert remaining == pytest.approx(-1.0, abs=0.01) + + +class TestCanAbandonManualPick: + def test_inside_window(self) -> None: + state = State( + manual_pick_app_id=5, + manual_pick_started_at=_iso_days_ago(_actions.MANUAL_GRACE_DAYS - 1), + ) + assert can_abandon_manual_pick(state) is True + + def test_outside_window(self) -> None: + state = State( + manual_pick_app_id=5, + manual_pick_started_at=_iso_days_ago(_actions.MANUAL_GRACE_DAYS + 1), + ) + assert can_abandon_manual_pick(state) is False + + def test_no_pick(self) -> None: + assert can_abandon_manual_pick(State()) is False + + +class TestAbandonManualPick: + def _fresh_pick(self) -> State: + return State( + manual_pick_app_id=440, + manual_pick_game_name="Team Fortress 2", + manual_pick_started_at=_iso_days_ago(1), + current_app_id=440, + current_game_name="Team Fortress 2", + ) + + def test_clears_lock_and_assignment(self) -> None: + state = self._fresh_pick() + assert abandon_manual_pick(state) is True + assert state.manual_pick_app_id is None + assert state.manual_pick_game_name == "" + assert state.manual_pick_started_at == "" + assert state.current_app_id is None + assert state.current_game_name == "" + assert is_manual_pick_locked(state) is False + + def test_records_cooldown(self) -> None: + state = self._fresh_pick() + abandon_manual_pick(state) + expiry = datetime.fromisoformat(state.skipped_until["440"]) + expected = datetime.now(timezone.utc) + timedelta( + days=_actions.ABANDON_COOLDOWN_DAYS + ) + assert abs((expiry - expected).total_seconds()) < 60 + + def test_persists_to_disk(self) -> None: + state = self._fresh_pick() + abandon_manual_pick(state) + assert State.load().manual_pick_app_id is None + + def test_refuses_after_grace_and_leaves_state_untouched(self) -> None: + state = self._fresh_pick() + state.manual_pick_started_at = _iso_days_ago(_actions.MANUAL_GRACE_DAYS + 1) + assert abandon_manual_pick(state) is False + assert state.manual_pick_app_id == 440 + assert state.current_app_id == 440 + assert state.skipped_until == {} + + def test_no_cooldown_when_pick_id_is_missing(self) -> None: + # Defensive branch: the grace check normally guarantees an app id, so + # force the inconsistent state to prove no cooldown key is invented. + state = State(current_app_id=None) + with patch.object(_actions, "can_abandon_manual_pick", return_value=True): + assert abandon_manual_pick(state) is True + assert state.skipped_until == {} + assert state.manual_pick_app_id is None + + def test_keeps_unrelated_assignment(self) -> None: + # A pick whose lock outlived a later reassignment must not clear the + # newer assignment when abandoned. + state = self._fresh_pick() + state.current_app_id = 70 + state.current_game_name = "Half-Life" + abandon_manual_pick(state) + assert state.current_app_id == 70 + assert state.current_game_name == "Half-Life" + + class TestStatusPayload: def _patch_leaves( self, diff --git a/steam_backlog_enforcer/tests/test_enforce_loop.py b/steam_backlog_enforcer/tests/test_enforce_loop.py index 82d0826..b36b359 100644 --- a/steam_backlog_enforcer/tests/test_enforce_loop.py +++ b/steam_backlog_enforcer/tests/test_enforce_loop.py @@ -16,7 +16,6 @@ from steam_backlog_enforcer._enforce_loop import ( get_all_owned_app_ids, ) from steam_backlog_enforcer.config import Config, State -from steam_backlog_enforcer.library_hider import SteamUnavailableError if TYPE_CHECKING: from pathlib import Path @@ -349,7 +348,7 @@ class TestEnforceHideGames: state = State(current_app_id=1) with ( patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]), - patch(f"{PKG}.hide_other_games", return_value=2), + patch(f"{PKG}.try_hide_other_games", return_value=(2, None)), patch(f"{PKG}._echo"), ): _enforce_hide_games(Config(), state) @@ -358,7 +357,7 @@ class TestEnforceHideGames: state = State(current_app_id=1) with ( patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2]), - patch(f"{PKG}.hide_other_games", return_value=0), + patch(f"{PKG}.try_hide_other_games", return_value=(0, None)), patch(f"{PKG}._echo") as mock_echo, ): _enforce_hide_games(Config(), state) @@ -384,8 +383,8 @@ class TestEnforceHideGames: with ( patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]), patch( - f"{PKG}.hide_other_games", - side_effect=SteamUnavailableError("Steam is not installed"), + f"{PKG}.try_hide_other_games", + return_value=(0, "Steam is not installed"), ), patch(f"{PKG}._echo") as mock_echo, ): diff --git a/steam_backlog_enforcer/tests/test_library_hider_part2.py b/steam_backlog_enforcer/tests/test_library_hider_part2.py index 15681f0..7d4c8df 100644 --- a/steam_backlog_enforcer/tests/test_library_hider_part2.py +++ b/steam_backlog_enforcer/tests/test_library_hider_part2.py @@ -8,12 +8,15 @@ from unittest.mock import MagicMock, patch from steam_backlog_enforcer.library_hider import ( _SPAWNED, + SteamUnavailableError, + SteamUpdateInProgressError, _reap_spawned, _resolve_desktop_user, _run_as_user, hide_other_games, restart_steam, steam_is_installed, + try_hide_other_games, unhide_all_games, ) @@ -289,6 +292,37 @@ class TestHideOtherGames: assert count == 2 +class TestTryHideOtherGames: + """Tests for the graceful wrapper around hide_other_games. + + Regression guard: an unreachable Steam (or a deferred restart while a game + update is in flight) used to escape as a traceback out of every + interactive command that reconciles the library. + """ + + def test_success_returns_count_and_no_reason(self) -> None: + with patch(f"{PKG}.hide_other_games", return_value=7): + assert try_hide_other_games([1, 2], 1) == (7, None) + + def test_steam_unavailable_is_reported_not_raised(self) -> None: + with patch( + f"{PKG}.hide_other_games", + side_effect=SteamUnavailableError("Steam is not installed"), + ): + hidden, reason = try_hide_other_games([1, 2], 1) + assert hidden == 0 + assert reason == "Steam is not installed" + + def test_update_in_progress_is_reported_not_raised(self) -> None: + with patch( + f"{PKG}.hide_other_games", + side_effect=SteamUpdateInProgressError("update in progress"), + ): + hidden, reason = try_hide_other_games([1, 2], 1) + assert hidden == 0 + assert reason == "update in progress" + + class TestUnhideAllGames: """Tests for unhide_all_games.""" diff --git a/steam_backlog_enforcer/tests/test_main.py b/steam_backlog_enforcer/tests/test_main.py index 6b3706c..8f2ce04 100644 --- a/steam_backlog_enforcer/tests/test_main.py +++ b/steam_backlog_enforcer/tests/test_main.py @@ -342,7 +342,7 @@ class TestCmdHide: state = State(current_app_id=1, current_game_name="G") with ( patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2]), - patch(f"{PKG}.hide_other_games", return_value=1), + patch(f"{PKG}.try_hide_other_games", return_value=(1, None)), patch(f"{PKG}._echo"), ): cmd_hide(Config(), state) @@ -351,11 +351,28 @@ class TestCmdHide: state = State(current_app_id=1, current_game_name="G") with ( patch(f"{PKG}.get_all_owned_app_ids", return_value=[1]), - patch(f"{PKG}.hide_other_games", return_value=0), + patch(f"{PKG}.try_hide_other_games", return_value=(0, None)), patch(f"{PKG}._echo"), ): cmd_hide(Config(), state) + def test_unreachable_steam_reports_skip(self) -> None: + # Regression guard: a deferred Steam restart used to escape as a + # traceback instead of degrading to a skip message. + state = State(current_app_id=1, current_game_name="G") + with ( + patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2]), + patch( + f"{PKG}.try_hide_other_games", + return_value=(0, "update in progress"), + ), + patch(f"{PKG}._echo") as mock_echo, + ): + cmd_hide(Config(), state) + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "skipped (update in progress)" in output + assert "Hidden" not in output + class TestCmdUnhide: """Tests for cmd_unhide.""" diff --git a/steam_backlog_enforcer/tests/test_main_part2.py b/steam_backlog_enforcer/tests/test_main_part2.py index 3ac3f88..58a224b 100644 --- a/steam_backlog_enforcer/tests/test_main_part2.py +++ b/steam_backlog_enforcer/tests/test_main_part2.py @@ -46,7 +46,7 @@ class TestFinalizeCompletion: patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick, patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]), - patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=2), + patch(f"{CMD_DONE_PKG}.try_hide_other_games", return_value=(2, None)), patch(f"{CMD_DONE_PKG}.send_notification"), patch.object(State, "save"), ): @@ -64,6 +64,41 @@ class TestFinalizeCompletion: _finalize_completion(config, state, "G", 1) assert 1 in state.finished_app_ids + def test_hide_skipped_when_steam_unreachable(self) -> None: + # Reconciliation is best-effort: an undrivable Steam must not abort + # the completion flow. + config = Config(steam_api_key="k", steam_id="i") + state = State(current_app_id=1, current_game_name="G") + snap = [_snap(2, "NewGame", 10, 0, 5.0)] + with ( + patch(f"{CMD_DONE_PKG}._echo") as mock_echo, + patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), + patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick, + patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]), + patch( + f"{CMD_DONE_PKG}.try_hide_other_games", + return_value=(0, "update in progress"), + ), + patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True), + patch(f"{CMD_DONE_PKG}.send_notification"), + patch.object(State, "save"), + ): + + def set_next( + _games: object, + s: State, + _c: object, + **_kwargs: object, + ) -> None: + s.current_app_id = 2 + s.current_game_name = "NewGame" + + mock_pick.side_effect = set_next + _finalize_completion(config, state, "G", 1) + assert "skipped (update in progress)" in " ".join( + str(c) for c in mock_echo.call_args_list + ) + def test_no_snapshot(self) -> None: config = Config() state = State(current_app_id=1, current_game_name="G") @@ -131,7 +166,7 @@ class TestFinalizeCompletion: patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick, patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]), - patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=0), + patch(f"{CMD_DONE_PKG}.try_hide_other_games", return_value=(0, None)), patch(f"{CMD_DONE_PKG}.send_notification"), patch.object(State, "save"), ): @@ -206,7 +241,7 @@ class TestFinalizeCompletion: patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), patch(f"{CMD_DONE_PKG}.pick_next_game", side_effect=set_next), patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]), - patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=1), + patch(f"{CMD_DONE_PKG}.try_hide_other_games", return_value=(1, None)), patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=False), patch(f"{CMD_DONE_PKG}.install_game") as mock_install, patch(f"{CMD_DONE_PKG}.send_notification"), @@ -236,7 +271,7 @@ class TestFinalizeCompletion: patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), patch(f"{CMD_DONE_PKG}.pick_next_game", side_effect=set_next), patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]), - patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=1), + patch(f"{CMD_DONE_PKG}.try_hide_other_games", return_value=(1, None)), patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True), patch(f"{CMD_DONE_PKG}.install_game") as mock_install, patch(f"{CMD_DONE_PKG}.send_notification"), @@ -268,7 +303,7 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.uninstall_other_games", return_value=2), patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True), patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]), - patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=1), + patch(f"{CMD_DONE_PKG}.try_hide_other_games", return_value=(1, None)), ): _enforce_on_done(config, state) @@ -284,10 +319,27 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.uninstall_other_games", return_value=0), patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True), patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[]), - patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=0), + patch(f"{CMD_DONE_PKG}.try_hide_other_games", return_value=(0, None)), ): _enforce_on_done(config, state) + def test_hide_skipped_when_steam_unreachable(self) -> None: + config = Config(kill_unauthorized_games=False, uninstall_other_games=False) + state = State(current_app_id=1, current_game_name="G") + with ( + patch(f"{CMD_DONE_PKG}._echo") as mock_echo, + patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True), + patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]), + patch( + f"{CMD_DONE_PKG}.try_hide_other_games", + return_value=(0, "Steam is not installed"), + ), + ): + _enforce_on_done(config, state) + assert "skipped (Steam is not installed)" in " ".join( + str(c) for c in mock_echo.call_args_list + ) + def test_reinstall_when_not_installed(self) -> None: config = Config( kill_unauthorized_games=False, @@ -300,7 +352,7 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=False), patch(f"{CMD_DONE_PKG}.install_game") as mock_install, patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]), - patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=0), + patch(f"{CMD_DONE_PKG}.try_hide_other_games", return_value=(0, None)), ): _enforce_on_done(config, state) mock_install.assert_called_once_with(1, "G", "s1", use_steam_protocol=True) diff --git a/steam_backlog_enforcer/tests/test_main_part3.py b/steam_backlog_enforcer/tests/test_main_part3.py index 7a50898..75f4ee4 100644 --- a/steam_backlog_enforcer/tests/test_main_part3.py +++ b/steam_backlog_enforcer/tests/test_main_part3.py @@ -240,7 +240,7 @@ class TestCmdPick: patch(f"{PKG}.load_hltb_cache", return_value={2: 5.0}), patch(f"{PKG}.pick_next_game"), patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]), - patch(f"{PKG}.hide_other_games", return_value=2) as mock_hide, + patch(f"{PKG}.try_hide_other_games", return_value=(2, None)) as mock_hide, patch(f"{PKG}._echo"), ): cmd_pick(Config(steam_api_key="k", steam_id="i"), state) @@ -254,12 +254,30 @@ class TestCmdPick: patch(f"{PKG}.load_hltb_cache", return_value={}), patch(f"{PKG}.pick_next_game"), patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]), - patch(f"{PKG}.hide_other_games", return_value=0), + patch(f"{PKG}.try_hide_other_games", return_value=(0, None)), patch(f"{PKG}._echo") as mock_echo, ): cmd_pick(Config(steam_api_key="k", steam_id="i"), state) mock_echo.assert_not_called() + def test_unreachable_steam_reports_skip(self) -> None: + snap = [_snap(2, "NewGame", 10, 0, 5.0)] + state = State(current_app_id=2, current_game_name="NewGame") + with ( + patch(f"{PKG}.load_snapshot", return_value=snap), + patch(f"{PKG}.load_hltb_cache", return_value={}), + patch(f"{PKG}.pick_next_game"), + patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]), + patch( + f"{PKG}.try_hide_other_games", + return_value=(0, "Steam is not installed"), + ), + patch(f"{PKG}._echo") as mock_echo, + ): + cmd_pick(Config(steam_api_key="k", steam_id="i"), state) + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "skipped (Steam is not installed)" in output + def test_no_hide_when_no_current_app(self) -> None: snap = [_snap(2, "NewGame", 10, 0, 5.0)] with ( @@ -279,7 +297,7 @@ class TestCmdPick: patch(f"{PKG}.load_hltb_cache", return_value={}), patch(f"{PKG}.pick_next_game"), patch(f"{PKG}.get_all_owned_app_ids", return_value=[]), - patch(f"{PKG}.hide_other_games") as mock_hide, + patch(f"{PKG}.try_hide_other_games") as mock_hide, ): cmd_pick(Config(steam_api_key="k", steam_id="i"), state) mock_hide.assert_not_called() diff --git a/steam_backlog_enforcer/tests/test_main_part4.py b/steam_backlog_enforcer/tests/test_main_part4.py index ff11960..696f51f 100644 --- a/steam_backlog_enforcer/tests/test_main_part4.py +++ b/steam_backlog_enforcer/tests/test_main_part4.py @@ -10,11 +10,13 @@ import pytest from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.main import ( + _MANUAL_GRACE_DAYS, _MANUAL_LOCK_DAYS, _enforce_manual_pick_lock, _is_manual_pick_locked, _resolve_game_name, _show_manual_pick_lock_message, + cmd_abandon_pick, cmd_pick_manual, cmd_status, main, @@ -221,7 +223,7 @@ class TestCmdPickManual: f"{PKG}.is_game_installed": False, f"{PKG}.install_game": None, f"{PKG}.get_all_owned_app_ids": [1, 2, 489830], - f"{PKG}.hide_other_games": 2, + f"{PKG}.try_hide_other_games": (2, None), } def test_invalid_app_id(self) -> None: @@ -275,7 +277,7 @@ class TestCmdPickManual: patch(f"{PKG}.is_game_installed", return_value=False), patch(f"{PKG}.install_game") as mock_install, patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 489830]), - patch(f"{PKG}.hide_other_games", return_value=1) as mock_hide, + patch(f"{PKG}.try_hide_other_games", return_value=(1, None)) as mock_hide, ): cmd_pick_manual(config, state, ["489830"]) @@ -328,7 +330,7 @@ class TestCmdPickManual: patch(f"{PKG}.uninstall_other_games", return_value=0), patch(f"{PKG}.is_game_installed", return_value=True), patch(f"{PKG}.get_all_owned_app_ids", return_value=[]), - patch(f"{PKG}.hide_other_games") as mock_hide, + patch(f"{PKG}.try_hide_other_games") as mock_hide, ): cmd_pick_manual(Config(), state, ["489830"]) mock_hide.assert_not_called() @@ -388,12 +390,35 @@ class TestCmdPickManual: patch(f"{PKG}.uninstall_other_games", return_value=0), patch(f"{PKG}.is_game_installed", return_value=True), patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2]), - patch(f"{PKG}.hide_other_games", return_value=0), + patch(f"{PKG}.try_hide_other_games", return_value=(0, None)), ): cmd_pick_manual(Config(), state, ["489830"]) output = " ".join(str(c) for c in mock_echo.call_args_list) assert "Library: hid" not in output + def test_unreachable_steam_reports_skip(self) -> None: + # The pick itself must survive a Steam that cannot be driven: this + # used to abort cmd_pick_manual with a traceback after the pick had + # already been saved. + state = State() + with ( + patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), + patch(f"{PKG}._echo") as mock_echo, + patch("builtins.input", return_value="YES"), + patch.object(State, "save"), + patch(f"{PKG}.uninstall_other_games", return_value=0), + patch(f"{PKG}.is_game_installed", return_value=True), + patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2]), + patch( + f"{PKG}.try_hide_other_games", + return_value=(0, "update in progress"), + ), + ): + cmd_pick_manual(Config(), state, ["489830"]) + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "skipped (update in progress)" in output + assert state.manual_pick_app_id == 489830 + # ────────────────────────────────────────────────────────────── # main() dispatch to pick-manual @@ -452,3 +477,165 @@ class TestCmdStatusLockHint: cmd_status(Config(), State()) output = " ".join(str(c) for c in mock_echo.call_args_list) assert "MANUAL PICK LOCK" not in output + + +# ────────────────────────────────────────────────────────────── +# cmd_abandon_pick (grace period escape hatch) +# ────────────────────────────────────────────────────────────── + +# Inside the 4-day grace window / already past it. +_IN_GRACE = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat() +_PAST_GRACE = ( + datetime.now(timezone.utc) - timedelta(days=_MANUAL_GRACE_DAYS + 1) +).isoformat() + + +def _abandonable_state(app_id: int = 100, started_at: str = _IN_GRACE) -> State: + state = _locked_state(app_id=app_id, started_at=started_at) + state.current_app_id = app_id + state.current_game_name = state.manual_pick_game_name + return state + + +class TestCmdAbandonPick: + def test_no_args_exits(self) -> None: + with patch(f"{PKG}._echo") as mock_echo, pytest.raises(SystemExit) as exc: + cmd_abandon_pick(Config(), _abandonable_state(), []) + assert exc.value.code == 1 + assert "Usage" in " ".join(str(c) for c in mock_echo.call_args_list) + + def test_non_numeric_app_id_exits(self) -> None: + with patch(f"{PKG}._echo") as mock_echo, pytest.raises(SystemExit): + cmd_abandon_pick(Config(), _abandonable_state(), ["abc"]) + assert "must be a number" in " ".join(str(c) for c in mock_echo.call_args_list) + + def test_no_active_pick_exits(self) -> None: + with patch(f"{PKG}._echo") as mock_echo, pytest.raises(SystemExit): + cmd_abandon_pick(Config(), State(), ["100"]) + assert "No manual pick" in " ".join(str(c) for c in mock_echo.call_args_list) + + def test_wrong_app_id_exits(self) -> None: + with patch(f"{PKG}._echo") as mock_echo, pytest.raises(SystemExit): + cmd_abandon_pick(Config(), _abandonable_state(app_id=100), ["999"]) + assert "not your manual pick" in " ".join( + str(c) for c in mock_echo.call_args_list + ) + + def test_expired_grace_exits_without_mutating(self) -> None: + state = _abandonable_state(started_at=_PAST_GRACE) + with ( + patch(f"{PKG}._echo") as mock_echo, + patch.object(State, "save") as mock_save, + pytest.raises(SystemExit) as exc, + ): + cmd_abandon_pick(Config(), state, ["100"]) + assert exc.value.code == 1 + assert "EXPIRED" in " ".join(str(c) for c in mock_echo.call_args_list) + mock_save.assert_not_called() + assert state.manual_pick_app_id == 100 + + def test_expired_grace_with_bad_timestamp_still_exits(self) -> None: + state = _abandonable_state(started_at="not-a-date") + with patch(f"{PKG}._echo"), pytest.raises(SystemExit): + cmd_abandon_pick(Config(), state, ["100"]) + + def test_aborted_when_not_yes(self) -> None: + state = _abandonable_state() + with ( + patch(f"{PKG}._echo"), + patch("builtins.input", return_value="no"), + patch.object(State, "save") as mock_save, + patch(f"{PKG}.uninstall_game") as mock_uninstall, + ): + cmd_abandon_pick(Config(), state, ["100"]) + mock_save.assert_not_called() + mock_uninstall.assert_not_called() + assert state.manual_pick_app_id == 100 + + def test_success_clears_lock_and_uninstalls(self) -> None: + state = _abandonable_state() + with ( + patch(f"{PKG}._echo") as mock_echo, + patch("builtins.input", return_value="YES"), + patch.object(State, "save"), + patch(f"{PKG}.is_game_installed", return_value=True), + patch(f"{PKG}.uninstall_game", return_value=True) as mock_uninstall, + ): + cmd_abandon_pick(Config(), state, ["100"]) + assert state.manual_pick_app_id is None + assert state.current_app_id is None + assert state.skipped_until["100"] != "" + mock_uninstall.assert_called_once_with(100, "TestGame") + assert "abandoned" in " ".join(str(c) for c in mock_echo.call_args_list) + + def test_skips_uninstall_when_not_installed(self) -> None: + state = _abandonable_state() + with ( + patch(f"{PKG}._echo"), + patch("builtins.input", return_value="YES"), + patch.object(State, "save"), + patch(f"{PKG}.is_game_installed", return_value=False), + patch(f"{PKG}.uninstall_game") as mock_uninstall, + ): + cmd_abandon_pick(Config(), state, ["100"]) + mock_uninstall.assert_not_called() + + def test_warns_when_uninstall_fails(self) -> None: + state = _abandonable_state() + with ( + patch(f"{PKG}._echo") as mock_echo, + patch("builtins.input", return_value="YES"), + patch.object(State, "save"), + patch(f"{PKG}.is_game_installed", return_value=True), + patch(f"{PKG}.uninstall_game", return_value=False), + ): + cmd_abandon_pick(Config(), state, ["100"]) + assert "Warning" in " ".join(str(c) for c in mock_echo.call_args_list) + + def test_race_between_check_and_apply_exits(self) -> None: + # The grace check passes, then the state-only core refuses: defensive + # branch that must exit rather than report a false success. + state = _abandonable_state() + with ( + patch(f"{PKG}._echo") as mock_echo, + patch("builtins.input", return_value="YES"), + patch(f"{PKG}.abandon_manual_pick", return_value=False), + pytest.raises(SystemExit) as exc, + ): + cmd_abandon_pick(Config(), state, ["100"]) + assert exc.value.code == 1 + assert "grace period closed" in " ".join( + str(c) for c in mock_echo.call_args_list + ) + + +class TestAbandonPickLockInteraction: + def test_abandon_pick_is_exempt_from_the_lock(self) -> None: + # The escape hatch must survive the pre-dispatch lock check. + _enforce_manual_pick_lock("abandon-pick", _locked_state()) + + def test_main_dispatches_abandon_pick_while_locked(self) -> None: + argv = ["prog", "abandon-pick", "100"] + with ( + patch.object(sys, "argv", argv), + patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")), + patch(f"{PKG}.State.load", return_value=_abandonable_state()), + patch(f"{PKG}.cmd_abandon_pick") as mock_cmd, + ): + main() + mock_cmd.assert_called_once() + + def test_lock_message_advertises_abandon_within_grace(self) -> None: + state = _abandonable_state(started_at=_IN_GRACE) + with patch(f"{PKG}._echo") as mock_echo: + _show_manual_pick_lock_message(state) + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "abandon-pick 100" in output + assert "grace period" in output + + def test_lock_message_hides_abandon_after_grace(self) -> None: + state = _abandonable_state(started_at=_PAST_GRACE) + with patch(f"{PKG}._echo") as mock_echo: + _show_manual_pick_lock_message(state) + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "abandon-pick" not in output diff --git a/steam_backlog_enforcer/tests/test_mcp.py b/steam_backlog_enforcer/tests/test_mcp.py index 5b28682..90a3b50 100644 --- a/steam_backlog_enforcer/tests/test_mcp.py +++ b/steam_backlog_enforcer/tests/test_mcp.py @@ -7,9 +7,11 @@ already-tested leaf functions. from __future__ import annotations +from datetime import datetime, timedelta, timezone from unittest.mock import patch from steam_backlog_enforcer import _mcp +from steam_backlog_enforcer.config import State from steam_backlog_enforcer.steam_api import GameInfo @@ -176,6 +178,61 @@ class TestPickManualGate: amp.assert_called_once_with(state.load.return_value, 440, "TF2") +class TestAbandonPickGate: + """The MCP escape hatch mirrors the CLI grace rules, state-only.""" + + def _state(self, *, days_ago: float = 1.0, app_id: int = 440) -> State: + started = (datetime.now(timezone.utc) - timedelta(days=days_ago)).isoformat() + return State( + manual_pick_app_id=app_id, + manual_pick_game_name="TF2", + manual_pick_started_at=started, + current_app_id=app_id, + current_game_name="TF2", + ) + + def test_no_active_pick(self) -> None: + with patch.object(_mcp.State, "load", return_value=State()): + out = _mcp.abandon_pick(440) + assert out["ok"] is False + assert "No manual pick" in out["reason"] + + def test_wrong_app_id(self) -> None: + with patch.object(_mcp.State, "load", return_value=self._state()): + out = _mcp.abandon_pick(999) + assert out["ok"] is False + assert "not the active manual pick" in out["reason"] + + def test_expired_grace(self) -> None: + state = self._state(days_ago=_mcp.MANUAL_GRACE_DAYS + 1) + with patch.object(_mcp.State, "load", return_value=state): + out = _mcp.abandon_pick(440) + assert out["ok"] is False + assert "grace period has expired" in out["reason"] + + def test_preview_does_not_mutate(self) -> None: + with ( + patch.object(_mcp.State, "load", return_value=self._state()), + patch.object(_mcp, "abandon_manual_pick") as amp, + ): + out = _mcp.abandon_pick(440) + assert out["preview"] is True + assert out["game_name"] == "TF2" + assert out["grace_days_left"] > 0 + amp.assert_not_called() + + def test_confirm_applies(self) -> None: + state = self._state() + with ( + patch.object(_mcp.State, "load", return_value=state), + patch.object(_mcp, "abandon_manual_pick") as amp, + ): + out = _mcp.abandon_pick(440, confirm=True) + assert out["applied"] is True + assert out["app_id"] == 440 + amp.assert_called_once_with(state) + + class TestBlockGamingGate: def test_invalid_days(self) -> None: out = _mcp.block_gaming(0)