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

steam_backlog_enforcer: fix stats command — show real Rush/Leisure/Worst data

Browse Source 
Four bugs fixed:
- HLTB search returned 0 results for ~87 games with special chars (™, ®, &,
  standalone -, (Legacy), RHCP, etc.) — add _sanitize_search_name() and
  extend _build_search_variants() with Steam-suffix and edition stripping
- fetch_hltb_detail_missing returned immediately because `app_id not in rush`
  was always False (all keys present with -1) — fix to `rush.get(id,-1) <= 0`
- save_hltb_cache overwrote rush/leisure on confidence-only partial saves —
  now reads existing cache and preserves data when extras dicts are empty
- _filter_qualifying_games excluded 57 games with stale snapshot hours (-1)
  even though HLTB hours cache had valid data — add cache fallback

Result: stats shows Rush 64,670h / Leisure 136,807h / Worst 228,594h for
all 785 qualifying games with full rush+leisure detail.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Krzysztof kuhy Rudnicki 2026-05-28 07:02:48 +02:00
parent f7d68bc062
commit 48b609e1a3
18 changed files with 1761 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
steam_backlog_enforcer/_hltb_detail.py
View File

@ -17,6 +17,7 @@ from python_pkg.steam_backlog_enforcer._hltb_types import (
MAX_CONCURRENT,
HLTBResult,
ProgressCb,
_HLTBExtras,
save_hltb_cache,
)
@ -70,6 +71,28 @@ def _platform_comp_high_candidates(game_data: dict[str, Any]) -> list[int]:
return candidates
def _extract_comp_100_avg_and_high(game_data: dict[str, Any]) -> tuple[float, float]:
"""Extract (average comp_100, high comp_100) from game detail data.
Returns hours as floats: (avg_hours, high_hours). Returns (-1, -1) when
insufficient data is present. The average is ``comp_100`` (seconds) from
``game[0]``; the high is ``comp_100_h``.
"""
games = game_data.get("game", [])
if not isinstance(games, list) or not games:
return -1, -1
if not isinstance(games[0], dict):
return -1, -1
base = games[0]
avg_s = _as_positive_int(base.get("comp_100", 0))
high_s = _as_positive_int(base.get("comp_100_h", 0))
avg_h = round(avg_s / 3600, 2) if avg_s > 0 else -1
high_h = round(high_s / 3600, 2) if high_s > 0 else avg_h
return avg_h, high_h
def _extract_base_leisure_hours(game_data: dict[str, Any]) -> float:
"""Extract base-game leisure hours from game detail data.
@ -186,22 +209,46 @@ async def _fetch_detail_one(
return None
def _process_game_detail(
game_data: dict[str, Any],
dlc_rels: list[tuple[int, float]],
dlc_hours_by_id: dict[int, float],
) -> tuple[float, float, float]:
"""Return (leisure_hours, rush_hours, leisure_100h) for one game's detail data."""
leisure = _extract_leisure_hours(game_data)
if leisure > 0:
leisure = _apply_dlc_leisure_overrides(leisure, dlc_rels, dlc_hours_by_id)
avg_h, high_h = _extract_comp_100_avg_and_high(game_data)
rush_h = -1.0
if avg_h > 0:
dlc_rush = sum(fh for _, fh in dlc_rels if fh > 0)
rush_h = round(avg_h + dlc_rush, 2)
l100 = -1.0
if high_h > 0:
l100 = _apply_dlc_leisure_overrides(high_h, dlc_rels, dlc_hours_by_id)
return leisure, rush_h, l100
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,
extras: _HLTBExtras | 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.
Also populates ``extras.rush`` (avg comp_100 + DLC) and
``extras.leisure_100h`` (comp_100_h + DLC leisure).
The ``polls`` and ``extras.count_comp`` are forwarded to
:func:`save_hltb_cache` so confidence metrics persist.
"""
if count_comp is None:
count_comp = {}
if extras is None:
extras = _HLTBExtras()
valid = [r for r in search_results if r.hltb_game_id > 0]
if not valid:
@ -231,22 +278,24 @@ async def _fetch_leisure_times(
for r, game_data in zip(valid, details, strict=False):
done += 1
if game_data is not None:
leisure = _extract_leisure_hours(game_data)
dlc_rels = dlc_relationships_by_app.get(r.app_id, [])
leisure, rush_h, l100 = _process_game_detail(
game_data, dlc_rels, dlc_hours_by_id
)
if leisure > 0:
leisure = _apply_dlc_leisure_overrides(
leisure,
dlc_relationships_by_app.get(r.app_id, []),
dlc_hours_by_id,
)
r.completionist_hours = leisure
cache[r.app_id] = leisure
found += 1
if rush_h > 0:
extras.rush[r.app_id] = rush_h
if l100 > 0:
extras.leisure_100h[r.app_id] = l100
if progress_cb is not None:
progress_cb(done, total, found, r.game_name)
if not done % _SAVE_INTERVAL:
save_hltb_cache(cache, polls, count_comp)
save_hltb_cache(cache, polls, extras)
def _collect_dlc_relationships(

105
steam_backlog_enforcer/_hltb_search.py
View File

@ -26,6 +26,7 @@ from python_pkg.steam_backlog_enforcer._hltb_types import (
HLTBResult,
ProgressCb,
_AuthInfo,
_HLTBExtras,
save_hltb_cache,
)
@ -98,11 +99,30 @@ def _similarity(a: str, b: str) -> float:
return SequenceMatcher(None, a.lower(), b.lower()).ratio()
_TM_RE = re.compile("[™®©\uff0a]")
_COLON_WORD_RE = re.compile(r":(\s|$)")
_STANDALONE_PUNCT_RE = re.compile(r"^[-/|]$")
_AMP_RE = re.compile(r"\s*&\s*")
def _sanitize_search_name(name: str) -> str:
"""Strip HLTB-breaking characters from a game name for searchTerms.
Removes trademark/copyright symbols, colons at word-end, standalone
punctuation tokens (dash, slash, pipe), and replaces & with 'and'.
"""
cleaned = _TM_RE.sub("", name)
cleaned = _AMP_RE.sub(" and ", cleaned)
cleaned = _COLON_WORD_RE.sub(" ", cleaned)
tokens = [t for t in cleaned.split() if not _STANDALONE_PUNCT_RE.match(t)]
return " ".join(tokens)
def _build_search_payload(game_name: str, auth: _AuthInfo | None = None) -> str:
"""Build the JSON POST body for an HLTB search."""
payload: dict[str, Any] = {
"searchType": "games",
"searchTerms": game_name.split(),
"searchTerms": _sanitize_search_name(game_name).split(),
"searchPage": 1,
"size": 20,
"searchOptions": {
@ -134,13 +154,66 @@ def _build_search_payload(game_name: str, auth: _AuthInfo | None = None) -> str:
return json.dumps(payload)
_STEAM_SUFFIX_RE = re.compile(
r"\s+(?:\(Legacy\)|\(Classic\)|\(beta\)|\(Remastered\)|Legacy|Classic|RHCP"
r"|\(Phase\s+\d+\))\s*$",
re.IGNORECASE,
)
def _build_search_variants(game_name: str) -> list[str]:
"""Return fallback search terms for one Steam game title."""
"""Return fallback search terms for one Steam game title.
Tries progressively simplified names so HLTB search finds a result even
when the Steam title contains edition suffixes, Steam-only labels, or
subtitle decorators that HLTB does not index under.
Order matters: most-specific first, then stripped-down fallbacks.
Simplifications are chained so e.g. "Foo - Bar Edition" →
"Foo - Bar" → "Foo" and "Foo - Bar Edition" → "Foo Edition" → "Foo".
"""
base = game_name.strip()
variants = [base]
seen: set[str] = set()
variants: list[str] = []
def _add(name: str) -> None:
s = name.strip()
if s and s not in seen:
seen.add(s)
variants.append(s)
_add(base)
# Strip Steam-only labels that HLTB never uses
no_steam = _STEAM_SUFFIX_RE.sub("", base).strip()
_add(no_steam)
# Strip trailing year "(YYYY)"
no_year = re.sub(r"\s*\(\d{4}\)$", "", base).strip()
if no_year and no_year != base:
variants.append(no_year)
_add(no_year)
# Strip " - subtitle" portion (e.g. "Brothers - A Tale of Two Sons" → "Brothers")
no_subtitle = re.sub(r"\s+-\s+.*$", "", base).strip()
_add(no_subtitle)
# Also strip edition from the subtitle-stripped name.
# e.g. "Rocksmith 2014 Edition - Remastered" → "Rocksmith 2014 Edition"
# → "Rocksmith 2014"
if no_subtitle != base:
_add(re.sub(r"\s+\w+\s+Edition\s*$", "", no_subtitle, flags=re.IGNORECASE))
_add(re.sub(r"\s+Edition\s*$", "", no_subtitle, flags=re.IGNORECASE))
# Strip "GOTY Edition" / "Gold Edition" / "Definitive Edition" etc. from base
no_edition = re.sub(r"\s+\w+\s+Edition\s*$", "", base, flags=re.IGNORECASE).strip()
_add(no_edition)
# Strip just " Edition" at end from base
no_bare_edition = re.sub(r"\s+Edition\s*$", "", base, flags=re.IGNORECASE).strip()
_add(no_bare_edition)
# Strip ": subtitle" portion (e.g. "Batman: Arkham Asylum" → "Batman")
no_colon_sub = re.sub(r"\s*:.*$", "", base).strip()
_add(no_colon_sub)
return variants
@ -322,6 +395,7 @@ class _SearchCtx:
counter: dict[str, int] = field(default_factory=dict)
total: int = 0
progress_cb: ProgressCb | None = None
hltb_game_id: dict[int, int] = field(default_factory=dict)
async def _search_one(
@ -358,6 +432,8 @@ async def _search_one(
ctx.cache[app_id] = result.completionist_hours
ctx.polls[app_id] = result.comp_100_count
ctx.count_comp[app_id] = result.count_comp
if result.hltb_game_id > 0:
ctx.hltb_game_id[app_id] = result.hltb_game_id
ctx.counter["found"] += 1
else:
ctx.cache[app_id] = -1
@ -369,7 +445,11 @@ async def _search_one(
# Incremental save every _SAVE_INTERVAL lookups.
if not done % _SAVE_INTERVAL:
save_hltb_cache(ctx.cache, ctx.polls, ctx.count_comp)
save_hltb_cache(
ctx.cache,
ctx.polls,
_HLTBExtras(count_comp=ctx.count_comp, hltb_game_id=ctx.hltb_game_id),
)
# Report progress.
if ctx.progress_cb is not None:
@ -383,9 +463,12 @@ async def _fetch_batch(
cache: dict[int, float],
polls: dict[int, int],
progress_cb: ProgressCb | None,
count_comp: dict[int, int] | None = None,
extras: _HLTBExtras | None = None,
) -> list[HLTBResult]:
"""Fetch HLTB data for a batch of games using one shared session."""
if extras is None:
extras = _HLTBExtras()
# 1. Discover the search URL (sync, one-time).
search_url = _get_hltb_search_url()
logger.info("HLTB search URL: %s", search_url)
@ -419,9 +502,6 @@ 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,
@ -436,11 +516,12 @@ async def _fetch_batch(
headers=headers,
cache=cache,
polls=polls,
count_comp=count_comp,
count_comp=extras.count_comp,
auth=auth,
counter=counter,
total=total,
progress_cb=progress_cb,
hltb_game_id=extras.hltb_game_id,
)
tasks = [
_search_one(
@ -465,7 +546,7 @@ async def _fetch_batch(
cache,
polls,
progress_cb=None,
count_comp=count_comp,
extras=extras,
)
return search_results

102
steam_backlog_enforcer/_hltb_types.py
View File

@ -45,6 +45,32 @@ class HLTBResult:
hltb_game_id: int = 0
comp_100_count: int = 0
count_comp: int = 0
rush_hours: float = -1
leisure_100h: float = -1
class _HLTBExtras:
"""Mutable accumulator for HLTB data beyond the core hours cache.
Passed through the fetch pipeline so callers stay within the 5-arg limit.
"""
def __init__(
self,
count_comp: dict[int, int] | None = None,
rush: dict[int, float] | None = None,
leisure_100h: dict[int, float] | None = None,
hltb_game_id: dict[int, int] | None = None,
) -> None:
"""Initialize with optional pre-populated dicts."""
self.count_comp: dict[int, int] = count_comp if count_comp is not None else {}
self.rush: dict[int, float] = rush if rush is not None else {}
self.leisure_100h: dict[int, float] = (
leisure_100h if leisure_100h is not None else {}
)
self.hltb_game_id: dict[int, int] = (
hltb_game_id if hltb_game_id is not None else {}
)
@dataclass
@ -64,7 +90,10 @@ def _read_raw_cache() -> dict[int, dict[str, Any]]:
"<app_id>": {
"hours": <float>,
"polls": <int>,
"count_comp": <int>
"count_comp": <int>,
"rush_hours": <float>,
"leisure_100h": <float>,
"hltb_game_id": <int>
}
}
@ -88,10 +117,20 @@ def _read_raw_cache() -> dict[int, dict[str, Any]]:
"hours": float(v.get("hours", -1)),
"polls": int(v.get("polls", 0)),
"count_comp": int(v.get("count_comp", 0)),
"rush_hours": float(v.get("rush_hours", -1)),
"leisure_100h": float(v.get("leisure_100h", -1)),
"hltb_game_id": int(v.get("hltb_game_id", 0)),
}
else:
try:
out[aid] = {"hours": float(v), "polls": 0, "count_comp": 0}
out[aid] = {
"hours": float(v),
"polls": 0,
"count_comp": 0,
"rush_hours": -1,
"leisure_100h": -1,
"hltb_game_id": 0,
}
except (TypeError, ValueError):
continue
return out
@ -121,19 +160,70 @@ def load_hltb_count_comp_cache() -> dict[int, int]:
return {aid: v["count_comp"] for aid, v in _read_raw_cache().items()}
def load_hltb_rush_cache() -> dict[int, float]:
"""Load the rush-hours (avg comp_100 + DLC) portion of the HLTB cache.
Returns: dict mapping app_id -> rush_hours (-1 = not yet computed).
"""
return {aid: v["rush_hours"] for aid, v in _read_raw_cache().items()}
def load_hltb_leisure_100h_cache() -> dict[int, float]:
"""Load the leisure-100h (comp_100_h + DLC) portion of the HLTB cache.
Returns: dict mapping app_id -> leisure_100h (-1 = not yet computed).
"""
return {aid: v["leisure_100h"] for aid, v in _read_raw_cache().items()}
def load_hltb_game_id_cache() -> dict[int, int]:
"""Load the HLTB game ID portion of the cache.
Returns: dict mapping app_id -> hltb_game_id (0 = not yet looked up).
"""
return {aid: v["hltb_game_id"] 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,
extras: _HLTBExtras | None = None,
) -> None:
"""Save the HLTB cache to disk, including confidence metrics."""
"""Save the HLTB cache to disk, including confidence and stats metrics."""
polls = polls or {}
count_comp = count_comp or {}
if extras is None:
extras = _HLTBExtras()
# Preserve existing per-game data when the caller didn't populate the maps.
# A partial save (e.g. confidence-only) must not clobber rush/leisure/game-id
# data that a prior detail fetch already wrote.
needs_existing = (
not extras.hltb_game_id or not extras.rush or not extras.leisure_100h
)
if needs_existing:
existing = _read_raw_cache()
game_id_map: dict[int, int] = extras.hltb_game_id or {
aid: v["hltb_game_id"] for aid, v in existing.items()
}
rush_map: dict[int, float] = extras.rush or {
aid: v["rush_hours"] for aid, v in existing.items() if v["rush_hours"] > 0
}
leisure_map: dict[int, float] = extras.leisure_100h or {
aid: v["leisure_100h"]
for aid, v in existing.items()
if v["leisure_100h"] > 0
}
else:
game_id_map = extras.hltb_game_id
rush_map = extras.rush
leisure_map = extras.leisure_100h
out = {
str(aid): {
"hours": hours,
"polls": polls.get(aid, 0),
"count_comp": count_comp.get(aid, 0),
"count_comp": extras.count_comp.get(aid, 0),
"rush_hours": rush_map.get(aid, -1),
"leisure_100h": leisure_map.get(aid, -1),
"hltb_game_id": game_id_map.get(aid, 0),
}
for aid, hours in cache.items()
}

7
steam_backlog_enforcer/_scanning_confidence.py
View File

@ -6,6 +6,7 @@ import logging
from typing import TYPE_CHECKING
from python_pkg.steam_backlog_enforcer._hltb_types import (
_HLTBExtras,
load_hltb_cache,
load_hltb_count_comp_cache,
load_hltb_polls_cache,
@ -105,7 +106,7 @@ def _refresh_candidate_confidence_batch(
cache.pop(aid, None)
polls.pop(aid, None)
count_comp.pop(aid, None)
save_hltb_cache(cache, polls, count_comp)
save_hltb_cache(cache, polls, _HLTBExtras(count_comp=count_comp))
fetch_hltb_confidence_cached(names)
@ -115,7 +116,9 @@ def _refresh_candidate_confidence_batch(
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)
save_hltb_cache(
refreshed_hours, refreshed_polls, _HLTBExtras(count_comp=refreshed_count_comp)
)
for game in refresh_slice:
game.comp_100_count = refreshed_polls.get(game.app_id, 0)

350
steam_backlog_enforcer/_stats.py Normal file
View File

@ -0,0 +1,350 @@
"""Backlog completion-time statistics for Steam Backlog Enforcer."""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
import logging
import secrets
from typing import TYPE_CHECKING
from urllib.parse import quote_plus
from python_pkg.steam_backlog_enforcer._hltb_types import (
HLTB_BASE_URL,
load_hltb_cache,
load_hltb_game_id_cache,
load_hltb_leisure_100h_cache,
load_hltb_rush_cache,
)
from python_pkg.steam_backlog_enforcer._scanning_confidence import (
_apply_cached_confidence_to_candidates,
_confidence_fail_reasons,
_refresh_candidate_confidence_batch,
)
from python_pkg.steam_backlog_enforcer.config import load_snapshot
from python_pkg.steam_backlog_enforcer.game_install import _echo
from python_pkg.steam_backlog_enforcer.hltb import fetch_hltb_detail_missing
from python_pkg.steam_backlog_enforcer.protondb import (
ProtonDBRating,
fetch_protondb_ratings,
)
from python_pkg.steam_backlog_enforcer.steam_api import GameInfo
if TYPE_CHECKING:
from python_pkg.steam_backlog_enforcer.config import Config, State
logger = logging.getLogger(__name__)
_HOURS_PER_DAY_PRESETS = (2.0, 4.0, 6.0, 8.0)
_LINE = "─" * 70
_HLTB_SEARCH_BASE = "https://howlongtobeat.com/?q="
@dataclass
class _GameTimes:
"""Per-game time estimates for stats display."""
game: GameInfo
worst_hours: float
rush_hours: float
leisure_100h: float
hltb_game_id: int = field(default=0)
def _filter_qualifying_games(
games: list[GameInfo],
state: State,
) -> tuple[list[_GameTimes], int, int, int]:
"""Return qualifying incomplete games with their time estimates.
Applies the same HLTB-confidence and Linux-compatibility filters as the
game picker. The current game and already-finished games are excluded.
Returns:
(qualified_list, hltb_skipped, linux_skipped, no_data_skipped)
"""
rush_cache = load_hltb_rush_cache()
leisure_100h_cache = load_hltb_leisure_100h_cache()
game_id_cache = load_hltb_game_id_cache()
hours_cache = load_hltb_cache()
exclude = set(state.finished_app_ids)
if state.current_app_id is not None:
exclude.add(state.current_app_id)
candidates = [g for g in games if not g.is_complete and g.app_id not in exclude]
_apply_cached_confidence_to_candidates(candidates)
_refresh_candidate_confidence_batch(candidates)
hltb_skipped = 0
linux_skipped = 0
no_data_skipped = 0
app_ids_to_check: list[int] = []
conf_ok: list[GameInfo] = []
for game in candidates:
if _confidence_fail_reasons(game):
hltb_skipped += 1
continue
conf_ok.append(game)
app_ids_to_check.append(game.app_id)
ratings: dict[int, ProtonDBRating] = {}
if app_ids_to_check:
ratings = fetch_protondb_ratings(app_ids_to_check)
qualified: list[_GameTimes] = []
for game in conf_ok:
rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id))
if not rating.is_playable:
linux_skipped += 1
continue
rush = rush_cache.get(game.app_id, -1)
leisure = leisure_100h_cache.get(game.app_id, -1)
# worst_hours = max of: snapshot completionist, HLTB hours cache (fallback
# when snapshot is stale/missing), and leisure_100h (slowest 100% time).
snap_hours = game.completionist_hours if game.completionist_hours > 0 else -1
cache_hours = hours_cache.get(game.app_id, -1)
worst_candidates = [v for v in (snap_hours, cache_hours, leisure) if v > 0]
worst = max(worst_candidates) if worst_candidates else -1
if worst <= 0 and rush <= 0 and leisure <= 0:
no_data_skipped += 1
continue
qualified.append(
_GameTimes(
game=game,
worst_hours=worst,
rush_hours=rush,
leisure_100h=leisure,
hltb_game_id=game_id_cache.get(game.app_id, 0),
)
)
return qualified, hltb_skipped, linux_skipped, no_data_skipped
def _ensure_rush_data(qualified: list[_GameTimes]) -> bool:
"""Auto-fetch rush/leisure detail for games that are missing it.
Returns True when a fetch was performed; the caller should then re-run
``_filter_qualifying_games`` to pick up the updated caches.
"""
total_q = len(qualified)
missing = sum(1 for e in qualified if e.rush_hours <= 0)
if not qualified or not missing:
return False
_echo(f"Fetching HLTB detail for {missing}/{total_q} games missing rush/leisure...")
game_pairs = [(e.game.app_id, e.game.name) for e in qualified]
fetch_hltb_detail_missing(game_pairs)
return True
def _print_worst_example(entries: list[_GameTimes]) -> None:
"""Print a randomly selected example from the worst-case qualified games."""
examples = [e for e in entries if e.worst_hours > 0]
if not examples:
return
example = secrets.choice(examples)
_echo(f"\n Example game: {example.game.name!r}")
_echo(f" Worst case: {example.worst_hours:.1f} h")
if example.rush_hours > 0:
_echo(f" Rush: {example.rush_hours:.1f} h")
if example.leisure_100h > 0:
_echo(f" Leisure: {example.leisure_100h:.1f} h")
if example.hltb_game_id > 0:
_echo(f" HLTB: {HLTB_BASE_URL}/game/{example.hltb_game_id}")
else:
_echo(f" HLTB: {_HLTB_SEARCH_BASE}{quote_plus(example.game.name)}")
def _sum_hours(entries: list[_GameTimes], attr: str) -> tuple[float, int]:
"""Sum a time attribute across entries; return (total_hours, missing_count).
Games where the attribute is ≤ 0 contribute 0 to the sum and are counted
in ``missing_count`` so the user knows the estimate may be an undercount.
"""
total = 0.0
missing = 0
for e in entries:
val: float = getattr(e, attr)
if val > 0:
total += val
else:
missing += 1
return round(total, 1), missing
def _format_completion_date(hours: float, daily_hours: float) -> str:
"""Return 'N days (YYYY-MM-DD)' for finishing hours at daily_hours per day."""
if hours <= 0 or daily_hours <= 0:
return "N/A"
days = int(hours / daily_hours)
target = datetime.now(timezone.utc) + timedelta(days=days)
return f"{days} days ({target.strftime('%Y-%m-%d')})"
def _print_scenario(
label: str,
total_hours: float,
missing: int,
total_games: int,
) -> None:
"""Print a single time-scenario block."""
_echo(f"\n {label}")
if total_hours <= 0:
_echo(" No data available.")
return
missing_note = (
f" ({missing}/{total_games} games had no data, hours underestimated)"
if missing
else ""
)
_echo(f" Total: {total_hours:,.1f} h{missing_note}")
for daily in _HOURS_PER_DAY_PRESETS:
estimate = _format_completion_date(total_hours, daily)
_echo(f" @ {daily:.0f} h/day → {estimate}")
def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None:
"""Print the pace-based completion estimate.
``games_done`` should be the count of 100%-complete games in the library
snapshot (``sum(1 for g in games if g.is_complete)``), not the enforcer's
own ``finished_app_ids`` list, which misses games completed outside the
enforcer flow.
"""
_echo("\n 1. AT YOUR CURRENT PACE")
if not state.enforcement_started_at:
_echo(" No start date recorded.")
_echo(" Set enforcement_started_at in state.json (ISO-8601 UTC)")
_echo(" to enable this estimate.")
return
try:
started = datetime.fromisoformat(state.enforcement_started_at)
except ValueError:
_echo(f" Invalid enforcement_started_at: {state.enforcement_started_at!r}")
return
now = datetime.now(timezone.utc)
days_elapsed = max(1, (now - started).days)
if games_done == 0:
_echo(f" Started: {started.strftime('%Y-%m-%d')}")
_echo(" No games finished yet — pace cannot be estimated.")
return
rate = games_done / days_elapsed
_echo(f" Started: {started.strftime('%Y-%m-%d')}")
_echo(f" Finished: {games_done} games in {days_elapsed} days")
_echo(
f" Pace: {rate:.4f} games/day (1 game every {1 / rate:.1f} days)"
)
_echo(f" Remaining: {remaining} games")
days_to_go = int(remaining / rate)
finish = now + timedelta(days=days_to_go)
_echo(f" Est. complete: {days_to_go} days ({finish.strftime('%Y-%m-%d')})")
def cmd_stats(_config: Config, state: State) -> None:
"""Display backlog completion-time statistics.
Filters games by the same HLTB-confidence and Linux-compatibility rules
used when picking the next game. Auto-fetches missing rush/leisure detail
data before printing. Shows four scenarios:
1. At your current pace (games finished per day since enforcement started).
2. Rush — avg comp_100 + DLC completion time per HLTB.
3. Leisure — comp_100_h (slowest 100 %) + DLC leisure per HLTB.
4. Worst — absolute maximum recorded time (any category) per HLTB.
"""
snapshot = load_snapshot()
if snapshot is None:
_echo("No snapshot found. Run 'scan' first.")
return
games = [GameInfo.from_snapshot(d) for d in snapshot]
# Count all 100%-achievement games in library (more accurate than
# finished_app_ids, which only tracks enforcer-assigned completions).
games_done = sum(1 for g in games if g.is_complete)
qualified, hltb_skip, linux_skip, no_data_skip = _filter_qualifying_games(
games, state
)
if _ensure_rush_data(qualified):
# Re-filter picks up updated rush/leisure caches; ProtonDB is now cached.
qualified, hltb_skip, linux_skip, no_data_skip = _filter_qualifying_games(
games, state
)
total_q = len(qualified)
_echo(f"\n{'═' * 70}")
_echo(" BACKLOG COMPLETION ESTIMATES")
_echo(f"{'═' * 70}")
_echo(f"\n Qualifying games: {total_q}")
if hltb_skip:
_echo(f" HLTB-skipped: {hltb_skip} (confidence too low)")
if linux_skip:
_echo(f" Linux-skipped: {linux_skip} (poor ProtonDB rating)")
if no_data_skip:
_echo(f" No-data-skipped: {no_data_skip} (no HLTB hours at all)")
missing_rush_final = sum(1 for e in qualified if e.rush_hours <= 0)
if missing_rush_final:
_echo(
f"\n Note: {missing_rush_final}/{total_q} games still missing"
" rush/leisure data (HLTB search may not have matched them)."
)
elif total_q:
_echo(
f"\n Detail data: rush + leisure available for all {total_q}"
" qualifying games."
)
if state.current_app_id:
_echo(
f"\n Current game: {state.current_game_name} (excluded from totals)"
)
_echo(f" Finished games: {games_done} (excluded from totals)")
_echo(f"\n{_LINE}")
_print_pace_scenario(state, total_q, games_done)
worst_total, worst_missing = _sum_hours(qualified, "worst_hours")
rush_total, rush_missing = _sum_hours(qualified, "rush_hours")
leisure_total, leisure_missing = _sum_hours(qualified, "leisure_100h")
_echo(f"\n{_LINE}")
_print_scenario(
"2. RUSH (avg comp_100 + DLC — typical fast completionist)",
rush_total,
rush_missing,
total_q,
)
_echo(f"\n{_LINE}")
_print_scenario(
"3. LEISURE (comp_100_h + DLC — slow/comfortable 100 %)",
leisure_total,
leisure_missing,
total_q,
)
_echo(f"\n{_LINE}")
_print_scenario(
"4. WORST CASE (max recorded time, any category, + DLC)",
worst_total,
worst_missing,
total_q,
)
_print_worst_example(qualified)
_echo(f"\n{_LINE}\n")

2
steam_backlog_enforcer/config.py
View File

@ -87,6 +87,8 @@ class State:
current_game_name: str = ""
finished_app_ids: list[int] = field(default_factory=list)
skipped_until: dict[str, str] = field(default_factory=dict)
enforcement_started_at: str = ""
"""ISO-8601 UTC timestamp set on the first game assignment."""
"""Map of ``str(app_id)`` → ISO-8601 UTC timestamp when the skip expires.
Games in this map are excluded from auto-assignment until the timestamp

