From 30fcaacc5de1609344682e253c964759593f9f52 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Fri, 25 Sep 2026 18:44:07 +0200 Subject: [PATCH] Deduplicate snapshot loading, pick application and library hiding into shared helpers require_snapshot_games / snapshot_game_name (_snapshot), apply_configured_pick (_actions) and hide_others_and_report (library_hider) replace copies spread across the CLI, MCP and stats paths. Tests retargeted to the new owners; test_main_pick_manual split to stay under 250 lines. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01D55DvFEHiwcPN59kwPqwiU --- steam_backlog_enforcer/_actions.py | 11 +- steam_backlog_enforcer/_cmd_done_finalize.py | 19 +-- steam_backlog_enforcer/_mcp_actions.py | 9 +- steam_backlog_enforcer/_mcp_server.py | 9 +- steam_backlog_enforcer/_snapshot.py | 19 +++ steam_backlog_enforcer/_stats.py | 11 +- steam_backlog_enforcer/_store_iptables.py | 24 +-- steam_backlog_enforcer/_store_tools.py | 25 +++ .../_total_block_iptables.py | 35 ++--- steam_backlog_enforcer/enforcer.py | 8 +- steam_backlog_enforcer/library_hider.py | 23 +++ steam_backlog_enforcer/main/picks.py | 38 ++--- steam_backlog_enforcer/main/status.py | 10 +- .../tests/test_main_part2.py | 34 +++- .../tests/test_main_part2b.py | 39 ++++- .../tests/test_main_part2c.py | 20 ++- .../tests/test_main_pick_cap.py | 25 ++- .../tests/test_main_pick_manual.py | 138 +++-------------- .../tests/test_main_pick_manual_cascade.py | 146 ++++++++++++++++++ .../tests/test_main_picks.py | 56 +++++-- .../tests/test_main_status.py | 13 +- steam_backlog_enforcer/tests/test_mcp.py | 11 +- .../tests/test_mcp_actions.py | 27 ++-- .../tests/test_stats_cmd_stats.py | 24 ++- ...efilter_called_when_ensure_rush_data_re.py | 16 +- 25 files changed, 478 insertions(+), 312 deletions(-) create mode 100644 steam_backlog_enforcer/tests/test_main_pick_manual_cascade.py diff --git a/steam_backlog_enforcer/_actions.py b/steam_backlog_enforcer/_actions.py index dbb0eb1..fd08956 100644 --- a/steam_backlog_enforcer/_actions.py +++ b/steam_backlog_enforcer/_actions.py @@ -24,7 +24,7 @@ from steam_backlog_enforcer._allowed_games import ( # accessible normally; this only needs to list names imported-then-reexported. if TYPE_CHECKING: - from steam_backlog_enforcer.config import State + from steam_backlog_enforcer.config import Config, State # 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. @@ -93,6 +93,15 @@ def manual_pick_slots_left(state: State, max_picks: int) -> int: return max(0, max_picks - len(active_manual_picks(state))) +def apply_configured_pick( + config: Config, state: State, app_id: int, game_name: str +) -> str | None: + """:func:`apply_manual_pick` under the cap the configuration sets.""" + return apply_manual_pick( + state, app_id, game_name, max_picks=config.max_manual_picks + ) + + def apply_manual_pick( state: State, app_id: int, diff --git a/steam_backlog_enforcer/_cmd_done_finalize.py b/steam_backlog_enforcer/_cmd_done_finalize.py index 2e6a987..5119c92 100644 --- a/steam_backlog_enforcer/_cmd_done_finalize.py +++ b/steam_backlog_enforcer/_cmd_done_finalize.py @@ -16,7 +16,6 @@ from steam_backlog_enforcer._cmd_done import ( _refresh_uncached_shortlist_hours, _report_assigned_confidence, ) -from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids from steam_backlog_enforcer._hltb_cached import fetch_hltb_times_cached from steam_backlog_enforcer._hltb_types import load_hltb_cache from steam_backlog_enforcer._pick_completion import mark_finished @@ -31,7 +30,7 @@ from steam_backlog_enforcer.game_install import ( is_game_installed, uninstall_other_games, ) -from steam_backlog_enforcer.library_hider import try_hide_other_games +from steam_backlog_enforcer.library_hider import hide_others_and_report from steam_backlog_enforcer.scanning import pick_next_game from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient @@ -73,13 +72,7 @@ def _finalize_completion( _echo(" No more games to assign!") return - owned_ids = get_all_owned_app_ids(config) - if owned_ids: - hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state)) - if skipped is not None: - _echo(f"\n Library hiding: skipped ({skipped})") - elif hidden > 0: - _echo(f"\n Library: hid {hidden} games") + hide_others_and_report(config, state) if not is_game_installed(state.current_app_id): logger.info( @@ -139,13 +132,7 @@ def _enforce_on_done(config: Config, state: State) -> None: # Reconcile library: hide non-assigned games and unhide the assigned one. # Without this, an interrupted earlier completion can leave the new # assigned game hidden and stale games visible. - owned_ids = get_all_owned_app_ids(config) - if owned_ids: - hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state)) - if skipped is not None: - _echo(f" Library hiding: skipped ({skipped})") - elif hidden > 0: - _echo(f" Library: hid {hidden} games") + hide_others_and_report(config, state, lead=" ") def cmd_done(config: Config, state: State) -> None: diff --git a/steam_backlog_enforcer/_mcp_actions.py b/steam_backlog_enforcer/_mcp_actions.py index c00bcdf..d11a280 100644 --- a/steam_backlog_enforcer/_mcp_actions.py +++ b/steam_backlog_enforcer/_mcp_actions.py @@ -12,7 +12,7 @@ from typing import Any from steam_backlog_enforcer._actions import ( ABANDON_COOLDOWN_DAYS, active_manual_picks, - apply_manual_pick, + apply_configured_pick, find_manual_pick, ) from steam_backlog_enforcer._allowed_games import MANUAL_LOCK_DAYS @@ -71,12 +71,7 @@ def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]: # Free the slots of picks already at 100%, so the cap below cannot refuse # on behalf of a game that is finished. Matches the CLI's pick-manual. retired = retire_completed_manual_picks(config, state) - refused = apply_manual_pick( - state, - app_id, - game_name, - max_picks=config.max_manual_picks, - ) + refused = apply_configured_pick(config, state, app_id, game_name) if refused is not None: return {"ok": False, "reason": refused} logger.info("pick_manual applied: %s (AppID=%s)", game_name, app_id) diff --git a/steam_backlog_enforcer/_mcp_server.py b/steam_backlog_enforcer/_mcp_server.py index cf80916..14fbbc9 100644 --- a/steam_backlog_enforcer/_mcp_server.py +++ b/steam_backlog_enforcer/_mcp_server.py @@ -31,7 +31,7 @@ from typing import TYPE_CHECKING from mcp.server import MCPServer -from steam_backlog_enforcer._snapshot import load_snapshot +from steam_backlog_enforcer._snapshot import snapshot_game_name if TYPE_CHECKING: from steam_backlog_enforcer.steam_api import GameInfo @@ -63,12 +63,7 @@ def _resolve_game_name(app_id: int) -> str | None: Returns: The game name, or ``None`` if not present in the snapshot. """ - snapshot = load_snapshot() - if snapshot: - for entry in snapshot: - if entry.get("app_id") == app_id: - return str(entry["name"]) - return None + return snapshot_game_name(app_id) def _backlog_sort_key(game: GameInfo) -> tuple[int, float]: diff --git a/steam_backlog_enforcer/_snapshot.py b/steam_backlog_enforcer/_snapshot.py index 1a86dd2..d327949 100644 --- a/steam_backlog_enforcer/_snapshot.py +++ b/steam_backlog_enforcer/_snapshot.py @@ -11,6 +11,8 @@ import json from typing import Any from steam_backlog_enforcer import config +from steam_backlog_enforcer._echo import _echo +from steam_backlog_enforcer._steam_models import GameInfo from steam_backlog_enforcer.config import _atomic_write @@ -38,3 +40,20 @@ def load_snapshot() -> list[dict[str, Any]] | None: ) return result return None + + +def snapshot_game_name(app_id: int) -> str | None: + """The name the last snapshot recorded for ``app_id``, or ``None``.""" + for entry in load_snapshot() or []: + if entry.get("app_id") == app_id: + return str(entry["name"]) + return None + + +def require_snapshot_games() -> list[GameInfo] | None: + """The last snapshot as games, or ``None`` after telling the user to scan.""" + snapshot = load_snapshot() + if snapshot is None: + _echo("No snapshot found. Run 'scan' first.") + return None + return [GameInfo.from_snapshot(d) for d in snapshot] diff --git a/steam_backlog_enforcer/_stats.py b/steam_backlog_enforcer/_stats.py index 4c1d36c..cd498f3 100644 --- a/steam_backlog_enforcer/_stats.py +++ b/steam_backlog_enforcer/_stats.py @@ -8,7 +8,7 @@ from typing import TYPE_CHECKING from steam_backlog_enforcer._hltb_types import ( _read_raw_cache, ) -from steam_backlog_enforcer._snapshot import load_snapshot +from steam_backlog_enforcer._snapshot import require_snapshot_games from steam_backlog_enforcer._stats_display import ( _format_completion_date, _print_pace_scenario, @@ -28,9 +28,6 @@ from steam_backlog_enforcer._web_dataset import ( count_complete_since_start, ) from steam_backlog_enforcer.game_install import _echo -from steam_backlog_enforcer.steam_api import ( - GameInfo, -) # _GameTimes lives in the _stats_types leaf so both this module and its # helper modules can share it; it is re-exported here because callers (and @@ -135,12 +132,10 @@ def cmd_stats(_config: Config, state: State) -> None: 4. Worst — absolute maximum recorded time (any category) per HLTB. 5. Your play style — extrapolated from completed-game calibration vs HLTB. """ - snapshot = load_snapshot() - if snapshot is None: - _echo("No snapshot found. Run 'scan' first.") + games = require_snapshot_games() + if games is None: return - games = [GameInfo.from_snapshot(d) for d in snapshot] games = _refresh_recently_played_completions(games, _config) # Count all 100%-achievement games in library (more accurate than # finished_app_ids, which only tracks enforcer-assigned completions). diff --git a/steam_backlog_enforcer/_store_iptables.py b/steam_backlog_enforcer/_store_iptables.py index c3d7b16..d834441 100644 --- a/steam_backlog_enforcer/_store_iptables.py +++ b/steam_backlog_enforcer/_store_iptables.py @@ -16,6 +16,8 @@ from steam_backlog_enforcer._store_tools import ( IPTABLES, IPTABLES_CHAIN, SUDO, + ensure_chain_in_output, + run_quiet, ) from steam_backlog_enforcer.config import ( BLOCKED_DOMAINS, @@ -82,20 +84,7 @@ def _block_store_iptables() -> bool: check=True, ) - # Hook our chain into OUTPUT if not already there. - result = subprocess.run( - [SUDO, IPTABLES, "-C", "OUTPUT", "-j", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=False, - ) - if result.returncode != 0: - subprocess.run( - [SUDO, IPTABLES, "-I", "OUTPUT", "-j", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=True, - ) + ensure_chain_in_output(IPTABLES_CHAIN) except OSError, subprocess.SubprocessError: logger.exception("Failed to block store via iptables") return False @@ -140,9 +129,4 @@ def flush_dns_cache() -> None: ] for cmd in commands: with contextlib.suppress(FileNotFoundError, OSError): - subprocess.run( - cmd, - capture_output=True, - timeout=5, - check=False, - ) + run_quiet(cmd) diff --git a/steam_backlog_enforcer/_store_tools.py b/steam_backlog_enforcer/_store_tools.py index 60584ee..cddb863 100644 --- a/steam_backlog_enforcer/_store_tools.py +++ b/steam_backlog_enforcer/_store_tools.py @@ -9,6 +9,7 @@ from __future__ import annotations from pathlib import Path import shutil +import subprocess # Path to the hosts install script. _REPO_ROOT resolves to $HOME (this # module lives two levels below it); the script itself is in the @@ -36,3 +37,27 @@ TEE = shutil.which("tee") or "/usr/bin/tee" # IP address used in /etc/hosts for blocking domains. HOSTS_REDIRECT_IP = ".".join(["0"] * 4) + + +def run_quiet( + cmd: list[str], *, timeout: float = 5 +) -> subprocess.CompletedProcess[bytes]: + """Run ``cmd`` with output captured and no exception on a non-zero exit.""" + return subprocess.run(cmd, capture_output=True, timeout=timeout, check=False) + + +def ensure_chain_in_output(chain: str) -> None: + """Hook ``chain`` into iptables' OUTPUT if it is not already there. + + Raises: + subprocess.CalledProcessError: If the insert itself fails -- a chain + that exists but is not reachable from OUTPUT blocks nothing. + """ + probe = run_quiet([SUDO, IPTABLES, "-C", "OUTPUT", "-j", chain]) + if probe.returncode != 0: + subprocess.run( + [SUDO, IPTABLES, "-I", "OUTPUT", "-j", chain], + capture_output=True, + timeout=5, + check=True, + ) diff --git a/steam_backlog_enforcer/_total_block_iptables.py b/steam_backlog_enforcer/_total_block_iptables.py index c0fad81..9424667 100644 --- a/steam_backlog_enforcer/_total_block_iptables.py +++ b/steam_backlog_enforcer/_total_block_iptables.py @@ -12,10 +12,10 @@ from __future__ import annotations import contextlib import json import logging -import shutil import socket import subprocess +from steam_backlog_enforcer._store_tools import IPTABLES, SUDO, ensure_chain_in_output from steam_backlog_enforcer._total_block_domains import ( _ALL_TOTAL_BLOCK_DOMAINS, NULL_ROUTE_IP, @@ -24,9 +24,6 @@ from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write logger = logging.getLogger(__name__) -_IPTABLES = shutil.which("iptables") or "/usr/sbin/iptables" -_SUDO = shutil.which("sudo") or "/usr/bin/sudo" - IPTABLES_CHAIN = "STEAM_TOTAL_BLOCK" _IPTABLES_IP_CACHE_FILE = CONFIG_DIR / "total_block_ip_cache.json" @@ -59,7 +56,7 @@ def _iptables_chain_intact(expected_ips: set[str]) -> bool: 3-second enforce tick. """ listing = subprocess.run( - [_SUDO, _IPTABLES, "-S", IPTABLES_CHAIN], + [SUDO, IPTABLES, "-S", IPTABLES_CHAIN], capture_output=True, text=True, timeout=5, @@ -80,7 +77,7 @@ def _iptables_chain_intact(expected_ips: set[str]) -> bool: return False hook = subprocess.run( - [_SUDO, _IPTABLES, "-C", "OUTPUT", "-j", IPTABLES_CHAIN], + [SUDO, IPTABLES, "-C", "OUTPUT", "-j", IPTABLES_CHAIN], capture_output=True, timeout=5, check=False, @@ -115,13 +112,13 @@ def apply_total_block_iptables() -> bool: resolved_ips: set[str] = set() try: subprocess.run( - [_SUDO, _IPTABLES, "-N", IPTABLES_CHAIN], + [SUDO, IPTABLES, "-N", IPTABLES_CHAIN], capture_output=True, timeout=5, check=False, ) subprocess.run( - [_SUDO, _IPTABLES, "-F", IPTABLES_CHAIN], + [SUDO, IPTABLES, "-F", IPTABLES_CHAIN], capture_output=True, timeout=5, check=True, @@ -137,25 +134,13 @@ def apply_total_block_iptables() -> bool: for ip in blocked_ips: subprocess.run( - [_SUDO, _IPTABLES, "-A", IPTABLES_CHAIN, "-d", ip, "-j", "DROP"], + [SUDO, IPTABLES, "-A", IPTABLES_CHAIN, "-d", ip, "-j", "DROP"], capture_output=True, timeout=5, check=True, ) - result = subprocess.run( - [_SUDO, _IPTABLES, "-C", "OUTPUT", "-j", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=False, - ) - if result.returncode != 0: - subprocess.run( - [_SUDO, _IPTABLES, "-I", "OUTPUT", "-j", IPTABLES_CHAIN], - capture_output=True, - timeout=5, - check=True, - ) + ensure_chain_in_output(IPTABLES_CHAIN) except OSError, subprocess.SubprocessError: logger.exception("Failed to apply total-block iptables rules") return False @@ -167,19 +152,19 @@ def remove_total_block_iptables() -> bool: """Remove the total-block iptables chain and its OUTPUT hook.""" try: subprocess.run( - [_SUDO, _IPTABLES, "-D", "OUTPUT", "-j", IPTABLES_CHAIN], + [SUDO, IPTABLES, "-D", "OUTPUT", "-j", IPTABLES_CHAIN], capture_output=True, timeout=5, check=False, ) subprocess.run( - [_SUDO, _IPTABLES, "-F", IPTABLES_CHAIN], + [SUDO, IPTABLES, "-F", IPTABLES_CHAIN], capture_output=True, timeout=5, check=False, ) subprocess.run( - [_SUDO, _IPTABLES, "-X", IPTABLES_CHAIN], + [SUDO, IPTABLES, "-X", IPTABLES_CHAIN], capture_output=True, timeout=5, check=False, diff --git a/steam_backlog_enforcer/enforcer.py b/steam_backlog_enforcer/enforcer.py index 2f596c9..8f8ba7f 100644 --- a/steam_backlog_enforcer/enforcer.py +++ b/steam_backlog_enforcer/enforcer.py @@ -13,6 +13,7 @@ from steam_backlog_enforcer._desktop_env import ( desktop_user_cmd, resolve_desktop_user, ) +from steam_backlog_enforcer._store_tools import run_quiet from steam_backlog_enforcer.game_uninstall import is_protected_app logger = logging.getLogger(__name__) @@ -163,12 +164,7 @@ def send_notification(title: str, body: str) -> None: resolve_desktop_user(), ) try: - subprocess.run( - cmd, - capture_output=True, - timeout=5, - check=False, - ) + run_quiet(cmd) except FileNotFoundError, OSError: logger.debug("notify-send not available.") except subprocess.TimeoutExpired: diff --git a/steam_backlog_enforcer/library_hider.py b/steam_backlog_enforcer/library_hider.py index 19ac0b5..a6d7e66 100644 --- a/steam_backlog_enforcer/library_hider.py +++ b/steam_backlog_enforcer/library_hider.py @@ -18,8 +18,12 @@ from __future__ import annotations import json import logging +from typing import TYPE_CHECKING +from steam_backlog_enforcer._actions import allowed_app_ids from steam_backlog_enforcer._cdp import _cdp_result_value, _evaluate_js +from steam_backlog_enforcer._echo import _echo +from steam_backlog_enforcer._owned_apps_cache import get_all_owned_app_ids from steam_backlog_enforcer._steam_errors import ( SteamUnavailableError, ) @@ -29,6 +33,9 @@ from steam_backlog_enforcer._steam_launch import ( steam_is_installed, ) +if TYPE_CHECKING: + from steam_backlog_enforcer.config import Config, State + logger = logging.getLogger(__name__) # Re-exported: the Steam lifecycle now lives in _steam_launch, but callers @@ -195,3 +202,19 @@ def unhide_all_games(owned_app_ids: list[int]) -> int: count: int = parsed["count"] logger.info("Unhidden %d games via CDP.", count) return count + + +def hide_others_and_report(config: Config, state: State, *, lead: str = "\n ") -> None: + """Hide every owned game the state does not allow, and say what happened. + + ``lead`` is the indentation (and blank line, by default) the report lines + start with, so the caller's output stays aligned with its own. + """ + owned_ids = get_all_owned_app_ids(config) + if not owned_ids: + return + hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state)) + if skipped is not None: + _echo(f"{lead}Library hiding: skipped ({skipped})") + elif hidden > 0: + _echo(f"{lead}Library: hid {hidden} games") diff --git a/steam_backlog_enforcer/main/picks.py b/steam_backlog_enforcer/main/picks.py index 0903137..6b50c3b 100644 --- a/steam_backlog_enforcer/main/picks.py +++ b/steam_backlog_enforcer/main/picks.py @@ -9,26 +9,25 @@ from steam_backlog_enforcer._actions import ( active_manual_picks, allowed_app_ids, allowed_games, - apply_manual_pick, + apply_configured_pick, manual_pick_slots_left, ) from steam_backlog_enforcer._allowed_games import ( MANUAL_LOCK_DAYS as _MANUAL_LOCK_DAYS, ) -from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids from steam_backlog_enforcer._hltb_types import load_hltb_cache from steam_backlog_enforcer._pick_completion import ( report_completion, warn_stale_assignment, ) -from steam_backlog_enforcer._snapshot import load_snapshot +from steam_backlog_enforcer._snapshot import load_snapshot, snapshot_game_name from steam_backlog_enforcer.game_install import ( _echo, install_game, is_game_installed, uninstall_other_games, ) -from steam_backlog_enforcer.library_hider import try_hide_other_games +from steam_backlog_enforcer.library_hider import hide_others_and_report from steam_backlog_enforcer.main._shared import _MANUAL_LOCK_EXEMPT_COMMANDS from steam_backlog_enforcer.scanning import pick_next_game from steam_backlog_enforcer.steam_api import ( @@ -57,13 +56,7 @@ def cmd_pick(config: Config, state: State) -> None: pick_next_game(games, state, config) if state.current_app_id is not None: - owned_ids = get_all_owned_app_ids(config) - if owned_ids: - hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state)) - if skipped is not None: - _echo(f"\n Library hiding: skipped ({skipped})") - elif hidden > 0: - _echo(f"\n Library: hid {hidden} games") + hide_others_and_report(config, state) def _resolve_game_name(config: Config, app_id: int) -> str | None: @@ -72,11 +65,9 @@ def _resolve_game_name(config: Config, app_id: int) -> str | None: Returns the game name, or None if not found. """ # Fast path: snapshot already on disk. - snapshot = load_snapshot() - if snapshot: - for entry in snapshot: - if entry.get("app_id") == app_id: - return str(entry["name"]) + name = snapshot_game_name(app_id) + if name is not None: + return name # Slower path: owned games API. try: @@ -145,13 +136,7 @@ def _apply_allowed_set(config: Config, state: State) -> None: _echo(f" Installing {name}...") install_game(app_id, name, config.steam_id, use_steam_protocol=True) - owned_ids = get_all_owned_app_ids(config) - if owned_ids: - hidden, skipped = try_hide_other_games(owned_ids, allowed) - if skipped is not None: - _echo(f" Library hiding: skipped ({skipped})") - elif hidden > 0: - _echo(f" Library: hid {hidden} games") + hide_others_and_report(config, state, lead=" ") def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None: @@ -211,12 +196,7 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None: # State mutation is the shared, stdout-free core (also used by the MCP # server); the destructive post-assignment cascade below stays CLI-only. - refused = apply_manual_pick( - state, - app_id, - game_name, - max_picks=config.max_manual_picks, - ) + refused = apply_configured_pick(config, state, app_id, game_name) if refused is not None: _echo(f"\nError: {refused}") sys.exit(1) diff --git a/steam_backlog_enforcer/main/status.py b/steam_backlog_enforcer/main/status.py index 1295186..60b7734 100644 --- a/steam_backlog_enforcer/main/status.py +++ b/steam_backlog_enforcer/main/status.py @@ -7,7 +7,7 @@ from typing import TYPE_CHECKING from steam_backlog_enforcer._actions import active_manual_picks from steam_backlog_enforcer._manual_pick_lifecycle import manual_pick_age_days from steam_backlog_enforcer._pick_completion import report_completion -from steam_backlog_enforcer._snapshot import load_snapshot +from steam_backlog_enforcer._snapshot import require_snapshot_games from steam_backlog_enforcer._total_block import get_total_block_status from steam_backlog_enforcer.game_install import ( _echo, @@ -15,11 +15,11 @@ from steam_backlog_enforcer.game_install import ( is_protected_app, ) from steam_backlog_enforcer.main._shared import _LIST_DISPLAY_LIMIT -from steam_backlog_enforcer.steam_api import GameInfo from steam_backlog_enforcer.store_blocker import is_store_blocked if TYPE_CHECKING: from steam_backlog_enforcer.config import Config, State + from steam_backlog_enforcer.steam_api import GameInfo def cmd_status(config: Config, state: State) -> None: @@ -74,12 +74,10 @@ def cmd_status(config: Config, state: State) -> None: def cmd_list(_config: Config, state: State) -> None: """List games from the last snapshot.""" - snapshot = load_snapshot() - if snapshot is None: - _echo("No snapshot found. Run 'scan' first.") + games = require_snapshot_games() + if games is None: return - games = [GameInfo.from_snapshot(d) for d in snapshot] incomplete = [g for g in games if not g.is_complete] complete = [g for g in games if g.is_complete] diff --git a/steam_backlog_enforcer/tests/test_main_part2.py b/steam_backlog_enforcer/tests/test_main_part2.py index a63cb82..5ca7a59 100644 --- a/steam_backlog_enforcer/tests/test_main_part2.py +++ b/steam_backlog_enforcer/tests/test_main_part2.py @@ -38,8 +38,14 @@ class TestFinalizeCompletion: patch(f"{CMD_DONE_PKG}._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=(2, None)), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2, 3], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", + return_value=(2, None), + ), patch(f"{CMD_DONE_PKG}.send_notification"), patch.object(State, "save"), ): @@ -65,11 +71,16 @@ class TestFinalizeCompletion: snap = [_snap(2, "NewGame", 10, 0, 5.0)] with ( patch(f"{CMD_DONE_PKG}._echo") as mock_echo, + # The skip line is printed by the shared hide helper. + patch("steam_backlog_enforcer.library_hider._echo", 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", + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2, 3], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", return_value=(0, "update in progress"), ), patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True), @@ -133,7 +144,10 @@ class TestFinalizeCompletion: patch(f"{CMD_DONE_PKG}._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=[]), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), patch(f"{CMD_DONE_PKG}.send_notification"), patch.object(State, "save"), ): @@ -158,8 +172,14 @@ class TestFinalizeCompletion: patch(f"{CMD_DONE_PKG}._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]), - patch(f"{CMD_DONE_PKG}.try_hide_other_games", return_value=(0, None)), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", + return_value=(0, None), + ), patch(f"{CMD_DONE_PKG}.send_notification"), patch.object(State, "save"), ): diff --git a/steam_backlog_enforcer/tests/test_main_part2b.py b/steam_backlog_enforcer/tests/test_main_part2b.py index 685bcd5..fb34747 100644 --- a/steam_backlog_enforcer/tests/test_main_part2b.py +++ b/steam_backlog_enforcer/tests/test_main_part2b.py @@ -51,8 +51,14 @@ 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}.try_hide_other_games", return_value=(1, None)), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", + return_value=(1, None), + ), ): _enforce_on_done(config, state) @@ -67,8 +73,14 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.enforce_allowed_game", return_value=[]), 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}.try_hide_other_games", return_value=(0, None)), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", + return_value=(0, None), + ), ): _enforce_on_done(config, state) @@ -77,10 +89,15 @@ class TestEnforceOnDone: state = State(current_app_id=1, current_game_name="G") with ( patch(f"{CMD_DONE_PKG}._echo") as mock_echo, + # The skip line is printed by the shared hide helper. + patch("steam_backlog_enforcer.library_hider._echo", 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", + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", return_value=(0, "Steam is not installed"), ), ): @@ -100,8 +117,14 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}._echo"), 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}.try_hide_other_games", return_value=(0, None)), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2], + ), + patch( + "steam_backlog_enforcer.library_hider.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_part2c.py b/steam_backlog_enforcer/tests/test_main_part2c.py index c0fb0a5..ec7774c 100644 --- a/steam_backlog_enforcer/tests/test_main_part2c.py +++ b/steam_backlog_enforcer/tests/test_main_part2c.py @@ -93,8 +93,14 @@ class TestFinalizeCompletionRetries: patch(f"{CMD_DONE_PKG}._echo"), 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}.try_hide_other_games", return_value=(1, None)), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2], + ), + patch( + "steam_backlog_enforcer.library_hider.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"), @@ -123,8 +129,14 @@ class TestFinalizeCompletionRetries: patch(f"{CMD_DONE_PKG}._echo"), 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}.try_hide_other_games", return_value=(1, None)), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2], + ), + patch( + "steam_backlog_enforcer.library_hider.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"), diff --git a/steam_backlog_enforcer/tests/test_main_pick_cap.py b/steam_backlog_enforcer/tests/test_main_pick_cap.py index 6dff6e5..a94c985 100644 --- a/steam_backlog_enforcer/tests/test_main_pick_cap.py +++ b/steam_backlog_enforcer/tests/test_main_pick_cap.py @@ -28,7 +28,10 @@ class TestPickManualCap: 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=[]), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), ): cmd_pick_manual(Config(max_manual_picks=2), state, ["489830"]) assert [p["app_id"] for p in state.manual_picks] == [100, 489830] @@ -59,7 +62,10 @@ class TestPickManualCap: 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=[]), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), pytest.raises(SystemExit), ): cmd_pick_manual(Config(max_manual_picks=2), state, ["489830"]) @@ -79,8 +85,14 @@ class TestPickManualCap: patch.object(State, "save"), patch(f"{PKG}.uninstall_other_games", return_value=0) as mock_uninstall, patch(f"{PKG}.is_game_installed", return_value=True), - patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 100, 489830]), - patch(f"{PKG}.try_hide_other_games", return_value=(1, None)) as mock_hide, + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 100, 489830], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", + return_value=(1, None), + ) as mock_hide, ): cmd_pick_manual(Config(max_manual_picks=2), state, ["489830"]) mock_uninstall.assert_called_once_with({100, 489830}) @@ -97,7 +109,10 @@ class TestPickManualCap: patch(f"{PKG}.uninstall_other_games", return_value=0), 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=[]), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), ): cmd_pick_manual(Config(max_manual_picks=2), state, ["489830"]) installed = {call.args[0] for call in mock_install.call_args_list} diff --git a/steam_backlog_enforcer/tests/test_main_pick_manual.py b/steam_backlog_enforcer/tests/test_main_pick_manual.py index 815f6b9..0aa42b6 100644 --- a/steam_backlog_enforcer/tests/test_main_pick_manual.py +++ b/steam_backlog_enforcer/tests/test_main_pick_manual.py @@ -11,16 +11,6 @@ PKG = "steam_backlog_enforcer.main.picks" class TestCmdPickManual: - def _base_patches(self) -> dict[str, object]: - return { - f"{PKG}._resolve_game_name": "Skyrim SE", - f"{PKG}.uninstall_other_games": 2, - f"{PKG}.is_game_installed": False, - f"{PKG}.install_game": None, - f"{PKG}.get_all_owned_app_ids": [1, 2, 489830], - f"{PKG}.try_hide_other_games": (2, None), - } - def test_invalid_app_id(self) -> None: with patch(f"{PKG}._echo") as mock_echo: cmd_pick_manual(Config(), State(), ["abc"]) @@ -55,7 +45,10 @@ class TestCmdPickManual: 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=[]), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), ): cmd_pick_manual(Config(), state, []) assert state.current_app_id == 489830 @@ -71,8 +64,14 @@ class TestCmdPickManual: patch(f"{PKG}.uninstall_other_games", return_value=2) as mock_uninstall, 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}.try_hide_other_games", return_value=(1, None)) as mock_hide, + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 489830], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", + return_value=(1, None), + ) as mock_hide, ): cmd_pick_manual(config, state, ["489830"]) @@ -85,67 +84,6 @@ class TestCmdPickManual: mock_install.assert_called_once() mock_hide.assert_called_once() - def test_no_uninstall_when_config_off(self) -> None: - state = State() - config = Config(uninstall_other_games=False) - with ( - patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), - patch(f"{PKG}._echo"), - patch("builtins.input", return_value="YES"), - patch.object(State, "save"), - patch(f"{PKG}.uninstall_other_games") as mock_uninstall, - patch(f"{PKG}.is_game_installed", return_value=True), - patch(f"{PKG}.get_all_owned_app_ids", return_value=[]), - ): - cmd_pick_manual(config, state, ["489830"]) - mock_uninstall.assert_not_called() - - def test_game_already_installed_skips_install(self) -> None: - state = State() - with ( - patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), - patch(f"{PKG}._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}.install_game") as mock_install, - patch(f"{PKG}.get_all_owned_app_ids", return_value=[]), - ): - cmd_pick_manual(Config(), state, ["489830"]) - mock_install.assert_not_called() - - def test_no_hide_when_no_owned_ids(self) -> None: - state = State() - with ( - patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), - patch(f"{PKG}._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=[]), - patch(f"{PKG}.try_hide_other_games") as mock_hide, - ): - cmd_pick_manual(Config(), state, ["489830"]) - mock_hide.assert_not_called() - - def test_uninstall_returns_zero_no_echo(self) -> None: - state = State() - config = Config(uninstall_other_games=True) - 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=[]), - ): - cmd_pick_manual(config, state, ["489830"]) - output = " ".join(str(c) for c in mock_echo.call_args_list) - assert "Uninstalled 0" not in output - def test_enforcement_started_at_set_when_empty(self) -> None: state = State(enforcement_started_at="") with ( @@ -155,7 +93,10 @@ class TestCmdPickManual: 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=[]), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), ): cmd_pick_manual(Config(), state, ["489830"]) assert state.enforcement_started_at != "" @@ -170,51 +111,10 @@ class TestCmdPickManual: 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=[]), - ): - cmd_pick_manual(Config(), state, ["489830"]) - assert state.enforcement_started_at == existing_ts - - def test_hide_returns_zero_no_echo(self) -> None: - 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, 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"), + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], ), ): 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 [p["app_id"] for p in state.manual_picks] == [489830] - - -# ────────────────────────────────────────────────────────────── -# main() dispatch to pick-manual -# ────────────────────────────────────────────────────────────── + assert state.enforcement_started_at == existing_ts diff --git a/steam_backlog_enforcer/tests/test_main_pick_manual_cascade.py b/steam_backlog_enforcer/tests/test_main_pick_manual_cascade.py new file mode 100644 index 0000000..460907d --- /dev/null +++ b/steam_backlog_enforcer/tests/test_main_pick_manual_cascade.py @@ -0,0 +1,146 @@ +"""Tests for pick-manual's post-confirmation cascade. + +Uninstall, install and library hiding after a confirmed pick. Split from +test_main_pick_manual.py (prompting and state) to keep both under 250 lines. +""" + +from unittest.mock import patch + +from steam_backlog_enforcer.config import Config, State +from steam_backlog_enforcer.main import ( + cmd_pick_manual, +) + +PKG = "steam_backlog_enforcer.main.picks" + + +class TestPickManualCascade: + def test_no_uninstall_when_config_off(self) -> None: + state = State() + config = Config(uninstall_other_games=False) + with ( + patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), + patch(f"{PKG}._echo"), + patch("builtins.input", return_value="YES"), + patch.object(State, "save"), + patch(f"{PKG}.uninstall_other_games") as mock_uninstall, + patch(f"{PKG}.is_game_installed", return_value=True), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), + ): + cmd_pick_manual(config, state, ["489830"]) + mock_uninstall.assert_not_called() + + def test_game_already_installed_skips_install(self) -> None: + state = State() + with ( + patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), + patch(f"{PKG}._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}.install_game") as mock_install, + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), + ): + cmd_pick_manual(Config(), state, ["489830"]) + mock_install.assert_not_called() + + def test_no_hide_when_no_owned_ids(self) -> None: + state = State() + with ( + patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"), + patch(f"{PKG}._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( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games" + ) as mock_hide, + ): + cmd_pick_manual(Config(), state, ["489830"]) + mock_hide.assert_not_called() + + def test_uninstall_returns_zero_no_echo(self) -> None: + state = State() + config = Config(uninstall_other_games=True) + 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( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), + ): + cmd_pick_manual(config, state, ["489830"]) + output = " ".join(str(c) for c in mock_echo.call_args_list) + assert "Uninstalled 0" not in output + + def test_hide_returns_zero_no_echo(self) -> None: + 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( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2], + ), + patch( + "steam_backlog_enforcer.library_hider.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, + # The skip line is printed by the shared hide helper. + patch("steam_backlog_enforcer.library_hider._echo", 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( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2], + ), + patch( + "steam_backlog_enforcer.library_hider.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 [p["app_id"] for p in state.manual_picks] == [489830] + + +# ────────────────────────────────────────────────────────────── +# main() dispatch to pick-manual +# ────────────────────────────────────────────────────────────── diff --git a/steam_backlog_enforcer/tests/test_main_picks.py b/steam_backlog_enforcer/tests/test_main_picks.py index 035d80c..46c04ea 100644 --- a/steam_backlog_enforcer/tests/test_main_picks.py +++ b/steam_backlog_enforcer/tests/test_main_picks.py @@ -34,7 +34,10 @@ class TestCmdPick: patch(f"{PKG}.load_snapshot", return_value=snapshot), patch(f"{PKG}.load_hltb_cache", return_value={2: 5.0}), patch(f"{PKG}.pick_next_game") as mock_pick, - patch(f"{PKG}.get_all_owned_app_ids", return_value=[]), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), ): config = Config(steam_api_key="k", steam_id="i") state = State() @@ -48,8 +51,14 @@ class TestCmdPick: patch(f"{PKG}.load_snapshot", return_value=snapshot), 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}.try_hide_other_games", return_value=(2, None)) as mock_hide, + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2, 3], + ), + patch( + "steam_backlog_enforcer.library_hider.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) @@ -62,8 +71,14 @@ class TestCmdPick: patch(f"{PKG}.load_snapshot", return_value=snapshot), 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, None)), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2, 3], + ), + patch( + "steam_backlog_enforcer.library_hider.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) @@ -76,12 +91,17 @@ class TestCmdPick: patch(f"{PKG}.load_snapshot", return_value=snapshot), 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", + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[1, 2, 3], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games", return_value=(0, "Steam is not installed"), ), patch(f"{PKG}._echo") as mock_echo, + # The skip line is printed by the shared hide helper. + patch("steam_backlog_enforcer.library_hider._echo", 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) @@ -93,7 +113,9 @@ class TestCmdPick: patch(f"{PKG}.load_snapshot", return_value=snapshot), patch(f"{PKG}.load_hltb_cache", return_value={}), patch(f"{PKG}.pick_next_game"), - patch(f"{PKG}.get_all_owned_app_ids") as mock_owned, + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids" + ) as mock_owned, ): cmd_pick(Config(steam_api_key="k", steam_id="i"), State()) mock_owned.assert_not_called() @@ -105,8 +127,13 @@ class TestCmdPick: patch(f"{PKG}.load_snapshot", return_value=snapshot), 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}.try_hide_other_games") as mock_hide, + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), + patch( + "steam_backlog_enforcer.library_hider.try_hide_other_games" + ) as mock_hide, ): cmd_pick(Config(steam_api_key="k", steam_id="i"), state) mock_hide.assert_not_called() @@ -124,7 +151,10 @@ class TestCmdPick: patch(f"{PKG}.load_snapshot", return_value=snapshot), patch(f"{PKG}.load_hltb_cache", return_value={2: 7.5}), patch(f"{PKG}.pick_next_game", side_effect=capture_pick), - patch(f"{PKG}.get_all_owned_app_ids", return_value=[]), + patch( + "steam_backlog_enforcer.library_hider.get_all_owned_app_ids", + return_value=[], + ), ): cmd_pick(config, state) @@ -143,7 +173,9 @@ class TestResolveGameName: "playtime_minutes": 0, } ] - with patch(f"{PKG}.load_snapshot", return_value=snapshot): + with patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ): result = _resolve_game_name(Config(), 440) assert result == "TF2" diff --git a/steam_backlog_enforcer/tests/test_main_status.py b/steam_backlog_enforcer/tests/test_main_status.py index c6bd4cb..78ebc06 100644 --- a/steam_backlog_enforcer/tests/test_main_status.py +++ b/steam_backlog_enforcer/tests/test_main_status.py @@ -49,9 +49,10 @@ class TestCmdList: def test_no_snapshot(self) -> None: with ( - patch(f"{PKG}.load_snapshot", return_value=None), + patch("steam_backlog_enforcer._snapshot.load_snapshot", return_value=None), patch(f"{PKG}.report_completion", return_value=[]), - patch(f"{PKG}._echo") as mock_echo, + # The "run scan" hint is printed by the shared snapshot helper. + patch("steam_backlog_enforcer._snapshot._echo") as mock_echo, ): cmd_list(Config(), State()) assert any("No snapshot" in str(c) for c in mock_echo.call_args_list) @@ -64,7 +65,9 @@ class TestCmdList: ] state = State(current_app_id=1) with ( - patch(f"{PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo"), ): @@ -73,7 +76,9 @@ class TestCmdList: def test_many_games(self) -> None: snapshot = [snap(i, f"Game{i}") for i in range(60)] with ( - patch(f"{PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch(f"{PKG}.report_completion", return_value=[]), patch(f"{PKG}._echo") as mock_echo, ): diff --git a/steam_backlog_enforcer/tests/test_mcp.py b/steam_backlog_enforcer/tests/test_mcp.py index 3effda6..d59e57c 100644 --- a/steam_backlog_enforcer/tests/test_mcp.py +++ b/steam_backlog_enforcer/tests/test_mcp.py @@ -6,7 +6,6 @@ from unittest.mock import patch from steam_backlog_enforcer import _mcp_actions as mcp_actions from steam_backlog_enforcer import _mcp_query as mcp_query -from steam_backlog_enforcer import _mcp_server as mcp_server from steam_backlog_enforcer.steam_api import GameInfo @@ -120,14 +119,16 @@ class TestListBacklog: class TestResolveAndSort: def test_resolve_found(self) -> None: - with patch.object( - mcp_server, "load_snapshot", return_value=[{"app_id": 440, "name": "TF2"}] + with patch( + "steam_backlog_enforcer._snapshot.load_snapshot", + return_value=[{"app_id": 440, "name": "TF2"}], ): assert mcp_actions._resolve_game_name(440) == "TF2" def test_resolve_missing(self) -> None: - with patch.object( - mcp_server, "load_snapshot", return_value=[{"app_id": 1, "name": "X"}] + with patch( + "steam_backlog_enforcer._snapshot.load_snapshot", + return_value=[{"app_id": 1, "name": "X"}], ): assert mcp_actions._resolve_game_name(440) is None diff --git a/steam_backlog_enforcer/tests/test_mcp_actions.py b/steam_backlog_enforcer/tests/test_mcp_actions.py index bf023d8..d4c8dc6 100644 --- a/steam_backlog_enforcer/tests/test_mcp_actions.py +++ b/steam_backlog_enforcer/tests/test_mcp_actions.py @@ -7,7 +7,6 @@ from unittest.mock import patch from steam_backlog_enforcer import _mcp_actions as mcp_actions from steam_backlog_enforcer import _mcp_query as mcp_query -from steam_backlog_enforcer import _mcp_server as mcp_server from steam_backlog_enforcer.config import State @@ -20,12 +19,11 @@ class TestPickManualGate: def test_preview_does_not_mutate(self) -> None: with ( - patch.object( - mcp_server, - "load_snapshot", + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=[{"app_id": 440, "name": "TF2"}], ), - patch.object(mcp_actions, "apply_manual_pick") as amp, + patch("steam_backlog_enforcer._actions.apply_manual_pick") as amp, ): out = mcp_actions.pick_manual(440) assert out["preview"] is True @@ -34,14 +32,15 @@ class TestPickManualGate: def test_confirm_applies(self) -> None: with ( - patch.object( - mcp_server, - "load_snapshot", + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=[{"app_id": 440, "name": "TF2"}], ), patch.object(mcp_actions, "State") as state, patch.object(mcp_actions, "Config") as config, - patch.object(mcp_actions, "apply_manual_pick", return_value=None) as amp, + patch( + "steam_backlog_enforcer._actions.apply_manual_pick", return_value=None + ) as amp, ): out = mcp_actions.pick_manual(440, confirm=True) assert out["applied"] is True @@ -113,14 +112,16 @@ class TestAbandonPickGate: def test_refused_at_cap(self) -> None: with ( - patch.object( - mcp_server, - "load_snapshot", + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=[{"app_id": 440, "name": "TF2"}], ), patch.object(mcp_actions, "State"), patch.object(mcp_actions, "Config"), - patch.object(mcp_actions, "apply_manual_pick", return_value="cap reached"), + patch( + "steam_backlog_enforcer._actions.apply_manual_pick", + return_value="cap reached", + ), ): out = mcp_actions.pick_manual(440, confirm=True) assert out["ok"] is False diff --git a/steam_backlog_enforcer/tests/test_stats_cmd_stats.py b/steam_backlog_enforcer/tests/test_stats_cmd_stats.py index 3833992..6554c5c 100644 --- a/steam_backlog_enforcer/tests/test_stats_cmd_stats.py +++ b/steam_backlog_enforcer/tests/test_stats_cmd_stats.py @@ -73,7 +73,9 @@ class TestCmdStats: ) echoed: list[str] = [] with ( - patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch( f"{_PKG}._filter_qualifying_games", return_value=([entry], hltb_skip, linux_skip, no_data_skip), @@ -92,8 +94,12 @@ class TestCmdStats: echoed: list[str] = [] state = State() with ( - patch(f"{_PKG}.load_snapshot", return_value=None), - patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + patch("steam_backlog_enforcer._snapshot.load_snapshot", return_value=None), + # The "run scan" hint is printed by the shared snapshot helper. + patch( + "steam_backlog_enforcer._snapshot._echo", + side_effect=lambda *a, **_: echoed.append(a[0]), + ), ): cmd_stats(self._config(), state) assert any("No snapshot found" in s for s in echoed) @@ -153,7 +159,9 @@ class TestCmdStats: ) echoed: list[str] = [] with ( - patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch( f"{_PKG}._filter_qualifying_games", return_value=([entry], 0, 0, 0), @@ -184,7 +192,9 @@ class TestCmdStats: ) echoed: list[str] = [] with ( - patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch( f"{_PKG}._filter_qualifying_games", return_value=([entry], 0, 0, 0), @@ -206,7 +216,9 @@ class TestCmdStats: snapshot = [self._snapshot_game()] echoed: list[str] = [] with ( - patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch( f"{_PKG}._filter_qualifying_games", return_value=([], 0, 0, 0), diff --git a/steam_backlog_enforcer/tests/test_stats_cmd_stats_refilter_called_when_ensure_rush_data_re.py b/steam_backlog_enforcer/tests/test_stats_cmd_stats_refilter_called_when_ensure_rush_data_re.py index 71f2407..f447449 100644 --- a/steam_backlog_enforcer/tests/test_stats_cmd_stats_refilter_called_when_ensure_rush_data_re.py +++ b/steam_backlog_enforcer/tests/test_stats_cmd_stats_refilter_called_when_ensure_rush_data_re.py @@ -75,7 +75,9 @@ class TestCmdStatsGroup2: ) echoed: list[str] = [] with ( - patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch( f"{_PKG}._filter_qualifying_games", return_value=([entry], hltb_skip, linux_skip, no_data_skip), @@ -107,7 +109,9 @@ class TestCmdStatsGroup2: return [entry], 0, 0, 0 with ( - patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch(f"{_PKG}._filter_qualifying_games", side_effect=count_filter), patch(f"{_PKG}._ensure_completed_rush_data", return_value=False), patch(f"{_PKG}._ensure_rush_data", return_value=True), @@ -162,7 +166,9 @@ class TestCmdStatsGroup2: captured["games_done"] = games_done with ( - patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch( f"{_PKG}._filter_qualifying_games", return_value=([entry], 0, 0, 0), @@ -201,7 +207,9 @@ class TestCmdStatsGroup2: captured["leisure"] = leisure with ( - patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + "steam_backlog_enforcer._snapshot.load_snapshot", return_value=snapshot + ), patch( f"{_PKG}._filter_qualifying_games", return_value=([entry], 0, 0, 0),