diff --git a/steam_backlog_enforcer/_cmd_done.py b/steam_backlog_enforcer/_cmd_done.py index 5808d3f..834eb41 100644 --- a/steam_backlog_enforcer/_cmd_done.py +++ b/steam_backlog_enforcer/_cmd_done.py @@ -3,12 +3,9 @@ from __future__ import annotations import logging +import sys from python_pkg.steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids -from python_pkg.steam_backlog_enforcer._scanning_confidence import ( - _confidence_fail_reasons, - _refresh_candidate_confidence, -) from python_pkg.steam_backlog_enforcer.config import Config, State, load_snapshot from python_pkg.steam_backlog_enforcer.enforcer import ( enforce_allowed_game, @@ -24,21 +21,49 @@ from python_pkg.steam_backlog_enforcer.hltb import ( fetch_hltb_confidence_cached, fetch_hltb_times_cached, load_hltb_cache, - load_hltb_count_comp_cache, load_hltb_polls_cache, save_hltb_cache, ) from python_pkg.steam_backlog_enforcer.library_hider import hide_other_games -from python_pkg.steam_backlog_enforcer.scanning import ( - _pick_next_shortest_candidate, - pick_next_game, -) +from python_pkg.steam_backlog_enforcer.scanning import pick_next_game from python_pkg.steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient _REASSIGN_REFRESH_LIMIT = 50 +_SKIP_DAYS = 7 logger = logging.getLogger(__name__) +def _prompt_keep_or_skip(game: GameInfo) -> bool: + """Ask the user whether to keep the freshly-picked ``game``. + + Returns ``True`` to accept the pick, ``False`` to skip it (which the + caller will translate into a 7-day skip entry on ``State``). When + stdin is not a TTY (e.g. background daemon, piped invocation), the + pick is accepted silently to preserve the legacy non-interactive + behaviour. + """ + if not sys.stdin.isatty(): + return True + hours_str = "" + if game.completionist_hours > 0: + hours_str = f" (~{game.completionist_hours:.1f}h leisure+dlc)" + _echo(f"\n Next pick: {game.name} (AppID={game.app_id}){hours_str}") + while True: + try: + answer = ( + input(f" Keep this game? [Y/n] (n = skip for {_SKIP_DAYS} days): ") + .strip() + .lower() + ) + except EOFError: + return True + if answer in {"", "y", "yes"}: + return True + if answer in {"n", "no"}: + return False + _echo(" Please answer 'y' or 'n'.") + + def _backfill_polls_for_finished( state: State, extra_app_id: int | None = None, @@ -124,17 +149,6 @@ def _apply_cached_hours_to_games( game.completionist_hours = hltb_cache[game.app_id] -def _apply_cached_confidence_to_games(games: list[GameInfo]) -> None: - """Overlay cached confidence counters onto snapshot-backed game objects.""" - polls_cache = load_hltb_polls_cache() - count_comp_cache = load_hltb_count_comp_cache() - for game in games: - if game.app_id in polls_cache: - game.comp_100_count = polls_cache[game.app_id] - if game.app_id in count_comp_cache: - game.count_comp = count_comp_cache[game.app_id] - - def _refresh_uncached_shortlist_hours( games: list[GameInfo], hltb_cache: dict[int, float], @@ -166,117 +180,6 @@ def _refresh_uncached_shortlist_hours( hltb_cache.update(refreshed) -def _should_reassign_candidate( - playable: GameInfo, - current_hours: float, - *, - force_reassign: bool, -) -> bool: - """Return whether a playable candidate should trigger reassignment.""" - if force_reassign: - return True - if current_hours > 0: - return playable.completionist_hours < current_hours - return True - - -def _echo_reassign_decision( - playable: GameInfo, - current_hours: float, - current_fail_reasons: list[str], - *, - force_reassign: bool, -) -> None: - """Emit a human-readable reassignment reason.""" - if force_reassign: - _echo( - f"\n Reassigning: current game confidence too low " - f"({'; '.join(current_fail_reasons)})" - ) - return - if current_hours > 0: - _echo( - f"\n Reassigning: {playable.name} is shorter" - f" (~{playable.completionist_hours:.1f}h vs ~{current_hours:.1f}h)" - ) - return - _echo( - f"\n Reassigning: current game has no usable HLTB time; " - f"picked {playable.name} (~{playable.completionist_hours:.1f}h)" - ) - - -def _try_reassign_shorter_game( - hltb_cache: dict[int, float], - app_id: int, - hours: float, - state: State, - config: Config, -) -> bool: - """Check if a shorter game is available and reassign if so.""" - snapshot_data = load_snapshot() - if not snapshot_data: - return False - all_games = [GameInfo.from_snapshot(d) for d in snapshot_data] - skip = set(state.finished_app_ids) - _refresh_uncached_shortlist_hours( - all_games, - hltb_cache, - skip, - upper_bound_hours=hours, - ) - _apply_cached_hours_to_games(all_games, hltb_cache) - _apply_cached_confidence_to_games(all_games) - current_game = next((g for g in all_games if g.app_id == app_id), None) - if current_game is not None and _confidence_fail_reasons(current_game): - _refresh_candidate_confidence(current_game) - current_fail_reasons = ( - _confidence_fail_reasons(current_game) if current_game is not None else [] - ) - force_reassign = bool(current_fail_reasons) - candidates = [ - g - for g in all_games - if not g.is_complete and g.app_id not in skip and g.completionist_hours > 0 - ] - if not force_reassign and hours > 0: - candidates = [g for g in candidates if g.completionist_hours < hours] - - candidates.sort(key=lambda g: g.completionist_hours) - candidates = [c for c in candidates if c.app_id != app_id] - if not candidates: - return False - - playable, _confidence_skipped, _linux_skipped = _pick_next_shortest_candidate( - candidates, - ) - if playable is None: - return False - - if not _should_reassign_candidate( - playable, - hours, - force_reassign=force_reassign, - ): - return False - _echo_reassign_decision( - playable, - hours, - current_fail_reasons, - force_reassign=force_reassign, - ) - pick_next_game(all_games, state, config) - - 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: - _echo(f"\n Library: hid {hidden} games") - - return True - - def _finalize_completion( config: Config, state: State, @@ -298,10 +201,10 @@ def _finalize_completion( games = [GameInfo.from_snapshot(d) for d in snapshot_data] hltb_cache = load_hltb_cache() - skip = set(state.finished_app_ids) + skip = set(state.finished_app_ids) | state.active_skipped_ids() _refresh_uncached_shortlist_hours(games, hltb_cache, skip) _apply_cached_hours_to_games(games, hltb_cache) - pick_next_game(games, state, config) + pick_next_game(games, state, config, on_select=_prompt_keep_or_skip) if state.current_app_id is None: _echo(" No more games to assign!") @@ -416,9 +319,6 @@ def cmd_done(config: Config, state: State) -> None: _echo(f" HLTB leisure+dlc estimate: {hours:.1f} hours") _report_assigned_confidence(app_id, state) - if _try_reassign_shorter_game(hltb_cache, app_id, hours, state, config): - return - if not game.is_complete: remaining = game.total_achievements - game.unlocked_achievements _echo(f"\n NOT COMPLETE: {remaining} achievements remaining. Keep going!") diff --git a/steam_backlog_enforcer/config.py b/steam_backlog_enforcer/config.py index a0be6eb..79c84fb 100644 --- a/steam_backlog_enforcer/config.py +++ b/steam_backlog_enforcer/config.py @@ -4,6 +4,7 @@ from __future__ import annotations import contextlib from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone import json import logging import os @@ -85,6 +86,41 @@ class State: current_app_id: int | None = None current_game_name: str = "" finished_app_ids: list[int] = field(default_factory=list) + skipped_until: dict[str, str] = field(default_factory=dict) + """Map of ``str(app_id)`` → ISO-8601 UTC timestamp when the skip expires. + + Games in this map are excluded from auto-assignment until the timestamp + elapses. Populated when the user declines a freshly-picked game via the + interactive prompt in ``cmd_done``. + """ + + def skip_for_days(self, app_id: int, days: int) -> None: + """Mark ``app_id`` as skipped for ``days`` days from now (UTC).""" + expires = datetime.now(timezone.utc) + timedelta(days=days) + self.skipped_until[str(app_id)] = expires.isoformat() + + def active_skipped_ids(self) -> set[int]: + """Return currently-skipped app IDs and prune expired entries. + + Mutates ``self.skipped_until`` to drop expired or malformed entries. + Callers should ``save()`` if they want the prune persisted. + """ + now = datetime.now(timezone.utc) + active: set[int] = set() + to_remove: list[str] = [] + for aid_str, ts in self.skipped_until.items(): + try: + expiry = datetime.fromisoformat(ts) + except ValueError: + to_remove.append(aid_str) + continue + if expiry > now: + active.add(int(aid_str)) + else: + to_remove.append(aid_str) + for aid_str in to_remove: + del self.skipped_until[aid_str] + return active def save(self) -> None: """Persist state to disk.""" diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index 7566272..ce18006 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -4,7 +4,7 @@ from __future__ import annotations import logging import time -from typing import Any +from typing import TYPE_CHECKING, Any from python_pkg.steam_backlog_enforcer._hltb_types import ( load_hltb_count_comp_cache, @@ -39,6 +39,9 @@ from python_pkg.steam_backlog_enforcer.protondb import ( ) from python_pkg.steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient +if TYPE_CHECKING: + from collections.abc import Callable + logger = logging.getLogger(__name__) _TAMPER_CHECK_LIMIT = 3 @@ -249,14 +252,77 @@ def _assign_chosen_game( ) -def pick_next_game(games: list[GameInfo], state: State, config: Config) -> None: +def _pick_next_game_sequential( + games: list[GameInfo], + state: State, + config: Config, + on_select: Callable[[GameInfo], bool], +) -> None: + """Pick the next-shortest playable game, asking the user per candidate. + + ``on_select`` is called with each prospective pick. Returning ``True`` + accepts the assignment; returning ``False`` records a 7-day skip on + ``state`` for that game and the next candidate is evaluated. + """ + while True: + skip = set(state.finished_app_ids) | state.active_skipped_ids() + candidates = [g for g in games if not g.is_complete and g.app_id not in skip] + if not candidates: + _echo(_NO_CONF_MSG) + state.current_app_id = None + state.current_game_name = "" + state.save() + return + + candidates.sort(key=_sort_key) + _apply_cached_confidence_to_candidates(candidates) + chosen, confidence_skipped, linux_skipped = _pick_next_shortest_candidate( + candidates + ) + if chosen is None: + _echo( + _NO_CONF_MSG + if confidence_skipped > 0 and linux_skipped == 0 + else "\nNo playable games left (all have poor ProtonDB ratings)!" + ) + state.current_app_id = None + state.current_game_name = "" + state.save() + return + + if not on_select(chosen): + state.skip_for_days(chosen.app_id, 7) + state.save() + _echo(f"\n Skipped {chosen.name} for 7 days; picking next...") + continue + + _assign_chosen_game(chosen, games, state, config) + return + + +def pick_next_game( + games: list[GameInfo], + state: State, + config: Config, + *, + on_select: Callable[[GameInfo], bool] | None = None, +) -> None: """Present a ranked list of eligible games and let the user pick one. Games are ranked by shortest completionist time first. Games with silver-or-worse ProtonDB ratings (or gold trending downward) are excluded as unplayable on Linux. + + If ``on_select`` is provided, the legacy 10-candidate picker is + bypassed: the function instead presents the shortest playable + candidate to ``on_select`` (typically a yes/no prompt) and, if the + callback rejects it, records a 7-day skip and re-evaluates. """ - skip = set(state.finished_app_ids) + if on_select is not None: + _pick_next_game_sequential(games, state, config, on_select) + return + + skip = set(state.finished_app_ids) | state.active_skipped_ids() candidates = [g for g in games if not g.is_complete and g.app_id not in skip] if not candidates: diff --git a/steam_backlog_enforcer/tests/test_cmd_done.py b/steam_backlog_enforcer/tests/test_cmd_done.py deleted file mode 100644 index 36230c6..0000000 --- a/steam_backlog_enforcer/tests/test_cmd_done.py +++ /dev/null @@ -1,447 +0,0 @@ -"""Tests for _cmd_done module.""" - -from __future__ import annotations - -from unittest.mock import patch - -from python_pkg.steam_backlog_enforcer._cmd_done import ( - _try_reassign_shorter_game, -) -from python_pkg.steam_backlog_enforcer.config import Config, State -from python_pkg.steam_backlog_enforcer.steam_api import GameInfo - -CMD_DONE_PKG = "python_pkg.steam_backlog_enforcer._cmd_done" - - -def _snap(**overrides: object) -> dict[str, object]: - snapshot: dict[str, object] = { - "app_id": 1, - "name": "G", - "total_achievements": 10, - "unlocked_achievements": 0, - "playtime_minutes": 60, - "completionist_hours": -1, - "comp_100_count": 3, - "count_comp": 15, - } - snapshot["app_id"] = overrides.get("app_id", 1) - snapshot.update(overrides) - return snapshot - - -class TestTryReassignShorterGame: - """Tests for _try_reassign_shorter_game.""" - - def test_no_snapshot(self) -> None: - with patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=None): - assert not _try_reassign_shorter_game({}, 1, 10.0, State(), Config()) - - def test_no_shorter_candidate(self) -> None: - snap = [ - _snap( - app_id=1, name="G", unlocked_achievements=5, completionist_hours=10.0 - ), - _snap(app_id=2, name="H", unlocked_achievements=5), - ] - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch(f"{CMD_DONE_PKG}._echo"), - ): - result = _try_reassign_shorter_game( - {1: 10.0}, - 1, - 10.0, - State(), - Config(), - ) - assert not result - - def test_reassigns(self) -> None: - snap = [ - _snap( - app_id=1, - name="Long", - unlocked_achievements=5, - completionist_hours=100.0, - ), - _snap( - app_id=2, name="Short", unlocked_achievements=5, completionist_hours=5.0 - ), - ] - state = State(current_app_id=2, current_game_name="Short") - short_game = GameInfo( - app_id=2, - name="Short", - total_achievements=10, - unlocked_achievements=5, - playtime_minutes=60, - completionist_hours=5.0, - ) - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch(f"{CMD_DONE_PKG}._echo"), - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(short_game, 0, 0), - ), - patch(f"{CMD_DONE_PKG}.pick_next_game"), - 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=5) as mock_hide, - ): - result = _try_reassign_shorter_game( - {1: 100.0, 2: 5.0}, - 1, - 100.0, - state, - Config(), - ) - assert result - mock_hide.assert_called_once_with([1, 2, 3], 2) - - def test_reassigns_no_hide_when_no_owned_ids(self) -> None: - snap = [ - _snap( - app_id=1, - name="Long", - unlocked_achievements=5, - completionist_hours=100.0, - ), - _snap( - app_id=2, name="Short", unlocked_achievements=5, completionist_hours=5.0 - ), - ] - state = State(current_app_id=2, current_game_name="Short") - short_game = GameInfo( - app_id=2, - name="Short", - total_achievements=10, - unlocked_achievements=5, - playtime_minutes=60, - completionist_hours=5.0, - ) - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch(f"{CMD_DONE_PKG}._echo") as mock_echo, - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(short_game, 0, 0), - ), - patch(f"{CMD_DONE_PKG}.pick_next_game"), - 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), - ): - result = _try_reassign_shorter_game( - {1: 100.0, 2: 5.0}, - 1, - 100.0, - state, - Config(), - ) - assert result - # hidden == 0, so "hid N games" should NOT be echoed - for call in mock_echo.call_args_list: - assert "hid" not in str(call) - - def test_reassigns_skip_hide_when_no_app_assigned(self) -> None: - snap = [ - _snap( - app_id=1, - name="Long", - unlocked_achievements=5, - completionist_hours=100.0, - ), - _snap( - app_id=2, name="Short", unlocked_achievements=5, completionist_hours=5.0 - ), - ] - state = State(current_app_id=None, current_game_name="") - short_game = GameInfo( - app_id=2, - name="Short", - total_achievements=10, - unlocked_achievements=5, - playtime_minutes=60, - completionist_hours=5.0, - ) - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch(f"{CMD_DONE_PKG}._echo"), - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(short_game, 0, 0), - ), - patch(f"{CMD_DONE_PKG}.pick_next_game"), - patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids") as mock_owned, - patch(f"{CMD_DONE_PKG}.hide_other_games") as mock_hide, - ): - result = _try_reassign_shorter_game( - {1: 100.0, 2: 5.0}, - 1, - 100.0, - state, - Config(), - ) - assert result - mock_owned.assert_not_called() - mock_hide.assert_not_called() - - def test_playable_none(self) -> None: - snap = [ - _snap( - app_id=1, - name="Long", - unlocked_achievements=5, - completionist_hours=100.0, - ), - _snap( - app_id=2, name="Short", unlocked_achievements=5, completionist_hours=5.0 - ), - ] - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(None, 0, 0), - ), - patch(f"{CMD_DONE_PKG}._echo"), - ): - result = _try_reassign_shorter_game( - {1: 100.0, 2: 5.0}, - 1, - 100.0, - State(), - Config(), - ) - assert not result - - def test_playable_longer(self) -> None: - """Playable candidate is longer than current — no reassign.""" - snap = [ - _snap( - app_id=1, - name="Short", - unlocked_achievements=5, - completionist_hours=10.0, - ), - _snap( - app_id=2, - name="Long", - unlocked_achievements=5, - completionist_hours=200.0, - ), - ] - long_game = GameInfo( - app_id=2, - name="Long", - total_achievements=10, - unlocked_achievements=5, - playtime_minutes=60, - completionist_hours=200.0, - ) - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(long_game, 0, 0), - ), - patch(f"{CMD_DONE_PKG}._echo"), - ): - result = _try_reassign_shorter_game( - {1: 10.0, 2: 200.0}, - 1, - 10.0, - State(), - Config(), - ) - assert not result - - def test_refreshes_stale_shorter_snapshot_entry(self) -> None: - """Uncached shorter snapshot candidates are refreshed before reassigning.""" - snap = [ - _snap( - app_id=1, - name="Current", - unlocked_achievements=5, - completionist_hours=20.1, - ), - _snap(app_id=2, name="Lacuna", completionist_hours=0.9), - ] - state = State(current_app_id=1, current_game_name="Current") - refreshed_short = GameInfo( - app_id=2, - name="Lacuna", - total_achievements=10, - unlocked_achievements=0, - playtime_minutes=60, - completionist_hours=18.8, - ) - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch( - f"{CMD_DONE_PKG}.fetch_hltb_times_cached", - return_value={2: 18.8}, - ) as mock_fetch_hltb, - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(refreshed_short, 0, 0), - ) as mock_pick_candidate, - patch(f"{CMD_DONE_PKG}.pick_next_game"), - patch(f"{CMD_DONE_PKG}._echo"), - patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[]), - patch(f"{CMD_DONE_PKG}.hide_other_games"), - ): - result = _try_reassign_shorter_game( - {1: 20.1}, - 1, - 20.1, - state, - Config(), - ) - - assert result - mock_fetch_hltb.assert_called_once_with([(2, "Lacuna")]) - mock_pick_candidate.assert_called_once() - - def test_reassigns_when_current_confidence_too_low(self) -> None: - """If current game fails confidence thresholds, reassign anyway.""" - snap = [ - _snap( - app_id=1, - name="Current", - unlocked_achievements=5, - completionist_hours=20.0, - comp_100_count=0, - count_comp=0, - ), - _snap( - app_id=2, - name="Confident", - unlocked_achievements=5, - completionist_hours=25.0, - ), - ] - state = State(current_app_id=2, current_game_name="Confident") - confident_game = GameInfo( - app_id=2, - name="Confident", - total_achievements=10, - unlocked_achievements=5, - playtime_minutes=60, - completionist_hours=25.0, - comp_100_count=3, - count_comp=15, - ) - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(confident_game, 0, 0), - ), - patch(f"{CMD_DONE_PKG}.pick_next_game"), - patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[]), - patch(f"{CMD_DONE_PKG}.hide_other_games"), - patch(f"{CMD_DONE_PKG}._echo") as mock_echo, - ): - result = _try_reassign_shorter_game( - {1: 20.0, 2: 25.0}, - 1, - 20.0, - state, - Config(), - ) - - assert result - assert any( - "confidence too low" in str(call).lower() - for call in mock_echo.call_args_list - ) - - def test_does_not_force_refresh_current_when_cached_confidence_is_good( - self, - ) -> None: - """Current-game confidence check should use cache-backed values first.""" - snap = [ - _snap( - app_id=1, - name="Current", - unlocked_achievements=5, - completionist_hours=20.0, - comp_100_count=0, - count_comp=0, - ), - _snap( - app_id=2, - name="Shorter", - unlocked_achievements=5, - completionist_hours=5.0, - comp_100_count=3, - count_comp=15, - ), - ] - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch(f"{CMD_DONE_PKG}.load_hltb_polls_cache", return_value={1: 36, 2: 20}), - patch( - f"{CMD_DONE_PKG}.load_hltb_count_comp_cache", - return_value={1: 200, 2: 50}, - ), - patch(f"{CMD_DONE_PKG}._refresh_candidate_confidence") as mock_refresh, - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(None, 0, 0), - ), - patch(f"{CMD_DONE_PKG}._echo"), - ): - result = _try_reassign_shorter_game( - {1: 20.0, 2: 5.0}, - 1, - 20.0, - State(), - Config(), - ) - - assert not result - mock_refresh.assert_not_called() - - def test_only_checks_strictly_shorter_candidates_when_not_forced(self) -> None: - """No confidence checks should run for non-shorter games.""" - snap = [ - _snap( - app_id=1, - name="Current", - unlocked_achievements=5, - completionist_hours=4.0, - comp_100_count=10, - count_comp=40, - ), - _snap( - app_id=2, - name="TooLong", - unlocked_achievements=5, - completionist_hours=8.0, - comp_100_count=1, - count_comp=8, - ), - ] - with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch(f"{CMD_DONE_PKG}.load_hltb_polls_cache", return_value={1: 10, 2: 1}), - patch( - f"{CMD_DONE_PKG}.load_hltb_count_comp_cache", return_value={1: 40, 2: 8} - ), - patch(f"{CMD_DONE_PKG}._pick_next_shortest_candidate") as mock_pick, - patch(f"{CMD_DONE_PKG}._echo"), - ): - result = _try_reassign_shorter_game( - {1: 4.0, 2: 8.0}, - 1, - 4.0, - State(), - Config(), - ) - - assert not result - mock_pick.assert_not_called() diff --git a/steam_backlog_enforcer/tests/test_cmd_done_part2.py b/steam_backlog_enforcer/tests/test_cmd_done_part2.py index 41584a2..b247c54 100644 --- a/steam_backlog_enforcer/tests/test_cmd_done_part2.py +++ b/steam_backlog_enforcer/tests/test_cmd_done_part2.py @@ -1,217 +1,87 @@ -"""Tests for _cmd_done module (part 2).""" +"""Tests for _cmd_done module (part 2): _prompt_keep_or_skip.""" from __future__ import annotations from unittest.mock import patch -from python_pkg.steam_backlog_enforcer._cmd_done import ( - _should_reassign_candidate, - _try_reassign_shorter_game, -) -from python_pkg.steam_backlog_enforcer.config import Config, State +from python_pkg.steam_backlog_enforcer._cmd_done import _prompt_keep_or_skip from python_pkg.steam_backlog_enforcer.steam_api import GameInfo CMD_DONE_PKG = "python_pkg.steam_backlog_enforcer._cmd_done" -def _snap(**overrides: object) -> dict[str, object]: - snapshot: dict[str, object] = { - "app_id": 1, - "name": "G", - "total_achievements": 10, - "unlocked_achievements": 0, - "playtime_minutes": 60, - "completionist_hours": -1, - "comp_100_count": 3, - "count_comp": 15, - } - snapshot["app_id"] = overrides.get("app_id", 1) - snapshot.update(overrides) - return snapshot +class TestPromptKeepOrSkip: + """Tests for _prompt_keep_or_skip.""" - -class TestTryReassignShorterGame2: - """Tests for _try_reassign_shorter_game (continued).""" - - def test_reassigns_when_current_hours_unknown(self) -> None: - """If current game has unknown hours, allow a confident replacement.""" - snap = [ - _snap(app_id=1, name="Current", unlocked_achievements=5), - _snap( - app_id=2, name="Known", unlocked_achievements=5, completionist_hours=9.0 - ), - ] - state = State(current_app_id=2, current_game_name="Known") - known_game = GameInfo( - app_id=2, - name="Known", + def _game(self, hours: float = 5.0) -> GameInfo: + return GameInfo( + app_id=42, + name="Test", total_achievements=10, unlocked_achievements=5, playtime_minutes=60, - completionist_hours=9.0, - comp_100_count=3, - count_comp=15, + completionist_hours=hours, ) + + def test_non_tty_accepts_silently(self) -> None: + with patch(f"{CMD_DONE_PKG}.sys.stdin") as mock_stdin: + mock_stdin.isatty.return_value = False + assert _prompt_keep_or_skip(self._game()) is True + + def test_yes_answers_accept(self) -> None: + for answer in ("y", "Y", "yes", "YES", ""): + with ( + patch(f"{CMD_DONE_PKG}.sys.stdin") as mock_stdin, + patch(f"{CMD_DONE_PKG}._echo"), + patch("builtins.input", return_value=answer), + ): + mock_stdin.isatty.return_value = True + assert _prompt_keep_or_skip(self._game()) is True, answer + + def test_no_answers_reject(self) -> None: + for answer in ("n", "N", "no", "NO"): + with ( + patch(f"{CMD_DONE_PKG}.sys.stdin") as mock_stdin, + patch(f"{CMD_DONE_PKG}._echo"), + patch("builtins.input", return_value=answer), + ): + mock_stdin.isatty.return_value = True + assert _prompt_keep_or_skip(self._game()) is False, answer + + def test_invalid_then_yes(self) -> None: + echoed: list[str] = [] with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), + patch(f"{CMD_DONE_PKG}.sys.stdin") as mock_stdin, patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(known_game, 0, 0), + f"{CMD_DONE_PKG}._echo", + side_effect=lambda *a, **_: echoed.append(a[0]), ), - patch(f"{CMD_DONE_PKG}.pick_next_game"), - patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[]), - patch(f"{CMD_DONE_PKG}.hide_other_games"), + patch("builtins.input", side_effect=["maybe", "y"]), ): - result = _try_reassign_shorter_game( - {2: 9.0}, - 1, - -1.0, - state, - Config(), - ) - - assert result - - def test_try_reassign_returns_false_when_playable_not_shorter(self) -> None: - """_try_reassign_shorter_game should not reassign to longer candidates.""" - snap = [ - _snap( - app_id=1, - name="Current", - unlocked_achievements=5, - completionist_hours=8.0, - comp_100_count=10, - count_comp=40, - ), - _snap( - app_id=2, - name="Longer", - unlocked_achievements=5, - completionist_hours=12.0, - comp_100_count=10, - count_comp=40, - ), - ] - longer = GameInfo( - app_id=2, - name="Longer", - total_achievements=10, - unlocked_achievements=5, - playtime_minutes=60, - completionist_hours=12.0, - comp_100_count=10, - count_comp=40, - ) + mock_stdin.isatty.return_value = True + assert _prompt_keep_or_skip(self._game()) is True + assert any("answer 'y' or 'n'" in line for line in echoed) + def test_eof_accepts(self) -> None: with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch( - f"{CMD_DONE_PKG}.load_hltb_polls_cache", - return_value={1: 10, 2: 10}, - ), - patch( - f"{CMD_DONE_PKG}.load_hltb_count_comp_cache", - return_value={1: 40, 2: 40}, - ), - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(longer, 0, 0), - ), - patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick_next, + patch(f"{CMD_DONE_PKG}.sys.stdin") as mock_stdin, patch(f"{CMD_DONE_PKG}._echo"), + patch("builtins.input", side_effect=EOFError), ): - result = _try_reassign_shorter_game( - hltb_cache={1: 8.0, 2: 12.0}, - app_id=1, - hours=8.0, - state=State(), - config=Config(), - ) - - assert not result - mock_pick_next.assert_not_called() - - def test_try_reassign_stops_when_should_reassign_is_false(self) -> None: - """Covers early return when policy says not to reassign.""" - snap = [ - _snap( - app_id=1, - name="Current", - unlocked_achievements=5, - completionist_hours=8.0, - comp_100_count=10, - count_comp=40, - ), - _snap( - app_id=2, - name="Candidate", - unlocked_achievements=5, - completionist_hours=6.0, - comp_100_count=10, - count_comp=40, - ), - ] - candidate = GameInfo( - app_id=2, - name="Candidate", - total_achievements=10, - unlocked_achievements=5, - playtime_minutes=60, - completionist_hours=6.0, - comp_100_count=10, - count_comp=40, - ) + mock_stdin.isatty.return_value = True + assert _prompt_keep_or_skip(self._game()) is True + def test_zero_hours_omits_hours_string(self) -> None: + echoed: list[str] = [] with ( - patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), + patch(f"{CMD_DONE_PKG}.sys.stdin") as mock_stdin, patch( - f"{CMD_DONE_PKG}.load_hltb_polls_cache", - return_value={1: 10, 2: 10}, + f"{CMD_DONE_PKG}._echo", + side_effect=lambda *a, **_: echoed.append(a[0]), ), - patch( - f"{CMD_DONE_PKG}.load_hltb_count_comp_cache", - return_value={1: 40, 2: 40}, - ), - patch( - f"{CMD_DONE_PKG}._pick_next_shortest_candidate", - return_value=(candidate, 0, 0), - ), - patch( - f"{CMD_DONE_PKG}._should_reassign_candidate", - return_value=False, - ), - patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick_next, - patch(f"{CMD_DONE_PKG}._echo"), + patch("builtins.input", return_value="y"), ): - result = _try_reassign_shorter_game( - hltb_cache={1: 8.0, 2: 6.0}, - app_id=1, - hours=8.0, - state=State(), - config=Config(), - ) - - assert not result - mock_pick_next.assert_not_called() - - -class TestShouldReassignCandidate: - """Tests for _should_reassign_candidate.""" - - def test_returns_false_when_candidate_not_shorter(self) -> None: - candidate = GameInfo( - app_id=2, - name="Candidate", - total_achievements=10, - unlocked_achievements=5, - playtime_minutes=60, - completionist_hours=9.0, - comp_100_count=3, - count_comp=15, - ) - should = _should_reassign_candidate( - candidate, - 8.0, - force_reassign=False, - ) - assert should is False + mock_stdin.isatty.return_value = True + _prompt_keep_or_skip(self._game(hours=0.0)) + # Without hours, the printed line should not contain "~" + assert not any("~" in line for line in echoed if "Next pick" in line) diff --git a/steam_backlog_enforcer/tests/test_config.py b/steam_backlog_enforcer/tests/test_config.py index 2b01085..aba7925 100644 --- a/steam_backlog_enforcer/tests/test_config.py +++ b/steam_backlog_enforcer/tests/test_config.py @@ -171,6 +171,42 @@ class TestState: assert st.current_app_id is None assert st.current_game_name == "" + def test_skip_for_days_records_iso_timestamp(self) -> None: + state = State() + state.skip_for_days(42, 7) + assert "42" in state.skipped_until + # Round-trip parse and check ~7 days in the future. + from datetime import datetime, timezone + + expiry = datetime.fromisoformat(state.skipped_until["42"]) + delta = (expiry - datetime.now(timezone.utc)).total_seconds() + assert 6 * 86400 < delta <= 7 * 86400 + 1 + + def test_active_skipped_ids_returns_active(self) -> None: + from datetime import datetime, timedelta, timezone + + state = State() + future = datetime.now(timezone.utc) + timedelta(days=3) + state.skipped_until["100"] = future.isoformat() + assert state.active_skipped_ids() == {100} + # Active entry retained. + assert "100" in state.skipped_until + + def test_active_skipped_ids_prunes_expired(self) -> None: + from datetime import datetime, timedelta, timezone + + state = State() + past = datetime.now(timezone.utc) - timedelta(days=1) + state.skipped_until["50"] = past.isoformat() + assert state.active_skipped_ids() == set() + assert "50" not in state.skipped_until + + def test_active_skipped_ids_prunes_malformed(self) -> None: + state = State() + state.skipped_until["77"] = "not-a-date" + assert state.active_skipped_ids() == set() + assert "77" not in state.skipped_until + class TestSnapshot: """Tests for snapshot save/load.""" diff --git a/steam_backlog_enforcer/tests/test_main_part2.py b/steam_backlog_enforcer/tests/test_main_part2.py index 35efb08..a36a64d 100644 --- a/steam_backlog_enforcer/tests/test_main_part2.py +++ b/steam_backlog_enforcer/tests/test_main_part2.py @@ -55,6 +55,7 @@ class TestFinalizeCompletion: _games: object, s: State, _c: object, + **_kwargs: object, ) -> None: s.current_app_id = 2 s.current_game_name = "NewGame" @@ -89,6 +90,7 @@ class TestFinalizeCompletion: _games: object, s: State, _c: object, + **_kwargs: object, ) -> None: s.current_app_id = None @@ -112,6 +114,7 @@ class TestFinalizeCompletion: _games: object, s: State, _c: object, + **_kwargs: object, ) -> None: s.current_app_id = 2 s.current_game_name = "Next" @@ -137,6 +140,7 @@ class TestFinalizeCompletion: _games: object, s: State, _c: object, + **_kwargs: object, ) -> None: s.current_app_id = 2 s.current_game_name = "Next" @@ -158,6 +162,7 @@ class TestFinalizeCompletion: games: list[GameInfo], s: State, _c: object, + **_kwargs: object, ) -> None: for game in games: seen[game.app_id] = game.completionist_hours @@ -191,6 +196,7 @@ class TestFinalizeCompletion: _games: object, s: State, _c: object, + **_kwargs: object, ) -> None: s.current_app_id = 2 s.current_game_name = "Next" @@ -220,6 +226,7 @@ class TestFinalizeCompletion: _games: object, s: State, _c: object, + **_kwargs: object, ) -> None: s.current_app_id = 2 s.current_game_name = "Next" @@ -330,7 +337,6 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client), patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 20.0}), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False), patch(f"{CMD_DONE_PKG}._enforce_on_done"), ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) @@ -350,7 +356,6 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client), patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 10.0}), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False), patch(f"{CMD_DONE_PKG}._finalize_completion") as mock_final, ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) @@ -375,7 +380,6 @@ class TestEnforceOnDone: f"{CMD_DONE_PKG}.fetch_hltb_times_cached", return_value={1: 15.0}, ), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False), patch(f"{CMD_DONE_PKG}._enforce_on_done"), ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) @@ -396,7 +400,6 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client), patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: -1.0}), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False), patch(f"{CMD_DONE_PKG}._enforce_on_done"), ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) @@ -406,7 +409,7 @@ class TestEnforceOnDone: app_id=1, name="G", total_achievements=10, - unlocked_achievements=5, + unlocked_achievements=10, playtime_minutes=60, ) mock_client = MagicMock() @@ -416,6 +419,6 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client), patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 50.0}), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=True), + patch(f"{CMD_DONE_PKG}._finalize_completion"), ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) diff --git a/steam_backlog_enforcer/tests/test_main_part3.py b/steam_backlog_enforcer/tests/test_main_part3.py index ee6a0d4..60276ff 100644 --- a/steam_backlog_enforcer/tests/test_main_part3.py +++ b/steam_backlog_enforcer/tests/test_main_part3.py @@ -70,7 +70,6 @@ class TestCmdDone: patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client), patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 20.0}), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False), patch(f"{CMD_DONE_PKG}._enforce_on_done"), ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) @@ -90,7 +89,6 @@ class TestCmdDone: patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client), patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 10.0}), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False), patch(f"{CMD_DONE_PKG}._finalize_completion") as mock_final, ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) @@ -115,7 +113,6 @@ class TestCmdDone: f"{CMD_DONE_PKG}.fetch_hltb_times_cached", return_value={1: 15.0}, ), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False), patch(f"{CMD_DONE_PKG}._enforce_on_done"), ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) @@ -136,7 +133,6 @@ class TestCmdDone: patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client), patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: -1.0}), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False), patch(f"{CMD_DONE_PKG}._enforce_on_done"), ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) @@ -146,7 +142,7 @@ class TestCmdDone: app_id=1, name="G", total_achievements=10, - unlocked_achievements=5, + unlocked_achievements=10, playtime_minutes=60, ) mock_client = MagicMock() @@ -156,7 +152,7 @@ class TestCmdDone: patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client), patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 50.0}), - patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=True), + patch(f"{CMD_DONE_PKG}._finalize_completion"), ): cmd_done(Config(steam_api_key="k", steam_id="i"), state) diff --git a/steam_backlog_enforcer/tests/test_scanning_part3.py b/steam_backlog_enforcer/tests/test_scanning_part3.py index fd850e9..3041dce 100644 --- a/steam_backlog_enforcer/tests/test_scanning_part3.py +++ b/steam_backlog_enforcer/tests/test_scanning_part3.py @@ -2,6 +2,7 @@ from __future__ import annotations +import contextlib from unittest.mock import patch from python_pkg.steam_backlog_enforcer.config import Config, State @@ -278,3 +279,125 @@ class TestPickNextGame: pick_next_game([g1], state, config) assert state.current_app_id == 1 assert any("Out of range" in line for line in echoed) + + +class TestPickNextGameSequential: + """Tests for the on_select sequential branch of pick_next_game.""" + + @staticmethod + def _common_patches(echoed: list[str]) -> contextlib.ExitStack: + stack = contextlib.ExitStack() + stack.enter_context( + patch( + "python_pkg.steam_backlog_enforcer.scanning._pick_playable_candidate", + side_effect=lambda c: c[0] if c else None, + ) + ) + stack.enter_context( + patch( + "python_pkg.steam_backlog_enforcer.scanning._echo", + side_effect=lambda *a, **_: echoed.append(a[0]), + ) + ) + stack.enter_context( + patch( + "python_pkg.steam_backlog_enforcer.scanning.is_game_installed", + return_value=True, + ) + ) + stack.enter_context( + patch( + "python_pkg.steam_backlog_enforcer.scanning.uninstall_other_games", + return_value=0, + ) + ) + stack.enter_context( + patch("python_pkg.steam_backlog_enforcer.config._atomic_write") + ) + return stack + + def test_on_select_accepts_pick(self) -> None: + g1 = _game(app_id=1, name="G1", hours=5.0) + config = Config(steam_api_key="k", steam_id="i") + state = State() + echoed: list[str] = [] + with self._common_patches(echoed): + pick_next_game([g1], state, config, on_select=lambda _g: True) + assert state.current_app_id == 1 + + def test_on_select_rejection_records_skip_and_picks_next(self) -> None: + g1 = _game(app_id=1, name="G1", hours=5.0) + g2 = _game(app_id=2, name="G2", hours=6.0) + config = Config(steam_api_key="k", steam_id="i") + state = State() + echoed: list[str] = [] + calls: list[int] = [] + + def on_select(game: GameInfo) -> bool: + calls.append(game.app_id) + return game.app_id == 2 # reject g1, accept g2 + + with self._common_patches(echoed): + pick_next_game([g1, g2], state, config, on_select=on_select) + + assert calls == [1, 2] + assert state.current_app_id == 2 + assert "1" in state.skipped_until + assert any("Skipped G1 for 7 days" in line for line in echoed) + + def test_on_select_no_candidates(self) -> None: + """Sequential branch with no candidates clears state.""" + complete = _game(app_id=1, hours=1.0, total=10, unlocked=10) + config = Config(steam_api_key="k", steam_id="i") + state = State(current_app_id=99, current_game_name="X") + echoed: list[str] = [] + with self._common_patches(echoed): + pick_next_game([complete], state, config, on_select=lambda _g: True) + assert state.current_app_id is None + + def test_on_select_no_playable_branch(self) -> None: + """Sequential branch when all candidates lack Linux compatibility.""" + g1 = _game(app_id=1, name="G1", hours=5.0) + config = Config(steam_api_key="k", steam_id="i") + state = State() + echoed: list[str] = [] + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._pick_playable_candidate", + return_value=None, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning._echo", + side_effect=lambda *a, **_: echoed.append(a[0]), + ), + patch("python_pkg.steam_backlog_enforcer.config._atomic_write"), + ): + pick_next_game([g1], state, config, on_select=lambda _g: True) + assert state.current_app_id is None + assert any("No playable games" in line for line in echoed) + + def test_on_select_no_confidence_branch(self) -> None: + """Sequential branch when all candidates fail HLTB confidence.""" + g1 = _game(app_id=1, name="G1", hours=5.0) + g1.comp_100_count = 0 + g1.count_comp = 0 + config = Config(steam_api_key="k", steam_id="i") + state = State() + echoed: list[str] = [] + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._pick_playable_candidate", + side_effect=lambda c: c[0] if c else None, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning._echo", + side_effect=lambda *a, **_: echoed.append(a[0]), + ), + patch( + "python_pkg.steam_backlog_enforcer._scanning_confidence._refresh_candidate_confidence", + ), + patch("python_pkg.steam_backlog_enforcer.config._atomic_write"), + ): + pick_next_game([g1], state, config, on_select=lambda _g: True) + assert state.current_app_id is None + assert any("No assignable games" in line for line in echoed)