97
steam_backlog_enforcer/hltb.py
View File

@ -30,9 +30,13 @@ from python_pkg.steam_backlog_enforcer._hltb_types import (
MAX_CONCURRENT,
HLTBResult,
ProgressCb,
_HLTBExtras,
load_hltb_cache,
load_hltb_count_comp_cache,
load_hltb_game_id_cache,
load_hltb_leisure_100h_cache,
load_hltb_polls_cache,
load_hltb_rush_cache,
save_hltb_cache,
)
@ -125,7 +129,7 @@ def fetch_hltb_times(
cache: dict[int, float] | None = None,
polls: dict[int, int] | None = None,
progress_cb: ProgressCb | None = None,
count_comp: dict[int, int] | None = None,
extras: _HLTBExtras | None = None,
) -> list[HLTBResult]:
"""Synchronous wrapper: fetch HLTB times for games."""
if not games:
@ -134,10 +138,14 @@ def fetch_hltb_times(
cache = {}
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)
_fetch_batch(
games,
cache,
polls,
progress_cb,
extras=extras,
)
)
@ -182,7 +190,11 @@ def fetch_hltb_times_cached(
"""
cache = load_hltb_cache()
polls = load_hltb_polls_cache()
count_comp = load_hltb_count_comp_cache()
extras = _HLTBExtras(
count_comp=load_hltb_count_comp_cache(),
rush=load_hltb_rush_cache(),
leisure_100h=load_hltb_leisure_100h_cache(),
)
uncached = [(app_id, name) for app_id, name in games if app_id not in cache]
if uncached:
@ -197,12 +209,12 @@ def fetch_hltb_times_cached(
cache=cache,
polls=polls,
progress_cb=progress_cb,
count_comp=count_comp,
extras=extras,
)
elapsed = time.monotonic() - t0
# Final save.
save_hltb_cache(cache, polls, count_comp)
save_hltb_cache(cache, polls, extras)
found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0)
rate = len(uncached) / elapsed if elapsed > 0 else 0
@ -245,7 +257,7 @@ def fetch_hltb_confidence_cached(
)
elapsed = time.monotonic() - t0
save_hltb_cache(cache, polls, count_comp)
save_hltb_cache(cache, polls, _HLTBExtras(count_comp=count_comp))
found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0)
rate = len(uncached) / elapsed if elapsed > 0 else 0
@ -262,6 +274,75 @@ def fetch_hltb_confidence_cached(
return cache
def fetch_hltb_detail_missing(
games: list[tuple[int, str]],
progress_cb: ProgressCb | None = None,
) -> int:
"""Fetch HLTB detail (rush + leisure) for games that are missing it.
Games already in the rush cache are skipped. For the rest, temporarily
removes them from the hours cache so ``fetch_hltb_times`` will visit their
detail pages. Restores prior hours for any game the re-fetch doesn't find.
Args:
games: list of (app_id, name) tuples to check.
progress_cb: optional progress callback.
Returns:
Number of games that now have rush-hour data after the fetch.
"""
rush = load_hltb_rush_cache()
missing = [(app_id, name) for app_id, name in games if rush.get(app_id, -1) <= 0]
if not missing:
return 0
cache = load_hltb_cache()
polls = load_hltb_polls_cache()
extras = _HLTBExtras(
count_comp=load_hltb_count_comp_cache(),
rush=rush,
leisure_100h=load_hltb_leisure_100h_cache(),
hltb_game_id=load_hltb_game_id_cache(),
)
# Remove from hours cache so fetch_hltb_times will visit the detail page.
prior_hours: dict[int, float] = {}
for app_id, _ in missing:
prior_hours[app_id] = cache.pop(app_id, -1.0)
logger.info(
"Fetching HLTB detail for %d games missing rush/leisure data...",
len(missing),
)
t0 = time.monotonic()
fetch_hltb_times(
missing,
cache=cache,
polls=polls,
progress_cb=progress_cb,
extras=extras,
)
elapsed = time.monotonic() - t0
# Restore prior hours for games the detail fetch didn't re-find.
for app_id, old_hours in prior_hours.items():
if old_hours > 0 and cache.get(app_id, -1.0) <= 0:
cache[app_id] = old_hours
save_hltb_cache(cache, polls, extras)
fetched = sum(1 for app_id, _ in missing if extras.rush.get(app_id, -1) > 0)
rate = len(missing) / elapsed if elapsed > 0 else 0
logger.info(
"HLTB detail fetch done: %d/%d got rush data in %.1fs (%.0f games/s)",
fetched,
len(missing),
elapsed,
rate,
)
return fetched
def get_hltb_submit_url(game_name: str) -> str | None:
"""Look up a game on HLTB and return its submit page URL.

2
steam_backlog_enforcer/main.py
View File

@ -13,6 +13,7 @@ from python_pkg.steam_backlog_enforcer._enforce_loop import (
get_all_owned_app_ids,
)
from python_pkg.steam_backlog_enforcer._hltb_types import load_hltb_cache
from python_pkg.steam_backlog_enforcer._stats import cmd_stats
from python_pkg.steam_backlog_enforcer._whitelist import (
WHITELIST_COOLDOWN_SECONDS,
add_pending_exception,
@ -397,6 +398,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
"setup": ("Run first-time setup", cmd_setup),
"done": ("Finish game, open HLTB, pick next", cmd_done),
"pick": ("Manually pick your next game from candidates", cmd_pick),
"stats": ("Show backlog completion-time estimates", cmd_stats),
}
# Extra commands with non-standard arg handling (shown in help but not in COMMANDS).

17
steam_backlog_enforcer/protondb.py
View File

@ -124,8 +124,12 @@ async def _fetch_one(
session: aiohttp.ClientSession,
sem: asyncio.Semaphore,
app_id: int,
) -> ProtonDBRating:
"""Fetch a single game's ProtonDB rating."""
) -> ProtonDBRating | None:
"""Fetch a single game's ProtonDB rating.
Returns None on network/server errors (not cached, will retry next run).
Returns ProtonDBRating with empty tier on HTTP 404 (no ProtonDB data).
"""
url = _PROTONDB_API.format(app_id=app_id)
async with sem:
try:
@ -142,9 +146,9 @@ async def _fetch_one(
confidence=data.get("confidence", ""),
total_reports=data.get("total", 0),
)
except (aiohttp.ClientError, asyncio.TimeoutError, OSError):
logger.warning("ProtonDB fetch failed for AppID=%d", app_id)
return ProtonDBRating(app_id=app_id)
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as e:
logger.warning("ProtonDB fetch failed for AppID=%d: %s", app_id, e)
return None # Don't cache transient failures — retry next run.
async def _fetch_batch(app_ids: list[int]) -> list[ProtonDBRating]:
@ -152,7 +156,8 @@ async def _fetch_batch(app_ids: list[int]) -> list[ProtonDBRating]:
sem = asyncio.Semaphore(MAX_CONCURRENT)
async with aiohttp.ClientSession() as session:
tasks = [_fetch_one(session, sem, aid) for aid in app_ids]
return await asyncio.gather(*tasks)
results = await asyncio.gather(*tasks)
return [r for r in results if r is not None]
def _rating_to_dict(r: ProtonDBRating) -> dict[str, Any]:

