kuhyx/steam-backlog-enforcer
1
0
Fork 0
mirror of https://github.com/kuhyx/steam-backlog-enforcer.git synced 2026-07-04 13:43:45 +02:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Apply focus-mode, screen-locker, and steam backlog updates

Browse Source
This commit is contained in:
Krzysztof kuhy Rudnicki 2026-05-03 22:30:48 +02:00
parent 1f22281993
commit 14d706ee16
17 changed files with 2554 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
steam_backlog_enforcer/_cmd_done.py
View File

@ -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

27
steam_backlog_enforcer/_enforce_loop.py
View File

@ -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
# ──────────────────────────────────────────────────────────────

10
steam_backlog_enforcer/_hltb_detail.py
View File

@ -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(

90
steam_backlog_enforcer/_hltb_types.py
View File

@ -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):
{
"<app_id>": {
"hours": <float>,
"polls": <int>,
"count_comp": <int>
}
}
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")

11
steam_backlog_enforcer/game_install.py
View File

@ -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:

334
steam_backlog_enforcer/hltb.py
View File

@ -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)
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:
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),
)
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", name, 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.

288
steam_backlog_enforcer/scanning.py
View File

@ -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,10 +191,18 @@ 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:
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 = ""
@ -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
# ──────────────────────────────────────────────────────────────

6
steam_backlog_enforcer/steam_api.py
View File

@ -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,
)

459
steam_backlog_enforcer/tests/test_cmd_done.py
View File

@ -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

22
steam_backlog_enforcer/tests/test_enforce_loop.py
View File

@ -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."""

14
steam_backlog_enforcer/tests/test_game_install.py
View File

@ -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."""

18
steam_backlog_enforcer/tests/test_hltb_detail.py
View File

@ -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

100
steam_backlog_enforcer/tests/test_hltb_part2.py
View File

@ -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()

100
steam_backlog_enforcer/tests/test_hltb_search.py
View File

@ -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 == []

6
steam_backlog_enforcer/tests/test_main_part2.py
View File

@ -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)

729
steam_backlog_enforcer/tests/test_polls_tracking.py Normal file
View File

@ -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()

294
steam_backlog_enforcer/tests/test_scanning.py
View File

@ -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()