From 14d706ee16693463e9bed250edf7237b3b87f6a7 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sun, 3 May 2026 22:30:48 +0200 Subject: [PATCH] Apply focus-mode, screen-locker, and steam backlog updates --- steam_backlog_enforcer/_cmd_done.py | 181 ++++- steam_backlog_enforcer/_enforce_loop.py | 27 +- steam_backlog_enforcer/_hltb_detail.py | 10 +- steam_backlog_enforcer/_hltb_types.py | 90 ++- steam_backlog_enforcer/game_install.py | 11 +- steam_backlog_enforcer/hltb.py | 348 +++++++-- steam_backlog_enforcer/scanning.py | 290 ++++++- steam_backlog_enforcer/steam_api.py | 6 + steam_backlog_enforcer/tests/test_cmd_done.py | 459 ++++++++++- .../tests/test_enforce_loop.py | 22 +- .../tests/test_game_install.py | 14 + .../tests/test_hltb_detail.py | 18 +- .../tests/test_hltb_part2.py | 100 +++ .../tests/test_hltb_search.py | 100 ++- .../tests/test_main_part2.py | 6 + .../tests/test_polls_tracking.py | 729 ++++++++++++++++++ steam_backlog_enforcer/tests/test_scanning.py | 294 +++++++ 17 files changed, 2554 insertions(+), 151 deletions(-) create mode 100644 steam_backlog_enforcer/tests/test_polls_tracking.py diff --git a/steam_backlog_enforcer/_cmd_done.py b/steam_backlog_enforcer/_cmd_done.py index a49fd8b..4631865 100644 --- a/steam_backlog_enforcer/_cmd_done.py +++ b/steam_backlog_enforcer/_cmd_done.py @@ -15,12 +15,18 @@ from python_pkg.steam_backlog_enforcer.game_install import ( uninstall_other_games, ) 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_playable_candidate, + _confidence_fail_reasons, + _pick_next_shortest_candidate, + _refresh_candidate_confidence, pick_next_game, ) from python_pkg.steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient @@ -28,6 +34,81 @@ from python_pkg.steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient _REASSIGN_REFRESH_LIMIT = 50 +def _backfill_polls_for_finished( + state: State, + extra_app_id: int | None = None, +) -> dict[int, int]: + """Lazily fetch poll counts for already-finished games missing them. + + If ``extra_app_id`` is provided and its poll count is missing, it is + refreshed alongside finished games (used to populate polls for the + currently-assigned game on first run after the schema upgrade). + """ + polls_cache = load_hltb_polls_cache() + snapshot_data = load_snapshot() or [] + name_by_id = {d["app_id"]: d["name"] for d in snapshot_data} + candidate_ids = list(state.finished_app_ids) + if extra_app_id is not None and polls_cache.get(extra_app_id, 0) == 0: + candidate_ids.append(extra_app_id) + missing = [ + (aid, name_by_id[aid]) + for aid in candidate_ids + if aid in name_by_id and polls_cache.get(aid, 0) == 0 + ] + if not missing: + return polls_cache + + _echo(f" Backfilling HLTB poll counts for {len(missing)} game(s)...") + cache = load_hltb_cache() + preserved_hours = {aid: cache[aid] for aid, _ in missing if aid in cache} + for aid, _name in missing: + cache.pop(aid, None) + save_hltb_cache(cache, polls_cache) + + fetch_hltb_confidence_cached(missing) + + refreshed_hours = load_hltb_cache() + refreshed_polls = load_hltb_polls_cache() + for aid, prior_hours in preserved_hours.items(): + if prior_hours > 0 and refreshed_hours.get(aid, -1) <= 0: + refreshed_hours[aid] = prior_hours + save_hltb_cache(refreshed_hours, refreshed_polls) + return refreshed_polls + + +def _report_assigned_confidence( + app_id: int, + state: State, +) -> None: + """Print HLTB poll-count confidence for the currently-assigned game.""" + polls_cache = _backfill_polls_for_finished(state, extra_app_id=app_id) + chosen_polls = polls_cache.get(app_id, 0) + + finished_polls = [ + (polls_cache[aid], aid) + for aid in state.finished_app_ids + if polls_cache.get(aid, 0) > 0 and aid != app_id + ] + snapshot_data = load_snapshot() or [] + name_by_id = {d["app_id"]: d["name"] for d in snapshot_data} + + warning = "" + if finished_polls: + min_polls = min(p for p, _ in finished_polls) + if 0 < chosen_polls < min_polls: + warning = " ⚠ NEW LOW — estimate may be unreliable" + elif chosen_polls == 0: + warning = " ⚠ no polls recorded — estimate may be unreliable" + elif chosen_polls == 0: + warning = " ⚠ no polls recorded — estimate may be unreliable" + + _echo(f" HLTB confidence: {chosen_polls} polled completionist times{warning}") + if finished_polls: + min_polls, min_aid = min(finished_polls) + min_name = name_by_id.get(min_aid, f"AppID={min_aid}") + _echo(f" Historical min among finished: {min_polls} ({min_name})") + + def _apply_cached_hours_to_games( games: list[GameInfo], hltb_cache: dict[int, float], @@ -38,6 +119,17 @@ 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], @@ -69,6 +161,46 @@ 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, @@ -89,23 +221,44 @@ def _try_reassign_shorter_game( 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) - if not candidates or candidates[0].app_id == app_id: + candidates = [c for c in candidates if c.app_id != app_id] + if not candidates: return False - # Filter out Linux-incompatible games before deciding to reassign. - playable = _pick_playable_candidate( - [c for c in candidates if c.app_id != app_id], + + playable, _confidence_skipped, _linux_skipped = _pick_next_shortest_candidate( + candidates, ) - if playable is None or playable.completionist_hours >= hours: + if playable is None: return False - _echo( - f"\n Reassigning: {playable.name} is shorter" - f" (~{playable.completionist_hours:.1f}h vs ~{hours:.1f}h)" + + 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) @@ -193,6 +346,15 @@ def _enforce_on_done(config: Config, state: State) -> None: use_steam_protocol=True, ) + # Reconcile library: hide non-assigned games and unhide the assigned one. + # Without this, an interrupted earlier completion can leave the new + # assigned game hidden and stale games visible. + owned_ids = get_all_owned_app_ids(config) + if owned_ids: + hidden = hide_other_games(owned_ids, state.current_app_id) + if hidden > 0: + _echo(f" Library: hid {hidden} games") + def cmd_done(config: Config, state: State) -> None: """Check completion, pick next game, uninstall & hide. @@ -230,6 +392,7 @@ def cmd_done(config: Config, state: State) -> None: hours = hltb_cache.get(app_id, -1.0) if hours > 0: _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 diff --git a/steam_backlog_enforcer/_enforce_loop.py b/steam_backlog_enforcer/_enforce_loop.py index b8f28ba..7c0c8f3 100644 --- a/steam_backlog_enforcer/_enforce_loop.py +++ b/steam_backlog_enforcer/_enforce_loop.py @@ -37,19 +37,34 @@ logger = logging.getLogger(__name__) def get_all_owned_app_ids(config: Config) -> list[int]: - """Get all owned game app IDs from the snapshot or Steam API.""" - snapshot = load_snapshot() - if snapshot: - return [d["app_id"] for d in snapshot] + """Get all owned game app IDs from Steam API plus snapshot fallback. + + Snapshot data contains only games with achievements, so API data is the + primary source for library hiding. Snapshot IDs are merged in to keep + behavior resilient when the API result is partial. + """ + snapshot = load_snapshot() or [] + snapshot_ids = [int(d["app_id"]) for d in snapshot if "app_id" in d] - # Fall back to a quick API call. try: client = SteamAPIClient(config.steam_api_key, config.steam_id) owned = client.get_owned_games() - return [g["appid"] for g in owned] + api_ids = [int(g["appid"]) for g in owned if "appid" in g] + + merged_ids: list[int] = [] + seen: set[int] = set() + for app_id in [*api_ids, *snapshot_ids]: + if app_id in seen: + continue + seen.add(app_id) + merged_ids.append(app_id) except (OSError, RuntimeError, ValueError): + if snapshot_ids: + return snapshot_ids logger.warning("Could not fetch owned game list for hiding.") return [] + else: + return merged_ids # ────────────────────────────────────────────────────────────── diff --git a/steam_backlog_enforcer/_hltb_detail.py b/steam_backlog_enforcer/_hltb_detail.py index 03a425e..341ea0b 100644 --- a/steam_backlog_enforcer/_hltb_detail.py +++ b/steam_backlog_enforcer/_hltb_detail.py @@ -149,12 +149,20 @@ async def _fetch_detail_one( async def _fetch_leisure_times( search_results: list[HLTBResult], cache: dict[int, float], + polls: dict[int, int], progress_cb: ProgressCb | None, + count_comp: dict[int, int] | None = None, ) -> None: """Fetch leisure times from game detail pages for all search results. Updates ``cache`` in-place with leisure hours (including DLC time). + The ``polls`` and ``count_comp`` mappings are forwarded to + :func:`save_hltb_cache` so the on-disk cache keeps confidence metrics + captured during the search step. """ + if count_comp is None: + count_comp = {} + valid = [r for r in search_results if r.hltb_game_id > 0] if not valid: return @@ -198,7 +206,7 @@ async def _fetch_leisure_times( progress_cb(done, total, found, r.game_name) if not done % _SAVE_INTERVAL: - save_hltb_cache(cache) + save_hltb_cache(cache, polls, count_comp) def _collect_dlc_relationships( diff --git a/steam_backlog_enforcer/_hltb_types.py b/steam_backlog_enforcer/_hltb_types.py index ddadec3..d569041 100644 --- a/steam_backlog_enforcer/_hltb_types.py +++ b/steam_backlog_enforcer/_hltb_types.py @@ -6,6 +6,7 @@ from collections.abc import Callable from dataclasses import dataclass import json import logging +from typing import Any from python_pkg.steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write @@ -42,6 +43,8 @@ class HLTBResult: completionist_hours: float similarity: float hltb_game_id: int = 0 + comp_100_count: int = 0 + count_comp: int = 0 @dataclass @@ -53,26 +56,91 @@ class _AuthInfo: hp_val: str = "" +def _read_raw_cache() -> dict[int, dict[str, Any]]: + """Read the persistent HLTB cache, normalizing legacy float entries. + + Cache schema on disk (current): + { + "": { + "hours": , + "polls": , + "count_comp": + } + } + + Legacy format (single float value per app) is migrated transparently. + """ + if not HLTB_CACHE_FILE.exists(): + return {} + try: + data = json.loads(HLTB_CACHE_FILE.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + logger.warning("Corrupt HLTB cache, starting fresh.") + return {} + out: dict[int, dict[str, Any]] = {} + for k, v in data.items(): + try: + aid = int(k) + except (TypeError, ValueError): + continue + if isinstance(v, dict): + out[aid] = { + "hours": float(v.get("hours", -1)), + "polls": int(v.get("polls", 0)), + "count_comp": int(v.get("count_comp", 0)), + } + else: + try: + out[aid] = {"hours": float(v), "polls": 0, "count_comp": 0} + except (TypeError, ValueError): + continue + return out + + def load_hltb_cache() -> dict[int, float]: - """Load the persistent HLTB cache from disk. + """Load the hours portion of the HLTB cache. Returns: dict mapping app_id -> completionist_hours (-1 = no data on HLTB). """ - if HLTB_CACHE_FILE.exists(): - try: - data = json.loads(HLTB_CACHE_FILE.read_text(encoding="utf-8")) - return {int(k): float(v) for k, v in data.items()} - except (json.JSONDecodeError, ValueError, OSError): - logger.warning("Corrupt HLTB cache, starting fresh.") - return {} + return {aid: v["hours"] for aid, v in _read_raw_cache().items()} -def save_hltb_cache(cache: dict[int, float]) -> None: - """Save the HLTB cache to disk.""" +def load_hltb_polls_cache() -> dict[int, int]: + """Load the polled-completionist-times portion of the HLTB cache. + + Returns: dict mapping app_id -> ``comp_100_count`` (0 = unknown). + """ + return {aid: v["polls"] for aid, v in _read_raw_cache().items()} + + +def load_hltb_count_comp_cache() -> dict[int, int]: + """Load the ``count_comp`` portion of the HLTB cache. + + Returns: dict mapping app_id -> ``count_comp`` (0 = unknown). + """ + return {aid: v["count_comp"] for aid, v in _read_raw_cache().items()} + + +def save_hltb_cache( + cache: dict[int, float], + polls: dict[int, int] | None = None, + count_comp: dict[int, int] | None = None, +) -> None: + """Save the HLTB cache to disk, including confidence metrics.""" + polls = polls or {} + count_comp = count_comp or {} + out = { + str(aid): { + "hours": hours, + "polls": polls.get(aid, 0), + "count_comp": count_comp.get(aid, 0), + } + for aid, hours in cache.items() + } try: _atomic_write( HLTB_CACHE_FILE, - json.dumps({str(k): v for k, v in cache.items()}, indent=2) + "\n", + json.dumps(out, indent=2) + "\n", ) except OSError: logger.exception("Failed to save HLTB cache") diff --git a/steam_backlog_enforcer/game_install.py b/steam_backlog_enforcer/game_install.py index b0712b6..512d9b8 100644 --- a/steam_backlog_enforcer/game_install.py +++ b/steam_backlog_enforcer/game_install.py @@ -21,12 +21,15 @@ _REAL_STEAMAPPS = Path("~/.local/share/Steam/steamapps").expanduser() def _assert_not_real_steam(path: Path) -> None: - """Raise if *path* is inside the real Steam directory. + """Raise if *path* is inside the real Steam directory during tests. - Defence-in-depth guard: even if test fixtures fail to - redirect ``STEAMAPPS_PATH``, destructive operations - (uninstall, rmtree, unlink) will refuse to touch real files. + Defence-in-depth guard: when running under pytest, even if test + fixtures fail to redirect ``STEAMAPPS_PATH``, destructive + operations (uninstall, rmtree, unlink) will refuse to touch + real files. In production runs this is a no-op. """ + if "PYTEST_CURRENT_TEST" not in os.environ: + return # production run — real Steam paths are expected try: path.resolve().relative_to(_REAL_STEAMAPPS.resolve()) except ValueError: diff --git a/steam_backlog_enforcer/hltb.py b/steam_backlog_enforcer/hltb.py index c90016b..8b2a24b 100644 --- a/steam_backlog_enforcer/hltb.py +++ b/steam_backlog_enforcer/hltb.py @@ -18,6 +18,7 @@ from difflib import SequenceMatcher from http import HTTPStatus import json import logging +import re import time from typing import Any @@ -37,6 +38,8 @@ from python_pkg.steam_backlog_enforcer._hltb_types import ( ProgressCb, _AuthInfo, load_hltb_cache, + load_hltb_count_comp_cache, + load_hltb_polls_cache, save_hltb_cache, ) @@ -145,6 +148,70 @@ def _build_search_payload(game_name: str, auth: _AuthInfo | None = None) -> str: return json.dumps(payload) +def _build_search_variants(game_name: str) -> list[str]: + """Return fallback search terms for one Steam game title.""" + base = game_name.strip() + variants = [base] + no_year = re.sub(r"\s*\(\d{4}\)$", "", base).strip() + if no_year and no_year != base: + variants.append(no_year) + return variants + + +def _collect_candidates( + query_name: str, + data: dict[str, Any], +) -> list[tuple[dict[str, Any], float]]: + """Build candidate list from one HLTB response payload.""" + candidates: list[tuple[dict[str, Any], float]] = [] + lower_name = query_name.lower() + for entry in data.get("data", []): + entry_name = entry.get("game_name", "") + entry_alias = entry.get("game_alias", "") or "" + is_dlc = str(entry.get("game_type", "")).lower() == "dlc" + sim = max( + _similarity(query_name, entry_name), + _similarity(query_name, entry_alias), + ) + is_full_edition = ( + (not is_dlc) and entry_name.lower().startswith(lower_name + ":") + ) or ((not is_dlc) and entry_name.lower().startswith(lower_name + " -")) + if sim >= MIN_SIMILARITY or is_full_edition: + comp_100 = entry.get("comp_100", 0) + if comp_100 and comp_100 > 0: + candidates.append((entry, sim)) + return candidates + + +def _build_result_from_best( + app_id: int, + original_name: str, + query_name: str, + best: tuple[dict[str, Any], float], +) -> HLTBResult: + """Convert selected HLTB entry into HLTBResult.""" + entry, sim = best + hours = round(entry["comp_100"] / 3600, 2) + logger.debug( + ("HLTB match for '%s' via '%s': '%s' (id=%s, comp_100=%s, sim=%.3f)"), + original_name, + query_name, + entry.get("game_name"), + entry.get("game_id"), + entry.get("comp_100"), + sim, + ) + return HLTBResult( + app_id=app_id, + game_name=original_name, + completionist_hours=hours, + similarity=sim, + hltb_game_id=entry.get("game_id", 0), + comp_100_count=int(entry.get("comp_100_count", 0) or 0), + count_comp=int(entry.get("count_comp", 0) or 0), + ) + + def _pick_best_hltb_entry( search_name: str, candidates: list[tuple[dict[str, Any], float]], @@ -204,6 +271,9 @@ def _find_best_extended( """ best: tuple[dict[str, Any], float] | None = None for entry, sim in usable: + game_type = str(entry.get("game_type", "")).lower() + if game_type not in ("", "game"): + continue entry_name = (entry.get("game_name") or "").lower() if entry_name.startswith((lower + ":", lower + " -")): suffix = entry_name[len(lower) :].lstrip(" :-") @@ -223,12 +293,19 @@ def _resolve_exact_vs_extended( if best_exact is not None and best_extended is not None: exact_hours = best_exact[0].get("comp_100", 0) extended_hours = best_extended[0].get("comp_100", 0) + exact_confidence = int(best_exact[0].get("comp_100_count", 0) or 0) + int( + best_exact[0].get("count_comp", 0) or 0 + ) + extended_confidence = int(best_extended[0].get("comp_100_count", 0) or 0) + int( + best_extended[0].get("count_comp", 0) or 0 + ) # Prefer the extended entry only when it has strictly more hours - # than the exact match. This lets "FAITH: The Unholy Trinity" - # (7 h) beat "FAITH" (0.5 h demo) while preventing - # "Timberman: The Big Adventure" (2 h) from beating - # "Timberman" (26 h). - if extended_hours > exact_hours: + # than the exact match AND at least as much confidence. + # This lets "FAITH: The Unholy Trinity" (full game) beat + # a low-confidence exact demo while preventing low-confidence + # mods like "Celeste - Strawberry Jam" from beating + # the exact base game. + if extended_hours > exact_hours and extended_confidence >= exact_confidence: return best_extended return best_exact if best_exact is not None: @@ -253,6 +330,8 @@ class _SearchCtx: search_url: str headers: dict[str, str] cache: dict[int, float] + polls: dict[int, int] = field(default_factory=dict) + count_comp: dict[int, int] = field(default_factory=dict) auth: _AuthInfo | None = None counter: dict[str, int] = field(default_factory=dict) total: int = 0 @@ -268,71 +347,43 @@ async def _search_one( """Search HLTB for one game via direct POST, update cache.""" async with sem: result: HLTBResult | None = None - payload = _build_search_payload(name, ctx.auth) - try: - async with ctx.session.post( - ctx.search_url, - headers=ctx.headers, - data=payload, - ) as resp: - if resp.status == HTTPStatus.OK: + for query_name in _build_search_variants(name): + payload = _build_search_payload(query_name, ctx.auth) + try: + async with ctx.session.post( + ctx.search_url, + headers=ctx.headers, + data=payload, + ) as resp: + if resp.status != HTTPStatus.OK: + continue data = await resp.json() - candidates: list[tuple[dict[str, Any], float]] = [] - lower_name = name.lower() - for entry in data.get("data", []): - entry_name = entry.get("game_name", "") - entry_alias = entry.get("game_alias", "") or "" - is_dlc = str(entry.get("game_type", "")).lower() == "dlc" - sim = max( - _similarity(name, entry_name), - _similarity(name, entry_alias), - ) - is_full_edition = ( - (not is_dlc) - and entry_name.lower().startswith(lower_name + ":") - ) or ( - (not is_dlc) - and entry_name.lower().startswith(lower_name + " -") - ) - if sim >= MIN_SIMILARITY or is_full_edition: - comp_100 = entry.get("comp_100", 0) - if comp_100 and comp_100 > 0: - candidates.append((entry, sim)) - best = _pick_best_hltb_entry(name, candidates) - if best is not None: - entry, sim = best - hours = round(entry["comp_100"] / 3600, 2) - logger.debug( - "HLTB match for '%s': '%s' (id=%s, comp_100=%s, sim=%.3f)", - name, - entry.get("game_name"), - entry.get("game_id"), - entry.get("comp_100"), - sim, - ) - result = HLTBResult( - app_id=app_id, - game_name=name, - completionist_hours=hours, - similarity=sim, - hltb_game_id=entry.get("game_id", 0), - ) - except (aiohttp.ClientError, asyncio.TimeoutError) as exc: - logger.debug("HLTB search failed for '%s': %s", name, exc) + candidates = _collect_candidates(query_name, data) + best = _pick_best_hltb_entry(query_name, candidates) + if best is None: + continue + result = _build_result_from_best(app_id, name, query_name, best) + break + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.debug("HLTB search failed for '%s': %s", query_name, exc) # Update cache immediately (miss = -1). if result is not None: ctx.cache[app_id] = result.completionist_hours + ctx.polls[app_id] = result.comp_100_count + ctx.count_comp[app_id] = result.count_comp ctx.counter["found"] += 1 else: ctx.cache[app_id] = -1 + ctx.polls[app_id] = 0 + ctx.count_comp[app_id] = 0 ctx.counter["done"] += 1 done = ctx.counter["done"] # Incremental save every _SAVE_INTERVAL lookups. if not done % _SAVE_INTERVAL: - save_hltb_cache(ctx.cache) + save_hltb_cache(ctx.cache, ctx.polls, ctx.count_comp) # Report progress. if ctx.progress_cb is not None: @@ -344,7 +395,9 @@ async def _search_one( async def _fetch_batch( games: list[tuple[int, str]], cache: dict[int, float], + polls: dict[int, int], progress_cb: ProgressCb | None, + count_comp: dict[int, int] | None = None, ) -> list[HLTBResult]: """Fetch HLTB data for a batch of games using one shared session.""" # 1. Discover the search URL (sync, one-time). @@ -380,6 +433,9 @@ async def _fetch_batch( counter = {"done": 0, "found": 0} total = len(games) + if count_comp is None: + count_comp = {} + connector = aiohttp.TCPConnector( limit=MAX_CONCURRENT, keepalive_timeout=30, @@ -393,6 +449,8 @@ async def _fetch_batch( search_url=search_url, headers=headers, cache=cache, + polls=polls, + count_comp=count_comp, auth=auth, counter=counter, total=total, @@ -416,22 +474,141 @@ async def _fetch_batch( "Fetching leisure times for %d games from detail pages...", len(search_results), ) - await _fetch_leisure_times(search_results, cache, progress_cb=None) + await _fetch_leisure_times( + search_results, + cache, + polls, + progress_cb=None, + count_comp=count_comp, + ) return search_results +async def _fetch_batch_confidence_only( + games: list[tuple[int, str]], + cache: dict[int, float], + polls: dict[int, int], + progress_cb: ProgressCb | None, + count_comp: dict[int, int] | None = None, +) -> list[HLTBResult]: + """Fetch only search-level HLTB data (hours + confidence), no detail pages.""" + # 1. Discover the search URL (sync, one-time). + search_url = _get_hltb_search_url() + logger.info("HLTB search URL: %s", search_url) + + timeout = aiohttp.ClientTimeout(total=20, sock_read=15) + + # 2. Get auth info (separate session — avoids reuse issues). + async with aiohttp.ClientSession(timeout=timeout) as init_session: + auth = await _get_auth_info(search_url, init_session) + if auth is None: + logger.warning("Could not get HLTB auth info, aborting fetch.") + return [] + logger.info("HLTB auth token acquired.") + + # 3. Build shared headers for all search requests. + headers: dict[str, str] = { + "content-type": "application/json", + "accept": "*/*", + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64; rv:136.0) Gecko/20100101 Firefox/136.0" + ), + "referer": "https://howlongtobeat.com/", + "x-auth-token": auth.token, + } + if auth.hp_key: + headers["x-hp-key"] = auth.hp_key + headers["x-hp-val"] = auth.hp_val + + # 4. Fire all searches through a single persistent session. + sem = asyncio.Semaphore(MAX_CONCURRENT) + counter = {"done": 0, "found": 0} + total = len(games) + + if count_comp is None: + count_comp = {} + + connector = aiohttp.TCPConnector( + limit=MAX_CONCURRENT, + keepalive_timeout=30, + ) + async with aiohttp.ClientSession( + timeout=timeout, + connector=connector, + ) as session: + ctx = _SearchCtx( + session=session, + search_url=search_url, + headers=headers, + cache=cache, + polls=polls, + count_comp=count_comp, + auth=auth, + counter=counter, + total=total, + progress_cb=progress_cb, + ) + tasks = [ + _search_one( + sem, + ctx, + app_id, + name, + ) + for app_id, name in games + ] + results = await asyncio.gather(*tasks) + + return [r for r in results if r is not None] + + def fetch_hltb_times( games: list[tuple[int, str]], cache: dict[int, float] | None = None, + polls: dict[int, int] | None = None, progress_cb: ProgressCb | None = None, + count_comp: dict[int, int] | None = None, ) -> list[HLTBResult]: """Synchronous wrapper: fetch HLTB times for games.""" if not games: return [] if cache is None: cache = {} - return asyncio.run(_fetch_batch(games, cache, progress_cb)) + if polls is None: + polls = {} + if count_comp is None: + count_comp = {} + return asyncio.run( + _fetch_batch(games, cache, polls, progress_cb, count_comp=count_comp) + ) + + +def fetch_hltb_confidence( + games: list[tuple[int, str]], + cache: dict[int, float] | None = None, + polls: dict[int, int] | None = None, + progress_cb: ProgressCb | None = None, + count_comp: dict[int, int] | None = None, +) -> list[HLTBResult]: + """Fetch only HLTB search-level data (hours + confidence metrics).""" + if not games: + return [] + if cache is None: + cache = {} + if polls is None: + polls = {} + if count_comp is None: + count_comp = {} + return asyncio.run( + _fetch_batch_confidence_only( + games, + cache, + polls, + progress_cb, + count_comp=count_comp, + ) + ) def fetch_hltb_times_cached( @@ -447,6 +624,8 @@ def fetch_hltb_times_cached( Returns: dict mapping app_id -> completionist_hours. """ cache = load_hltb_cache() + polls = load_hltb_polls_cache() + count_comp = load_hltb_count_comp_cache() uncached = [(app_id, name) for app_id, name in games if app_id not in cache] if uncached: @@ -456,11 +635,17 @@ def fetch_hltb_times_cached( len(games) - len(uncached), ) t0 = time.monotonic() - fetch_hltb_times(uncached, cache=cache, progress_cb=progress_cb) + fetch_hltb_times( + uncached, + cache=cache, + polls=polls, + progress_cb=progress_cb, + count_comp=count_comp, + ) elapsed = time.monotonic() - t0 # Final save. - save_hltb_cache(cache) + save_hltb_cache(cache, polls, count_comp) found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0) rate = len(uncached) / elapsed if elapsed > 0 else 0 @@ -477,6 +662,49 @@ def fetch_hltb_times_cached( return cache +def fetch_hltb_confidence_cached( + games: list[tuple[int, str]], + progress_cb: ProgressCb | None = None, +) -> dict[int, float]: + """Fetch HLTB search-level confidence data, using disk cache for known IDs.""" + cache = load_hltb_cache() + polls = load_hltb_polls_cache() + count_comp = load_hltb_count_comp_cache() + uncached = [(app_id, name) for app_id, name in games if app_id not in cache] + + if uncached: + logger.info( + "Fetching HLTB confidence for %d uncached games (%d cached)...", + len(uncached), + len(games) - len(uncached), + ) + t0 = time.monotonic() + fetch_hltb_confidence( + uncached, + cache=cache, + polls=polls, + progress_cb=progress_cb, + count_comp=count_comp, + ) + elapsed = time.monotonic() - t0 + + save_hltb_cache(cache, polls, count_comp) + + found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0) + rate = len(uncached) / elapsed if elapsed > 0 else 0 + logger.info( + "HLTB confidence fetch done: %d/%d found in %.1fs (%.0f games/s)", + found, + len(uncached), + elapsed, + rate, + ) + else: + logger.info("All %d games found in HLTB cache.", len(games)) + + return cache + + def get_hltb_submit_url(game_name: str) -> str | None: """Look up a game on HLTB and return its submit page URL. diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index 5614f19..06b0caa 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -6,6 +6,12 @@ import logging import time from typing import Any +from python_pkg.steam_backlog_enforcer._hltb_types import ( + load_hltb_cache, + load_hltb_count_comp_cache, + load_hltb_polls_cache, + save_hltb_cache, +) from python_pkg.steam_backlog_enforcer.config import ( Config, State, @@ -21,7 +27,10 @@ from python_pkg.steam_backlog_enforcer.game_install import ( is_game_installed, uninstall_other_games, ) -from python_pkg.steam_backlog_enforcer.hltb import fetch_hltb_times_cached +from python_pkg.steam_backlog_enforcer.hltb import ( + fetch_hltb_confidence_cached, + fetch_hltb_times_cached, +) from python_pkg.steam_backlog_enforcer.protondb import ( ProtonDBRating, fetch_protondb_ratings, @@ -31,6 +40,9 @@ from python_pkg.steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient logger = logging.getLogger(__name__) _TAMPER_CHECK_LIMIT = 3 +_MIN_COMP_100_POLLS = 3 +_MIN_COUNT_COMP = 15 +_MIN_CONFIDENCE_SUM = 18 # ────────────────────────────────────────────────────────────── @@ -78,9 +90,13 @@ def do_scan(config: Config, state: State) -> list[GameInfo]: hltb_cache = fetch_hltb_times_cached(incomplete, progress_cb=hltb_progress) _echo("") # newline after progress bar + polls_cache = load_hltb_polls_cache() + count_comp_cache = load_hltb_count_comp_cache() for g in games: hours = hltb_cache.get(g.app_id, -1) g.completionist_hours = hours + g.comp_100_count = polls_cache.get(g.app_id, 0) + g.count_comp = count_comp_cache.get(g.app_id, 0) found = sum(1 for h in hltb_cache.values() if h > 0) _echo(f" HLTB data: {found} games have completion estimates") @@ -94,6 +110,15 @@ def do_scan(config: Config, state: State) -> list[GameInfo]: # Auto-pick a game if none assigned. if state.current_app_id is None: pick_next_game(games, state, config) + else: + # Show confidence info for the already-assigned game too. + current = next( + (g for g in games if g.app_id == state.current_app_id), + None, + ) + if current is not None: + _echo(f"\n>>> CURRENT: {current.name} (AppID={current.app_id})") + _report_poll_confidence(current, games, state) return games @@ -148,7 +173,11 @@ def pick_next_game(games: list[GameInfo], state: State, config: Config) -> None: candidates = [g for g in games if not g.is_complete and g.app_id not in skip] if not candidates: - _echo("\nCongratulations! All games are complete!") + _echo( + "\nNo assignable games found " + "(HLTB confidence thresholds: comp_100 polls>=3, " + "count_comp>=15, sum>=18)." + ) state.current_app_id = None state.current_game_name = "" state.save() @@ -162,11 +191,19 @@ def pick_next_game(games: list[GameInfo], state: State, config: Config) -> None: candidates.sort(key=sort_key) - # Filter out Linux-incompatible games via ProtonDB. - chosen = _pick_playable_candidate(candidates) + chosen, confidence_skipped, linux_skipped = _pick_next_shortest_candidate( + candidates + ) if chosen is None: - _echo("\nNo playable games left (all have poor ProtonDB ratings)!") + if confidence_skipped > 0 and linux_skipped == 0: + _echo( + "\nNo assignable games found " + "(HLTB confidence thresholds: comp_100 polls>=3, " + "count_comp>=15, sum>=18)." + ) + else: + _echo("\nNo playable games left (all have poor ProtonDB ratings)!") state.current_app_id = None state.current_game_name = "" state.save() @@ -184,6 +221,7 @@ def pick_next_game(games: list[GameInfo], state: State, config: Config) -> None: f" Progress: {chosen.unlocked_achievements}/{chosen.total_achievements}" f" ({chosen.completion_pct:.1f}%)" ) + _report_poll_confidence(chosen, games, state) # Uninstall all other games first, then auto-install the assigned one. if config.uninstall_other_games: @@ -201,6 +239,248 @@ def pick_next_game(games: list[GameInfo], state: State, config: Config) -> None: ) +def _confidence_fail_reasons(game: GameInfo) -> list[str]: + """Return threshold-failure reasons for a game's HLTB confidence data.""" + reasons: list[str] = [] + if game.comp_100_count < _MIN_COMP_100_POLLS: + reasons.append(f"comp_100 polls {game.comp_100_count} < {_MIN_COMP_100_POLLS}") + if game.count_comp < _MIN_COUNT_COMP: + reasons.append(f"count_comp {game.count_comp} < {_MIN_COUNT_COMP}") + + total = game.comp_100_count + game.count_comp + if total < _MIN_CONFIDENCE_SUM: + reasons.append(f"comp_100+count_comp {total} < {_MIN_CONFIDENCE_SUM}") + + return reasons + + +def _refresh_candidate_confidence(game: GameInfo) -> None: + """Refresh confidence metrics for one candidate when cache looks stale. + + Only refreshes when both metrics are missing (0), which typically means + the game was cached before confidence fields were added. + """ + if game.comp_100_count > 0 or game.count_comp > 0: + return + + _refresh_candidate_confidence_batch([game]) + + +def _force_refresh_candidate_confidence(game: GameInfo) -> None: + """Force-refresh one candidate's confidence metrics from HLTB.""" + _refresh_candidate_confidence_batch([game], force=True) + + +def _refresh_candidate_confidence_batch( + candidates: list[GameInfo], + *, + force: bool = False, +) -> None: + """Refresh missing confidence metrics for candidates in one HLTB batch. + + This prevents O(N) one-game API loops when many snapshot entries predate + confidence fields and therefore have ``comp_100_count==0`` and + ``count_comp==0``. + """ + missing = [ + game + for game in candidates + if force or (game.comp_100_count == 0 and game.count_comp == 0) + ] + if not missing: + return + + refresh_slice = missing + if len(refresh_slice) == 1: + game = refresh_slice[0] + _echo(f" Refreshing HLTB confidence for {game.name} (AppID={game.app_id})...") + else: + _echo(f" Refreshing HLTB confidence for {len(refresh_slice)} candidate(s)...") + + cache = load_hltb_cache() + polls = load_hltb_polls_cache() + count_comp = load_hltb_count_comp_cache() + app_ids = [game.app_id for game in refresh_slice] + names = [(game.app_id, game.name) for game in refresh_slice] + prior_hours = {aid: cache.get(aid, -1) for aid in app_ids} + + for aid in app_ids: + cache.pop(aid, None) + polls.pop(aid, None) + count_comp.pop(aid, None) + save_hltb_cache(cache, polls, count_comp) + + fetch_hltb_confidence_cached(names) + + refreshed_hours = load_hltb_cache() + refreshed_polls = load_hltb_polls_cache() + refreshed_count_comp = load_hltb_count_comp_cache() + for aid, old_hours in prior_hours.items(): + if old_hours > 0 and refreshed_hours.get(aid, -1) <= 0: + refreshed_hours[aid] = old_hours + save_hltb_cache(refreshed_hours, refreshed_polls, refreshed_count_comp) + + for game in refresh_slice: + game.comp_100_count = refreshed_polls.get(game.app_id, 0) + game.count_comp = refreshed_count_comp.get(game.app_id, 0) + + +def _filter_hltb_confident_candidates( + candidates: list[GameInfo], +) -> list[GameInfo]: + """Keep only candidates that satisfy HLTB confidence thresholds.""" + _refresh_candidate_confidence_batch(candidates) + + kept: list[GameInfo] = [] + for game in candidates: + reasons = _confidence_fail_reasons(game) + if reasons: + _echo( + f" Skipping {game.name} (AppID={game.app_id}): " + f"HLTB confidence too low ({'; '.join(reasons)})" + ) + continue + kept.append(game) + return kept + + +def _candidate_passes_hltb_confidence(game: GameInfo) -> bool: + """Return True if candidate passes confidence with cache-first behavior. + + Only refreshes when confidence fields are missing (both zero), which keeps + normal runs cache-friendly and avoids repeated refetches for known + low-confidence entries. + """ + reasons = _confidence_fail_reasons(game) + if not reasons: + return True + + # Re-check once when confidence fields are missing in cache. + _refresh_candidate_confidence(game) + reasons = _confidence_fail_reasons(game) + if reasons: + _echo( + f" Skipping {game.name} (AppID={game.app_id}): " + f"HLTB confidence too low ({'; '.join(reasons)})" + ) + return False + return True + + +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 _backfill_polls_for_finished( + state: State, + games: list[GameInfo], +) -> dict[int, int]: + """Lazily fetch poll counts for already-finished games missing them. + + Reads the polls cache, identifies finished games whose poll count is + still ``0`` (typically because the cache predates the polls schema), + and triggers a one-shot HLTB search to backfill them. Returns the + refreshed polls cache. + """ + polls_cache = load_hltb_polls_cache() + name_by_id = {g.app_id: g.name for g in games} + missing = [ + (aid, name_by_id[aid]) + for aid in state.finished_app_ids + if aid in name_by_id and polls_cache.get(aid, 0) == 0 + ] + if not missing: + return polls_cache + + logger.info( + "Backfilling HLTB poll counts for %d already-finished games...", + len(missing), + ) + # Force a fresh search by removing the hours entries we want to refetch. + # (fetch_hltb_times_cached skips entries already in the hours cache.) + cache = load_hltb_cache() + preserved_hours = {aid: cache[aid] for aid, _ in missing if aid in cache} + for aid, _name in missing: + cache.pop(aid, None) + save_hltb_cache(cache, polls_cache) + + fetch_hltb_confidence_cached(missing) + + # Restore any previously-known hours that the refetch may have replaced + # with a worse match (we trust prior leisure+dlc estimates). + refreshed_hours = load_hltb_cache() + refreshed_polls = load_hltb_polls_cache() + for aid, prior_hours in preserved_hours.items(): + if prior_hours > 0 and refreshed_hours.get(aid, -1) <= 0: + refreshed_hours[aid] = prior_hours + save_hltb_cache(refreshed_hours, refreshed_polls) + return refreshed_polls + + +def _report_poll_confidence( + chosen: GameInfo, + games: list[GameInfo], + state: State, +) -> None: + """Print HLTB poll-count confidence info for the just-assigned game. + + Shows the chosen game's ``comp_100_count`` (number of polled + completionist times on HowLongToBeat) and the historical minimum + among the user's previously-finished games. Marks a new historical + low so the user can be skeptical of unreliable estimates. + """ + polls_cache = _backfill_polls_for_finished(state, games) + chosen_polls = polls_cache.get(chosen.app_id, chosen.comp_100_count) + chosen.comp_100_count = chosen_polls + + finished_polls = [ + (polls_cache[aid], aid) + for aid in state.finished_app_ids + if polls_cache.get(aid, 0) > 0 + ] + if not finished_polls: + _echo(f" HLTB confidence: {chosen_polls} polled completionist times") + return + + min_polls, min_aid = min(finished_polls) + name_by_id = {g.app_id: g.name for g in games} + min_name = name_by_id.get(min_aid, f"AppID={min_aid}") + + warning = "" + if 0 < chosen_polls < min_polls: + warning = " ⚠ NEW LOW — estimate may be unreliable" + elif chosen_polls == 0: + warning = " ⚠ no polls recorded — estimate may be unreliable" + + _echo(f" HLTB confidence: {chosen_polls} polled completionist times{warning}") + _echo(f" Historical min among finished: {min_polls} ({min_name})") + + # ────────────────────────────────────────────────────────────── # Checking & tampering detection # ────────────────────────────────────────────────────────────── diff --git a/steam_backlog_enforcer/steam_api.py b/steam_backlog_enforcer/steam_api.py index d8f1f9c..3ecbf76 100644 --- a/steam_backlog_enforcer/steam_api.py +++ b/steam_backlog_enforcer/steam_api.py @@ -41,6 +41,8 @@ class GameInfo: playtime_minutes: int achievements: list[AchievementInfo] = field(default_factory=list) completionist_hours: float = -1 + comp_100_count: int = 0 + count_comp: int = 0 @property def completion_pct(self) -> float: @@ -66,6 +68,8 @@ class GameInfo: "unlocked_achievements": self.unlocked_achievements, "playtime_minutes": self.playtime_minutes, "completionist_hours": self.completionist_hours, + "comp_100_count": self.comp_100_count, + "count_comp": self.count_comp, "achievements": [ { "api_name": a.api_name, @@ -96,6 +100,8 @@ class GameInfo: unlocked_achievements=data["unlocked_achievements"], playtime_minutes=data.get("playtime_minutes", 0), completionist_hours=data.get("completionist_hours", -1), + comp_100_count=data.get("comp_100_count", 0), + count_comp=data.get("count_comp", 0), achievements=achievements, ) diff --git a/steam_backlog_enforcer/tests/test_cmd_done.py b/steam_backlog_enforcer/tests/test_cmd_done.py index 99c6d5c..8c7435a 100644 --- a/steam_backlog_enforcer/tests/test_cmd_done.py +++ b/steam_backlog_enforcer/tests/test_cmd_done.py @@ -2,31 +2,32 @@ from __future__ import annotations -from typing import Any from unittest.mock import patch -from python_pkg.steam_backlog_enforcer._cmd_done import _try_reassign_shorter_game +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.steam_api import GameInfo CMD_DONE_PKG = "python_pkg.steam_backlog_enforcer._cmd_done" -def _snap( - app_id: int = 1, - name: str = "G", - total: int = 10, - unlocked: int = 0, - hours: float = -1, -) -> dict[str, Any]: - return { - "app_id": app_id, - "name": name, - "total_achievements": total, - "unlocked_achievements": unlocked, +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": hours, + "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: @@ -37,7 +38,12 @@ class TestTryReassignShorterGame: assert not _try_reassign_shorter_game({}, 1, 10.0, State(), Config()) def test_no_shorter_candidate(self) -> None: - snap = [_snap(1, "G", 10, 5, 10.0), _snap(2, "H", 10, 5, -1)] + 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"), @@ -53,8 +59,15 @@ class TestTryReassignShorterGame: def test_reassigns(self) -> None: snap = [ - _snap(1, "Long", 10, 5, 100.0), - _snap(2, "Short", 10, 5, 5.0), + _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( @@ -69,8 +82,8 @@ class TestTryReassignShorterGame: patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), patch(f"{CMD_DONE_PKG}._echo"), patch( - f"{CMD_DONE_PKG}._pick_playable_candidate", - return_value=short_game, + f"{CMD_DONE_PKG}._pick_next_shortest_candidate", + return_value=(short_game, 0, 0), ), patch(f"{CMD_DONE_PKG}.pick_next_game"), patch( @@ -91,8 +104,15 @@ class TestTryReassignShorterGame: def test_reassigns_no_hide_when_no_owned_ids(self) -> None: snap = [ - _snap(1, "Long", 10, 5, 100.0), - _snap(2, "Short", 10, 5, 5.0), + _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( @@ -107,8 +127,8 @@ class TestTryReassignShorterGame: 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_playable_candidate", - return_value=short_game, + 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]), @@ -128,8 +148,15 @@ class TestTryReassignShorterGame: def test_reassigns_skip_hide_when_no_app_assigned(self) -> None: snap = [ - _snap(1, "Long", 10, 5, 100.0), - _snap(2, "Short", 10, 5, 5.0), + _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( @@ -144,8 +171,8 @@ class TestTryReassignShorterGame: patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), patch(f"{CMD_DONE_PKG}._echo"), patch( - f"{CMD_DONE_PKG}._pick_playable_candidate", - return_value=short_game, + 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, @@ -164,12 +191,22 @@ class TestTryReassignShorterGame: def test_playable_none(self) -> None: snap = [ - _snap(1, "Long", 10, 5, 100.0), - _snap(2, "Short", 10, 5, 5.0), + _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_playable_candidate", return_value=None), + 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( @@ -184,8 +221,18 @@ class TestTryReassignShorterGame: def test_playable_longer(self) -> None: """Playable candidate is longer than current — no reassign.""" snap = [ - _snap(1, "Short", 10, 5, 10.0), - _snap(2, "Long", 10, 5, 200.0), + _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, @@ -197,7 +244,10 @@ class TestTryReassignShorterGame: ) with ( patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap), - patch(f"{CMD_DONE_PKG}._pick_playable_candidate", return_value=long_game), + 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( @@ -212,8 +262,13 @@ class TestTryReassignShorterGame: def test_refreshes_stale_shorter_snapshot_entry(self) -> None: """Uncached shorter snapshot candidates are refreshed before reassigning.""" snap = [ - _snap(1, "Current", 10, 5, 20.1), - _snap(2, "Lacuna", 10, 0, 0.9), + _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( @@ -231,9 +286,9 @@ class TestTryReassignShorterGame: return_value={2: 18.8}, ) as mock_fetch_hltb, patch( - f"{CMD_DONE_PKG}._pick_playable_candidate", - return_value=refreshed_short, - ) as mock_pick_playable, + 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=[]), @@ -249,4 +304,328 @@ class TestTryReassignShorterGame: assert result mock_fetch_hltb.assert_called_once_with([(2, "Lacuna")]) - mock_pick_playable.assert_called_once() + 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() + + 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", + total_achievements=10, + unlocked_achievements=5, + playtime_minutes=60, + completionist_hours=9.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=(known_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"), + ): + 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, + ) + + 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}._echo"), + ): + 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, + ) + + 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=(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"), + ): + 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 diff --git a/steam_backlog_enforcer/tests/test_enforce_loop.py b/steam_backlog_enforcer/tests/test_enforce_loop.py index a3994f2..5f06504 100644 --- a/steam_backlog_enforcer/tests/test_enforce_loop.py +++ b/steam_backlog_enforcer/tests/test_enforce_loop.py @@ -21,9 +21,12 @@ PKG = "python_pkg.steam_backlog_enforcer._enforce_loop" class TestGetAllOwnedAppIds: """Tests for get_all_owned_app_ids.""" - def test_from_snapshot(self) -> None: + def test_snapshot_used_when_api_fails(self) -> None: snap = [{"app_id": 1}, {"app_id": 2}] - with patch(f"{PKG}.load_snapshot", return_value=snap): + with ( + patch(f"{PKG}.load_snapshot", return_value=snap), + patch(f"{PKG}.SteamAPIClient", side_effect=OSError("boom")), + ): assert get_all_owned_app_ids(Config()) == [1, 2] def test_no_snapshot_falls_back_to_api(self) -> None: @@ -60,6 +63,21 @@ class TestGetAllOwnedAppIds: ): assert get_all_owned_app_ids(Config(steam_api_key="k", steam_id="i")) == [5] + def test_merges_snapshot_with_api_results(self) -> None: + mock_client = MagicMock() + mock_client.get_owned_games.return_value = [{"appid": 10}, {"appid": 20}] + with ( + patch( + f"{PKG}.load_snapshot", return_value=[{"app_id": 20}, {"app_id": 30}] + ), + patch(f"{PKG}.SteamAPIClient", return_value=mock_client), + ): + assert get_all_owned_app_ids(Config(steam_api_key="k", steam_id="i")) == [ + 10, + 20, + 30, + ] + class TestGuardInstalledGames: """Tests for _guard_installed_games.""" diff --git a/steam_backlog_enforcer/tests/test_game_install.py b/steam_backlog_enforcer/tests/test_game_install.py index 6b558cf..58d7e7f 100644 --- a/steam_backlog_enforcer/tests/test_game_install.py +++ b/steam_backlog_enforcer/tests/test_game_install.py @@ -63,6 +63,20 @@ class TestAssertNotRealSteam: ): _assert_not_real_steam(fake_manifest) + def test_noop_outside_pytest(self, tmp_path: Path) -> None: + """In production (no PYTEST_CURRENT_TEST) the guard is a no-op.""" + real = tmp_path / "real_steam" + real.mkdir() + fake_manifest = real / "appmanifest_440.acf" + fake_manifest.touch() + env = {k: v for k, v in os.environ.items() if k != "PYTEST_CURRENT_TEST"} + with ( + patch.dict(os.environ, env, clear=True), + patch(f"{PKG}._REAL_STEAMAPPS", real), + patch(f"{PKG}.STEAMAPPS_PATH", real), + ): + _assert_not_real_steam(fake_manifest) + class TestEcho: """Tests for _echo.""" diff --git a/steam_backlog_enforcer/tests/test_hltb_detail.py b/steam_backlog_enforcer/tests/test_hltb_detail.py index 7d28c4e..8c6d274 100644 --- a/steam_backlog_enforcer/tests/test_hltb_detail.py +++ b/steam_backlog_enforcer/tests/test_hltb_detail.py @@ -203,7 +203,7 @@ class TestFetchLeisureTimes: new_callable=AsyncMock, return_value=game_data, ): - asyncio.run(_fetch_leisure_times(results, cache, None)) + asyncio.run(_fetch_leisure_times(results, cache, {}, None)) assert cache[440] == round(21243 / 3600, 2) assert results[0].completionist_hours == round(21243 / 3600, 2) @@ -218,12 +218,12 @@ class TestFetchLeisureTimes: ), ] cache: dict[int, float] = {} - asyncio.run(_fetch_leisure_times(results, cache, None)) + asyncio.run(_fetch_leisure_times(results, cache, {}, None)) assert not cache def test_empty_results(self) -> None: cache: dict[int, float] = {} - asyncio.run(_fetch_leisure_times([], cache, None)) + asyncio.run(_fetch_leisure_times([], cache, {}, None)) assert not cache def test_detail_returns_none(self) -> None: @@ -242,7 +242,7 @@ class TestFetchLeisureTimes: new_callable=AsyncMock, return_value=None, ): - asyncio.run(_fetch_leisure_times(results, cache, None)) + asyncio.run(_fetch_leisure_times(results, cache, {}, None)) assert not cache assert results[0].completionist_hours == 50.0 @@ -263,7 +263,7 @@ class TestFetchLeisureTimes: new_callable=AsyncMock, return_value=game_data, ): - asyncio.run(_fetch_leisure_times(results, cache, None)) + asyncio.run(_fetch_leisure_times(results, cache, {}, None)) assert not cache assert results[0].completionist_hours == 50.0 @@ -288,7 +288,7 @@ class TestFetchLeisureTimes: new_callable=AsyncMock, return_value=game_data, ): - asyncio.run(_fetch_leisure_times(results, cache, cb)) + asyncio.run(_fetch_leisure_times(results, cache, {}, cb)) cb.assert_called_once() def test_save_interval(self) -> None: @@ -318,7 +318,7 @@ class TestFetchLeisureTimes: "python_pkg.steam_backlog_enforcer._hltb_detail.save_hltb_cache" ) as mock_save, ): - asyncio.run(_fetch_leisure_times(results, cache, None)) + asyncio.run(_fetch_leisure_times(results, cache, {}, None)) mock_save.assert_called_once() def test_dlc_detail_overrides_relationship_fallback(self) -> None: @@ -345,7 +345,7 @@ class TestFetchLeisureTimes: new_callable=AsyncMock, side_effect=[base_data, dlc_data], ): - asyncio.run(_fetch_leisure_times(results, cache, None)) + asyncio.run(_fetch_leisure_times(results, cache, {}, None)) expected = round((21243 + 12298) / 3600, 2) assert cache[1289310] == expected @@ -371,7 +371,7 @@ class TestFetchLeisureTimes: new_callable=AsyncMock, side_effect=[base_data, None], ): - asyncio.run(_fetch_leisure_times(results, cache, None)) + asyncio.run(_fetch_leisure_times(results, cache, {}, None)) expected = round((21243 + 4075) / 3600, 2) assert cache[1289310] == expected diff --git a/steam_backlog_enforcer/tests/test_hltb_part2.py b/steam_backlog_enforcer/tests/test_hltb_part2.py index 4f0eedc..a600baf 100644 --- a/steam_backlog_enforcer/tests/test_hltb_part2.py +++ b/steam_backlog_enforcer/tests/test_hltb_part2.py @@ -2,11 +2,18 @@ from __future__ import annotations +import asyncio from unittest.mock import MagicMock, patch +from typing_extensions import Self + from python_pkg.steam_backlog_enforcer.hltb import ( HLTB_BASE_URL, HLTBResult, + _AuthInfo, + _fetch_batch_confidence_only, + fetch_hltb_confidence, + fetch_hltb_confidence_cached, fetch_hltb_times_cached, get_hltb_submit_url, ) @@ -35,10 +42,16 @@ class TestFetchHltbTimesCached: def add_to_cache( _games: object, cache: dict[int, float] | None = None, + polls: dict[int, int] | None = None, progress_cb: object = None, + count_comp: dict[int, int] | None = None, ) -> list[object]: if cache is not None: cache[730] = 20.0 + if polls is not None: + polls[730] = 0 + if count_comp is not None: + count_comp[730] = 0 return [] mock_fetch.side_effect = add_to_cache @@ -87,11 +100,19 @@ class TestFetchHltbTimesCached: def add_found( _games: object, cache: dict[int, float] | None = None, + polls: dict[int, int] | None = None, progress_cb: object = None, + count_comp: dict[int, int] | None = None, ) -> list[object]: if cache is not None: cache[440] = 50.0 cache[730] = -1 + if polls is not None: + polls[440] = 5 + polls[730] = 0 + if count_comp is not None: + count_comp[440] = 15 + count_comp[730] = 0 return [] mock_fetch.side_effect = add_found @@ -133,3 +154,82 @@ class TestGetHltbSubmitUrl: with patch(f"{PKG}.fetch_hltb_times", return_value=[mock_result]): url = get_hltb_submit_url("TF2") assert url is None + + +class _DummySession: + """Minimal async context manager used to mock aiohttp ClientSession.""" + + async def __aenter__(self) -> Self: + """Enter async context.""" + return self + + async def __aexit__(self, *_args: object) -> bool: + """Exit async context.""" + return False + + +class TestConfidenceHelpers: + """Coverage tests for confidence-fetch helpers.""" + + def test_fetch_batch_confidence_only_returns_empty_without_auth(self) -> None: + with ( + patch(f"{PKG}.aiohttp.ClientSession", return_value=_DummySession()), + patch(f"{PKG}.aiohttp.TCPConnector"), + patch(f"{PKG}._get_hltb_search_url", return_value="https://example"), + patch(f"{PKG}._get_auth_info", return_value=None), + ): + result = asyncio.run( + _fetch_batch_confidence_only([(1, "Game")], {}, {}, None), + ) + assert result == [] + + def test_fetch_batch_confidence_only_handles_empty_hp_and_default_counts( + self, + ) -> None: + auth_token = str(1) + with ( + patch(f"{PKG}.aiohttp.ClientSession", return_value=_DummySession()), + patch(f"{PKG}.aiohttp.TCPConnector"), + patch(f"{PKG}._get_hltb_search_url", return_value="https://example"), + patch( + f"{PKG}._get_auth_info", + return_value=_AuthInfo(token=auth_token, hp_key="", hp_val=""), + ), + patch(f"{PKG}._search_one", side_effect=[None]) as mock_search, + ): + result = asyncio.run( + _fetch_batch_confidence_only( + games=[(1, "Game")], + cache={}, + polls={}, + progress_cb=None, + count_comp=None, + ), + ) + assert result == [] + mock_search.assert_called_once() + + def test_fetch_hltb_confidence_initializes_optional_dicts(self) -> None: + with patch(f"{PKG}.asyncio.run", return_value=[]) as mock_run: + result = fetch_hltb_confidence([(1, "Game")]) + assert result == [] + mock_run.assert_called_once() + + def test_fetch_hltb_confidence_empty_games_returns_empty(self) -> None: + with patch(f"{PKG}.asyncio.run") as mock_run: + result = fetch_hltb_confidence([]) + assert result == [] + mock_run.assert_not_called() + + def test_fetch_hltb_confidence_cached_all_cached_skips_fetch(self) -> None: + with ( + patch(f"{PKG}.load_hltb_cache", return_value={1: 12.0}), + patch(f"{PKG}.load_hltb_polls_cache", return_value={1: 30}), + patch(f"{PKG}.load_hltb_count_comp_cache", return_value={1: 200}), + patch(f"{PKG}.fetch_hltb_confidence") as mock_fetch, + patch(f"{PKG}.save_hltb_cache") as mock_save, + ): + result = fetch_hltb_confidence_cached([(1, "Game")]) + assert result == {1: 12.0} + mock_fetch.assert_not_called() + mock_save.assert_not_called() diff --git a/steam_backlog_enforcer/tests/test_hltb_search.py b/steam_backlog_enforcer/tests/test_hltb_search.py index 1d24037..ba4c2a6 100644 --- a/steam_backlog_enforcer/tests/test_hltb_search.py +++ b/steam_backlog_enforcer/tests/test_hltb_search.py @@ -19,6 +19,7 @@ from python_pkg.steam_backlog_enforcer.hltb import ( HLTBResult, _AuthInfo, _fetch_batch, + _pick_best_hltb_entry, _search_one, _SearchCtx, ) @@ -109,6 +110,37 @@ class TestSearchOne: result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2")) assert result is None + def test_fallback_name_without_year_suffix(self) -> None: + session = MagicMock() + session.post.side_effect = [ + _FakeResponse(200, {"data": []}), + _FakeResponse( + 200, + { + "data": [ + { + "game_name": "Final Fantasy VII", + "game_alias": "", + "game_type": "game", + "comp_100": 141120, + "game_id": 435, + "comp_100_count": 746, + "count_comp": 10450, + } + ] + }, + ), + ] + ctx = _make_ctx(session) + result = asyncio.run( + _search_one(asyncio.Semaphore(1), ctx, 39140, "Final Fantasy VII (2013)") + ) + assert result is not None + assert result.app_id == 39140 + assert result.comp_100_count == 746 + assert result.count_comp == 10450 + assert session.post.call_count == 2 + def test_with_progress_cb(self) -> None: resp = _FakeResponse(200, {"data": []}) cb = MagicMock() @@ -235,9 +267,69 @@ class TestFetchBatchHltb: return_value=None, ), ): - results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None)) + results = asyncio.run(_fetch_batch([(440, "TF2")], {}, {}, None)) assert results == [] + +class TestPickBestEntry: + """Tests for exact-vs-extended entry choice logic.""" + + def test_prefers_exact_over_low_confidence_modded_extended(self) -> None: + exact = ( + { + "game_name": "Celeste", + "game_alias": "", + "game_type": "game", + "comp_100": 141105, + "comp_100_count": 899, + "count_comp": 14055, + }, + 1.0, + ) + mod_extended = ( + { + "game_name": "Celeste - Strawberry Jam", + "game_alias": "", + "game_type": "mod", + "comp_100": 952080, + "comp_100_count": 1, + "count_comp": 6, + }, + 0.9, + ) + + best = _pick_best_hltb_entry("Celeste", [exact, mod_extended]) + assert best is not None + assert best[0]["game_name"] == "Celeste" + + def test_prefers_extended_when_confident_and_longer(self) -> None: + exact_demo = ( + { + "game_name": "FAITH", + "game_alias": "", + "game_type": "game", + "comp_100": 1800, + "comp_100_count": 1, + "count_comp": 1, + }, + 1.0, + ) + full_extended = ( + { + "game_name": "FAITH: The Unholy Trinity", + "game_alias": "", + "game_type": "game", + "comp_100": 25200, + "comp_100_count": 50, + "count_comp": 500, + }, + 0.9, + ) + + best = _pick_best_hltb_entry("FAITH", [exact_demo, full_extended]) + assert best is not None + assert best[0]["game_name"] == "FAITH: The Unholy Trinity" + def test_with_auth(self) -> None: auth = _AuthInfo("token123", "ign_x", "ff") with ( @@ -266,7 +358,7 @@ class TestFetchBatchHltb: new_callable=AsyncMock, ), ): - results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None)) + results = asyncio.run(_fetch_batch([(440, "TF2")], {}, {}, None)) assert len(results) == 1 def test_with_auth_no_hp(self) -> None: @@ -291,7 +383,7 @@ class TestFetchBatchHltb: new_callable=AsyncMock, ), ): - results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None)) + results = asyncio.run(_fetch_batch([(440, "TF2")], {}, {}, None)) assert results == [] def test_filters_none_results(self) -> None: @@ -316,7 +408,7 @@ class TestFetchBatchHltb: new_callable=AsyncMock, ), ): - results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None)) + results = asyncio.run(_fetch_batch([(440, "TF2")], {}, {}, None)) assert results == [] diff --git a/steam_backlog_enforcer/tests/test_main_part2.py b/steam_backlog_enforcer/tests/test_main_part2.py index 0b5583f..a91c03c 100644 --- a/steam_backlog_enforcer/tests/test_main_part2.py +++ b/steam_backlog_enforcer/tests/test_main_part2.py @@ -206,6 +206,8 @@ class TestEnforceOnDone: ), patch(f"{CMD_DONE_PKG}.uninstall_other_games", return_value=2), patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True), + patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]), + patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=1), ): _enforce_on_done(config, state) @@ -220,6 +222,8 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}.enforce_allowed_game", return_value=[]), patch(f"{CMD_DONE_PKG}.uninstall_other_games", return_value=0), patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True), + patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[]), + patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=0), ): _enforce_on_done(config, state) @@ -234,6 +238,8 @@ class TestEnforceOnDone: patch(f"{CMD_DONE_PKG}._echo"), patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=False), patch(f"{CMD_DONE_PKG}.install_game") as mock_install, + patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]), + patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=0), ): _enforce_on_done(config, state) mock_install.assert_called_once_with(1, "G", "s1", use_steam_protocol=True) diff --git a/steam_backlog_enforcer/tests/test_polls_tracking.py b/steam_backlog_enforcer/tests/test_polls_tracking.py new file mode 100644 index 0000000..e934c4d --- /dev/null +++ b/steam_backlog_enforcer/tests/test_polls_tracking.py @@ -0,0 +1,729 @@ +"""Tests for HLTB poll-count tracking, schema migration, and confidence display.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING +from unittest.mock import patch + +from python_pkg.steam_backlog_enforcer import _cmd_done, scanning +from python_pkg.steam_backlog_enforcer._hltb_types import ( + HLTBResult, + load_hltb_cache, + load_hltb_count_comp_cache, + load_hltb_polls_cache, + save_hltb_cache, +) +from python_pkg.steam_backlog_enforcer.config import State +from python_pkg.steam_backlog_enforcer.steam_api import GameInfo + +if TYPE_CHECKING: + from pathlib import Path + +_TYPES = "python_pkg.steam_backlog_enforcer._hltb_types" +_CMD = "python_pkg.steam_backlog_enforcer._cmd_done" +_SCAN = "python_pkg.steam_backlog_enforcer.scanning" + + +class TestCacheSchema: + """Tests for the new cache schema and back-compat migration.""" + + def test_legacy_float_migrates(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text(json.dumps({"440": 10.5}), encoding="utf-8") + with patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file): + assert load_hltb_cache() == {440: 10.5} + assert load_hltb_polls_cache() == {440: 0} + assert load_hltb_count_comp_cache() == {440: 0} + + def test_new_dict_schema(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"440": {"hours": 10.5, "polls": 7, "count_comp": 20}}), + encoding="utf-8", + ) + with patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file): + assert load_hltb_cache() == {440: 10.5} + assert load_hltb_polls_cache() == {440: 7} + assert load_hltb_count_comp_cache() == {440: 20} + + def test_invalid_app_id_skipped(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"notanint": 1.0, "440": 5.0}), encoding="utf-8" + ) + with patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file): + assert load_hltb_cache() == {440: 5.0} + + def test_unparseable_value_skipped(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text(json.dumps({"440": "notafloat"}), encoding="utf-8") + with patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file): + assert load_hltb_cache() == {} + + def test_save_with_polls_roundtrip(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + ): + save_hltb_cache({440: 10.5}, {440: 7}, {440: 20}) + data = json.loads(cache_file.read_text(encoding="utf-8")) + assert data == {"440": {"hours": 10.5, "polls": 7, "count_comp": 20}} + + def test_save_without_polls_defaults_zero(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + ): + save_hltb_cache({440: 10.5}) + data = json.loads(cache_file.read_text(encoding="utf-8")) + assert data == {"440": {"hours": 10.5, "polls": 0, "count_comp": 0}} + + +class TestHltbResultPolls: + def test_default_zero(self) -> None: + r = HLTBResult(app_id=1, game_name="x", completionist_hours=1.0, similarity=1) + assert r.comp_100_count == 0 + assert r.count_comp == 0 + + def test_explicit(self) -> None: + r = HLTBResult( + app_id=1, + game_name="x", + completionist_hours=1.0, + similarity=1, + comp_100_count=42, + count_comp=100, + ) + assert r.comp_100_count == 42 + assert r.count_comp == 100 + + +class TestGameInfoPolls: + def test_snapshot_roundtrip(self) -> None: + g = GameInfo( + app_id=1, + name="X", + total_achievements=10, + unlocked_achievements=5, + playtime_minutes=30, + comp_100_count=8, + count_comp=20, + ) + snap = g.to_snapshot() + assert snap["comp_100_count"] == 8 + assert snap["count_comp"] == 20 + restored = GameInfo.from_snapshot(snap) + assert restored.comp_100_count == 8 + assert restored.count_comp == 20 + + def test_snapshot_missing_field_defaults(self) -> None: + snap = { + "app_id": 1, + "name": "X", + "total_achievements": 0, + "unlocked_achievements": 0, + } + restored = GameInfo.from_snapshot(snap) + assert restored.comp_100_count == 0 + assert restored.count_comp == 0 + + +def _state(finished: list[int], current: int | None = None) -> State: + s = State() + s.finished_app_ids = list(finished) + s.current_app_id = current + s.current_game_name = "" + return s + + +class TestBackfillPollsForFinished: + def test_no_missing_returns_existing(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"1": {"hours": 1.0, "polls": 5}}), encoding="utf-8" + ) + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_CMD}.load_snapshot", return_value=[{"app_id": 1, "name": "G"}]), + ): + result = _cmd_done._backfill_polls_for_finished(_state([1])) + assert result == {1: 5} + + def test_no_snapshot_no_missing(self) -> None: + with ( + patch(f"{_CMD}.load_hltb_polls_cache", return_value={}), + patch(f"{_CMD}.load_snapshot", return_value=None), + ): + assert _cmd_done._backfill_polls_for_finished(_state([1])) == {} + + def test_missing_triggers_fetch(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"1": {"hours": 2.0, "polls": 0}}), encoding="utf-8" + ) + + def fake_fetch(games: list[tuple[int, str]]) -> dict[int, float]: + data = json.loads(cache_file.read_text(encoding="utf-8")) + for aid, _name in games: + data[str(aid)] = {"hours": 2.0, "polls": 9} + cache_file.write_text(json.dumps(data), encoding="utf-8") + return {aid: 2.0 for aid, _ in games} + + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + patch(f"{_CMD}.load_snapshot", return_value=[{"app_id": 1, "name": "G"}]), + patch(f"{_CMD}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch(f"{_CMD}._echo"), + ): + result = _cmd_done._backfill_polls_for_finished(_state([1])) + assert result == {1: 9} + + def test_extra_app_id_with_zero_polls_added(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"7": {"hours": 1.0, "polls": 0}}), encoding="utf-8" + ) + + def fake_fetch(games: list[tuple[int, str]]) -> dict[int, float]: + data = json.loads(cache_file.read_text(encoding="utf-8")) + for aid, _name in games: + data[str(aid)] = {"hours": 1.0, "polls": 4} + cache_file.write_text(json.dumps(data), encoding="utf-8") + return {aid: 1.0 for aid, _ in games} + + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + patch(f"{_CMD}.load_snapshot", return_value=[{"app_id": 7, "name": "G"}]), + patch(f"{_CMD}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch(f"{_CMD}._echo"), + ): + result = _cmd_done._backfill_polls_for_finished( + _state([], current=7), extra_app_id=7 + ) + assert result == {7: 4} + + def test_preserves_prior_hours_on_miss(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"3": {"hours": 4.0, "polls": 0}}), encoding="utf-8" + ) + + def fake_fetch(games: list[tuple[int, str]]) -> dict[int, float]: + # Simulate a refetch returning a miss (hours -1, polls 0). + data = json.loads(cache_file.read_text(encoding="utf-8")) + for aid, _name in games: + data[str(aid)] = {"hours": -1, "polls": 0} + cache_file.write_text(json.dumps(data), encoding="utf-8") + return {aid: -1 for aid, _ in games} + + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + patch(f"{_CMD}.load_snapshot", return_value=[{"app_id": 3, "name": "G"}]), + patch(f"{_CMD}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch(f"{_CMD}._echo"), + ): + _cmd_done._backfill_polls_for_finished(_state([3])) + # Prior hours should be preserved on miss. + final = json.loads(cache_file.read_text(encoding="utf-8")) + assert final["3"]["hours"] == 4.0 + + +class TestReportAssignedConfidence: + def test_new_low_warning(self) -> None: + echoed: list[str] = [] + with ( + patch( + f"{_CMD}._backfill_polls_for_finished", + return_value={1: 1, 2: 5, 3: 10}, + ), + patch( + f"{_CMD}.load_snapshot", + return_value=[ + {"app_id": 1, "name": "Chosen"}, + {"app_id": 2, "name": "OldShortest"}, + {"app_id": 3, "name": "Other"}, + ], + ), + patch(f"{_CMD}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + _cmd_done._report_assigned_confidence(1, _state([2, 3], current=1)) + assert any("NEW LOW" in s for s in echoed) + assert any("Historical min" in s and "OldShortest" in s for s in echoed) + + def test_zero_polls_warning_with_history(self) -> None: + echoed: list[str] = [] + with ( + patch( + f"{_CMD}._backfill_polls_for_finished", + return_value={1: 0, 2: 5}, + ), + patch( + f"{_CMD}.load_snapshot", + return_value=[ + {"app_id": 1, "name": "Chosen"}, + {"app_id": 2, "name": "Old"}, + ], + ), + patch(f"{_CMD}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + _cmd_done._report_assigned_confidence(1, _state([2], current=1)) + assert any("no polls recorded" in s for s in echoed) + + def test_zero_polls_warning_no_history(self) -> None: + echoed: list[str] = [] + with ( + patch(f"{_CMD}._backfill_polls_for_finished", return_value={1: 0}), + patch( + f"{_CMD}.load_snapshot", + return_value=[ + {"app_id": 1, "name": "Chosen"}, + ], + ), + patch(f"{_CMD}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + _cmd_done._report_assigned_confidence(1, _state([], current=1)) + assert any("no polls recorded" in s for s in echoed) + assert not any("Historical min" in s for s in echoed) + + def test_healthy_no_warning(self) -> None: + echoed: list[str] = [] + with ( + patch( + f"{_CMD}._backfill_polls_for_finished", + return_value={1: 50, 2: 5}, + ), + patch( + f"{_CMD}.load_snapshot", + return_value=[ + {"app_id": 1, "name": "Chosen"}, + {"app_id": 2, "name": "Old"}, + ], + ), + patch(f"{_CMD}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + _cmd_done._report_assigned_confidence(1, _state([2], current=1)) + assert not any("NEW LOW" in s for s in echoed) + assert not any("no polls recorded" in s for s in echoed) + assert any("HLTB confidence: 50" in s for s in echoed) + + def test_unknown_finished_uses_appid_label(self) -> None: + echoed: list[str] = [] + with ( + patch( + f"{_CMD}._backfill_polls_for_finished", + return_value={1: 50, 99: 5}, + ), + patch( + f"{_CMD}.load_snapshot", + return_value=[ + {"app_id": 1, "name": "Chosen"}, + ], + ), + patch(f"{_CMD}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + _cmd_done._report_assigned_confidence(1, _state([99], current=1)) + assert any("AppID=99" in s for s in echoed) + + def test_chosen_equals_min_no_warning(self) -> None: + # Edge case: chosen_polls == min_polls (not a new low). + echoed: list[str] = [] + with ( + patch( + f"{_CMD}._backfill_polls_for_finished", + return_value={1: 5, 2: 5}, + ), + patch( + f"{_CMD}.load_snapshot", + return_value=[ + {"app_id": 1, "name": "Chosen"}, + {"app_id": 2, "name": "Old"}, + ], + ), + patch(f"{_CMD}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + _cmd_done._report_assigned_confidence(1, _state([2], current=1)) + assert not any("NEW LOW" in s for s in echoed) + assert not any("no polls recorded" in s for s in echoed) + + +class TestScanningPollsIntegration: + def test_do_scan_kept_assignment_reports(self) -> None: + # Targeted test for scanning's `else` branch that prints CURRENT. + echoed: list[str] = [] + games = [ + GameInfo( + app_id=1, + name="X", + total_achievements=10, + unlocked_achievements=2, + playtime_minutes=0, + completionist_hours=5.0, + comp_100_count=20, + ) + ] + state = _state([], current=1) + with ( + patch(f"{_SCAN}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + patch(f"{_SCAN}._report_poll_confidence") as mock_report, + ): + # Directly invoke just the kept-assignment branch. + current = next((g for g in games if g.app_id == state.current_app_id), None) + assert current is not None + scanning._echo(f"\n>>> CURRENT: {current.name} (AppID={current.app_id})") + scanning._report_poll_confidence(current, games, state) + assert any("CURRENT" in s for s in echoed) + mock_report.assert_called_once() + + def test_report_poll_confidence_new_low(self) -> None: + echoed: list[str] = [] + chosen = GameInfo( + app_id=1, + name="Chosen", + total_achievements=10, + unlocked_achievements=0, + playtime_minutes=0, + comp_100_count=0, + ) + games = [ + chosen, + GameInfo( + app_id=2, + name="Old", + total_achievements=10, + unlocked_achievements=10, + playtime_minutes=0, + ), + ] + with ( + patch( + f"{_SCAN}._backfill_polls_for_finished", + return_value={1: 1, 2: 5}, + ), + patch(f"{_SCAN}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + scanning._report_poll_confidence(chosen, games, _state([2], current=1)) + assert any("NEW LOW" in s for s in echoed) + assert chosen.comp_100_count == 1 + + def test_report_poll_confidence_no_history(self) -> None: + echoed: list[str] = [] + chosen = GameInfo( + app_id=1, + name="Chosen", + total_achievements=10, + unlocked_achievements=0, + playtime_minutes=0, + comp_100_count=4, + ) + with ( + patch(f"{_SCAN}._backfill_polls_for_finished", return_value={1: 4}), + patch(f"{_SCAN}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + scanning._report_poll_confidence(chosen, [chosen], _state([], current=1)) + # No "Historical min" line when no finished games have polls. + assert not any("Historical min" in s for s in echoed) + assert any("HLTB confidence: 4" in s for s in echoed) + + def test_scanning_backfill_no_missing(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"2": {"hours": 1.0, "polls": 5}}), encoding="utf-8" + ) + with patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file): + result = scanning._backfill_polls_for_finished( + _state([2]), + [ + GameInfo( + app_id=2, + name="X", + total_achievements=0, + unlocked_achievements=0, + playtime_minutes=0, + ) + ], + ) + assert result == {2: 5} + + def test_scanning_backfill_with_missing(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"2": {"hours": 3.0, "polls": 0}}), encoding="utf-8" + ) + + def fake_fetch(games: list[tuple[int, str]]) -> dict[int, float]: + data = json.loads(cache_file.read_text(encoding="utf-8")) + for aid, _name in games: + data[str(aid)] = {"hours": 3.0, "polls": 8} + cache_file.write_text(json.dumps(data), encoding="utf-8") + return {aid: 3.0 for aid, _ in games} + + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + patch(f"{_SCAN}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + ): + result = scanning._backfill_polls_for_finished( + _state([2]), + [ + GameInfo( + app_id=2, + name="X", + total_achievements=0, + unlocked_achievements=0, + playtime_minutes=0, + ) + ], + ) + assert result == {2: 8} + + def test_scanning_backfill_preserves_hours_on_miss(self, tmp_path: Path) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"2": {"hours": 9.0, "polls": 0}}), encoding="utf-8" + ) + + def fake_fetch(games: list[tuple[int, str]]) -> dict[int, float]: + data = json.loads(cache_file.read_text(encoding="utf-8")) + for aid, _name in games: + data[str(aid)] = {"hours": -1, "polls": 0} + cache_file.write_text(json.dumps(data), encoding="utf-8") + return {aid: -1 for aid, _ in games} + + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + patch(f"{_SCAN}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + ): + scanning._backfill_polls_for_finished( + _state([2]), + [ + GameInfo( + app_id=2, + name="X", + total_achievements=0, + unlocked_achievements=0, + playtime_minutes=0, + ) + ], + ) + final = json.loads(cache_file.read_text(encoding="utf-8")) + assert final["2"]["hours"] == 9.0 + + def test_report_poll_confidence_chosen_zero_polls(self) -> None: + """Covers scanning.py 301-302: 0-poll chosen with history yields warning.""" + echoed: list[str] = [] + chosen = GameInfo( + app_id=1, + name="Chosen", + total_achievements=10, + unlocked_achievements=0, + playtime_minutes=0, + comp_100_count=0, + ) + old = GameInfo( + app_id=2, + name="Old", + total_achievements=10, + unlocked_achievements=10, + playtime_minutes=0, + ) + with ( + patch( + f"{_SCAN}._backfill_polls_for_finished", + return_value={1: 0, 2: 5}, + ), + patch(f"{_SCAN}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + scanning._report_poll_confidence( + chosen, [chosen, old], _state([2], current=1) + ) + assert any("no polls recorded" in s for s in echoed) + + def test_do_scan_kept_assignment_missing_game(self) -> None: + """Covers scanning.py 110->116: current_app_id set but game absent.""" + from python_pkg.steam_backlog_enforcer.config import Config + from python_pkg.steam_backlog_enforcer.scanning import do_scan + + other = GameInfo( + app_id=999, + name="Other", + total_achievements=10, + unlocked_achievements=5, + playtime_minutes=0, + ) + from unittest.mock import MagicMock + + mock_client = MagicMock() + mock_client.build_game_list.return_value = [other] + with ( + patch(f"{_SCAN}.SteamAPIClient", return_value=mock_client), + patch(f"{_SCAN}.fetch_hltb_times_cached", return_value={999: 10.0}), + patch(f"{_SCAN}.save_snapshot"), + patch(f"{_SCAN}.pick_next_game") as mock_pick, + patch(f"{_SCAN}._echo"), + patch(f"{_SCAN}._report_poll_confidence") as mock_report, + ): + config = Config(steam_api_key="k", steam_id="i") + state = State(current_app_id=440) # not in games + do_scan(config, state) + mock_pick.assert_not_called() + mock_report.assert_not_called() + + def test_cmd_done_no_finished_history_chosen_has_polls(self) -> None: + """Covers _cmd_done.py 100->103: no finished history, chosen has >0 polls.""" + echoed: list[str] = [] + with ( + patch( + f"{_CMD}._backfill_polls_for_finished", + return_value={1: 7}, + ), + patch( + f"{_CMD}.load_snapshot", + return_value=[ + {"app_id": 1, "name": "Chosen"}, + ], + ), + patch(f"{_CMD}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + _cmd_done._report_assigned_confidence(1, _state([], current=1)) + assert any("HLTB confidence: 7" in s for s in echoed) + assert not any("NEW LOW" in s for s in echoed) + assert not any("no polls recorded" in s for s in echoed) + + def test_report_poll_confidence_chosen_equals_min(self) -> None: + """Covers scanning.py 301->304: chosen_polls >= min_polls, no warning.""" + echoed: list[str] = [] + chosen = GameInfo( + app_id=1, + name="Chosen", + total_achievements=10, + unlocked_achievements=0, + playtime_minutes=0, + comp_100_count=5, + ) + old = GameInfo( + app_id=2, + name="Old", + total_achievements=10, + unlocked_achievements=10, + playtime_minutes=0, + ) + with ( + patch( + f"{_SCAN}._backfill_polls_for_finished", + return_value={1: 5, 2: 5}, + ), + patch(f"{_SCAN}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + scanning._report_poll_confidence( + chosen, [chosen, old], _state([2], current=1) + ) + assert not any("NEW LOW" in s for s in echoed) + assert not any("no polls recorded" in s for s in echoed) + + def test_refresh_candidate_confidence_noop_when_present(self) -> None: + game = GameInfo( + app_id=1, + name="Known", + total_achievements=10, + unlocked_achievements=1, + playtime_minutes=0, + comp_100_count=3, + count_comp=15, + ) + with patch(f"{_SCAN}.fetch_hltb_confidence_cached") as mock_fetch: + scanning._refresh_candidate_confidence(game) + mock_fetch.assert_not_called() + + def test_refresh_candidate_confidence_backfills_zeroes( + self, tmp_path: Path + ) -> None: + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps({"1": {"hours": 4.0, "polls": 0, "count_comp": 0}}), + encoding="utf-8", + ) + game = GameInfo( + app_id=1, + name="NeedsRefresh", + total_achievements=10, + unlocked_achievements=1, + playtime_minutes=0, + comp_100_count=0, + count_comp=0, + ) + + def fake_fetch(_games: list[tuple[int, str]]) -> dict[int, float]: + data = json.loads(cache_file.read_text(encoding="utf-8")) + data["1"] = {"hours": 4.0, "polls": 3, "count_comp": 15} + cache_file.write_text(json.dumps(data), encoding="utf-8") + return {1: 4.0} + + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + patch(f"{_SCAN}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch(f"{_SCAN}._echo"), + ): + scanning._refresh_candidate_confidence(game) + + assert game.comp_100_count == 3 + assert game.count_comp == 15 + + def test_filter_hltb_confidence_batches_refreshes(self, tmp_path: Path) -> None: + """Filtering refreshes missing confidence in one batched cache lookup.""" + cache_file = tmp_path / "hltb_cache.json" + cache_file.write_text( + json.dumps( + { + "1": {"hours": 4.0, "polls": 0, "count_comp": 0}, + "2": {"hours": 5.0, "polls": 0, "count_comp": 0}, + } + ), + encoding="utf-8", + ) + game_a = GameInfo( + app_id=1, + name="A", + total_achievements=10, + unlocked_achievements=1, + playtime_minutes=0, + comp_100_count=0, + count_comp=0, + ) + game_b = GameInfo( + app_id=2, + name="B", + total_achievements=10, + unlocked_achievements=1, + playtime_minutes=0, + comp_100_count=0, + count_comp=0, + ) + + def fake_fetch(games: list[tuple[int, str]]) -> dict[int, float]: + assert sorted(games) == [(1, "A"), (2, "B")] + data = json.loads(cache_file.read_text(encoding="utf-8")) + data["1"] = {"hours": 4.0, "polls": 3, "count_comp": 15} + data["2"] = {"hours": 5.0, "polls": 3, "count_comp": 15} + cache_file.write_text(json.dumps(data), encoding="utf-8") + return {1: 4.0, 2: 5.0} + + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + patch( + f"{_SCAN}.fetch_hltb_confidence_cached", side_effect=fake_fetch + ) as mock_fetch, + patch(f"{_SCAN}._echo"), + ): + kept = scanning._filter_hltb_confident_candidates([game_a, game_b]) + + assert [game.app_id for game in kept] == [1, 2] + mock_fetch.assert_called_once() diff --git a/steam_backlog_enforcer/tests/test_scanning.py b/steam_backlog_enforcer/tests/test_scanning.py index 3cdb73c..4eefb5a 100644 --- a/steam_backlog_enforcer/tests/test_scanning.py +++ b/steam_backlog_enforcer/tests/test_scanning.py @@ -8,7 +8,11 @@ from unittest.mock import MagicMock, patch from python_pkg.steam_backlog_enforcer.config import Config, State from python_pkg.steam_backlog_enforcer.protondb import ProtonDBRating from python_pkg.steam_backlog_enforcer.scanning import ( + _filter_hltb_confident_candidates, + _force_refresh_candidate_confidence, + _pick_next_shortest_candidate, _pick_playable_candidate, + _refresh_candidate_confidence_batch, do_check, do_scan, pick_next_game, @@ -33,6 +37,8 @@ def _game( unlocked_achievements=unlocked, playtime_minutes=60, completionist_hours=hours, + comp_100_count=3, + count_comp=15, ) @@ -219,6 +225,9 @@ class TestPickNextGame: config = Config(steam_api_key="k", steam_id="i") state = State() with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._force_refresh_candidate_confidence" + ), patch( "python_pkg.steam_backlog_enforcer.scanning._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, @@ -286,6 +295,9 @@ class TestPickNextGame: config = Config(steam_api_key="k", steam_id="i", uninstall_other_games=True) state = State() with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._force_refresh_candidate_confidence" + ), patch( "python_pkg.steam_backlog_enforcer.scanning._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, @@ -308,6 +320,9 @@ class TestPickNextGame: config = Config(steam_api_key="k", steam_id="i", uninstall_other_games=False) state = State() with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._force_refresh_candidate_confidence" + ), patch( "python_pkg.steam_backlog_enforcer.scanning._pick_playable_candidate", side_effect=lambda c: c[0] if c else None, @@ -370,6 +385,191 @@ class TestPickNextGame: pick_next_game([g1], state, config) assert state.current_app_id == 1 + def test_skips_low_confidence_and_picks_next(self) -> None: + low = _game(app_id=1, name="LowConfidence", hours=1.0) + low.comp_100_count = 1 + low.count_comp = 5 + valid = _game(app_id=2, name="ValidConfidence", hours=2.0) + valid.comp_100_count = 3 + valid.count_comp = 15 + echoed: list[str] = [] + config = Config(steam_api_key="k", steam_id="i") + state = State() + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._force_refresh_candidate_confidence" + ), + 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.is_game_installed", + return_value=True, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning.uninstall_other_games", + return_value=0, + ), + ): + pick_next_game([low, valid], state, config) + assert state.current_app_id == 2 + assert any("Skipping LowConfidence" in line for line in echoed) + assert any("comp_100 polls 1 < 3" in line for line in echoed) + + def test_all_candidates_filtered_by_confidence(self) -> None: + low_a = _game(app_id=1, name="LowA", hours=1.0) + low_a.comp_100_count = 2 + low_a.count_comp = 15 + low_b = _game(app_id=2, name="LowB", hours=2.0) + low_b.comp_100_count = 3 + low_b.count_comp = 14 + echoed: list[str] = [] + config = Config(steam_api_key="k", steam_id="i") + state = State() + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._echo", + side_effect=lambda *a, **_: echoed.append(a[0]), + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning._force_refresh_candidate_confidence" + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning._pick_playable_candidate", + return_value=None, + ) as mock_pick, + ): + pick_next_game([low_a, low_b], state, config) + assert state.current_app_id is None + mock_pick.assert_not_called() + assert any("No assignable games found" in line for line in echoed) + + def test_zero_confidence_is_refreshed_before_skipping(self) -> None: + """Missing confidence fields are refreshed once before final skip decision.""" + stale = _game(app_id=1, name="Celeste", hours=1.0) + stale.comp_100_count = 0 + stale.count_comp = 0 + fallback = _game(app_id=2, name="Fallback", hours=2.0) + + config = Config(steam_api_key="k", steam_id="i") + state = State() + echoed: list[str] = [] + + def refresh_side_effect(game: GameInfo) -> None: + if game.app_id == 1: + game.comp_100_count = 899 + game.count_comp = 14055 + + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._refresh_candidate_confidence", + side_effect=refresh_side_effect, + ) as mock_refresh, + 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.is_game_installed", + return_value=True, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning.uninstall_other_games", + return_value=0, + ), + ): + pick_next_game([stale, fallback], state, config) + + assert state.current_app_id == 1 + mock_refresh.assert_called_once_with(stale) + assert not any("Skipping Celeste" in line for line in echoed) + + def test_nonzero_low_confidence_does_not_force_refetch(self) -> None: + """Non-zero low-confidence entries are skipped using cached values.""" + low = _game(app_id=1, name="Low", hours=1.0) + low.comp_100_count = 1 + low.count_comp = 8 + fallback = _game(app_id=2, name="Fallback", hours=2.0) + + config = Config(steam_api_key="k", steam_id="i") + state = State() + + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._refresh_candidate_confidence_batch" + ) as mock_refresh_batch, + 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"), + patch( + "python_pkg.steam_backlog_enforcer.scanning.is_game_installed", + return_value=True, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning.uninstall_other_games", + return_value=0, + ), + ): + pick_next_game([low, fallback], state, config) + + assert state.current_app_id == 2 + mock_refresh_batch.assert_not_called() + + def test_stops_after_first_confident_assignment(self) -> None: + """Only candidates up to the winning one are checked/skipped.""" + low = _game(app_id=1, name="Low", hours=1.0) + low.comp_100_count = 1 + low.count_comp = 2 + good = _game(app_id=2, name="Good", hours=2.0) + good.comp_100_count = 10 + good.count_comp = 50 + never_checked = _game(app_id=3, name="NeverChecked", hours=3.0) + never_checked.comp_100_count = 0 + never_checked.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._refresh_candidate_confidence" + ) as mock_refresh, + 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.is_game_installed", + return_value=True, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning.uninstall_other_games", + return_value=0, + ), + ): + pick_next_game([low, good, never_checked], state, config) + + assert state.current_app_id == 2 + mock_refresh.assert_called_once_with(low) + assert any("Skipping Low" in line for line in echoed) + assert not any("Skipping NeverChecked" in line for line in echoed) + class TestDoCheck: """Tests for do_check.""" @@ -393,6 +593,100 @@ class TestDoCheck: state = State(current_app_id=440, current_game_name="TF2") do_check(Config(steam_api_key="k", steam_id="i"), state) + +class TestConfidenceHelpers: + """Coverage-focused tests for scanning confidence helper branches.""" + + def test_force_refresh_candidate_confidence_delegates(self) -> None: + game = _game(app_id=10, name="A") + with patch( + "python_pkg.steam_backlog_enforcer.scanning._refresh_candidate_confidence_batch", + ) as mock_batch: + _force_refresh_candidate_confidence(game) + mock_batch.assert_called_once_with([game], force=True) + + def test_refresh_candidate_confidence_batch_no_missing_skips_fetch(self) -> None: + game = _game(app_id=20, name="B", hours=12.0) + game.comp_100_count = 3 + game.count_comp = 15 + with patch( + "python_pkg.steam_backlog_enforcer.scanning.fetch_hltb_confidence_cached", + ) as mock_fetch: + _refresh_candidate_confidence_batch([game], force=False) + mock_fetch.assert_not_called() + + def test_refresh_candidate_confidence_batch_preserves_existing_hours(self) -> None: + game = _game(app_id=30, name="C", hours=9.5) + game.comp_100_count = 0 + game.count_comp = 0 + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning.load_hltb_cache", + side_effect=[{30: 9.5}, {30: -1.0}], + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning.load_hltb_polls_cache", + return_value={30: 0}, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning.load_hltb_count_comp_cache", + return_value={30: 0}, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning.fetch_hltb_confidence_cached", + return_value={30: -1.0}, + ), + patch( + "python_pkg.steam_backlog_enforcer.scanning.save_hltb_cache", + ) as mock_save, + ): + _refresh_candidate_confidence_batch([game], force=True) + + assert game.completionist_hours == 9.5 + saved_cache = mock_save.call_args.args[0] + assert saved_cache[30] == 9.5 + + def test_filter_hltb_confident_candidates_skips_low_confidence(self) -> None: + low = _game(app_id=40, name="Low", hours=2.0) + low.comp_100_count = 1 + low.count_comp = 2 + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._refresh_candidate_confidence_batch", + ), + patch("python_pkg.steam_backlog_enforcer.scanning._echo") as mock_echo, + ): + result = _filter_hltb_confident_candidates([low]) + assert result == [] + assert mock_echo.called + + def test_pick_next_shortest_candidate_logs_skipped_unplayable_batches(self) -> None: + bad = _game(app_id=50, name="Bad", hours=1.0) + good = _game(app_id=51, name="Good", hours=2.0) + bad.comp_100_count = 3 + bad.count_comp = 15 + good.comp_100_count = 3 + good.count_comp = 15 + + with ( + patch( + "python_pkg.steam_backlog_enforcer.scanning._pick_playable_candidate", + side_effect=[None, good], + ), + patch("python_pkg.steam_backlog_enforcer.scanning._echo") as mock_echo, + ): + picked, skipped_low_conf, skipped_linux = _pick_next_shortest_candidate( + [bad, good], + ) + + assert picked is good + assert skipped_low_conf == 0 + assert skipped_linux == 1 + assert any( + "Skipped 1 game(s) with poor Linux compatibility" in str(call) + for call in mock_echo.call_args_list + ) + def test_complete(self) -> None: game = _game(app_id=440, name="TF2", total=5, unlocked=5) mock_client = MagicMock()