3
steam_backlog_enforcer/scanning.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
from datetime import datetime, timezone
import logging
import time
from typing import TYPE_CHECKING, Any
@ -229,6 +230,8 @@ def _assign_chosen_game(
"""Save assignment, announce it, and handle install/uninstall."""
state.current_app_id = chosen.app_id
state.current_game_name = chosen.name
if not state.enforcement_started_at:
state.enforcement_started_at = datetime.now(timezone.utc).isoformat()
state.save()
hours_str = (
f" (~{chosen.completionist_hours:.1f}h leisure+dlc)"

113
steam_backlog_enforcer/tests/test_hltb_detail.py
View File

@ -15,12 +15,18 @@ from python_pkg.steam_backlog_enforcer._hltb_detail import (
_as_positive_int,
_collect_dlc_relationships,
_extract_base_leisure_hours,
_extract_comp_100_avg_and_high,
_extract_dlc_relationships,
_fetch_detail_one,
_fetch_dlc_leisure_hours,
_fetch_leisure_times,
_process_game_detail,
)
from python_pkg.steam_backlog_enforcer._hltb_types import (
_SAVE_INTERVAL,
HLTBResult,
_HLTBExtras,
)
from python_pkg.steam_backlog_enforcer._hltb_types import _SAVE_INTERVAL, HLTBResult
class TestInternalHelpers:
@ -183,6 +189,76 @@ class TestInternalHelpers:
assert asyncio.run(_run()) == {}
class TestExtractComp100AvgAndHigh:
"""Tests for _extract_comp_100_avg_and_high."""
def test_returns_minus_one_for_empty_game_list(self) -> None:
assert _extract_comp_100_avg_and_high({"game": []}) == (-1, -1)
def test_returns_minus_one_for_non_list_game(self) -> None:
assert _extract_comp_100_avg_and_high({"game": "bad"}) == (-1, -1)
def test_returns_minus_one_when_game0_not_dict(self) -> None:
assert _extract_comp_100_avg_and_high({"game": [42]}) == (-1, -1)
def test_returns_avg_and_high(self) -> None:
data: dict[str, Any] = {"game": [{"comp_100": 7200, "comp_100_h": 10800}]}
avg_h, high_h = _extract_comp_100_avg_and_high(data)
assert avg_h == round(7200 / 3600, 2)
assert high_h == round(10800 / 3600, 2)
def test_high_falls_back_to_avg_when_zero(self) -> None:
data: dict[str, Any] = {"game": [{"comp_100": 7200, "comp_100_h": 0}]}
avg_h, high_h = _extract_comp_100_avg_and_high(data)
assert avg_h == round(7200 / 3600, 2)
assert high_h == avg_h
def test_avg_zero_returns_minus_one_avg(self) -> None:
data: dict[str, Any] = {"game": [{"comp_100": 0, "comp_100_h": 0}]}
avg_h, high_h = _extract_comp_100_avg_and_high(data)
assert avg_h == -1
assert high_h == -1
class TestProcessGameDetail:
"""Tests for _process_game_detail."""
def test_returns_leisure_rush_and_l100(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 10800, "comp_100": 7200}],
"relationships": [],
}
leisure, rush_h, l100 = _process_game_detail(data, [], {})
assert leisure == round(10800 / 3600, 2)
assert rush_h == round(7200 / 3600, 2)
assert l100 == round(10800 / 3600, 2)
def test_negative_leisure_when_no_data(self) -> None:
leisure, rush_h, l100 = _process_game_detail({"game": []}, [], {})
assert leisure == -1
assert rush_h == -1.0
assert l100 == -1.0
def test_rush_includes_dlc_fallback(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100": 7200, "comp_100_h": 0}],
"relationships": [],
}
dlc_rels = [(99, 1.5)]
_leisure, rush_h, _l100 = _process_game_detail(data, dlc_rels, {})
assert rush_h == round(7200 / 3600 + 1.5, 2)
def test_l100_uses_dlc_override(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 10800, "comp_100": 7200}],
"relationships": [],
}
dlc_rels = [(77, 2.0)]
dlc_hours_by_id = {77: 3.0}
_leisure, _rush_h, l100 = _process_game_detail(data, dlc_rels, dlc_hours_by_id)
assert l100 == round(10800 / 3600 + (3.0 - 2.0), 2)
class _FakeTextResponse:
"""Async context manager mimicking aiohttp response for text."""
@ -441,8 +517,34 @@ class TestFetchLeisureTimes:
assert cache[1289310] == expected
assert results[0].completionist_hours == expected
def test_with_explicit_count_comp(self) -> None:
"""Pass a non-None count_comp to cover the False branch of the None check."""
def test_extras_populated_with_rush_and_l100(self) -> None:
"""rush_h and l100 are stored in extras when game has comp_100 data."""
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 10800, "comp_100": 7200}],
"relationships": [],
}
cache: dict[int, float] = {}
extras = _HLTBExtras(count_comp={440: 5})
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
):
asyncio.run(_fetch_leisure_times(results, cache, {}, None, extras=extras))
assert extras.rush[440] == round(7200 / 3600, 2)
assert extras.leisure_100h[440] == round(10800 / 3600, 2)
def test_with_explicit_extras(self) -> None:
"""Pass a pre-populated _HLTBExtras to cover the non-None extras branch."""
results = [
HLTBResult(
app_id=440,
@ -457,12 +559,11 @@ class TestFetchLeisureTimes:
"relationships": [],
}
cache: dict[int, float] = {}
extras = _HLTBExtras(count_comp={440: 5})
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
):
asyncio.run(
_fetch_leisure_times(results, cache, {}, None, count_comp={440: 5})
)
asyncio.run(_fetch_leisure_times(results, cache, {}, None, extras=extras))
assert cache[440] == 1.0

