mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-11-11 12:27:33 +01:00
feat(enforcer): add 4-day grace period to undo a manual pick
A manual pick had no escape hatch: once locked in, every command except done/check/status/enforce/setup/serve exited 1 for 14 days or until 100% achievements. A wrong pick meant two weeks stuck with it. Adds a short mistake-correction window: - MANUAL_GRACE_DAYS = 4 — within 4 days, 'abandon-pick <app_id>' backs out of the pick. Outside it 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' joins _MANUAL_LOCK_EXEMPT_COMMANDS — the pre-dispatch lock check would otherwise block the only way out. It is dropped from the advertised command list once the window closes. - Abandoning clears the lock and the assignment, uninstalls the game, and reuses 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 cascade), matching apply_manual_pick, so the gated MCP abandon_pick tool reuses it. Also fixes a crash surfaced while picking a replacement game: the SteamUnavailableError family (including SteamUpdateInProgressError, raised when a game update defers a Steam restart) was only caught in the enforce loop. cmd_hide, cmd_pick, cmd_pick_manual and both cmd_done reconciliation points called hide_other_games bare, so an undrivable Steam aborted the command with a traceback — after the pick had already been saved. All six sites now go through the new stdout-free try_hide_other_games wrapper and report "Library hiding: skipped (<reason>)" instead. Verified live: lock message advertises the window, expired/wrong-id/no-arg paths exit 1, and abandoning released the AoE2 lock, uninstalled it and recorded the 30-day cooldown. 1030 tests pass at 100% coverage. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
be2b7c7b93
commit
eb1472a433
@ -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),
|
||||
}
|
||||
|
||||
@ -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")
|
||||
|
||||
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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).
|
||||
|
||||
@ -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 <app_id>`
|
||||
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.
|
||||
|
||||
@ -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.
|
||||
|
||||
|
||||
@ -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 <app_id>\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 <app_id>' to choose one yourself.\n")
|
||||
|
||||
|
||||
_BLOCK_GAMING_USAGE = (
|
||||
"Usage: block-gaming <days>\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> 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)
|
||||
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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,
|
||||
):
|
||||
|
||||
@ -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."""
|
||||
|
||||
|
||||
@ -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."""
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user