diff --git a/steam_backlog_enforcer/_scanning_assign.py b/steam_backlog_enforcer/_scanning_assign.py new file mode 100644 index 0000000..937fd90 --- /dev/null +++ b/steam_backlog_enforcer/_scanning_assign.py @@ -0,0 +1,142 @@ +"""Interactive pick prompt and assignment of the chosen game. + +Split out of :mod:`steam_backlog_enforcer.scanning` to keep both files under +the 250-line cap. Leaf helpers: nothing here calls back into ``scanning``. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +import logging +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._scanning_candidates import ( + _pick_next_shortest_candidate, + _sort_key, +) +from steam_backlog_enforcer._scanning_confidence import ( + _apply_cached_confidence_to_candidates, + _report_poll_confidence, +) +from steam_backlog_enforcer.game_install import ( + _echo, + install_game, + is_game_installed, + uninstall_other_games, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + from steam_backlog_enforcer.config import Config, State + from steam_backlog_enforcer.steam_api import GameInfo + +logger = logging.getLogger(__name__) + +_NO_CONF_MSG = ( + "\nNo assignable games found " + "(HLTB confidence thresholds: comp_100 polls>=3, " + "count_comp>=15, sum>=18)." +) + + +def _prompt_user_pick(qualified: list[GameInfo]) -> int: + """Present numbered list, return 0-based index of user's choice.""" + for i, g in enumerate(qualified, 1): + hours_str = ( + f" (~{g.completionist_hours:.1f}h)" if g.completionist_hours > 0 else "" + ) + _echo(f" {i}. {g.name} (AppID={g.app_id}){hours_str}") + while True: + raw = input("Select game number: ") + try: + idx = int(raw) + except ValueError: + _echo(f"Invalid input: {raw!r}") + continue + if idx < 1 or idx > len(qualified): + _echo(f"Out of range: {idx}") + continue + return idx - 1 + + +def _assign_chosen_game( + chosen: GameInfo, + games: list[GameInfo], + state: State, + config: Config, +) -> None: + """Save assignment, announce it, and handle install/uninstall.""" + state.current_app_id = chosen.app_id + state.current_game_name = chosen.name + if not state.enforcement_started_at: + state.enforcement_started_at = datetime.now(timezone.utc).isoformat() + state.save() + hours_str = ( + f" (~{chosen.completionist_hours:.1f}h leisure+dlc)" + if chosen.completionist_hours > 0 + else "" + ) + _echo(f"\n>>> ASSIGNED: {chosen.name} (AppID={chosen.app_id}){hours_str}") + _echo( + f" Progress: {chosen.unlocked_achievements}/{chosen.total_achievements}" + f" ({chosen.completion_pct:.1f}%)" + ) + _report_poll_confidence(chosen, games, state) + if config.uninstall_other_games: + count = uninstall_other_games(chosen.app_id) + if count: + _echo(f"\n Uninstalled {count} non-assigned games") + if not is_game_installed(chosen.app_id): + _echo(f"\n Auto-installing {chosen.name}...") + install_game( + chosen.app_id, chosen.name, config.steam_id, use_steam_protocol=True + ) + + +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 diff --git a/steam_backlog_enforcer/_scanning_candidates.py b/steam_backlog_enforcer/_scanning_candidates.py new file mode 100644 index 0000000..79e05dd --- /dev/null +++ b/steam_backlog_enforcer/_scanning_candidates.py @@ -0,0 +1,149 @@ +"""Candidate selection helpers for ``scan`` and ``pick_next_game``. + +Split out of :mod:`steam_backlog_enforcer.scanning` to keep both files under +the 250-line cap. Leaf helpers: nothing here calls back into ``scanning``. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._scanning_confidence import ( + _candidate_passes_hltb_confidence, +) +from steam_backlog_enforcer.game_install import _echo +from steam_backlog_enforcer.protondb import ( + ProtonDBRating, + fetch_protondb_ratings, +) + +if TYPE_CHECKING: + from steam_backlog_enforcer.steam_api import GameInfo + +logger = logging.getLogger(__name__) + +_PROTONDB_BATCH_SIZE = 20 + +_PICK_LIST_SIZE = 10 + + +def _sort_key(g: GameInfo) -> tuple[int, float]: + """Sort by known HLTB time (shortest first), then unknown games.""" + if g.completionist_hours > 0: + return (0, g.completionist_hours) + return (1, g.name.lower().encode().hex().__hash__()) + + +def _pick_playable_candidate( + candidates: list[GameInfo], +) -> GameInfo | None: + """Return the first candidate with an acceptable ProtonDB rating. + + Checks candidates in batches (sorted by HLTB hours, shortest first). + Games rated silver-or-worse, or gold-trending-down, are skipped. + """ + offset = 0 + while offset < len(candidates): + batch = candidates[offset : offset + _PROTONDB_BATCH_SIZE] + app_ids = [g.app_id for g in batch] + ratings = fetch_protondb_ratings(app_ids) + + for game in batch: + rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id)) + if rating.is_playable: + if offset > 0 or game is not batch[0]: + _echo( + f" Skipped {offset + batch.index(game)} game(s) " + f"with poor Linux compatibility" + ) + return game + logger.info( + "Skipping %s (AppID=%d): ProtonDB %s (trending %s)", + game.name, + game.app_id, + rating.tier, + rating.trending_tier, + ) + + offset += _PROTONDB_BATCH_SIZE + + return None + + +def _collect_qualified_candidates( + candidates: list[GameInfo], +) -> tuple[list[GameInfo], int, int]: + """Collect up to _PICK_LIST_SIZE playable, HLTB-confident candidates.""" + qualified: list[GameInfo] = [] + confidence_skipped = 0 + linux_skipped = 0 + for game in candidates: + if len(qualified) >= _PICK_LIST_SIZE: + break + if not _candidate_passes_hltb_confidence(game): + confidence_skipped += 1 + continue + playable = _pick_playable_candidate([game]) + if playable is not None: + qualified.append(playable) + else: + linux_skipped += 1 + return qualified, confidence_skipped, linux_skipped + + +def _pick_next_shortest_candidate( + candidates: list[GameInfo], +) -> tuple[GameInfo | None, int, int]: + """Pick next game by checking confidence one candidate at a time. + + The list must be pre-sorted by desired priority (shortest first). + """ + confidence_skipped = 0 + linux_skipped = 0 + for game in candidates: + if not _candidate_passes_hltb_confidence(game): + confidence_skipped += 1 + continue + + # Reuse existing ProtonDB compatibility gate for one candidate. + playable = _pick_playable_candidate([game]) + if playable is not None: + if linux_skipped > 0: + _echo( + f" Skipped {linux_skipped} game(s) with poor Linux compatibility" + ) + return playable, confidence_skipped, linux_skipped + linux_skipped += 1 + + if linux_skipped > 0: + _echo(f" Skipped {linux_skipped} game(s) with poor Linux compatibility") + return None, confidence_skipped, linux_skipped + + +def _collect_top_candidates( + candidates: list[GameInfo], + n: int = 3, +) -> tuple[list[GameInfo], int, int]: + """Collect up to n candidates that pass the Linux compatibility gate. + + Args: + candidates: Pre-sorted list of candidate games. + n: Maximum number of qualified games to collect. + + Returns: + Tuple of (qualified_list, conf_skipped, linux_skipped). + """ + qualified: list[GameInfo] = [] + linux_skipped = 0 + for game in candidates: + if len(qualified) >= n: + break + playable = _pick_playable_candidate([game]) + if playable is not None: + qualified.append(playable) + else: + linux_skipped += 1 + if linux_skipped > 0: + _echo(f" Skipped {linux_skipped} game(s) with poor Linux compatibility") + return qualified, 0, linux_skipped diff --git a/steam_backlog_enforcer/_scanning_tampering.py b/steam_backlog_enforcer/_scanning_tampering.py new file mode 100644 index 0000000..fe5a155 --- /dev/null +++ b/steam_backlog_enforcer/_scanning_tampering.py @@ -0,0 +1,80 @@ +"""Tampering detection for installed games. + +Split out of :mod:`steam_backlog_enforcer.scanning` to keep both files under +the 250-line cap. Leaf helpers: nothing here calls back into ``scanning``. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any + +from steam_backlog_enforcer.config import load_snapshot +from steam_backlog_enforcer.enforcer import send_notification +from steam_backlog_enforcer.game_install import _echo +from steam_backlog_enforcer.steam_api import SteamAPIClient + +if TYPE_CHECKING: + from steam_backlog_enforcer.config import Config, State + +logger = logging.getLogger(__name__) + +_TAMPER_CHECK_LIMIT = 3 + + +def _check_game_tampering( + client: SteamAPIClient, + entry: dict[str, Any], + state: State, +) -> tuple[str, int, int] | None: + """Check if a single game has unexpected achievement progress. + + Args: + client: Steam API client. + entry: Snapshot entry for the game. + state: Current enforcer state. + + Returns: + Tuple of (name, app_id, diff) if tampering detected, else None. + """ + app_id = entry["app_id"] + if app_id == state.current_app_id: + return None + if entry["unlocked_achievements"] >= entry["total_achievements"]: + return None + if entry.get("playtime_minutes", 0) <= 0: + return None + game = client.refresh_single_game( + app_id, entry["name"], entry.get("playtime_minutes", 0) + ) + if game and game.unlocked_achievements > entry["unlocked_achievements"]: + diff = game.unlocked_achievements - entry["unlocked_achievements"] + return (entry["name"], app_id, diff) + return None + + +def detect_tampering(config: Config, state: State) -> None: + """Check if achievements were unlocked on non-assigned games.""" + old_snapshot = load_snapshot() + if old_snapshot is None: + return + + client = SteamAPIClient(config.steam_api_key, config.steam_id) + + # Quick check: only re-fetch a few random non-assigned games. + suspicious: list[tuple[str, int, int]] = [] + for entry in old_snapshot: + result = _check_game_tampering(client, entry, state) + if result: + suspicious.append(result) + if len(suspicious) >= _TAMPER_CHECK_LIMIT: + break + + if suspicious: + _echo("\n TAMPERING DETECTED:") + for name, app_id, diff in suspicious: + _echo(f" {name} (AppID={app_id}): +{diff} new achievements!") + send_notification( + "Tampering Detected!", + f"Achievements unlocked on {len(suspicious)} non-assigned games!", + ) diff --git a/steam_backlog_enforcer/_stats.py b/steam_backlog_enforcer/_stats.py index 93c7a67..3d09ddc 100644 --- a/steam_backlog_enforcer/_stats.py +++ b/steam_backlog_enforcer/_stats.py @@ -2,45 +2,42 @@ from __future__ import annotations -from concurrent.futures import ThreadPoolExecutor, as_completed -from dataclasses import dataclass, field -from datetime import datetime, timedelta, timezone import logging -import secrets from typing import TYPE_CHECKING -from urllib.parse import quote_plus from steam_backlog_enforcer._hltb_types import ( - HLTB_BASE_URL, _read_raw_cache, - load_hltb_cache, - load_hltb_game_id_cache, - load_hltb_leisure_100h_cache, - load_hltb_rush_cache, ) -from steam_backlog_enforcer._scanning_confidence import ( - _apply_cached_confidence_to_candidates, - _confidence_fail_reasons, - _refresh_candidate_confidence_batch, +from steam_backlog_enforcer._stats_display import ( + _format_completion_date, + _print_pace_scenario, + _print_worst_example, + _sum_hours, ) +from steam_backlog_enforcer._stats_gathering import ( + _ensure_completed_rush_data, + _ensure_rush_data, + _filter_qualifying_games, + _refresh_recently_played_completions, +) +from steam_backlog_enforcer._stats_types import _GameTimes from steam_backlog_enforcer._web_dataset import ( PaceVsHLTB, compute_pace_vs_hltb, count_complete_since_start, ) -from steam_backlog_enforcer.config import SNAPSHOT_FILE, load_snapshot +from steam_backlog_enforcer.config import load_snapshot from steam_backlog_enforcer.game_install import _echo -from steam_backlog_enforcer.hltb import fetch_hltb_detail_missing -from steam_backlog_enforcer.protondb import ( - ProtonDBRating, - fetch_protondb_ratings, -) from steam_backlog_enforcer.steam_api import ( GameInfo, - SteamAPIClient, - SteamAPIError, ) +# _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 +# tests) have always imported it from _stats. Without __all__, mypy's +# --no-implicit-reexport rejects those imports. +__all__ = ["_GameTimes"] + if TYPE_CHECKING: from steam_backlog_enforcer.config import Config, State @@ -50,181 +47,6 @@ _HOURS_PER_DAY_PRESETS = (2.0, 4.0, 6.0, 8.0) _LINE = "─" * 70 -_HLTB_SEARCH_BASE = "https://howlongtobeat.com/?q=" - - -@dataclass -class _GameTimes: - """Per-game time estimates for stats display.""" - - game: GameInfo - worst_hours: float - rush_hours: float - leisure_100h: float - hltb_game_id: int = field(default=0) - - -def _filter_qualifying_games( - games: list[GameInfo], - state: State, -) -> tuple[list[_GameTimes], int, int, int]: - """Return qualifying incomplete games with their time estimates. - - Applies the same HLTB-confidence and Linux-compatibility filters as the - game picker. The current game and already-finished games are excluded. - - Returns: - (qualified_list, hltb_skipped, linux_skipped, no_data_skipped) - """ - rush_cache = load_hltb_rush_cache() - leisure_100h_cache = load_hltb_leisure_100h_cache() - game_id_cache = load_hltb_game_id_cache() - hours_cache = load_hltb_cache() - - exclude = set(state.finished_app_ids) - if state.current_app_id is not None: - exclude.add(state.current_app_id) - - candidates = [g for g in games if not g.is_complete and g.app_id not in exclude] - _apply_cached_confidence_to_candidates(candidates) - _refresh_candidate_confidence_batch(candidates) - - hltb_skipped = 0 - linux_skipped = 0 - no_data_skipped = 0 - app_ids_to_check: list[int] = [] - - conf_ok: list[GameInfo] = [] - for game in candidates: - if _confidence_fail_reasons(game): - hltb_skipped += 1 - continue - conf_ok.append(game) - app_ids_to_check.append(game.app_id) - - ratings: dict[int, ProtonDBRating] = {} - if app_ids_to_check: - ratings = fetch_protondb_ratings(app_ids_to_check) - - qualified: list[_GameTimes] = [] - for game in conf_ok: - rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id)) - if not rating.is_playable: - linux_skipped += 1 - continue - - rush = rush_cache.get(game.app_id, -1) - leisure = leisure_100h_cache.get(game.app_id, -1) - - # worst_hours = max of: snapshot completionist, HLTB hours cache (fallback - # when snapshot is stale/missing), and leisure_100h (slowest 100% time). - snap_hours = game.completionist_hours if game.completionist_hours > 0 else -1 - cache_hours = hours_cache.get(game.app_id, -1) - worst_candidates = [v for v in (snap_hours, cache_hours, leisure) if v > 0] - worst = max(worst_candidates) if worst_candidates else -1 - - if worst <= 0 and rush <= 0 and leisure <= 0: - no_data_skipped += 1 - continue - - qualified.append( - _GameTimes( - game=game, - worst_hours=worst, - rush_hours=rush, - leisure_100h=leisure, - hltb_game_id=game_id_cache.get(game.app_id, 0), - ) - ) - - return qualified, hltb_skipped, linux_skipped, no_data_skipped - - -def _ensure_rush_data(qualified: list[_GameTimes]) -> bool: - """Auto-fetch rush/leisure detail for games that are missing it. - - Returns True when a fetch was performed; the caller should then re-run - ``_filter_qualifying_games`` to pick up the updated caches. - """ - total_q = len(qualified) - missing = sum(1 for e in qualified if e.rush_hours <= 0) - if not qualified or not missing: - return False - _echo(f"Fetching HLTB detail for {missing}/{total_q} games missing rush/leisure...") - game_pairs = [(e.game.app_id, e.game.name) for e in qualified] - fetch_hltb_detail_missing(game_pairs) - return True - - -def _ensure_completed_rush_data(games: list[GameInfo]) -> bool: - """Fetch rush/leisure detail for completed games used for pace calibration. - - Completed games aren't processed by ``_ensure_rush_data`` (which only - handles incomplete qualifying games), so this separate pass fills in - their rush/leisure data for ``compute_pace_vs_hltb``. - - Returns True when at least one new fetch was performed. - """ - pairs = [ - (g.app_id, g.name) for g in games if g.is_complete and g.playtime_minutes > 0 - ] - if not pairs: - return False - _echo( - f"Fetching HLTB detail for {len(pairs)} completed games (pace calibration)..." - ) - fetched = fetch_hltb_detail_missing(pairs) - return fetched > 0 - - -def _print_worst_example(entries: list[_GameTimes]) -> None: - """Print a randomly selected example from the worst-case qualified games.""" - examples = [e for e in entries if e.worst_hours > 0] - if not examples: - return - example = secrets.choice(examples) - _echo(f"\n Example game: {example.game.name!r}") - _echo(f" Worst case: {example.worst_hours:.1f} h") - if example.rush_hours > 0: - _echo(f" Rush: {example.rush_hours:.1f} h") - if example.leisure_100h > 0: - _echo(f" Leisure: {example.leisure_100h:.1f} h") - hltb_game_id = example.hltb_game_id - if hltb_game_id == 0: - # On-demand backfill: one search to get the HLTB game ID for this game. - fetch_hltb_detail_missing([(example.game.app_id, example.game.name)]) - hltb_game_id = load_hltb_game_id_cache().get(example.game.app_id, 0) - if hltb_game_id > 0: - _echo(f" HLTB: {HLTB_BASE_URL}/game/{hltb_game_id}") - else: - _echo(f" HLTB: {_HLTB_SEARCH_BASE}{quote_plus(example.game.name)}") - - -def _sum_hours(entries: list[_GameTimes], attr: str) -> tuple[float, int]: - """Sum a time attribute across entries; return (total_hours, missing_count). - - Games where the attribute is ≤ 0 contribute 0 to the sum and are counted - in ``missing_count`` so the user knows the estimate may be an undercount. - """ - total = 0.0 - missing = 0 - for e in entries: - val: float = getattr(e, attr) - if val > 0: - total += val - else: - missing += 1 - return round(total, 1), missing - - -def _format_completion_date(hours: float, daily_hours: float) -> str: - """Return 'N days (YYYY-MM-DD)' for finishing hours at daily_hours per day.""" - if hours <= 0 or daily_hours <= 0: - return "N/A" - days = int(hours / daily_hours) - target = datetime.now(timezone.utc) + timedelta(days=days) - return f"{days} days ({target.strftime('%Y-%m-%d')})" - def _print_scenario( label: str, @@ -249,50 +71,6 @@ def _print_scenario( _echo(f" @ {daily:.0f} h/day → {estimate}") -def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None: - """Print the pace-based completion estimate. - - ``games_done`` must be the count of games completed ON OR AFTER - ``state.enforcement_started_at`` (use ``count_complete_since_start``). - Pre-enforcement completions inflate the rate and are excluded. - """ - _echo("\n 1. AT YOUR CURRENT PACE") - if not state.enforcement_started_at: - _echo(" No start date recorded.") - _echo(" Set enforcement_started_at in state.json (ISO-8601 UTC)") - _echo(" to enable this estimate.") - return - - try: - started = datetime.fromisoformat(state.enforcement_started_at) - except ValueError: - _echo(f" Invalid enforcement_started_at: {state.enforcement_started_at!r}") - return - - now = datetime.now(timezone.utc) - days_elapsed = max(1, (now - started).days) - - if games_done == 0: - _echo(f" Started: {started.strftime('%Y-%m-%d')}") - _echo(" No games finished yet — pace cannot be estimated.") - return - - rate = games_done / days_elapsed - _echo(f" Started: {started.strftime('%Y-%m-%d')}") - _echo( - f" Finished: {games_done} games in {days_elapsed} days " - "(since enforcement start)" - ) - _echo( - f" Pace: {rate:.4f} games/day (1 game every {1 / rate:.1f} days)" - ) - _echo(f" Remaining: {remaining} games") - - days_to_go = int(remaining / rate) - finish = now + timedelta(days=days_to_go) - _echo(f" Est. complete: {days_to_go} days ({finish.strftime('%Y-%m-%d')})") - - def _print_player_speed_scenario( pace: PaceVsHLTB | None, rush_total: float, @@ -344,74 +122,6 @@ def _print_player_speed_scenario( _echo(f" @ {daily:.0f} h/day → {estimate}") -def _refresh_recently_played_completions( - games: list[GameInfo], - config: Config, -) -> list[GameInfo]: - """Refresh achievement data for incomplete games played since last scan. - - Makes 1 ``GetOwnedGames`` request + 1 ``GetPlayerAchievements`` per - recently-played incomplete game. Finds games newly completed since the - last ``scan`` without re-scanning the whole library. - - Returns a new list with updated GameInfo objects for any game that was - played after the snapshot was written; all other games are unchanged. - """ - try: - snapshot_mtime = SNAPSHOT_FILE.stat().st_mtime - except OSError: - return games - - try: - client = SteamAPIClient(config.steam_api_key, config.steam_id) - owned_raw = client.get_owned_games() - except SteamAPIError: - logger.debug("Steam API unavailable; skipping completion refresh.") - return games - last_played_map = {g["appid"]: g.get("rtime_last_played", 0) for g in owned_raw} - - to_refresh = [ - g - for g in games - if not g.is_complete and last_played_map.get(g.app_id, 0) > snapshot_mtime - ] - - if not to_refresh: - return games - - _echo( - f"Refreshing {len(to_refresh)} recently-played game(s)" - " for up-to-date completion status..." - ) - - game_map = {g.app_id: g for g in games} - - def _refresh_one(game: GameInfo) -> GameInfo: - achievements = client.get_achievement_details(game.app_id) - if not achievements: - return game - unlocked = sum(1 for a in achievements if a.achieved) - return GameInfo( - app_id=game.app_id, - name=game.name, - total_achievements=len(achievements), - unlocked_achievements=unlocked, - playtime_minutes=game.playtime_minutes, - achievements=achievements, - completionist_hours=game.completionist_hours, - comp_100_count=game.comp_100_count, - count_comp=game.count_comp, - ) - - with ThreadPoolExecutor(max_workers=20) as pool: - futures = {pool.submit(_refresh_one, g): g for g in to_refresh} - for future in as_completed(futures): - refreshed = future.result() - game_map[refreshed.app_id] = refreshed - - return list(game_map.values()) - - def cmd_stats(_config: Config, state: State) -> None: """Display backlog completion-time statistics. diff --git a/steam_backlog_enforcer/_stats_display.py b/steam_backlog_enforcer/_stats_display.py new file mode 100644 index 0000000..fa94b82 --- /dev/null +++ b/steam_backlog_enforcer/_stats_display.py @@ -0,0 +1,116 @@ +"""Formatting helpers for the ``stats`` command's printed report. + +Split out of :mod:`steam_backlog_enforcer._stats` to keep both files under +the 250-line cap. Leaf helpers: nothing here calls back into ``_stats``. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +import secrets +from typing import TYPE_CHECKING +from urllib.parse import quote_plus + +from steam_backlog_enforcer._hltb_types import ( + HLTB_BASE_URL, + load_hltb_game_id_cache, +) +from steam_backlog_enforcer._stats_types import _HLTB_SEARCH_BASE, _GameTimes +from steam_backlog_enforcer.game_install import _echo +from steam_backlog_enforcer.hltb import fetch_hltb_detail_missing + +if TYPE_CHECKING: + from steam_backlog_enforcer.config import State + + +def _print_worst_example(entries: list[_GameTimes]) -> None: + """Print a randomly selected example from the worst-case qualified games.""" + examples = [e for e in entries if e.worst_hours > 0] + if not examples: + return + example = secrets.choice(examples) + _echo(f"\n Example game: {example.game.name!r}") + _echo(f" Worst case: {example.worst_hours:.1f} h") + if example.rush_hours > 0: + _echo(f" Rush: {example.rush_hours:.1f} h") + if example.leisure_100h > 0: + _echo(f" Leisure: {example.leisure_100h:.1f} h") + hltb_game_id = example.hltb_game_id + if hltb_game_id == 0: + # On-demand backfill: one search to get the HLTB game ID for this game. + fetch_hltb_detail_missing([(example.game.app_id, example.game.name)]) + hltb_game_id = load_hltb_game_id_cache().get(example.game.app_id, 0) + if hltb_game_id > 0: + _echo(f" HLTB: {HLTB_BASE_URL}/game/{hltb_game_id}") + else: + _echo(f" HLTB: {_HLTB_SEARCH_BASE}{quote_plus(example.game.name)}") + + +def _sum_hours(entries: list[_GameTimes], attr: str) -> tuple[float, int]: + """Sum a time attribute across entries; return (total_hours, missing_count). + + Games where the attribute is ≤ 0 contribute 0 to the sum and are counted + in ``missing_count`` so the user knows the estimate may be an undercount. + """ + total = 0.0 + missing = 0 + for e in entries: + val: float = getattr(e, attr) + if val > 0: + total += val + else: + missing += 1 + return round(total, 1), missing + + +def _format_completion_date(hours: float, daily_hours: float) -> str: + """Return 'N days (YYYY-MM-DD)' for finishing hours at daily_hours per day.""" + if hours <= 0 or daily_hours <= 0: + return "N/A" + days = int(hours / daily_hours) + target = datetime.now(timezone.utc) + timedelta(days=days) + return f"{days} days ({target.strftime('%Y-%m-%d')})" + + +def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None: + """Print the pace-based completion estimate. + + ``games_done`` must be the count of games completed ON OR AFTER + ``state.enforcement_started_at`` (use ``count_complete_since_start``). + Pre-enforcement completions inflate the rate and are excluded. + """ + _echo("\n 1. AT YOUR CURRENT PACE") + if not state.enforcement_started_at: + _echo(" No start date recorded.") + _echo(" Set enforcement_started_at in state.json (ISO-8601 UTC)") + _echo(" to enable this estimate.") + return + + try: + started = datetime.fromisoformat(state.enforcement_started_at) + except ValueError: + _echo(f" Invalid enforcement_started_at: {state.enforcement_started_at!r}") + return + + now = datetime.now(timezone.utc) + days_elapsed = max(1, (now - started).days) + + if games_done == 0: + _echo(f" Started: {started.strftime('%Y-%m-%d')}") + _echo(" No games finished yet — pace cannot be estimated.") + return + + rate = games_done / days_elapsed + _echo(f" Started: {started.strftime('%Y-%m-%d')}") + _echo( + f" Finished: {games_done} games in {days_elapsed} days " + "(since enforcement start)" + ) + _echo( + f" Pace: {rate:.4f} games/day (1 game every {1 / rate:.1f} days)" + ) + _echo(f" Remaining: {remaining} games") + + days_to_go = int(remaining / rate) + finish = now + timedelta(days=days_to_go) + _echo(f" Est. complete: {days_to_go} days ({finish.strftime('%Y-%m-%d')})") diff --git a/steam_backlog_enforcer/_stats_gathering.py b/steam_backlog_enforcer/_stats_gathering.py new file mode 100644 index 0000000..d36bb6e --- /dev/null +++ b/steam_backlog_enforcer/_stats_gathering.py @@ -0,0 +1,222 @@ +"""Game selection and HLTB/completion data top-ups for ``stats``. + +Split out of :mod:`steam_backlog_enforcer._stats` to keep both files under +the 250-line cap. Leaf helpers: nothing here calls back into ``_stats``. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor, as_completed +import logging +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._hltb_types import ( + load_hltb_cache, + load_hltb_game_id_cache, + load_hltb_leisure_100h_cache, + load_hltb_rush_cache, +) +from steam_backlog_enforcer._scanning_confidence import ( + _apply_cached_confidence_to_candidates, + _confidence_fail_reasons, + _refresh_candidate_confidence_batch, +) +from steam_backlog_enforcer._stats_types import _GameTimes +from steam_backlog_enforcer.config import SNAPSHOT_FILE +from steam_backlog_enforcer.game_install import _echo +from steam_backlog_enforcer.hltb import fetch_hltb_detail_missing +from steam_backlog_enforcer.protondb import ( + ProtonDBRating, + fetch_protondb_ratings, +) +from steam_backlog_enforcer.steam_api import ( + GameInfo, + SteamAPIClient, + SteamAPIError, +) + +if TYPE_CHECKING: + from steam_backlog_enforcer.config import Config, State + +logger = logging.getLogger(__name__) + + +def _filter_qualifying_games( + games: list[GameInfo], + state: State, +) -> tuple[list[_GameTimes], int, int, int]: + """Return qualifying incomplete games with their time estimates. + + Applies the same HLTB-confidence and Linux-compatibility filters as the + game picker. The current game and already-finished games are excluded. + + Returns: + (qualified_list, hltb_skipped, linux_skipped, no_data_skipped) + """ + rush_cache = load_hltb_rush_cache() + leisure_100h_cache = load_hltb_leisure_100h_cache() + game_id_cache = load_hltb_game_id_cache() + hours_cache = load_hltb_cache() + + exclude = set(state.finished_app_ids) + if state.current_app_id is not None: + exclude.add(state.current_app_id) + + candidates = [g for g in games if not g.is_complete and g.app_id not in exclude] + _apply_cached_confidence_to_candidates(candidates) + _refresh_candidate_confidence_batch(candidates) + + hltb_skipped = 0 + linux_skipped = 0 + no_data_skipped = 0 + app_ids_to_check: list[int] = [] + + conf_ok: list[GameInfo] = [] + for game in candidates: + if _confidence_fail_reasons(game): + hltb_skipped += 1 + continue + conf_ok.append(game) + app_ids_to_check.append(game.app_id) + + ratings: dict[int, ProtonDBRating] = {} + if app_ids_to_check: + ratings = fetch_protondb_ratings(app_ids_to_check) + + qualified: list[_GameTimes] = [] + for game in conf_ok: + rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id)) + if not rating.is_playable: + linux_skipped += 1 + continue + + rush = rush_cache.get(game.app_id, -1) + leisure = leisure_100h_cache.get(game.app_id, -1) + + # worst_hours = max of: snapshot completionist, HLTB hours cache (fallback + # when snapshot is stale/missing), and leisure_100h (slowest 100% time). + snap_hours = game.completionist_hours if game.completionist_hours > 0 else -1 + cache_hours = hours_cache.get(game.app_id, -1) + worst_candidates = [v for v in (snap_hours, cache_hours, leisure) if v > 0] + worst = max(worst_candidates) if worst_candidates else -1 + + if worst <= 0 and rush <= 0 and leisure <= 0: + no_data_skipped += 1 + continue + + qualified.append( + _GameTimes( + game=game, + worst_hours=worst, + rush_hours=rush, + leisure_100h=leisure, + hltb_game_id=game_id_cache.get(game.app_id, 0), + ) + ) + + return qualified, hltb_skipped, linux_skipped, no_data_skipped + + +def _ensure_rush_data(qualified: list[_GameTimes]) -> bool: + """Auto-fetch rush/leisure detail for games that are missing it. + + Returns True when a fetch was performed; the caller should then re-run + ``_filter_qualifying_games`` to pick up the updated caches. + """ + total_q = len(qualified) + missing = sum(1 for e in qualified if e.rush_hours <= 0) + if not qualified or not missing: + return False + _echo(f"Fetching HLTB detail for {missing}/{total_q} games missing rush/leisure...") + game_pairs = [(e.game.app_id, e.game.name) for e in qualified] + fetch_hltb_detail_missing(game_pairs) + return True + + +def _ensure_completed_rush_data(games: list[GameInfo]) -> bool: + """Fetch rush/leisure detail for completed games used for pace calibration. + + Completed games aren't processed by ``_ensure_rush_data`` (which only + handles incomplete qualifying games), so this separate pass fills in + their rush/leisure data for ``compute_pace_vs_hltb``. + + Returns True when at least one new fetch was performed. + """ + pairs = [ + (g.app_id, g.name) for g in games if g.is_complete and g.playtime_minutes > 0 + ] + if not pairs: + return False + _echo( + f"Fetching HLTB detail for {len(pairs)} completed games (pace calibration)..." + ) + fetched = fetch_hltb_detail_missing(pairs) + return fetched > 0 + + +def _refresh_recently_played_completions( + games: list[GameInfo], + config: Config, +) -> list[GameInfo]: + """Refresh achievement data for incomplete games played since last scan. + + Makes 1 ``GetOwnedGames`` request + 1 ``GetPlayerAchievements`` per + recently-played incomplete game. Finds games newly completed since the + last ``scan`` without re-scanning the whole library. + + Returns a new list with updated GameInfo objects for any game that was + played after the snapshot was written; all other games are unchanged. + """ + try: + snapshot_mtime = SNAPSHOT_FILE.stat().st_mtime + except OSError: + return games + + try: + client = SteamAPIClient(config.steam_api_key, config.steam_id) + owned_raw = client.get_owned_games() + except SteamAPIError: + logger.debug("Steam API unavailable; skipping completion refresh.") + return games + last_played_map = {g["appid"]: g.get("rtime_last_played", 0) for g in owned_raw} + + to_refresh = [ + g + for g in games + if not g.is_complete and last_played_map.get(g.app_id, 0) > snapshot_mtime + ] + + if not to_refresh: + return games + + _echo( + f"Refreshing {len(to_refresh)} recently-played game(s)" + " for up-to-date completion status..." + ) + + game_map = {g.app_id: g for g in games} + + def _refresh_one(game: GameInfo) -> GameInfo: + achievements = client.get_achievement_details(game.app_id) + if not achievements: + return game + unlocked = sum(1 for a in achievements if a.achieved) + return GameInfo( + app_id=game.app_id, + name=game.name, + total_achievements=len(achievements), + unlocked_achievements=unlocked, + playtime_minutes=game.playtime_minutes, + achievements=achievements, + completionist_hours=game.completionist_hours, + comp_100_count=game.comp_100_count, + count_comp=game.count_comp, + ) + + with ThreadPoolExecutor(max_workers=20) as pool: + futures = {pool.submit(_refresh_one, g): g for g in to_refresh} + for future in as_completed(futures): + refreshed = future.result() + game_map[refreshed.app_id] = refreshed + + return list(game_map.values()) diff --git a/steam_backlog_enforcer/_stats_types.py b/steam_backlog_enforcer/_stats_types.py new file mode 100644 index 0000000..b6647a9 --- /dev/null +++ b/steam_backlog_enforcer/_stats_types.py @@ -0,0 +1,26 @@ +"""Shared value types for the ``stats`` command. + +A leaf module: imported by :mod:`steam_backlog_enforcer._stats` and by its +helper modules, so it must not import either of them back. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from steam_backlog_enforcer.steam_api import GameInfo + +_HLTB_SEARCH_BASE = "https://howlongtobeat.com/?q=" + + +@dataclass +class _GameTimes: + """Per-game time estimates for stats display.""" + + game: GameInfo + worst_hours: float + rush_hours: float + leisure_100h: float + hltb_game_id: int = field(default=0) diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index 5ef544e..f9c5c17 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -2,20 +2,35 @@ from __future__ import annotations -from datetime import datetime, timezone import logging import time -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING from steam_backlog_enforcer._hltb_types import ( load_hltb_count_comp_cache, load_hltb_polls_cache, ) +from steam_backlog_enforcer._scanning_assign import ( + _NO_CONF_MSG, + _assign_chosen_game, + _pick_next_game_sequential, + _prompt_user_pick, +) +from steam_backlog_enforcer._scanning_candidates import ( + _collect_qualified_candidates, + _collect_top_candidates, + _pick_next_shortest_candidate, + _pick_playable_candidate, + _sort_key, +) from steam_backlog_enforcer._scanning_confidence import ( _apply_cached_confidence_to_candidates, - _candidate_passes_hltb_confidence, _report_poll_confidence, ) +from steam_backlog_enforcer._scanning_tampering import ( + _check_game_tampering, + detect_tampering, +) from steam_backlog_enforcer.config import ( Config, State, @@ -27,25 +42,28 @@ from steam_backlog_enforcer.enforcer import ( ) from steam_backlog_enforcer.game_install import ( _echo, - install_game, - is_game_installed, - uninstall_other_games, ) from steam_backlog_enforcer.hltb import ( fetch_hltb_times_cached, ) -from steam_backlog_enforcer.protondb import ( - ProtonDBRating, - fetch_protondb_ratings, -) from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient if TYPE_CHECKING: from collections.abc import Callable +# These helpers moved to the _scanning_* leaf modules, but callers and tests +# have always imported them from here. mypy's --no-implicit-reexport needs +# them listed explicitly. +__all__ = [ + "_check_game_tampering", + "_collect_top_candidates", + "_pick_next_shortest_candidate", + "_pick_playable_candidate", + "detect_tampering", +] + logger = logging.getLogger(__name__) -_TAMPER_CHECK_LIMIT = 3 # ────────────────────────────────────────────────────────────── # Scanning & game selection @@ -125,182 +143,6 @@ def do_scan(config: Config, state: State) -> list[GameInfo]: # How many candidates to check per ProtonDB batch. -_PROTONDB_BATCH_SIZE = 20 - - -def _pick_playable_candidate( - candidates: list[GameInfo], -) -> GameInfo | None: - """Return the first candidate with an acceptable ProtonDB rating. - - Checks candidates in batches (sorted by HLTB hours, shortest first). - Games rated silver-or-worse, or gold-trending-down, are skipped. - """ - offset = 0 - while offset < len(candidates): - batch = candidates[offset : offset + _PROTONDB_BATCH_SIZE] - app_ids = [g.app_id for g in batch] - ratings = fetch_protondb_ratings(app_ids) - - for game in batch: - rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id)) - if rating.is_playable: - if offset > 0 or game is not batch[0]: - _echo( - f" Skipped {offset + batch.index(game)} game(s) " - f"with poor Linux compatibility" - ) - return game - logger.info( - "Skipping %s (AppID=%d): ProtonDB %s (trending %s)", - game.name, - game.app_id, - rating.tier, - rating.trending_tier, - ) - - offset += _PROTONDB_BATCH_SIZE - - return None - - -_PICK_LIST_SIZE = 10 - -_NO_CONF_MSG = ( - "\nNo assignable games found " - "(HLTB confidence thresholds: comp_100 polls>=3, " - "count_comp>=15, sum>=18)." -) - - -def _sort_key(g: GameInfo) -> tuple[int, float]: - """Sort by known HLTB time (shortest first), then unknown games.""" - if g.completionist_hours > 0: - return (0, g.completionist_hours) - return (1, g.name.lower().encode().hex().__hash__()) - - -def _collect_qualified_candidates( - candidates: list[GameInfo], -) -> tuple[list[GameInfo], int, int]: - """Collect up to _PICK_LIST_SIZE playable, HLTB-confident candidates.""" - qualified: list[GameInfo] = [] - confidence_skipped = 0 - linux_skipped = 0 - for game in candidates: - if len(qualified) >= _PICK_LIST_SIZE: - break - if not _candidate_passes_hltb_confidence(game): - confidence_skipped += 1 - continue - playable = _pick_playable_candidate([game]) - if playable is not None: - qualified.append(playable) - else: - linux_skipped += 1 - return qualified, confidence_skipped, linux_skipped - - -def _prompt_user_pick(qualified: list[GameInfo]) -> int: - """Present numbered list, return 0-based index of user's choice.""" - for i, g in enumerate(qualified, 1): - hours_str = ( - f" (~{g.completionist_hours:.1f}h)" if g.completionist_hours > 0 else "" - ) - _echo(f" {i}. {g.name} (AppID={g.app_id}){hours_str}") - while True: - raw = input("Select game number: ") - try: - idx = int(raw) - except ValueError: - _echo(f"Invalid input: {raw!r}") - continue - if idx < 1 or idx > len(qualified): - _echo(f"Out of range: {idx}") - continue - return idx - 1 - - -def _assign_chosen_game( - chosen: GameInfo, - games: list[GameInfo], - state: State, - config: Config, -) -> None: - """Save assignment, announce it, and handle install/uninstall.""" - state.current_app_id = chosen.app_id - state.current_game_name = chosen.name - if not state.enforcement_started_at: - state.enforcement_started_at = datetime.now(timezone.utc).isoformat() - state.save() - hours_str = ( - f" (~{chosen.completionist_hours:.1f}h leisure+dlc)" - if chosen.completionist_hours > 0 - else "" - ) - _echo(f"\n>>> ASSIGNED: {chosen.name} (AppID={chosen.app_id}){hours_str}") - _echo( - f" Progress: {chosen.unlocked_achievements}/{chosen.total_achievements}" - f" ({chosen.completion_pct:.1f}%)" - ) - _report_poll_confidence(chosen, games, state) - if config.uninstall_other_games: - count = uninstall_other_games(chosen.app_id) - if count: - _echo(f"\n Uninstalled {count} non-assigned games") - if not is_game_installed(chosen.app_id): - _echo(f"\n Auto-installing {chosen.name}...") - install_game( - chosen.app_id, chosen.name, config.steam_id, use_steam_protocol=True - ) - - -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( @@ -356,63 +198,6 @@ def pick_next_game( _assign_chosen_game(qualified[idx], games, state, config) -def _pick_next_shortest_candidate( - candidates: list[GameInfo], -) -> tuple[GameInfo | None, int, int]: - """Pick next game by checking confidence one candidate at a time. - - The list must be pre-sorted by desired priority (shortest first). - """ - confidence_skipped = 0 - linux_skipped = 0 - for game in candidates: - if not _candidate_passes_hltb_confidence(game): - confidence_skipped += 1 - continue - - # Reuse existing ProtonDB compatibility gate for one candidate. - playable = _pick_playable_candidate([game]) - if playable is not None: - if linux_skipped > 0: - _echo( - f" Skipped {linux_skipped} game(s) with poor Linux compatibility" - ) - return playable, confidence_skipped, linux_skipped - linux_skipped += 1 - - if linux_skipped > 0: - _echo(f" Skipped {linux_skipped} game(s) with poor Linux compatibility") - return None, confidence_skipped, linux_skipped - - -def _collect_top_candidates( - candidates: list[GameInfo], - n: int = 3, -) -> tuple[list[GameInfo], int, int]: - """Collect up to n candidates that pass the Linux compatibility gate. - - Args: - candidates: Pre-sorted list of candidate games. - n: Maximum number of qualified games to collect. - - Returns: - Tuple of (qualified_list, conf_skipped, linux_skipped). - """ - qualified: list[GameInfo] = [] - linux_skipped = 0 - for game in candidates: - if len(qualified) >= n: - break - playable = _pick_playable_candidate([game]) - if playable is not None: - qualified.append(playable) - else: - linux_skipped += 1 - if linux_skipped > 0: - _echo(f" Skipped {linux_skipped} game(s) with poor Linux compatibility") - return qualified, 0, linux_skipped - - # ────────────────────────────────────────────────────────────── # Checking & tampering detection # ────────────────────────────────────────────────────────────── @@ -461,61 +246,3 @@ def do_check(config: Config, state: State) -> None: # Tampering detection on snapshot. detect_tampering(config, state) - - -def _check_game_tampering( - client: SteamAPIClient, - entry: dict[str, Any], - state: State, -) -> tuple[str, int, int] | None: - """Check if a single game has unexpected achievement progress. - - Args: - client: Steam API client. - entry: Snapshot entry for the game. - state: Current enforcer state. - - Returns: - Tuple of (name, app_id, diff) if tampering detected, else None. - """ - app_id = entry["app_id"] - if app_id == state.current_app_id: - return None - if entry["unlocked_achievements"] >= entry["total_achievements"]: - return None - if entry.get("playtime_minutes", 0) <= 0: - return None - game = client.refresh_single_game( - app_id, entry["name"], entry.get("playtime_minutes", 0) - ) - if game and game.unlocked_achievements > entry["unlocked_achievements"]: - diff = game.unlocked_achievements - entry["unlocked_achievements"] - return (entry["name"], app_id, diff) - return None - - -def detect_tampering(config: Config, state: State) -> None: - """Check if achievements were unlocked on non-assigned games.""" - old_snapshot = load_snapshot() - if old_snapshot is None: - return - - client = SteamAPIClient(config.steam_api_key, config.steam_id) - - # Quick check: only re-fetch a few random non-assigned games. - suspicious: list[tuple[str, int, int]] = [] - for entry in old_snapshot: - result = _check_game_tampering(client, entry, state) - if result: - suspicious.append(result) - if len(suspicious) >= _TAMPER_CHECK_LIMIT: - break - - if suspicious: - _echo("\n TAMPERING DETECTED:") - for name, app_id, diff in suspicious: - _echo(f" {name} (AppID={app_id}): +{diff} new achievements!") - send_notification( - "Tampering Detected!", - f"Achievements unlocked on {len(suspicious)} non-assigned games!", - ) diff --git a/steam_backlog_enforcer/tests/test_scanning.py b/steam_backlog_enforcer/tests/test_scanning.py index 1b253d1..17a26be 100644 --- a/steam_backlog_enforcer/tests/test_scanning.py +++ b/steam_backlog_enforcer/tests/test_scanning.py @@ -154,7 +154,7 @@ class TestPickPlayableCandidate: game = _game(app_id=440, name="TF2") with ( patch( - "steam_backlog_enforcer.scanning.fetch_protondb_ratings", + "steam_backlog_enforcer._scanning_candidates.fetch_protondb_ratings", return_value={ 440: ProtonDBRating(app_id=440, tier="gold"), }, @@ -171,7 +171,7 @@ class TestPickPlayableCandidate: good = _game(app_id=2, name="Good") with ( patch( - "steam_backlog_enforcer.scanning.fetch_protondb_ratings", + "steam_backlog_enforcer._scanning_candidates.fetch_protondb_ratings", return_value={ 1: ProtonDBRating(app_id=1, tier="borked"), 2: ProtonDBRating(app_id=2, tier="platinum"), @@ -188,7 +188,7 @@ class TestPickPlayableCandidate: game = _game(app_id=1, name="Bad") with ( patch( - "steam_backlog_enforcer.scanning.fetch_protondb_ratings", + "steam_backlog_enforcer._scanning_candidates.fetch_protondb_ratings", return_value={ 1: ProtonDBRating(app_id=1, tier="borked"), }, @@ -206,7 +206,7 @@ class TestPickPlayableCandidate: game = _game(app_id=440, name="TF2") with ( patch( - "steam_backlog_enforcer.scanning.fetch_protondb_ratings", + "steam_backlog_enforcer._scanning_candidates.fetch_protondb_ratings", return_value={ 440: ProtonDBRating(app_id=440, tier="platinum"), }, diff --git a/steam_backlog_enforcer/tests/test_scanning_cached_confidence_overlay_avoids_refetch.py b/steam_backlog_enforcer/tests/test_scanning_cached_confidence_overlay_avoids_refetch.py index 80321a5..cd712a9 100644 --- a/steam_backlog_enforcer/tests/test_scanning_cached_confidence_overlay_avoids_refetch.py +++ b/steam_backlog_enforcer/tests/test_scanning_cached_confidence_overlay_avoids_refetch.py @@ -58,16 +58,16 @@ class TestPickNextGameGroup2: "steam_backlog_enforcer._scanning_confidence._refresh_candidate_confidence_batch" ) as mock_refresh_batch, patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), @@ -95,16 +95,16 @@ class TestPickNextGameGroup2: with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=playable_side_effect, ), patch("steam_backlog_enforcer.scanning._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), @@ -122,16 +122,16 @@ class TestPickNextGameGroup2: state = State() with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="2"), @@ -147,19 +147,19 @@ class TestPickNextGameGroup2: echoed: list[str] = [] with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch( - "steam_backlog_enforcer.scanning._echo", + "steam_backlog_enforcer._scanning_assign._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", side_effect=["abc", "1"]), @@ -176,19 +176,19 @@ class TestPickNextGameGroup2: echoed: list[str] = [] with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch( - "steam_backlog_enforcer.scanning._echo", + "steam_backlog_enforcer._scanning_assign._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", side_effect=["99", "1"]), diff --git a/steam_backlog_enforcer/tests/test_scanning_confidence_helpers.py b/steam_backlog_enforcer/tests/test_scanning_confidence_helpers.py index 729b743..ff85cb9 100644 --- a/steam_backlog_enforcer/tests/test_scanning_confidence_helpers.py +++ b/steam_backlog_enforcer/tests/test_scanning_confidence_helpers.py @@ -115,10 +115,10 @@ class TestConfidenceHelpers: with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=[None, good], ), - patch("steam_backlog_enforcer.scanning._echo") as mock_echo, + patch("steam_backlog_enforcer._scanning_candidates._echo") as mock_echo, ): picked, skipped_low_conf, skipped_linux = _pick_next_shortest_candidate( [bad, good], diff --git a/steam_backlog_enforcer/tests/test_scanning_confidence_helpers_pick_next_shortest_candidate_no_echo_whe.py b/steam_backlog_enforcer/tests/test_scanning_confidence_helpers_pick_next_shortest_candidate_no_echo_whe.py index d66fa57..d186071 100644 --- a/steam_backlog_enforcer/tests/test_scanning_confidence_helpers_pick_next_shortest_candidate_no_echo_whe.py +++ b/steam_backlog_enforcer/tests/test_scanning_confidence_helpers_pick_next_shortest_candidate_no_echo_whe.py @@ -41,10 +41,10 @@ class TestConfidenceHelpersGroup2: good = _game(app_id=51, name="Good", hours=2.0) with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", return_value=good, ), - patch("steam_backlog_enforcer.scanning._echo") as mock_echo, + patch("steam_backlog_enforcer._scanning_candidates._echo") as mock_echo, ): picked, _skipped_low_conf, skipped_linux = _pick_next_shortest_candidate( [good], @@ -62,7 +62,7 @@ class TestConfidenceHelpersGroup2: patch( "steam_backlog_enforcer._scanning_confidence._refresh_candidate_confidence" ), - patch("steam_backlog_enforcer.scanning._echo"), + patch("steam_backlog_enforcer._scanning_candidates._echo"), ): picked, skipped_low_conf, skipped_linux = _pick_next_shortest_candidate( [low_conf], @@ -76,10 +76,10 @@ class TestConfidenceHelpersGroup2: g1 = _game(app_id=10, name="Borked", hours=1.0) with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", return_value=None, ), - patch("steam_backlog_enforcer.scanning._echo") as mock_echo, + patch("steam_backlog_enforcer._scanning_candidates._echo") as mock_echo, ): picked, _skipped_low_conf, skipped_linux = _pick_next_shortest_candidate( [g1], @@ -100,7 +100,7 @@ class TestConfidenceHelpersGroup2: "steam_backlog_enforcer.scanning.SteamAPIClient", return_value=mock_client, ), - patch("steam_backlog_enforcer.scanning._echo"), + patch("steam_backlog_enforcer._scanning_candidates._echo"), patch( "steam_backlog_enforcer.scanning.send_notification", ), @@ -127,7 +127,7 @@ class TestConfidenceHelpersGroup2: "steam_backlog_enforcer.scanning.SteamAPIClient", return_value=mock_client, ), - patch("steam_backlog_enforcer.scanning._echo"), + patch("steam_backlog_enforcer._scanning_candidates._echo"), patch( "steam_backlog_enforcer.scanning.send_notification", ), @@ -150,7 +150,7 @@ class TestConfidenceHelpersGroup2: "steam_backlog_enforcer.scanning.SteamAPIClient", return_value=mock_client, ), - patch("steam_backlog_enforcer.scanning._echo"), + patch("steam_backlog_enforcer._scanning_candidates._echo"), patch("steam_backlog_enforcer.scanning.detect_tampering"), ): state = State(current_app_id=440, current_game_name="TF2") diff --git a/steam_backlog_enforcer/tests/test_scanning_part2.py b/steam_backlog_enforcer/tests/test_scanning_part2.py index 09d7d99..9c2aadc 100644 --- a/steam_backlog_enforcer/tests/test_scanning_part2.py +++ b/steam_backlog_enforcer/tests/test_scanning_part2.py @@ -11,7 +11,7 @@ from steam_backlog_enforcer.scanning import ( detect_tampering, ) -PKG = "steam_backlog_enforcer.scanning" +PKG = "steam_backlog_enforcer._scanning_tampering" def _entry( diff --git a/steam_backlog_enforcer/tests/test_scanning_part3.py b/steam_backlog_enforcer/tests/test_scanning_part3.py index 030a786..618298a 100644 --- a/steam_backlog_enforcer/tests/test_scanning_part3.py +++ b/steam_backlog_enforcer/tests/test_scanning_part3.py @@ -54,7 +54,7 @@ class TestPickNextGame: side_effect=refresh_side_effect, ) as mock_refresh, patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch( @@ -66,11 +66,11 @@ class TestPickNextGame: side_effect=lambda *a, **_: echoed.append(a[0]), ), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), @@ -96,16 +96,16 @@ class TestPickNextGame: "steam_backlog_enforcer._scanning_confidence._refresh_candidate_confidence_batch" ) as mock_refresh_batch, patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), diff --git a/steam_backlog_enforcer/tests/test_scanning_part4.py b/steam_backlog_enforcer/tests/test_scanning_part4.py index 8f62609..2ffc1aa 100644 --- a/steam_backlog_enforcer/tests/test_scanning_part4.py +++ b/steam_backlog_enforcer/tests/test_scanning_part4.py @@ -38,7 +38,7 @@ class TestCollectTopCandidates: """Returns at most n qualified candidates.""" games = [_game(app_id=i, name=f"G{i}", hours=float(i)) for i in range(1, 6)] with patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ): qualified, conf_skip, linux_skip = _collect_top_candidates(games, n=3) @@ -53,10 +53,10 @@ class TestCollectTopCandidates: g2 = _game(app_id=2, name="Good", hours=2.0) with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: None if c[0].app_id == 1 else c[0], ), - patch("steam_backlog_enforcer.scanning._echo"), + patch("steam_backlog_enforcer._scanning_candidates._echo"), ): qualified, conf_skip, linux_skip = _collect_top_candidates([g1, g2], n=10) assert [g.app_id for g in qualified] == [2] @@ -75,10 +75,10 @@ class TestCollectTopCandidates: g = _game(app_id=1, name="Good", hours=1.0) with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), - patch("steam_backlog_enforcer.scanning._echo") as mock_echo, + patch("steam_backlog_enforcer._scanning_candidates._echo") as mock_echo, ): _collect_top_candidates([g], n=10) mock_echo.assert_not_called() diff --git a/steam_backlog_enforcer/tests/test_scanning_pick_next_game.py b/steam_backlog_enforcer/tests/test_scanning_pick_next_game.py index 16c354c..00bd4b3 100644 --- a/steam_backlog_enforcer/tests/test_scanning_pick_next_game.py +++ b/steam_backlog_enforcer/tests/test_scanning_pick_next_game.py @@ -41,17 +41,17 @@ class TestPickNextGame: state = State() with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch("steam_backlog_enforcer._scanning_confidence._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), @@ -76,16 +76,16 @@ class TestPickNextGame: state = State(finished_app_ids=[1]) with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), @@ -100,7 +100,7 @@ class TestPickNextGame: state = State() with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", return_value=None, ), patch("steam_backlog_enforcer.scanning._echo"), @@ -115,17 +115,17 @@ class TestPickNextGame: state = State() with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch("steam_backlog_enforcer._scanning_confidence._echo"), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=2, ), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch("builtins.input", return_value="1"), @@ -140,16 +140,18 @@ class TestPickNextGame: state = State() with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch("steam_backlog_enforcer._scanning_confidence._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=False, ), - patch("steam_backlog_enforcer.scanning.install_game") as mock_install, + patch( + "steam_backlog_enforcer._scanning_assign.install_game" + ) as mock_install, patch("builtins.input", return_value="1"), ): pick_next_game([g1], state, config) diff --git a/steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py b/steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py index b893a0b..59990a4 100644 --- a/steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py +++ b/steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py @@ -38,25 +38,25 @@ class TestPickNextGameSequential: stack = contextlib.ExitStack() stack.enter_context( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ) ) stack.enter_context( patch( - "steam_backlog_enforcer.scanning._echo", + "steam_backlog_enforcer._scanning_assign._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ) ) stack.enter_context( patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ) ) stack.enter_context( patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ) ) @@ -113,11 +113,11 @@ class TestPickNextGameSequential: echoed: list[str] = [] with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", return_value=None, ), patch( - "steam_backlog_enforcer.scanning._echo", + "steam_backlog_enforcer._scanning_assign._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), patch("steam_backlog_enforcer.config._atomic_write"), @@ -136,11 +136,11 @@ class TestPickNextGameSequential: echoed: list[str] = [] with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch( - "steam_backlog_enforcer.scanning._echo", + "steam_backlog_enforcer._scanning_assign._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), patch( diff --git a/steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py b/steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py index 2d90c1f..7f9b712 100644 --- a/steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py +++ b/steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py @@ -41,16 +41,16 @@ class TestPickNextGameGroup2: state = State() with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), @@ -65,16 +65,16 @@ class TestPickNextGameGroup2: state = State() with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch("steam_backlog_enforcer.scanning._echo"), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), @@ -95,7 +95,7 @@ class TestPickNextGameGroup2: state = State() with ( patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, ), patch( @@ -107,11 +107,11 @@ class TestPickNextGameGroup2: side_effect=lambda *a, **_: echoed.append(a[0]), ), patch( - "steam_backlog_enforcer.scanning.is_game_installed", + "steam_backlog_enforcer._scanning_assign.is_game_installed", return_value=True, ), patch( - "steam_backlog_enforcer.scanning.uninstall_other_games", + "steam_backlog_enforcer._scanning_assign.uninstall_other_games", return_value=0, ), patch("builtins.input", return_value="1"), @@ -142,7 +142,7 @@ class TestPickNextGameGroup2: side_effect=lambda *a, **_: echoed.append(a[0]), ), patch( - "steam_backlog_enforcer.scanning._pick_playable_candidate", + "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", return_value=None, ) as mock_pick, ): diff --git a/steam_backlog_enforcer/tests/test_stats.py b/steam_backlog_enforcer/tests/test_stats.py index baa3066..78f9084 100644 --- a/steam_backlog_enforcer/tests/test_stats.py +++ b/steam_backlog_enforcer/tests/test_stats.py @@ -12,7 +12,7 @@ from steam_backlog_enforcer.config import State from steam_backlog_enforcer.protondb import ProtonDBRating from steam_backlog_enforcer.steam_api import GameInfo -_PKG = "steam_backlog_enforcer._stats" +_PKG = "steam_backlog_enforcer._stats_gathering" def _game( diff --git a/steam_backlog_enforcer/tests/test_stats_ensure_rush_data.py b/steam_backlog_enforcer/tests/test_stats_ensure_rush_data.py index 02576d9..2b119fe 100644 --- a/steam_backlog_enforcer/tests/test_stats_ensure_rush_data.py +++ b/steam_backlog_enforcer/tests/test_stats_ensure_rush_data.py @@ -12,7 +12,7 @@ from steam_backlog_enforcer._stats import ( from steam_backlog_enforcer.protondb import ProtonDBRating from steam_backlog_enforcer.steam_api import GameInfo -_PKG = "steam_backlog_enforcer._stats" +_PKG = "steam_backlog_enforcer._stats_gathering" def _game( diff --git a/steam_backlog_enforcer/tests/test_stats_print_worst_example.py b/steam_backlog_enforcer/tests/test_stats_print_worst_example.py index fdcea19..1c2b40e 100644 --- a/steam_backlog_enforcer/tests/test_stats_print_worst_example.py +++ b/steam_backlog_enforcer/tests/test_stats_print_worst_example.py @@ -11,7 +11,7 @@ from steam_backlog_enforcer._stats import ( from steam_backlog_enforcer.protondb import ProtonDBRating from steam_backlog_enforcer.steam_api import GameInfo -_PKG = "steam_backlog_enforcer._stats" +_PKG = "steam_backlog_enforcer._stats_display" def _game( diff --git a/steam_backlog_enforcer/tests/test_stats_refresh_recently_played_completions.py b/steam_backlog_enforcer/tests/test_stats_refresh_recently_played_completions.py index f945a1b..1e9e355 100644 --- a/steam_backlog_enforcer/tests/test_stats_refresh_recently_played_completions.py +++ b/steam_backlog_enforcer/tests/test_stats_refresh_recently_played_completions.py @@ -8,7 +8,7 @@ from steam_backlog_enforcer.config import Config from steam_backlog_enforcer.protondb import ProtonDBRating from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIError -_PKG = "steam_backlog_enforcer._stats" +_PKG = "steam_backlog_enforcer._stats_gathering" def _game( diff --git a/steam_backlog_enforcer/tests/test_stats_sum_hours.py b/steam_backlog_enforcer/tests/test_stats_sum_hours.py index e53e69f..0d6b7fd 100644 --- a/steam_backlog_enforcer/tests/test_stats_sum_hours.py +++ b/steam_backlog_enforcer/tests/test_stats_sum_hours.py @@ -16,7 +16,9 @@ from steam_backlog_enforcer.config import State from steam_backlog_enforcer.protondb import ProtonDBRating from steam_backlog_enforcer.steam_api import GameInfo -_PKG = "steam_backlog_enforcer._stats" +_PKG = "steam_backlog_enforcer._stats_display" +# _print_scenario was not moved: it still lives in _stats. +_PKG_STATS = "steam_backlog_enforcer._stats" def _game( @@ -118,14 +120,18 @@ class TestPrintScenario: def test_no_data_prints_no_data_message(self) -> None: """Test no data prints no data message.""" echoed: list[str] = [] - with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + with patch( + f"{_PKG_STATS}._echo", side_effect=lambda *a, **_: echoed.append(a[0]) + ): _print_scenario("2. RUSH", 0.0, 0, 5) assert any("No data available" in s for s in echoed) def test_with_data_no_missing(self) -> None: """Test with data no missing.""" echoed: list[str] = [] - with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + with patch( + f"{_PKG_STATS}._echo", side_effect=lambda *a, **_: echoed.append(a[0]) + ): _print_scenario("2. RUSH", 100.0, 0, 5) assert any("Total:" in s for s in echoed) assert not any("had no data" in s for s in echoed) @@ -133,7 +139,9 @@ class TestPrintScenario: def test_with_data_and_missing(self) -> None: """Test with data and missing.""" echoed: list[str] = [] - with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + with patch( + f"{_PKG_STATS}._echo", side_effect=lambda *a, **_: echoed.append(a[0]) + ): _print_scenario("2. RUSH", 100.0, 2, 5) assert any("had no data" in s for s in echoed) diff --git a/steam_backlog_enforcer/tests/test_stats_unplayable_rating_counts_linux_skipped.py b/steam_backlog_enforcer/tests/test_stats_unplayable_rating_counts_linux_skipped.py index 3efa6c4..9326254 100644 --- a/steam_backlog_enforcer/tests/test_stats_unplayable_rating_counts_linux_skipped.py +++ b/steam_backlog_enforcer/tests/test_stats_unplayable_rating_counts_linux_skipped.py @@ -12,7 +12,7 @@ from steam_backlog_enforcer.config import State from steam_backlog_enforcer.protondb import ProtonDBRating from steam_backlog_enforcer.steam_api import GameInfo -_PKG = "steam_backlog_enforcer._stats" +_PKG = "steam_backlog_enforcer._stats_gathering" def _game(