145
steam_backlog_enforcer/tests/test_hltb_part2.py
View File

@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
from typing_extensions import Self
@ -14,10 +15,14 @@ from python_pkg.steam_backlog_enforcer.hltb import (
_fetch_batch_confidence_only,
fetch_hltb_confidence,
fetch_hltb_confidence_cached,
fetch_hltb_detail_missing,
fetch_hltb_times_cached,
get_hltb_submit_url,
)
if TYPE_CHECKING:
from python_pkg.steam_backlog_enforcer._hltb_types import _HLTBExtras
PKG = "python_pkg.steam_backlog_enforcer.hltb"
@ -44,14 +49,14 @@ class TestFetchHltbTimesCached:
cache: dict[int, float] | None = None,
polls: dict[int, int] | None = None,
progress_cb: object = None,
count_comp: dict[int, int] | None = None,
extras: _HLTBExtras | 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
if extras is not None:
extras.count_comp[730] = 0
return []
mock_fetch.side_effect = add_to_cache
@ -102,7 +107,7 @@ class TestFetchHltbTimesCached:
cache: dict[int, float] | None = None,
polls: dict[int, int] | None = None,
progress_cb: object = None,
count_comp: dict[int, int] | None = None,
extras: _HLTBExtras | None = None,
) -> list[object]:
if cache is not None:
cache[440] = 50.0
@ -110,9 +115,9 @@ class TestFetchHltbTimesCached:
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
if extras is not None:
extras.count_comp[440] = 15
extras.count_comp[730] = 0
return []
mock_fetch.side_effect = add_found
@ -233,3 +238,129 @@ class TestConfidenceHelpers:
assert result == {1: 12.0}
mock_fetch.assert_not_called()
mock_save.assert_not_called()
class TestFetchHltbDetailMissing:
"""Tests for fetch_hltb_detail_missing."""
def test_no_missing_returns_zero(self) -> None:
"""All games in rush cache → early return without fetching."""
with (
patch(f"{PKG}.load_hltb_rush_cache", return_value={440: 15.0}),
patch(f"{PKG}.fetch_hltb_times") as mock_fetch,
):
result = fetch_hltb_detail_missing([(440, "TF2")])
assert result == 0
mock_fetch.assert_not_called()
def test_fetches_missing_and_returns_count(self) -> None:
"""Games not in rush cache are fetched; returns count with rush data."""
def add_rush(
_games: object,
cache: dict[int, float] | None = None,
polls: dict[int, int] | None = None,
progress_cb: object = None,
extras: _HLTBExtras | None = None,
) -> list[object]:
if extras is not None:
extras.rush[730] = 10.0
if cache is not None:
cache[730] = 25.0
return []
with (
patch(f"{PKG}.load_hltb_rush_cache", return_value={440: 15.0}),
patch(f"{PKG}.load_hltb_cache", return_value={730: 20.0}),
patch(f"{PKG}.load_hltb_polls_cache", return_value={}),
patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}),
patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}),
patch(f"{PKG}.load_hltb_game_id_cache", return_value={}),
patch(f"{PKG}.fetch_hltb_times", side_effect=add_rush),
patch(f"{PKG}.save_hltb_cache") as mock_save,
patch(f"{PKG}.time.monotonic", side_effect=[0.0, 2.0]),
):
result = fetch_hltb_detail_missing([(440, "TF2"), (730, "CS")])
assert result == 1
mock_save.assert_called_once()
def test_restores_prior_hours_when_not_refound(self) -> None:
"""Hours are restored when re-fetch finds nothing for the game."""
saved: dict[int, float] = {}
def capture_save(
cache: dict[int, float],
_polls: object,
_extras: object = None,
) -> None:
saved.update(cache)
with (
patch(f"{PKG}.load_hltb_rush_cache", return_value={}),
patch(f"{PKG}.load_hltb_cache", return_value={730: 20.0}),
patch(f"{PKG}.load_hltb_polls_cache", return_value={}),
patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}),
patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}),
patch(f"{PKG}.load_hltb_game_id_cache", return_value={}),
patch(f"{PKG}.fetch_hltb_times"), # no-op, cache stays empty
patch(f"{PKG}.save_hltb_cache", side_effect=capture_save),
patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]),
):
fetch_hltb_detail_missing([(730, "CS")])
assert saved[730] == 20.0
def test_does_not_restore_when_refound(self) -> None:
"""Prior hours are NOT restored when re-fetch successfully finds game."""
def add_hours_and_rush(
_games: object,
cache: dict[int, float] | None = None,
polls: dict[int, int] | None = None,
progress_cb: object = None,
extras: _HLTBExtras | None = None,
) -> list[object]:
if cache is not None:
cache[730] = 30.0
if extras is not None:
extras.rush[730] = 12.0
return []
saved: dict[int, float] = {}
def capture_save(
cache: dict[int, float],
_polls: object,
_extras: object = None,
) -> None:
saved.update(cache)
with (
patch(f"{PKG}.load_hltb_rush_cache", return_value={}),
patch(f"{PKG}.load_hltb_cache", return_value={730: 20.0}),
patch(f"{PKG}.load_hltb_polls_cache", return_value={}),
patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}),
patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}),
patch(f"{PKG}.load_hltb_game_id_cache", return_value={}),
patch(f"{PKG}.fetch_hltb_times", side_effect=add_hours_and_rush),
patch(f"{PKG}.save_hltb_cache", side_effect=capture_save),
patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]),
):
result = fetch_hltb_detail_missing([(730, "CS")])
assert result == 1
assert saved[730] == 30.0
def test_zero_elapsed_rate(self) -> None:
"""Covers the elapsed == 0 branch in the rate calculation."""
with (
patch(f"{PKG}.load_hltb_rush_cache", return_value={}),
patch(f"{PKG}.load_hltb_cache", return_value={}),
patch(f"{PKG}.load_hltb_polls_cache", return_value={}),
patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}),
patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}),
patch(f"{PKG}.load_hltb_game_id_cache", return_value={}),
patch(f"{PKG}.fetch_hltb_times"),
patch(f"{PKG}.save_hltb_cache"),
patch(f"{PKG}.time.monotonic", side_effect=[5.0, 5.0]),
):
result = fetch_hltb_detail_missing([(730, "CS")])
assert result == 0

20
steam_backlog_enforcer/tests/test_hltb_search.py
View File

@ -84,6 +84,26 @@ class TestSearchOne:
assert result is not None
assert result.app_id == 440
def test_found_without_game_id(self) -> None:
"""Found result with hltb_game_id=0 does not populate ctx.hltb_game_id."""
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2",
"game_alias": "",
"comp_100": 180000,
"game_id": 0,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
assert 440 not in ctx.hltb_game_id
def test_not_found(self) -> None:
resp = _FakeResponse(200, {"data": []})
ctx = _make_ctx(_make_session(resp))

18
steam_backlog_enforcer/tests/test_hltb_search_part2.py
View File

@ -14,6 +14,7 @@ from python_pkg.steam_backlog_enforcer._hltb_detail import (
_parse_game_page,
)
from python_pkg.steam_backlog_enforcer._hltb_search import (
_build_search_variants,
_fetch_batch,
_pick_best_hltb_entry,
)
@ -305,3 +306,20 @@ class TestExtractLeisureHours:
"relationships": "not-a-list",
}
assert _extract_leisure_hours(data) == 1.0
class TestBuildSearchVariants:
"""Tests for _build_search_variants."""
def test_subtitle_with_edition_strips_edition_from_subtitle_part(self) -> None:
# "Rocksmith 2014 Edition - Remastered" → no_subtitle = "Rocksmith 2014 Edition"
# (which != base), so lines 201-202 also add "Rocksmith" and "Rocksmith 2014"
variants = _build_search_variants("Rocksmith 2014 Edition - Remastered")
assert "Rocksmith 2014 Edition" in variants
assert "Rocksmith 2014" in variants
assert "Rocksmith" in variants
def test_no_subtitle_skips_edition_strip(self) -> None:
# No " - " → no_subtitle == base → lines 201-202 are not executed
variants = _build_search_variants("Portal 2")
assert "Portal 2" in variants

36
steam_backlog_enforcer/tests/test_polls_tracking.py
View File

@ -9,8 +9,10 @@ from unittest.mock import patch
from python_pkg.steam_backlog_enforcer import _cmd_done
from python_pkg.steam_backlog_enforcer._hltb_types import (
HLTBResult,
_HLTBExtras,
load_hltb_cache,
load_hltb_count_comp_cache,
load_hltb_game_id_cache,
load_hltb_polls_cache,
save_hltb_cache,
)
@ -67,9 +69,18 @@ class TestCacheSchema:
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})
save_hltb_cache({440: 10.5}, {440: 7}, _HLTBExtras(count_comp={440: 20}))
data = json.loads(cache_file.read_text(encoding="utf-8"))
assert data == {"440": {"hours": 10.5, "polls": 7, "count_comp": 20}}
assert data == {
"440": {
"hours": 10.5,
"polls": 7,
"count_comp": 20,
"rush_hours": -1,
"leisure_100h": -1,
"hltb_game_id": 0,
}
}
def test_save_without_polls_defaults_zero(self, tmp_path: Path) -> None:
cache_file = tmp_path / "hltb_cache.json"
@ -79,7 +90,26 @@ class TestCacheSchema:
):
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}}
assert data == {
"440": {
"hours": 10.5,
"polls": 0,
"count_comp": 0,
"rush_hours": -1,
"leisure_100h": -1,
"hltb_game_id": 0,
}
}
def test_load_game_id_cache(self, tmp_path: Path) -> None:
"""load_hltb_game_id_cache returns the hltb_game_id portion of the cache."""
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}, extras=_HLTBExtras(hltb_game_id={440: 99}))
assert load_hltb_game_id_cache() == {440: 99}
class TestHltbResultPolls:

21
steam_backlog_enforcer/tests/test_protondb.py
View File

@ -222,7 +222,7 @@ class TestFetchOne:
sem = asyncio.Semaphore(1)
result = asyncio.run(_fetch_one(mock_session, sem, 440))
assert result.tier == ""
assert result is None
class TestFetchBatch:
@ -239,6 +239,25 @@ class TestFetchBatch:
assert len(result) == 1
assert result[0].tier == "gold"
def test_filters_none_results(self) -> None:
"""Network failures (None) are filtered out of the batch result."""
rating = ProtonDBRating(app_id=440, tier="gold")
async def mock_fetch_one(
_session: aiohttp.ClientSession,
_sem: asyncio.Semaphore,
app_id: int,
) -> ProtonDBRating | None:
return rating if app_id == 440 else None
with patch(
"python_pkg.steam_backlog_enforcer.protondb._fetch_one",
side_effect=mock_fetch_one,
):
result = asyncio.run(_fetch_batch([440, 999]))
assert len(result) == 1
assert result[0].app_id == 440
class TestFetchProtondbRatings:
"""Tests for fetch_protondb_ratings."""

11
steam_backlog_enforcer/tests/test_scanning_part3.py
View File

@ -401,3 +401,14 @@ class TestPickNextGameSequential:
pick_next_game([g1], state, config, on_select=lambda _g: True)
assert state.current_app_id is None
assert any("No assignable games" in line for line in echoed)
def test_enforcement_started_at_not_overwritten_when_set(self) -> None:
"""enforcement_started_at is preserved when already populated."""
g1 = _game(app_id=1, name="G1", hours=5.0)
config = Config(steam_api_key="k", steam_id="i")
existing_ts = "2024-01-01T00:00:00+00:00"
state = State(enforcement_started_at=existing_ts)
with self._common_patches([]):
pick_next_game([g1], state, config, on_select=lambda _g: True)
assert state.current_app_id == 1
assert state.enforcement_started_at == existing_ts

701
steam_backlog_enforcer/tests/test_stats.py Normal file
View File

@ -0,0 +1,701 @@
"""Tests for _stats module — 100% branch coverage."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from python_pkg.steam_backlog_enforcer._stats import (
_ensure_rush_data,
_filter_qualifying_games,
_format_completion_date,
_GameTimes,
_print_pace_scenario,
_print_scenario,
_print_worst_example,
_sum_hours,
cmd_stats,
)
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.steam_api import GameInfo
_PKG = "python_pkg.steam_backlog_enforcer._stats"
def _game(
app_id: int = 1,
name: str = "G",
hours: float = 10.0,
total: int = 10,
unlocked: int = 0,
) -> GameInfo:
return GameInfo(
app_id=app_id,
name=name,
total_achievements=total,
unlocked_achievements=unlocked,
playtime_minutes=60,
completionist_hours=hours,
comp_100_count=5,
count_comp=20,
)
def _unplayable_rating(app_id: int) -> ProtonDBRating:
return ProtonDBRating(app_id=app_id, tier="borked")
class TestFilterQualifyingGames:
"""Tests for _filter_qualifying_games."""
def _run(
self,
games: list[GameInfo],
state: State,
rush_cache: dict[int, float] | None = None,
leisure_cache: dict[int, float] | None = None,
game_id_cache: dict[int, int] | None = None,
) -> tuple[list[_GameTimes], int, int, int]:
with (
patch(f"{_PKG}.load_hltb_rush_cache", return_value=rush_cache or {}),
patch(
f"{_PKG}.load_hltb_leisure_100h_cache",
return_value=leisure_cache or {},
),
patch(
f"{_PKG}.load_hltb_game_id_cache",
return_value=game_id_cache or {},
),
patch(f"{_PKG}._apply_cached_confidence_to_candidates"),
patch(f"{_PKG}._refresh_candidate_confidence_batch"),
patch(f"{_PKG}._confidence_fail_reasons", return_value=[]),
patch(f"{_PKG}.fetch_protondb_ratings", return_value={}),
):
return _filter_qualifying_games(games, state)
def test_current_app_id_excluded(self) -> None:
state = State(current_app_id=1)
g1 = _game(app_id=1)
g2 = _game(app_id=2)
qualified, _, _, _ = self._run([g1, g2], state)
ids = [e.game.app_id for e in qualified]
assert 1 not in ids
assert 2 in ids
def test_no_current_app_id_branch(self) -> None:
"""current_app_id is None — the exclude.add branch is not taken."""
state = State(current_app_id=None)
g = _game(app_id=3)
qualified, _, _, _ = self._run([g], state)
assert len(qualified) == 1
def test_finished_app_ids_excluded(self) -> None:
state = State()
state.finished_app_ids = [1]
g1 = _game(app_id=1)
g2 = _game(app_id=2)
qualified, _, _, _ = self._run([g1, g2], state)
assert all(e.game.app_id != 1 for e in qualified)
def test_complete_games_excluded(self) -> None:
"""Games where is_complete is True are excluded from candidates."""
state = State()
complete = _game(app_id=1, total=5, unlocked=5)
incomplete = _game(app_id=2, total=5, unlocked=0)
qualified, _, _, _ = self._run([complete, incomplete], state)
assert len(qualified) == 1
assert qualified[0].game.app_id == 2
def test_low_confidence_counts_hltb_skipped(self) -> None:
state = State()
g = _game(app_id=1)
with (
patch(f"{_PKG}.load_hltb_rush_cache", return_value={}),
patch(f"{_PKG}.load_hltb_leisure_100h_cache", return_value={}),
patch(f"{_PKG}.load_hltb_game_id_cache", return_value={}),
patch(f"{_PKG}._apply_cached_confidence_to_candidates"),
patch(f"{_PKG}._refresh_candidate_confidence_batch"),
patch(f"{_PKG}._confidence_fail_reasons", return_value=["low"]),
patch(f"{_PKG}.fetch_protondb_ratings", return_value={}),
):
qualified, hltb_skip, _, _ = _filter_qualifying_games([g], state)
assert hltb_skip == 1
assert len(qualified) == 0
def test_no_candidates_skips_protondb_call(self) -> None:
"""When confidence filters all out, fetch_protondb_ratings is not called."""
state = State()
g = _game(app_id=1)
with (
patch(f"{_PKG}.load_hltb_rush_cache", return_value={}),
patch(f"{_PKG}.load_hltb_leisure_100h_cache", return_value={}),
patch(f"{_PKG}.load_hltb_game_id_cache", return_value={}),
patch(f"{_PKG}._apply_cached_confidence_to_candidates"),
patch(f"{_PKG}._refresh_candidate_confidence_batch"),
patch(f"{_PKG}._confidence_fail_reasons", return_value=["low"]),
patch(f"{_PKG}.fetch_protondb_ratings") as mock_proton,
):
_filter_qualifying_games([g], state)
mock_proton.assert_not_called()
def test_unplayable_rating_counts_linux_skipped(self) -> None:
state = State()
g = _game(app_id=1)
ratings = {1: _unplayable_rating(1)}
with (
patch(f"{_PKG}.load_hltb_rush_cache", return_value={}),
patch(f"{_PKG}.load_hltb_leisure_100h_cache", return_value={}),
patch(f"{_PKG}.load_hltb_game_id_cache", return_value={}),
patch(f"{_PKG}._apply_cached_confidence_to_candidates"),
patch(f"{_PKG}._refresh_candidate_confidence_batch"),
patch(f"{_PKG}._confidence_fail_reasons", return_value=[]),
patch(f"{_PKG}.fetch_protondb_ratings", return_value=ratings),
):
qualified, _, linux_skip, _ = _filter_qualifying_games([g], state)
assert linux_skip == 1
assert len(qualified) == 0
def test_no_data_counts_no_data_skipped(self) -> None:
"""Game with all -1 hours is counted as no_data_skipped."""
state = State()
g = _game(app_id=1, hours=-1.0)
qualified, _, _, no_data_skip = self._run([g], state)
assert no_data_skip == 1
assert len(qualified) == 0
def test_worst_hours_positive_when_completionist_hours_positive(self) -> None:
state = State()
g = _game(app_id=1, hours=25.0)
qualified, _, _, _ = self._run([g], state, rush_cache={1: 10.0})
assert qualified[0].worst_hours == 25.0
def test_worst_hours_from_leisure_when_completionist_zero(self) -> None:
"""worst_hours falls back to leisure_100h when completionist_hours is zero."""
state = State()
g = _game(app_id=1, hours=0.0)
qualified, _, _, _ = self._run(
[g], state, rush_cache={1: 5.0}, leisure_cache={1: 6.0}
)
assert qualified[0].worst_hours == 6.0
def test_worst_hours_is_max_when_leisure_exceeds_completionist(self) -> None:
"""worst_hours is max(completionist, leisure_100h) when leisure is higher."""
state = State()
g = _game(app_id=1, hours=25.0)
qualified, _, _, _ = self._run(
[g], state, rush_cache={1: 10.0}, leisure_cache={1: 40.0}
)
assert qualified[0].worst_hours == 40.0
def test_worst_hours_negative_when_all_zero(self) -> None:
"""worst_hours = -1 when both completionist_hours and leisure_100h are zero."""
state = State()
g = _game(app_id=1, hours=0.0)
qualified, _, _, _ = self._run([g], state, rush_cache={1: 5.0})
assert qualified[0].worst_hours == -1
def test_rush_and_leisure_from_cache(self) -> None:
state = State()
g = _game(app_id=1, hours=30.0)
qualified, _, _, _ = self._run(
[g], state, rush_cache={1: 12.0}, leisure_cache={1: 40.0}
)
assert qualified[0].rush_hours == 12.0
assert qualified[0].leisure_100h == 40.0
def test_missing_cache_entry_defaults_to_minus_one(self) -> None:
state = State()
g = _game(app_id=1, hours=20.0)
qualified, _, _, _ = self._run([g], state)
assert qualified[0].rush_hours == -1
assert qualified[0].leisure_100h == -1
def test_only_rush_nonzero_qualifies(self) -> None:
"""Game qualifies if only rush_hours is positive (worst <= 0, leisure <= 0)."""
state = State()
g = _game(app_id=1, hours=-1.0)
qualified, _, _, no_data_skip = self._run([g], state, rush_cache={1: 8.0})
assert no_data_skip == 0
assert len(qualified) == 1
def test_game_id_populated_from_cache(self) -> None:
"""hltb_game_id is taken from game_id_cache."""
state = State()
g = _game(app_id=1, hours=20.0)
qualified, _, _, _ = self._run([g], state, game_id_cache={1: 57514})
assert qualified[0].hltb_game_id == 57514
def test_game_id_defaults_to_zero_when_not_in_cache(self) -> None:
"""hltb_game_id defaults to 0 when not in cache."""
state = State()
g = _game(app_id=1, hours=20.0)
qualified, _, _, _ = self._run([g], state)
assert qualified[0].hltb_game_id == 0
class TestSumHours:
"""Tests for _sum_hours."""
def _make_entry(self, worst: float, rush: float, leisure: float) -> _GameTimes:
return _GameTimes(
game=_game(), worst_hours=worst, rush_hours=rush, leisure_100h=leisure
)
def test_empty_list(self) -> None:
total, missing = _sum_hours([], "worst_hours")
assert total == 0.0
assert missing == 0
def test_all_positive(self) -> None:
entries = [
self._make_entry(10.0, 8.0, 12.0),
self._make_entry(20.0, 15.0, 25.0),
]
total, missing = _sum_hours(entries, "worst_hours")
assert total == 30.0
assert missing == 0
def test_some_negative(self) -> None:
entries = [
self._make_entry(10.0, -1.0, 12.0),
self._make_entry(-1.0, 8.0, 25.0),
]
total, missing = _sum_hours(entries, "worst_hours")
assert total == 10.0
assert missing == 1
def test_all_negative(self) -> None:
entries = [self._make_entry(-1.0, -1.0, -1.0)]
total, missing = _sum_hours(entries, "rush_hours")
assert total == 0.0
assert missing == 1
class TestFormatCompletionDate:
"""Tests for _format_completion_date."""
def test_zero_hours_returns_na(self) -> None:
assert _format_completion_date(0.0, 4.0) == "N/A"
def test_negative_hours_returns_na(self) -> None:
assert _format_completion_date(-5.0, 4.0) == "N/A"
def test_zero_daily_hours_returns_na(self) -> None:
assert _format_completion_date(100.0, 0.0) == "N/A"
def test_negative_daily_hours_returns_na(self) -> None:
assert _format_completion_date(100.0, -1.0) == "N/A"
def test_normal_returns_days_and_date(self) -> None:
result = _format_completion_date(40.0, 4.0)
# 40 / 4 = 10 days
assert result.startswith("10 days (")
assert ")" in result
class TestPrintScenario:
"""Tests for _print_scenario."""
def test_no_data_prints_no_data_message(self) -> None:
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_scenario("2. RUSH", 0.0, 0, 5)
assert any("No data available" in s for s in echoed)
def test_with_data_no_missing(self) -> None:
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_scenario("2. RUSH", 100.0, 0, 5)
assert any("Total:" in s for s in echoed)
assert not any("had no data" in s for s in echoed)
def test_with_data_and_missing(self) -> None:
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_scenario("2. RUSH", 100.0, 2, 5)
assert any("had no data" in s for s in echoed)
class TestPrintPaceScenario:
"""Tests for _print_pace_scenario."""
def test_no_start_date(self) -> None:
state = State()
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_pace_scenario(state, 10, 0)
assert any("No start date recorded" in s for s in echoed)
def test_invalid_start_date(self) -> None:
state = State(enforcement_started_at="not-a-date")
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_pace_scenario(state, 10, 0)
assert any("Invalid enforcement_started_at" in s for s in echoed)
def test_no_games_finished(self) -> None:
started = datetime.now(timezone.utc) - timedelta(days=30)
state = State(enforcement_started_at=started.isoformat())
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_pace_scenario(state, 10, 0)
assert any("No games finished yet" in s for s in echoed)
def test_normal_pace(self) -> None:
started = datetime.now(timezone.utc) - timedelta(days=60)
state = State(enforcement_started_at=started.isoformat())
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_pace_scenario(state, 5, 3)
assert any("Pace:" in s for s in echoed)
assert any("Est. complete:" in s for s in echoed)
class TestCmdStats:
"""Tests for cmd_stats."""
def _config(self) -> Config:
return Config(steam_api_key="k", steam_id="i")
def test_no_snapshot(self) -> None:
echoed: list[str] = []
state = State()
with (
patch(f"{_PKG}.load_snapshot", return_value=None),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
):
cmd_stats(self._config(), state)
assert any("No snapshot found" in s for s in echoed)
def _snapshot_game(self, app_id: int = 1, hours: float = 20.0) -> dict[str, object]:
return {
"app_id": app_id,
"name": f"Game{app_id}",
"total_achievements": 10,
"unlocked_achievements": 0,
"playtime_minutes": 60,
"completionist_hours": hours,
"comp_100_count": 5,
"count_comp": 20,
}
def _run_cmd_stats(
self,
state: State,
hltb_skip: int = 0,
linux_skip: int = 0,
no_data_skip: int = 0,
) -> list[str]:
snapshot = [self._snapshot_game()]
game = GameInfo.from_snapshot(snapshot[0])
entry = _GameTimes(
game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0
)
echoed: list[str] = []
with (
patch(f"{_PKG}.load_snapshot", return_value=snapshot),
patch(
f"{_PKG}._filter_qualifying_games",
return_value=([entry], hltb_skip, linux_skip, no_data_skip),
),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"),
):
cmd_stats(self._config(), state)
return echoed
def test_with_no_current_game(self) -> None:
state = State()
echoed = self._run_cmd_stats(state)
assert any("Qualifying games" in s for s in echoed)
assert not any("Current game:" in s for s in echoed)
def test_with_current_game(self) -> None:
state = State(current_app_id=42, current_game_name="Hollow Knight")
echoed = self._run_cmd_stats(state)
assert any("Current game:" in s and "Hollow Knight" in s for s in echoed)
def test_hltb_skipped_shown(self) -> None:
state = State()
echoed = self._run_cmd_stats(state, hltb_skip=3)
assert any("HLTB-skipped" in s for s in echoed)
def test_linux_skipped_shown(self) -> None:
state = State()
echoed = self._run_cmd_stats(state, linux_skip=2)
assert any("Linux-skipped" in s for s in echoed)
def test_no_data_skipped_shown(self) -> None:
state = State()
echoed = self._run_cmd_stats(state, no_data_skip=1)
assert any("No-data-skipped" in s for s in echoed)
def test_zero_skips_not_shown(self) -> None:
state = State()
echoed = self._run_cmd_stats(state)
assert not any("HLTB-skipped" in s for s in echoed)
assert not any("Linux-skipped" in s for s in echoed)
assert not any("No-data-skipped" in s for s in echoed)
def test_finished_games_count_uses_snapshot_complete(self) -> None:
"""'Finished games' count uses snapshot is_complete, not finished_app_ids."""
state = State()
# finished_app_ids has 1 entry, but snapshot has 2 complete games — count = 2.
state.finished_app_ids = [99]
snapshot_complete = {
**self._snapshot_game(app_id=2),
"unlocked_achievements": 10,
}
snapshot = [self._snapshot_game(app_id=1), snapshot_complete]
game = GameInfo.from_snapshot(self._snapshot_game())
entry = _GameTimes(
game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0
)
echoed: list[str] = []
with (
patch(f"{_PKG}.load_snapshot", return_value=snapshot),
patch(
f"{_PKG}._filter_qualifying_games",
return_value=([entry], 0, 0, 0),
),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"),
):
cmd_stats(self._config(), state)
assert any("Finished games" in s and "1" in s for s in echoed)
def test_detail_data_complete_message_shown(self) -> None:
"""'Detail data: ...' shown when all qualifying games have rush hours."""
state = State()
echoed = self._run_cmd_stats(state)
# entry has rush_hours=15.0 > 0, so missing_rush_final == 0 and total_q == 1
assert any("Detail data" in s for s in echoed)
def test_note_missing_rush_shown_when_rush_absent(self) -> None:
"""'Note: X games still missing...' shown when rush_hours <= 0 after fetch."""
state = State()
snapshot = [self._snapshot_game()]
game = GameInfo.from_snapshot(snapshot[0])
entry = _GameTimes(
game=game, worst_hours=20.0, rush_hours=-1.0, leisure_100h=-1.0
)
echoed: list[str] = []
with (
patch(f"{_PKG}.load_snapshot", return_value=snapshot),
patch(
f"{_PKG}._filter_qualifying_games",
return_value=([entry], 0, 0, 0),
),
patch(f"{_PKG}._ensure_rush_data", return_value=False),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"),
patch(f"{_PKG}._print_worst_example"),
):
cmd_stats(self._config(), state)
assert any("still missing" in s for s in echoed)
def test_no_detail_message_when_no_qualifying_games(self) -> None:
"""Neither 'Note' nor 'Detail data' shown when qualified list is empty."""
state = State()
snapshot = [self._snapshot_game()]
echoed: list[str] = []
with (
patch(f"{_PKG}.load_snapshot", return_value=snapshot),
patch(
f"{_PKG}._filter_qualifying_games",
return_value=([], 0, 0, 0),
),
patch(f"{_PKG}._ensure_rush_data", return_value=False),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"),
patch(f"{_PKG}._print_worst_example"),
):
cmd_stats(self._config(), state)
assert not any("Detail data" in s for s in echoed)
assert not any("still missing" in s for s in echoed)
def test_refilter_called_when_ensure_rush_data_returns_true(self) -> None:
"""_filter_qualifying_games called twice when _ensure_rush_data returns True."""
state = State()
snapshot = [self._snapshot_game()]
game = GameInfo.from_snapshot(snapshot[0])
entry = _GameTimes(
game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0
)
filter_calls: list[int] = []
def count_filter(
_games: object, _state: object
) -> tuple[list[_GameTimes], int, int, int]:
filter_calls.append(1)
return [entry], 0, 0, 0
with (
patch(f"{_PKG}.load_snapshot", return_value=snapshot),
patch(f"{_PKG}._filter_qualifying_games", side_effect=count_filter),
patch(f"{_PKG}._ensure_rush_data", return_value=True),
patch(f"{_PKG}._echo"),
patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"),
patch(f"{_PKG}._print_worst_example"),
):
cmd_stats(self._config(), state)
assert len(filter_calls) == 2
def test_games_done_passed_to_pace_from_snapshot_complete(self) -> None:
"""_print_pace_scenario receives is_complete count from snapshot."""
state = State()
# Snapshot: 1 complete game (unlocked=total=10), 1 incomplete.
snapshot_complete = {
**self._snapshot_game(app_id=2),
"unlocked_achievements": 10,
}
snapshot = [self._snapshot_game(app_id=1), snapshot_complete]
game = GameInfo.from_snapshot(self._snapshot_game())
entry = _GameTimes(
game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0
)
captured: dict[str, int] = {}
def capture_pace(_state: object, _remaining: object, games_done: int) -> None:
captured["games_done"] = games_done
with (
patch(f"{_PKG}.load_snapshot", return_value=snapshot),
patch(
f"{_PKG}._filter_qualifying_games",
return_value=([entry], 0, 0, 0),
),
patch(f"{_PKG}._echo"),
patch(f"{_PKG}._print_pace_scenario", side_effect=capture_pace),
patch(f"{_PKG}._print_scenario"),
patch(f"{_PKG}._print_worst_example"),
):
cmd_stats(self._config(), state)
assert captured["games_done"] == 1
class TestEnsureRushData:
"""Tests for _ensure_rush_data."""
def _entry(self, rush: float) -> _GameTimes:
return _GameTimes(
game=_game(), worst_hours=10.0, rush_hours=rush, leisure_100h=5.0
)
def test_empty_qualified_returns_false(self) -> None:
with patch(f"{_PKG}.fetch_hltb_detail_missing") as mock_fetch:
result = _ensure_rush_data([])
assert result is False
mock_fetch.assert_not_called()
def test_all_have_rush_returns_false(self) -> None:
entries = [self._entry(10.0), self._entry(5.0)]
with patch(f"{_PKG}.fetch_hltb_detail_missing") as mock_fetch:
result = _ensure_rush_data(entries)
assert result is False
mock_fetch.assert_not_called()
def test_missing_rush_fetches_and_returns_true(self) -> None:
entries = [self._entry(-1.0)]
with (
patch(f"{_PKG}.fetch_hltb_detail_missing") as mock_fetch,
patch(f"{_PKG}._echo"),
):
result = _ensure_rush_data(entries)
assert result is True
mock_fetch.assert_called_once()
class TestPrintWorstExample:
"""Tests for _print_worst_example."""
def test_empty_list_does_nothing(self) -> None:
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_worst_example([])
assert echoed == []
def test_example_with_rush_and_leisure(self) -> None:
entry = _GameTimes(
game=_game(name="Portal"),
worst_hours=15.0,
rush_hours=5.0,
leisure_100h=20.0,
)
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_worst_example([entry])
assert any("Portal" in s for s in echoed)
assert any("Rush" in s for s in echoed)
assert any("Leisure" in s for s in echoed)
def test_example_without_rush(self) -> None:
entry = _GameTimes(
game=_game(name="X"), worst_hours=15.0, rush_hours=-1.0, leisure_100h=20.0
)
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_worst_example([entry])
assert not any("Rush" in s for s in echoed)
assert any("Leisure" in s for s in echoed)
def test_example_without_leisure(self) -> None:
entry = _GameTimes(
game=_game(name="Y"), worst_hours=15.0, rush_hours=5.0, leisure_100h=-1.0
)
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_worst_example([entry])
assert any("Rush" in s for s in echoed)
assert not any("Leisure" in s for s in echoed)
def test_hltb_search_url_shown_when_no_game_id(self) -> None:
"""Falls back to search URL when hltb_game_id is 0."""
entry = _GameTimes(
game=_game(name="Portal 2"),
worst_hours=15.0,
rush_hours=-1.0,
leisure_100h=-1.0,
)
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_worst_example([entry])
assert any("howlongtobeat.com" in s and "Portal+2" in s for s in echoed)
def test_hltb_direct_link_shown_when_game_id_known(self) -> None:
"""Direct HLTB game link shown when hltb_game_id is populated."""
entry = _GameTimes(
game=_game(name="Devil May Cry 5"),
worst_hours=186.0,
rush_hours=50.0,
leisure_100h=186.0,
hltb_game_id=57514,
)
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_worst_example([entry])
assert any("howlongtobeat.com/game/57514" in s for s in echoed)
assert not any("?q=" in s for s in echoed)
def test_entries_with_zero_worst_hours_excluded_from_examples(self) -> None:
"""Games with worst_hours <= 0 are not selected as the example."""
bad = _GameTimes(
game=_game(name="Skip"), worst_hours=0.0, rush_hours=-1.0, leisure_100h=-1.0
)
good = _GameTimes(
game=_game(name="Pick"),
worst_hours=10.0,
rush_hours=-1.0,
leisure_100h=-1.0,
)
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_worst_example([bad, good])
assert any("Pick" in s for s in echoed)
assert not any("Skip" in s for s in echoed)