feat: add pick-manual command with 2-week enforcement lock

User can now pick any owned game by Steam app_id via `pick-manual <id>`.
The script resolves the game name, asks for YES confirmation, then locks
all other commands for 14 days or until the game is 100% complete.
Post-assignment steps (uninstall others, install, hide library) mirror
the automatic pick flow. Lock is checked before every command including
add-exception. Also fixes pre-existing test failures in hltb, stats,
and web_dataset modules and adds 100% coverage for all changed code.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Krzysztof kuhy Rudnicki 2026-05-30 17:15:37 +02:00
parent 41deb90324
commit 7ac07c4b7a
19 changed files with 2025 additions and 64 deletions

3
CLAUDE.md Normal file
View File

@ -0,0 +1,3 @@
do NOT run tests unless specifically instructed to do so or before commiting
ALWAYS confirm that the feature you add / bug you fixed behaves as it should by running the program after your changes (not tests!) and inspecting output comparing it with what user wanted, after confirming by yourself ask user if the program behaves as they indendent
After running tests fix all coverage gaps and issues, do not ignore unless specifically instructed to do so

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
import logging import logging
@ -11,6 +12,7 @@ from urllib.parse import quote_plus
from steam_backlog_enforcer._hltb_types import ( from steam_backlog_enforcer._hltb_types import (
HLTB_BASE_URL, HLTB_BASE_URL,
_read_raw_cache,
load_hltb_cache, load_hltb_cache,
load_hltb_game_id_cache, load_hltb_game_id_cache,
load_hltb_leisure_100h_cache, load_hltb_leisure_100h_cache,
@ -21,14 +23,19 @@ from steam_backlog_enforcer._scanning_confidence import (
_confidence_fail_reasons, _confidence_fail_reasons,
_refresh_candidate_confidence_batch, _refresh_candidate_confidence_batch,
) )
from steam_backlog_enforcer.config import load_snapshot from steam_backlog_enforcer._web_dataset import (
PaceVsHLTB,
compute_pace_vs_hltb,
count_complete_since_start,
)
from steam_backlog_enforcer.config import SNAPSHOT_FILE, load_snapshot
from steam_backlog_enforcer.game_install import _echo from steam_backlog_enforcer.game_install import _echo
from steam_backlog_enforcer.hltb import fetch_hltb_detail_missing from steam_backlog_enforcer.hltb import fetch_hltb_detail_missing
from steam_backlog_enforcer.protondb import ( from steam_backlog_enforcer.protondb import (
ProtonDBRating, ProtonDBRating,
fetch_protondb_ratings, fetch_protondb_ratings,
) )
from steam_backlog_enforcer.steam_api import GameInfo from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient
if TYPE_CHECKING: if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.config import Config, State
@ -145,6 +152,27 @@ def _ensure_rush_data(qualified: list[_GameTimes]) -> bool:
return True return True
def _ensure_completed_rush_data(games: list[GameInfo]) -> bool:
"""Fetch rush/leisure detail for completed games used for pace calibration.
Completed games aren't processed by ``_ensure_rush_data`` (which only
handles incomplete qualifying games), so this separate pass fills in
their rush/leisure data for ``compute_pace_vs_hltb``.
Returns True when at least one new fetch was performed.
"""
pairs = [
(g.app_id, g.name) for g in games if g.is_complete and g.playtime_minutes > 0
]
if not pairs:
return False
_echo(
f"Fetching HLTB detail for {len(pairs)} completed games (pace calibration)..."
)
fetched = fetch_hltb_detail_missing(pairs)
return fetched > 0
def _print_worst_example(entries: list[_GameTimes]) -> None: def _print_worst_example(entries: list[_GameTimes]) -> None:
"""Print a randomly selected example from the worst-case qualified games.""" """Print a randomly selected example from the worst-case qualified games."""
examples = [e for e in entries if e.worst_hours > 0] examples = [e for e in entries if e.worst_hours > 0]
@ -157,8 +185,13 @@ def _print_worst_example(entries: list[_GameTimes]) -> None:
_echo(f" Rush: {example.rush_hours:.1f} h") _echo(f" Rush: {example.rush_hours:.1f} h")
if example.leisure_100h > 0: if example.leisure_100h > 0:
_echo(f" Leisure: {example.leisure_100h:.1f} h") _echo(f" Leisure: {example.leisure_100h:.1f} h")
if example.hltb_game_id > 0: hltb_game_id = example.hltb_game_id
_echo(f" HLTB: {HLTB_BASE_URL}/game/{example.hltb_game_id}") if hltb_game_id == 0:
# On-demand backfill: one search to get the HLTB game ID for this game.
fetch_hltb_detail_missing([(example.game.app_id, example.game.name)])
hltb_game_id = load_hltb_game_id_cache().get(example.game.app_id, 0)
if hltb_game_id > 0:
_echo(f" HLTB: {HLTB_BASE_URL}/game/{hltb_game_id}")
else: else:
_echo(f" HLTB: {_HLTB_SEARCH_BASE}{quote_plus(example.game.name)}") _echo(f" HLTB: {_HLTB_SEARCH_BASE}{quote_plus(example.game.name)}")
@ -215,10 +248,9 @@ def _print_scenario(
def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None: def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None:
"""Print the pace-based completion estimate. """Print the pace-based completion estimate.
``games_done`` should be the count of 100%-complete games in the library ``games_done`` must be the count of games completed ON OR AFTER
snapshot (``sum(1 for g in games if g.is_complete)``), not the enforcer's ``state.enforcement_started_at`` (use ``count_complete_since_start``).
own ``finished_app_ids`` list, which misses games completed outside the Pre-enforcement completions inflate the rate and are excluded.
enforcer flow.
""" """
_echo("\n 1. AT YOUR CURRENT PACE") _echo("\n 1. AT YOUR CURRENT PACE")
if not state.enforcement_started_at: if not state.enforcement_started_at:
@ -243,7 +275,9 @@ def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None:
rate = games_done / days_elapsed rate = games_done / days_elapsed
_echo(f" Started: {started.strftime('%Y-%m-%d')}") _echo(f" Started: {started.strftime('%Y-%m-%d')}")
_echo(f" Finished: {games_done} games in {days_elapsed} days") _echo(
f" Finished: {games_done} games in {days_elapsed} days (since enforcement start)"
)
_echo( _echo(
f" Pace: {rate:.4f} games/day (1 game every {1 / rate:.1f} days)" f" Pace: {rate:.4f} games/day (1 game every {1 / rate:.1f} days)"
) )
@ -254,17 +288,139 @@ def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None:
_echo(f" Est. complete: {days_to_go} days ({finish.strftime('%Y-%m-%d')})") _echo(f" Est. complete: {days_to_go} days ({finish.strftime('%Y-%m-%d')})")
def _print_player_speed_scenario(
pace: PaceVsHLTB | None,
rush_total: float,
leisure_total: float,
) -> None:
"""Print player pace vs HLTB averages and an extrapolated backlog estimate."""
_echo(f"\n{_LINE}")
_echo("\n 5. YOUR PLAY STYLE vs HLTB AVERAGES")
if pace is None or pace.calibration_count == 0:
_echo(" No calibration data available.")
_echo(
" Finish some games (100 % achievements) and re-run 'stats'"
" to enable this estimate."
)
return
_echo(f"\n Calibration games: {pace.calibration_count}")
if pace.ratio_vs_rush > 0:
_echo(f" vs Rush: {pace.ratio_vs_rush:.2f}x rush pace")
if pace.ratio_vs_leisure > 0:
_echo(f" vs Leisure: {pace.ratio_vs_leisure:.2f}x leisure pace")
if pace.interpolation_t != -1.0:
_echo(
f" Interpolation t: {pace.interpolation_t:.3f}"
" (0 = rush speed, 1 = leisure speed)"
)
style_labels = {
"faster_than_rush": "Faster than rush",
"rush_to_leisure": "Between rush and leisure",
"slower_than_leisure": "Slower than leisure",
"unknown": "Unknown",
}
style = style_labels.get(pace.player_style, pace.player_style)
_echo(f" Play style: {style}")
if pace.interpolation_t != -1.0 and rush_total > 0 and leisure_total > 0:
est = rush_total + pace.interpolation_t * (leisure_total - rush_total)
elif pace.ratio_vs_rush > 0 and rush_total > 0:
est = rush_total * pace.ratio_vs_rush
else:
est = -1.0
if est > 0:
_echo(f"\n Estimated backlog total at your pace: {est:,.1f} h")
for daily in _HOURS_PER_DAY_PRESETS:
estimate = _format_completion_date(est, daily)
_echo(f" @ {daily:.0f} h/day → {estimate}")
def _refresh_recently_played_completions(
games: list[GameInfo],
config: Config,
) -> list[GameInfo]:
"""Refresh achievement data for incomplete games played since last scan.
Makes 1 ``GetOwnedGames`` request + 1 ``GetPlayerAchievements`` per
recently-played incomplete game. Finds games newly completed since the
last ``scan`` without re-scanning the whole library.
Returns a new list with updated GameInfo objects for any game that was
played after the snapshot was written; all other games are unchanged.
"""
try:
snapshot_mtime = SNAPSHOT_FILE.stat().st_mtime
except OSError:
return games
from steam_backlog_enforcer.steam_api import SteamAPIError
try:
client = SteamAPIClient(config.steam_api_key, config.steam_id)
owned_raw = client.get_owned_games()
except SteamAPIError:
logger.debug("Steam API unavailable; skipping completion refresh.")
return games
last_played_map = {g["appid"]: g.get("rtime_last_played", 0) for g in owned_raw}
to_refresh = [
g
for g in games
if not g.is_complete and last_played_map.get(g.app_id, 0) > snapshot_mtime
]
if not to_refresh:
return games
_echo(
f"Refreshing {len(to_refresh)} recently-played game(s)"
" for up-to-date completion status..."
)
game_map = {g.app_id: g for g in games}
def _refresh_one(game: GameInfo) -> GameInfo:
achievements = client.get_achievement_details(game.app_id)
if not achievements:
return game
unlocked = sum(1 for a in achievements if a.achieved)
return GameInfo(
app_id=game.app_id,
name=game.name,
total_achievements=len(achievements),
unlocked_achievements=unlocked,
playtime_minutes=game.playtime_minutes,
achievements=achievements,
completionist_hours=game.completionist_hours,
comp_100_count=game.comp_100_count,
count_comp=game.count_comp,
)
with ThreadPoolExecutor(max_workers=20) as pool:
futures = {pool.submit(_refresh_one, g): g for g in to_refresh}
for future in as_completed(futures):
refreshed = future.result()
game_map[refreshed.app_id] = refreshed
return list(game_map.values())
def cmd_stats(_config: Config, state: State) -> None: def cmd_stats(_config: Config, state: State) -> None:
"""Display backlog completion-time statistics. """Display backlog completion-time statistics.
Filters games by the same HLTB-confidence and Linux-compatibility rules Filters games by the same HLTB-confidence and Linux-compatibility rules
used when picking the next game. Auto-fetches missing rush/leisure detail used when picking the next game. Auto-fetches missing rush/leisure detail
data before printing. Shows four scenarios: data before printing. Shows five scenarios:
1. At your current pace (games finished per day since enforcement started). 1. At your current pace (games finished per day since enforcement started).
2. Rush avg comp_100 + DLC completion time per HLTB. 2. Rush avg comp_100 + DLC completion time per HLTB.
3. Leisure comp_100_h (slowest 100 %) + DLC leisure per HLTB. 3. Leisure comp_100_h (slowest 100 %) + DLC leisure per HLTB.
4. Worst absolute maximum recorded time (any category) per HLTB. 4. Worst absolute maximum recorded time (any category) per HLTB.
5. Your play style extrapolated from completed-game calibration vs HLTB.
""" """
snapshot = load_snapshot() snapshot = load_snapshot()
if snapshot is None: if snapshot is None:
@ -272,9 +428,18 @@ def cmd_stats(_config: Config, state: State) -> None:
return return
games = [GameInfo.from_snapshot(d) for d in snapshot] games = [GameInfo.from_snapshot(d) for d in snapshot]
games = _refresh_recently_played_completions(games, _config)
# Count all 100%-achievement games in library (more accurate than # Count all 100%-achievement games in library (more accurate than
# finished_app_ids, which only tracks enforcer-assigned completions). # finished_app_ids, which only tracks enforcer-assigned completions).
games_done = sum(1 for g in games if g.is_complete) games_done = sum(1 for g in games if g.is_complete)
# Only count games completed on/after enforcement start for pace — pre-start
# completions are not representative of the enforcer period's throughput.
games_done_since_start = count_complete_since_start(
games, state.enforcement_started_at
)
# Ensure completed games have rush/leisure data for pace calibration.
_ensure_completed_rush_data(games)
qualified, hltb_skip, linux_skip, no_data_skip = _filter_qualifying_games( qualified, hltb_skip, linux_skip, no_data_skip = _filter_qualifying_games(
games, state games, state
@ -316,7 +481,7 @@ def cmd_stats(_config: Config, state: State) -> None:
_echo(f" Finished games: {games_done} (excluded from totals)") _echo(f" Finished games: {games_done} (excluded from totals)")
_echo(f"\n{_LINE}") _echo(f"\n{_LINE}")
_print_pace_scenario(state, total_q, games_done) _print_pace_scenario(state, total_q, games_done_since_start)
worst_total, worst_missing = _sum_hours(qualified, "worst_hours") worst_total, worst_missing = _sum_hours(qualified, "worst_hours")
rush_total, rush_missing = _sum_hours(qualified, "rush_hours") rush_total, rush_missing = _sum_hours(qualified, "rush_hours")
@ -347,4 +512,9 @@ def cmd_stats(_config: Config, state: State) -> None:
) )
_print_worst_example(qualified) _print_worst_example(qualified)
# Pace calibration uses the freshly-updated cache (both fetches above ran).
raw_cache = _read_raw_cache()
pace_vs_hltb = compute_pace_vs_hltb(games, raw_cache)
_print_player_speed_scenario(pace_vs_hltb, rush_total, leisure_total)
_echo(f"\n{_LINE}\n") _echo(f"\n{_LINE}\n")

View File

@ -67,6 +67,7 @@ class WebStateInfo:
current_app_id: int | None current_app_id: int | None
current_game_name: str current_game_name: str
games_done: int games_done: int
games_done_since_start: int
days_elapsed: int days_elapsed: int
enforcement_started_at: str enforcement_started_at: str
pace_games_per_day: float pace_games_per_day: float
@ -97,6 +98,31 @@ class DefaultSummary:
worst_total: float worst_total: float
@dataclass
class PaceVsHLTB:
"""Player pace calibrated against HLTB rush/leisure averages.
Derived from completed games that have HLTB detail data. All ratio /
interpolation fields use ``-1`` to mean "insufficient data", matching the
cache convention used elsewhere.
Fields:
calibration_count: number of completed games used for calibration.
ratio_vs_rush: actual_hours / rush_hours across calibration games.
ratio_vs_leisure: actual_hours / leisure_hours (-1 if no leisure data).
interpolation_t: position between rush (0.0) and leisure (1.0) speed.
Negative means faster than rush; >1 means slower than leisure.
-1 means insufficient data.
player_style: human-readable style label.
"""
calibration_count: int
ratio_vs_rush: float
ratio_vs_leisure: float
interpolation_t: float
player_style: str
@dataclass @dataclass
class WebDataset: class WebDataset:
"""Full payload served to the browser.""" """Full payload served to the browser."""
@ -105,6 +131,7 @@ class WebDataset:
state: WebStateInfo state: WebStateInfo
defaults: WebDefaults defaults: WebDefaults
default_summary: DefaultSummary default_summary: DefaultSummary
pace_vs_hltb: PaceVsHLTB | None
generated_at: str = field(default="") generated_at: str = field(default="")
@ -213,7 +240,38 @@ def _default_summary(rows: list[WebGame]) -> DefaultSummary:
) )
def _state_info(state: State, games_done: int) -> WebStateInfo: def count_complete_since_start(games: list[GameInfo], started_at: str) -> int:
"""Count complete games whose last achievement was unlocked on/after started_at.
Games with no achievement timestamp data are excluded their completion
date is unknown, and they were most likely finished before Steam began
recording unlock timestamps (i.e. before the enforcement period).
Returns 0 when started_at is empty or unparseable.
"""
if not started_at:
return 0
try:
started = datetime.fromisoformat(started_at)
except ValueError:
return 0
started_ts = int(started.timestamp())
count = 0
for game in games:
if not game.is_complete:
continue
achieved_times = [
a.unlock_time for a in game.achievements if a.achieved and a.unlock_time > 0
]
if not achieved_times:
continue
if max(achieved_times) >= started_ts:
count += 1
return count
def _state_info(
state: State, games_done: int, games_done_since_start: int
) -> WebStateInfo:
"""Build pace metadata, mirroring ``_print_pace_scenario`` inputs.""" """Build pace metadata, mirroring ``_print_pace_scenario`` inputs."""
days_elapsed = 0 days_elapsed = 0
pace = 0.0 pace = 0.0
@ -225,18 +283,109 @@ def _state_info(state: State, games_done: int) -> WebStateInfo:
if started is not None: if started is not None:
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
days_elapsed = max(1, (now - started).days) days_elapsed = max(1, (now - started).days)
if games_done > 0: if games_done_since_start > 0:
pace = round(games_done / days_elapsed, 4) pace = round(games_done_since_start / days_elapsed, 4)
return WebStateInfo( return WebStateInfo(
current_app_id=state.current_app_id, current_app_id=state.current_app_id,
current_game_name=state.current_game_name, current_game_name=state.current_game_name,
games_done=games_done, games_done=games_done,
games_done_since_start=games_done_since_start,
days_elapsed=days_elapsed, days_elapsed=days_elapsed,
enforcement_started_at=state.enforcement_started_at, enforcement_started_at=state.enforcement_started_at,
pace_games_per_day=pace, pace_games_per_day=pace,
) )
def _collect_calibration_pairs(
raw_games: list[GameInfo],
raw_cache: dict[int, dict[str, Any]],
) -> tuple[list[tuple[float, float]], list[tuple[float, float, float]]]:
"""Separate complete games into rush-only and rush+leisure sample sets."""
rush_pairs: list[tuple[float, float]] = []
both_pairs: list[tuple[float, float, float]] = []
for game in raw_games:
if not game.is_complete or game.playtime_minutes <= 0:
continue
entry = raw_cache.get(game.app_id, {})
rush = float(entry.get("rush_hours", -1))
leisure = float(entry.get("leisure_100h", -1))
actual = game.playtime_minutes / 60.0
if rush > 0:
rush_pairs.append((actual, rush))
if rush > 0 and leisure > 0:
both_pairs.append((actual, rush, leisure))
return rush_pairs, both_pairs
def _interpolate_from_both(
both_pairs: list[tuple[float, float, float]],
) -> tuple[float, float]:
"""Return (ratio_vs_leisure, interpolation_t) from (actual, rush, leisure) triples.
Returns -1.0 for interpolation_t when leisure <= rush (degenerate data).
"""
sum_actual = sum(p[0] for p in both_pairs)
sum_rush = sum(p[1] for p in both_pairs)
sum_leisure = sum(p[2] for p in both_pairs)
ratio_vs_leisure = round(sum_actual / sum_leisure, 3)
if sum_leisure > sum_rush:
t = round((sum_actual - sum_rush) / (sum_leisure - sum_rush), 3)
else:
t = -1.0
return ratio_vs_leisure, t
def _classify_player_style(interpolation_t: float, ratio_vs_rush: float) -> str:
"""Map calibration metrics to a player-style label."""
if interpolation_t != -1.0:
if interpolation_t < 0:
return "faster_than_rush"
if interpolation_t <= 1.0:
return "rush_to_leisure"
return "slower_than_leisure"
return "faster_than_rush" if ratio_vs_rush < 1.0 else "unknown"
def compute_pace_vs_hltb(
raw_games: list[GameInfo],
raw_cache: dict[int, dict[str, Any]],
) -> PaceVsHLTB | None:
"""Compute player pace relative to HLTB rush/leisure averages.
Uses completed games (100 % achievements, positive playtime) as calibration
samples. Steam playtime includes idle time, so ratios > 1 are expected for
most players.
Args:
raw_games: All games from the snapshot (completed + incomplete).
raw_cache: The full HLTB cache (from ``_read_raw_cache()``).
Returns:
A ``PaceVsHLTB`` when at least one completed game has rush data,
``None`` when there is no calibration data at all.
"""
rush_pairs, both_pairs = _collect_calibration_pairs(raw_games, raw_cache)
if not rush_pairs:
return None
ratio_vs_rush = round(
sum(p[0] for p in rush_pairs) / sum(p[1] for p in rush_pairs), 3
)
if both_pairs:
ratio_vs_leisure, interpolation_t = _interpolate_from_both(both_pairs)
else:
ratio_vs_leisure = -1.0
interpolation_t = -1.0
return PaceVsHLTB(
calibration_count=len(rush_pairs),
ratio_vs_rush=ratio_vs_rush,
ratio_vs_leisure=ratio_vs_leisure,
interpolation_t=interpolation_t,
player_style=_classify_player_style(interpolation_t, ratio_vs_rush),
)
def build_web_dataset(state: State) -> WebDataset: def build_web_dataset(state: State) -> WebDataset:
"""Build the full web dataset from on-disk caches (no network calls). """Build the full web dataset from on-disk caches (no network calls).
@ -253,6 +402,9 @@ def build_web_dataset(state: State) -> WebDataset:
[GameInfo.from_snapshot(d) for d in snapshot] if snapshot is not None else [] [GameInfo.from_snapshot(d) for d in snapshot] if snapshot is not None else []
) )
games_done = sum(1 for g in raw_games if g.is_complete) games_done = sum(1 for g in raw_games if g.is_complete)
games_done_since_start = count_complete_since_start(
raw_games, state.enforcement_started_at
)
exclude = set(state.finished_app_ids) exclude = set(state.finished_app_ids)
if state.current_app_id is not None: if state.current_app_id is not None:
@ -260,9 +412,12 @@ def build_web_dataset(state: State) -> WebDataset:
rows = _build_games(raw_games, exclude) rows = _build_games(raw_games, exclude)
raw_cache = _read_raw_cache()
pace_vs_hltb = compute_pace_vs_hltb(raw_games, raw_cache)
return WebDataset( return WebDataset(
games=rows, games=rows,
state=_state_info(state, games_done), state=_state_info(state, games_done, games_done_since_start),
defaults=WebDefaults( defaults=WebDefaults(
min_comp_100_polls=_MIN_COMP_100_POLLS, min_comp_100_polls=_MIN_COMP_100_POLLS,
min_count_comp=_MIN_COUNT_COMP, min_count_comp=_MIN_COUNT_COMP,
@ -271,6 +426,7 @@ def build_web_dataset(state: State) -> WebDataset:
hours_per_day_presets=list(HOURS_PER_DAY_PRESETS), hours_per_day_presets=list(HOURS_PER_DAY_PRESETS),
), ),
default_summary=_default_summary(rows), default_summary=_default_summary(rows),
pace_vs_hltb=pace_vs_hltb,
generated_at=datetime.now(timezone.utc).isoformat(), generated_at=datetime.now(timezone.utc).isoformat(),
) )

View File

@ -95,6 +95,10 @@ class State:
elapses. Populated when the user declines a freshly-picked game via the elapses. Populated when the user declines a freshly-picked game via the
interactive prompt in ``cmd_done``. interactive prompt in ``cmd_done``.
""" """
manual_pick_app_id: int | None = None
manual_pick_game_name: str = ""
manual_pick_started_at: str = ""
"""ISO-8601 UTC timestamp when the user manually locked in a game."""
def skip_for_days(self, app_id: int, days: int) -> None: def skip_for_days(self, app_id: int, days: int) -> None:
"""Mark ``app_id`` as skipped for ``days`` days from now (UTC).""" """Mark ``app_id`` as skipped for ``days`` days from now (UTC)."""

View File

@ -0,0 +1,19 @@
Add an option to pick a specific game manually (by providing its steam id)
Picking a game should work like this:
1. user invokes script with specific flag for picking a game manually
2. user provides game steam id (in testing use 489830)
3. Script shows what game it believes this id means (in this case it should show The Elder Scrolls V: Skyrim Special Edition)
4. user confirms that this is the game they want to pick and confirm that they will not be able to use the script for up to 2 weeks or until the game is completed (100% achievments)
When user picks a game manually this should override the current pick if it exists
After picking manually backlog enforcer should make a note of that and very aggresively disallow user to do anything else
for a period of 2 weeks or untill user completes a given game
Logic should be as follows:
1. backlog checks if a user picked game manually
a. if not -> continue as before
2. if yes check if the game is completed
a. if yes -> continue as before
3. if NOT show info that user picked a specific game manually and they have to finish it before using ANY OTHER functionality of backlog enforcer
test the functionality with 489830 (The Elder Scrolls V: Skyrim Special Edition)
as always first write full functionality confirm that it works alone and with the user and only AFTER that write tests and coverage and fix issues

View File

@ -280,9 +280,12 @@ def fetch_hltb_detail_missing(
) -> int: ) -> int:
"""Fetch HLTB detail (rush + leisure) for games that are missing it. """Fetch HLTB detail (rush + leisure) for games that are missing it.
Games already in the rush cache are skipped. For the rest, temporarily Also backfills ``hltb_game_id`` for any game that already has rush/leisure
removes them from the hours cache so ``fetch_hltb_times`` will visit their data but whose HLTB game ID was never stored (e.g. from an old cache).
detail pages. Restores prior hours for any game the re-fetch doesn't find. Games with both rush data and a game_id are skipped entirely. 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: Args:
games: list of (app_id, name) tuples to check. games: list of (app_id, name) tuples to check.
@ -292,7 +295,18 @@ def fetch_hltb_detail_missing(
Number of games that now have rush-hour data after the fetch. Number of games that now have rush-hour data after the fetch.
""" """
rush = load_hltb_rush_cache() rush = load_hltb_rush_cache()
missing = [(app_id, name) for app_id, name in games if rush.get(app_id, -1) <= 0] game_id_cache = load_hltb_game_id_cache()
missing_rush = [
(app_id, name) for app_id, name in games if rush.get(app_id, -1) <= 0
]
# Also re-search games that have rush data but no HLTB game ID yet so the
# direct URL can be shown in stats output.
missing_id_only = [
(app_id, name)
for app_id, name in games
if rush.get(app_id, -1) > 0 and game_id_cache.get(app_id, 0) == 0
]
missing = missing_rush + missing_id_only
if not missing: if not missing:
return 0 return 0
@ -302,7 +316,7 @@ def fetch_hltb_detail_missing(
count_comp=load_hltb_count_comp_cache(), count_comp=load_hltb_count_comp_cache(),
rush=rush, rush=rush,
leisure_100h=load_hltb_leisure_100h_cache(), leisure_100h=load_hltb_leisure_100h_cache(),
hltb_game_id=load_hltb_game_id_cache(), hltb_game_id=game_id_cache,
) )
# Remove from hours cache so fetch_hltb_times will visit the detail page. # Remove from hours cache so fetch_hltb_times will visit the detail page.
@ -310,10 +324,21 @@ def fetch_hltb_detail_missing(
for app_id, _ in missing: for app_id, _ in missing:
prior_hours[app_id] = cache.pop(app_id, -1.0) prior_hours[app_id] = cache.pop(app_id, -1.0)
logger.info( n_rush = len(missing_rush)
"Fetching HLTB detail for %d games missing rush/leisure data...", n_id = len(missing_id_only)
len(missing), if n_rush and n_id:
) logger.info(
"Fetching HLTB detail for %d games missing rush/leisure data"
" + %d games missing game ID...",
n_rush,
n_id,
)
elif n_rush:
logger.info(
"Fetching HLTB detail for %d games missing rush/leisure data...", n_rush
)
else:
logger.info("Backfilling HLTB game ID for %d game(s)...", n_id)
t0 = time.monotonic() t0 = time.monotonic()
fetch_hltb_times( fetch_hltb_times(
missing, missing,
@ -331,12 +356,12 @@ def fetch_hltb_detail_missing(
save_hltb_cache(cache, polls, extras) save_hltb_cache(cache, polls, extras)
fetched = sum(1 for app_id, _ in missing if extras.rush.get(app_id, -1) > 0) fetched = sum(1 for app_id, _ in missing_rush if extras.rush.get(app_id, -1) > 0)
rate = len(missing) / elapsed if elapsed > 0 else 0 rate = len(missing) / elapsed if elapsed > 0 else 0
logger.info( logger.info(
"HLTB detail fetch done: %d/%d got rush data in %.1fs (%.0f games/s)", "HLTB detail fetch done: %d/%d got rush data in %.1fs (%.0f games/s)",
fetched, fetched,
len(missing), len(missing_rush),
elapsed, elapsed,
rate, rate,
) )

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
from datetime import datetime, timedelta, timezone
import logging import logging
import sys import sys
import time import time
@ -45,7 +46,7 @@ from steam_backlog_enforcer.scanning import (
do_scan, do_scan,
pick_next_game, pick_next_game,
) )
from steam_backlog_enforcer.steam_api import GameInfo from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient, SteamAPIError
from steam_backlog_enforcer.store_blocker import ( from steam_backlog_enforcer.store_blocker import (
block_store, block_store,
is_store_blocked, is_store_blocked,
@ -65,6 +66,85 @@ logger = logging.getLogger(__name__)
_LIST_DISPLAY_LIMIT = 50 _LIST_DISPLAY_LIMIT = 50
_MIN_CLI_ARGS = 2 _MIN_CLI_ARGS = 2
# Days before the manual-pick lock automatically expires.
_MANUAL_LOCK_DAYS = 14
# Commands that remain usable while the manual pick lock is active.
# Principle: only what is needed to release the lock (done/check) or
# that cannot change the game assignment (status, enforce, setup, serve).
_MANUAL_LOCK_EXEMPT_COMMANDS = frozenset(
{"done", "check", "status", "enforce", "setup", "serve"}
)
# ──────────────────────────────────────────────────────────────
# Manual pick lock helpers
# ──────────────────────────────────────────────────────────────
def _is_manual_pick_locked(state: State) -> bool:
"""Return True if the manual-pick lock is currently in force."""
if state.manual_pick_app_id is None:
return False
# Lock released once the game appears in finished_app_ids.
if state.manual_pick_app_id in state.finished_app_ids:
return False
# Lock released after 14 days from the pick timestamp.
if state.manual_pick_started_at:
try:
started = datetime.fromisoformat(state.manual_pick_started_at)
deadline = started + timedelta(days=_MANUAL_LOCK_DAYS)
if datetime.now(timezone.utc) >= deadline:
return False
except ValueError:
pass
return True
def _show_manual_pick_lock_message(state: State) -> None:
"""Print the aggressive lock-active message to stdout."""
_echo("\n" + "=" * 60)
_echo(" *** MANUAL PICK LOCK ACTIVE ***")
_echo("=" * 60)
_echo(
f"\nYou manually picked: {state.manual_pick_game_name}"
f" (AppID={state.manual_pick_app_id})"
)
if state.manual_pick_started_at:
try:
started = datetime.fromisoformat(state.manual_pick_started_at)
deadline = started + timedelta(days=_MANUAL_LOCK_DAYS)
days_left = (deadline - datetime.now(timezone.utc)).days
_echo(f"Locked since: {started.strftime('%Y-%m-%d')}")
_echo(
f"Deadline: {deadline.strftime('%Y-%m-%d')}"
f" ({max(0, days_left)} day(s) remaining)"
)
except ValueError:
pass
_echo(
"\nYou CANNOT use any other feature until you finish this game"
"\n(100% achievements) or the 2-week deadline passes."
"\n\nTo release the lock: finish the game, then run 'done' or 'check'."
f"\n\nAllowed commands: {', '.join(sorted(_MANUAL_LOCK_EXEMPT_COMMANDS))}"
)
_echo("=" * 60 + "\n")
def _enforce_manual_pick_lock(command: str, state: State) -> None:
"""Exit with a lock message if command is blocked by the manual pick."""
if not _is_manual_pick_locked(state):
return
if command in _MANUAL_LOCK_EXEMPT_COMMANDS:
return
_show_manual_pick_lock_message(state)
sys.exit(1)
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
# CLI commands # CLI commands
@ -94,6 +174,9 @@ def cmd_status(_config: Config, state: State) -> None:
is_assigned_installed = any(aid == state.current_app_id for aid, _ in installed) is_assigned_installed = any(aid == state.current_app_id for aid, _ in installed)
_echo(f"Assigned game installed: {is_assigned_installed}") _echo(f"Assigned game installed: {is_assigned_installed}")
if _is_manual_pick_locked(state):
_echo("\n[MANUAL PICK LOCK is active — most commands are blocked]")
def cmd_list(_config: Config, state: State) -> None: def cmd_list(_config: Config, state: State) -> None:
"""List games from the last snapshot.""" """List games from the last snapshot."""
@ -180,6 +263,9 @@ def cmd_reset(config: Config, state: State) -> None:
state.current_app_id = None state.current_app_id = None
state.current_game_name = "" state.current_game_name = ""
state.finished_app_ids = [] state.finished_app_ids = []
state.manual_pick_app_id = None
state.manual_pick_game_name = ""
state.manual_pick_started_at = ""
state.save() state.save()
_echo("State reset. Store unblocked.") _echo("State reset. Store unblocked.")
@ -387,6 +473,106 @@ def cmd_serve(_config: Config, _state: State) -> None:
serve() serve()
def _resolve_game_name(config: Config, app_id: int) -> str | None:
"""Look up a game name by app_id, checking snapshot then Steam API.
Returns the game name, or None if not found.
"""
# Fast path: snapshot already on disk.
snapshot = load_snapshot()
if snapshot:
for entry in snapshot:
if entry.get("app_id") == app_id:
return str(entry["name"])
# Slower path: owned games API.
try:
client = SteamAPIClient(config.steam_api_key, config.steam_id)
owned = client.get_owned_games()
for g in owned:
if g.get("appid") == app_id:
return str(g.get("name", f"Unknown ({app_id})"))
except (SteamAPIError, OSError, RuntimeError, ValueError):
return None
return None
def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None:
"""Manually pick a game by Steam app_id, locking the enforcer for 2 weeks.
Args:
config: Enforcer configuration.
state: Current enforcer state.
args: Remaining CLI args (first element should be the app_id).
"""
raw_id = args[0] if args else input("Enter Steam app_id: ").strip()
try:
app_id = int(raw_id)
except ValueError:
_echo(f"Error: app_id must be a number, got '{raw_id}'.")
return
_echo(f"Looking up AppID={app_id}...")
game_name = _resolve_game_name(config, app_id)
if game_name is None:
_echo(
f"Error: AppID={app_id} not found in your Steam library or snapshot.\n"
"Run 'scan' first, or verify the app_id is correct."
)
return
_echo(f"\nGame found: {game_name} (AppID={app_id})")
_echo(
f"\nWARNING: Picking this game will:"
f"\n - Override your current assignment"
f"\n - Lock ALL other commands for {_MANUAL_LOCK_DAYS} DAYS or until"
f"\n you reach 100% achievements"
f"\n - Only 'done', 'check', 'status', 'enforce', 'setup', 'serve'"
f"\n will remain usable during this period"
)
_echo()
confirm = input(
f"Type YES to confirm you will play {game_name} until completion: "
).strip()
if confirm != "YES":
_echo("Aborted.")
return
state.manual_pick_app_id = app_id
state.manual_pick_game_name = game_name
state.manual_pick_started_at = datetime.now(timezone.utc).isoformat()
state.current_app_id = app_id
state.current_game_name = game_name
if not state.enforcement_started_at:
state.enforcement_started_at = datetime.now(timezone.utc).isoformat()
state.save()
_echo(f"\nManual pick confirmed: {game_name} (AppID={app_id})")
_echo(f"Lock active from now until 100% achievements or {_MANUAL_LOCK_DAYS} days.")
_echo("Run 'done' or 'check' once you have 100% to release the lock.\n")
# Post-assignment: mirror what _assign_chosen_game + cmd_pick do.
if config.uninstall_other_games:
_echo(" Uninstalling non-assigned games...")
count = uninstall_other_games(app_id)
if count:
_echo(f" Uninstalled {count} non-assigned game(s)")
if not is_game_installed(app_id):
_echo(f" Installing {game_name}...")
install_game(app_id, game_name, config.steam_id, use_steam_protocol=True)
else:
_echo(f" {game_name} is already installed.")
owned_ids = get_all_owned_app_ids(config)
if owned_ids:
hidden = hide_other_games(owned_ids, app_id)
if hidden > 0:
_echo(f" Library: hid {hidden} games")
COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = { COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
"scan": ("Scan library & assign a game", do_scan), "scan": ("Scan library & assign a game", do_scan),
"check": ("Check assigned game completion", do_check), "check": ("Check assigned game completion", do_check),
@ -411,6 +597,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
# Extra commands with non-standard arg handling (shown in help but not in COMMANDS). # Extra commands with non-standard arg handling (shown in help but not in COMMANDS).
_EXTRA_COMMAND_DESCRIPTIONS: dict[str, str] = { _EXTRA_COMMAND_DESCRIPTIONS: dict[str, str] = {
"add-exception": "Request 24h-locked whitelist exception (use --reason)", "add-exception": "Request 24h-locked whitelist exception (use --reason)",
"pick-manual": f"Pick a game by app_id, lock enforcer for {_MANUAL_LOCK_DAYS} days",
} }
_ALL_COMMANDS: dict[str, str] = { _ALL_COMMANDS: dict[str, str] = {
@ -430,18 +617,27 @@ def main() -> None:
command = sys.argv[1] command = sys.argv[1]
# add-exception has its own argument structure; handle before config load.
if command == "add-exception":
cmd_add_exception(sys.argv[2:])
return
config = Config.load() config = Config.load()
if command != "setup" and not config.steam_api_key: if command not in {"setup", "add-exception"} and not config.steam_api_key:
_echo("Not configured. Run 'setup' first.") _echo("Not configured. Run 'setup' first.")
sys.exit(1) sys.exit(1)
state = State.load() state = State.load()
# Enforce the manual-pick lock before dispatching any command.
# This also covers add-exception (previously dispatched before state load).
_enforce_manual_pick_lock(command, state)
# add-exception and pick-manual have non-standard argument structures.
if command == "add-exception":
cmd_add_exception(sys.argv[2:])
return
if command == "pick-manual":
cmd_pick_manual(config, state, sys.argv[2:])
return
_, func = COMMANDS[command] _, func = COMMANDS[command]
func(config, state) func(config, state)

View File

@ -270,9 +270,10 @@ class TestFetchHltbDetailMissing:
"""Tests for fetch_hltb_detail_missing.""" """Tests for fetch_hltb_detail_missing."""
def test_no_missing_returns_zero(self) -> None: def test_no_missing_returns_zero(self) -> None:
"""All games in rush cache → early return without fetching.""" """All games in rush cache with known game IDs → early return."""
with ( with (
patch(f"{PKG}.load_hltb_rush_cache", return_value={440: 15.0}), patch(f"{PKG}.load_hltb_rush_cache", return_value={440: 15.0}),
patch(f"{PKG}.load_hltb_game_id_cache", return_value={440: 12345}),
patch(f"{PKG}.fetch_hltb_times") as mock_fetch, patch(f"{PKG}.fetch_hltb_times") as mock_fetch,
): ):
result = fetch_hltb_detail_missing([(440, "TF2")]) result = fetch_hltb_detail_missing([(440, "TF2")])
@ -390,3 +391,19 @@ class TestFetchHltbDetailMissing:
): ):
result = fetch_hltb_detail_missing([(730, "CS")]) result = fetch_hltb_detail_missing([(730, "CS")])
assert result == 0 assert result == 0
def test_id_only_missing_logs_else_branch(self) -> None:
"""Rush data present but game ID missing → else branch in log selection."""
with (
patch(f"{PKG}.load_hltb_rush_cache", return_value={440: 15.0}),
patch(f"{PKG}.load_hltb_cache", return_value={440: 15.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"),
patch(f"{PKG}.save_hltb_cache"),
patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]),
):
result = fetch_hltb_detail_missing([(440, "TF2")])
assert result == 0

View File

@ -0,0 +1,454 @@
"""Tests for main CLI module — part 4 (manual pick lock + cmd_pick_manual)."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import sys
from unittest.mock import patch
import pytest
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.main import (
_MANUAL_LOCK_DAYS,
_enforce_manual_pick_lock,
_is_manual_pick_locked,
_resolve_game_name,
_show_manual_pick_lock_message,
cmd_pick_manual,
cmd_status,
main,
)
from steam_backlog_enforcer.steam_api import SteamAPIError
PKG = "steam_backlog_enforcer.main"
# A start timestamp that is always within the 14-day lock window.
_STARTED_AT = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
# A start timestamp that is always past the 14-day deadline.
_EXPIRED_AT = (
datetime.now(timezone.utc) - timedelta(days=_MANUAL_LOCK_DAYS + 1)
).isoformat()
def _locked_state(
app_id: int = 100,
name: str = "TestGame",
started_at: str = _STARTED_AT,
) -> State:
return State(
manual_pick_app_id=app_id,
manual_pick_game_name=name,
manual_pick_started_at=started_at,
)
# ──────────────────────────────────────────────────────────────
# _is_manual_pick_locked
# ──────────────────────────────────────────────────────────────
class TestIsManualPickLocked:
def test_no_manual_pick_not_locked(self) -> None:
assert _is_manual_pick_locked(State()) is False
def test_game_finished_not_locked(self) -> None:
state = _locked_state(app_id=100)
state.finished_app_ids = [100]
assert _is_manual_pick_locked(state) is False
def test_deadline_passed_not_locked(self) -> None:
state = _locked_state(started_at=_EXPIRED_AT)
assert _is_manual_pick_locked(state) is False
def test_active_lock_returns_true(self) -> None:
state = _locked_state(started_at=_STARTED_AT)
assert _is_manual_pick_locked(state) is True
def test_no_started_at_stays_locked(self) -> None:
# Missing timestamp → cannot determine deadline → stays locked.
state = _locked_state(started_at="")
assert _is_manual_pick_locked(state) is True
def test_invalid_started_at_stays_locked(self) -> None:
state = _locked_state(started_at="not-a-date")
assert _is_manual_pick_locked(state) is True
# ──────────────────────────────────────────────────────────────
# _show_manual_pick_lock_message
# ──────────────────────────────────────────────────────────────
class TestShowManualPickLockMessage:
def test_shows_game_info(self) -> None:
state = _locked_state(app_id=42, name="MyGame", started_at=_STARTED_AT)
with patch(f"{PKG}._echo") as mock_echo:
_show_manual_pick_lock_message(state)
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "MyGame" in output
assert "42" in output
def test_shows_deadline_when_started_at_valid(self) -> None:
state = _locked_state(started_at=_STARTED_AT)
with patch(f"{PKG}._echo") as mock_echo:
_show_manual_pick_lock_message(state)
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "Deadline" in output
def test_no_crash_on_invalid_started_at(self) -> None:
state = _locked_state(started_at="bad-date")
with patch(f"{PKG}._echo"):
_show_manual_pick_lock_message(state) # must not raise
def test_no_crash_on_empty_started_at(self) -> None:
state = _locked_state(started_at="")
with patch(f"{PKG}._echo"):
_show_manual_pick_lock_message(state) # must not raise
# ──────────────────────────────────────────────────────────────
# _enforce_manual_pick_lock
# ──────────────────────────────────────────────────────────────
class TestEnforceManualPickLock:
def test_no_lock_passes(self) -> None:
_enforce_manual_pick_lock("scan", State()) # no exit
def test_exempt_command_passes_while_locked(self) -> None:
state = _locked_state()
_enforce_manual_pick_lock("done", state) # no exit
_enforce_manual_pick_lock("status", state) # no exit
def test_blocked_command_exits(self) -> None:
state = _locked_state()
with (
patch(f"{PKG}._show_manual_pick_lock_message"),
pytest.raises(SystemExit) as exc_info,
):
_enforce_manual_pick_lock("scan", state)
assert exc_info.value.code == 1
def test_add_exception_blocked_when_locked(self) -> None:
state = _locked_state()
with (
patch(f"{PKG}._show_manual_pick_lock_message"),
pytest.raises(SystemExit),
):
_enforce_manual_pick_lock("add-exception", state)
def test_pick_manual_blocked_when_already_locked(self) -> None:
state = _locked_state()
with (
patch(f"{PKG}._show_manual_pick_lock_message"),
pytest.raises(SystemExit),
):
_enforce_manual_pick_lock("pick-manual", state)
# ──────────────────────────────────────────────────────────────
# _resolve_game_name
# ──────────────────────────────────────────────────────────────
class TestResolveGameName:
def test_found_in_snapshot(self) -> None:
snapshot = [
{
"app_id": 440,
"name": "TF2",
"total_achievements": 0,
"unlocked_achievements": 0,
"playtime_minutes": 0,
}
]
with patch(f"{PKG}.load_snapshot", return_value=snapshot):
result = _resolve_game_name(Config(), 440)
assert result == "TF2"
def test_not_in_snapshot_found_via_api(self) -> None:
with (
patch(f"{PKG}.load_snapshot", return_value=[]),
patch(f"{PKG}.SteamAPIClient") as mock_cls,
):
mock_cls.return_value.get_owned_games.return_value = [
{"appid": 730, "name": "Counter-Strike 2"}
]
result = _resolve_game_name(Config(), 730)
assert result == "Counter-Strike 2"
def test_api_raises_returns_none(self) -> None:
with (
patch(f"{PKG}.load_snapshot", return_value=[]),
patch(f"{PKG}.SteamAPIClient") as mock_cls,
):
mock_cls.return_value.get_owned_games.side_effect = SteamAPIError("fail")
result = _resolve_game_name(Config(), 999)
assert result is None
def test_not_found_anywhere_returns_none(self) -> None:
with (
patch(f"{PKG}.load_snapshot", return_value=[{"app_id": 1, "name": "X"}]),
patch(f"{PKG}.SteamAPIClient") as mock_cls,
):
mock_cls.return_value.get_owned_games.return_value = [{"appid": 1}]
result = _resolve_game_name(Config(), 999)
assert result is None
def test_no_snapshot_falls_through_to_api(self) -> None:
with (
patch(f"{PKG}.load_snapshot", return_value=None),
patch(f"{PKG}.SteamAPIClient") as mock_cls,
):
mock_cls.return_value.get_owned_games.return_value = [
{"appid": 440, "name": "TF2"}
]
result = _resolve_game_name(Config(), 440)
assert result == "TF2"
# ──────────────────────────────────────────────────────────────
# cmd_pick_manual
# ──────────────────────────────────────────────────────────────
class TestCmdPickManual:
def _base_patches(self) -> dict[str, object]:
return {
f"{PKG}._resolve_game_name": "Skyrim SE",
f"{PKG}.uninstall_other_games": 2,
f"{PKG}.is_game_installed": False,
f"{PKG}.install_game": None,
f"{PKG}.get_all_owned_app_ids": [1, 2, 489830],
f"{PKG}.hide_other_games": 2,
}
def test_invalid_app_id(self) -> None:
with patch(f"{PKG}._echo") as mock_echo:
cmd_pick_manual(Config(), State(), ["abc"])
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "Error" in output
def test_game_not_found(self) -> None:
with (
patch(f"{PKG}._resolve_game_name", return_value=None),
patch(f"{PKG}._echo") as mock_echo,
):
cmd_pick_manual(Config(), State(), ["489830"])
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "not found" in output
def test_aborted_when_not_yes(self) -> None:
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo"),
patch("builtins.input", return_value="no"),
patch(f"{PKG}.State.save") as mock_save,
):
cmd_pick_manual(Config(), State(), ["489830"])
mock_save.assert_not_called()
def test_prompts_for_id_when_no_args(self) -> None:
state = State()
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo"),
patch("builtins.input", side_effect=["489830", "YES"]),
patch.object(State, "save"),
patch(f"{PKG}.uninstall_other_games", return_value=0),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{PKG}.get_all_owned_app_ids", return_value=[]),
):
cmd_pick_manual(Config(), state, [])
assert state.current_app_id == 489830
def test_success_sets_state_and_runs_post_steps(self) -> None:
state = State()
config = Config(uninstall_other_games=True)
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo"),
patch("builtins.input", return_value="YES"),
patch.object(State, "save") as mock_save,
patch(f"{PKG}.uninstall_other_games", return_value=2) as mock_uninstall,
patch(f"{PKG}.is_game_installed", return_value=False),
patch(f"{PKG}.install_game") as mock_install,
patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 489830]),
patch(f"{PKG}.hide_other_games", return_value=1) as mock_hide,
):
cmd_pick_manual(config, state, ["489830"])
assert state.manual_pick_app_id == 489830
assert state.manual_pick_game_name == "Skyrim SE"
assert state.manual_pick_started_at != ""
assert state.current_app_id == 489830
mock_save.assert_called_once()
mock_uninstall.assert_called_once_with(489830)
mock_install.assert_called_once()
mock_hide.assert_called_once()
def test_no_uninstall_when_config_off(self) -> None:
state = State()
config = Config(uninstall_other_games=False)
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo"),
patch("builtins.input", return_value="YES"),
patch.object(State, "save"),
patch(f"{PKG}.uninstall_other_games") as mock_uninstall,
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{PKG}.get_all_owned_app_ids", return_value=[]),
):
cmd_pick_manual(config, state, ["489830"])
mock_uninstall.assert_not_called()
def test_game_already_installed_skips_install(self) -> None:
state = State()
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo"),
patch("builtins.input", return_value="YES"),
patch.object(State, "save"),
patch(f"{PKG}.uninstall_other_games", return_value=0),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{PKG}.install_game") as mock_install,
patch(f"{PKG}.get_all_owned_app_ids", return_value=[]),
):
cmd_pick_manual(Config(), state, ["489830"])
mock_install.assert_not_called()
def test_no_hide_when_no_owned_ids(self) -> None:
state = State()
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo"),
patch("builtins.input", return_value="YES"),
patch.object(State, "save"),
patch(f"{PKG}.uninstall_other_games", return_value=0),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{PKG}.get_all_owned_app_ids", return_value=[]),
patch(f"{PKG}.hide_other_games") as mock_hide,
):
cmd_pick_manual(Config(), state, ["489830"])
mock_hide.assert_not_called()
def test_uninstall_returns_zero_no_echo(self) -> None:
state = State()
config = Config(uninstall_other_games=True)
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo") as mock_echo,
patch("builtins.input", return_value="YES"),
patch.object(State, "save"),
patch(f"{PKG}.uninstall_other_games", return_value=0),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{PKG}.get_all_owned_app_ids", return_value=[]),
):
cmd_pick_manual(config, state, ["489830"])
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "Uninstalled 0" not in output
def test_enforcement_started_at_set_when_empty(self) -> None:
state = State(enforcement_started_at="")
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo"),
patch("builtins.input", return_value="YES"),
patch.object(State, "save"),
patch(f"{PKG}.uninstall_other_games", return_value=0),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{PKG}.get_all_owned_app_ids", return_value=[]),
):
cmd_pick_manual(Config(), state, ["489830"])
assert state.enforcement_started_at != ""
def test_enforcement_started_at_not_overwritten(self) -> None:
existing_ts = "2026-01-01T00:00:00+00:00"
state = State(enforcement_started_at=existing_ts)
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo"),
patch("builtins.input", return_value="YES"),
patch.object(State, "save"),
patch(f"{PKG}.uninstall_other_games", return_value=0),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{PKG}.get_all_owned_app_ids", return_value=[]),
):
cmd_pick_manual(Config(), state, ["489830"])
assert state.enforcement_started_at == existing_ts
def test_hide_returns_zero_no_echo(self) -> None:
state = State()
with (
patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
patch(f"{PKG}._echo") as mock_echo,
patch("builtins.input", return_value="YES"),
patch.object(State, "save"),
patch(f"{PKG}.uninstall_other_games", return_value=0),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2]),
patch(f"{PKG}.hide_other_games", return_value=0),
):
cmd_pick_manual(Config(), state, ["489830"])
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "Library: hid" not in output
# ──────────────────────────────────────────────────────────────
# main() dispatch to pick-manual
# ──────────────────────────────────────────────────────────────
class TestMainDispatchPickManual:
def test_dispatches_pick_manual(self) -> None:
argv = ["prog", "pick-manual", "489830"]
with (
patch.object(sys, "argv", argv),
patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")),
patch(f"{PKG}.State.load", return_value=State()),
patch(f"{PKG}.cmd_pick_manual") as mock_cmd,
):
main()
mock_cmd.assert_called_once()
def test_pick_manual_blocked_when_locked(self) -> None:
state = _locked_state()
argv = ["prog", "pick-manual", "730"]
with (
patch.object(sys, "argv", argv),
patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")),
patch(f"{PKG}.State.load", return_value=state),
patch(f"{PKG}._show_manual_pick_lock_message"),
pytest.raises(SystemExit) as exc_info,
):
main()
assert exc_info.value.code == 1
# ──────────────────────────────────────────────────────────────
# cmd_status shows lock hint when locked
# ──────────────────────────────────────────────────────────────
class TestCmdStatusLockHint:
def test_shows_lock_hint_when_locked(self) -> None:
state = _locked_state()
with (
patch(f"{PKG}.is_store_blocked", return_value=False),
patch(f"{PKG}.get_installed_games", return_value=[]),
patch(f"{PKG}._echo") as mock_echo,
):
cmd_status(Config(), state)
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "MANUAL PICK LOCK" in output
def test_no_lock_hint_when_not_locked(self) -> None:
with (
patch(f"{PKG}.is_store_blocked", return_value=False),
patch(f"{PKG}.get_installed_games", return_value=[]),
patch(f"{PKG}._echo") as mock_echo,
):
cmd_status(Config(), State())
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "MANUAL PICK LOCK" not in output

View File

@ -6,16 +6,19 @@ from datetime import datetime, timedelta, timezone
from unittest.mock import patch from unittest.mock import patch
from steam_backlog_enforcer._stats import ( from steam_backlog_enforcer._stats import (
_ensure_completed_rush_data,
_ensure_rush_data, _ensure_rush_data,
_filter_qualifying_games, _filter_qualifying_games,
_format_completion_date, _format_completion_date,
_GameTimes, _GameTimes,
_print_pace_scenario, _print_pace_scenario,
_print_player_speed_scenario,
_print_scenario, _print_scenario,
_print_worst_example, _print_worst_example,
_sum_hours, _sum_hours,
cmd_stats, cmd_stats,
) )
from steam_backlog_enforcer._web_dataset import PaceVsHLTB
from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.protondb import ProtonDBRating from steam_backlog_enforcer.protondb import ProtonDBRating
from steam_backlog_enforcer.steam_api import GameInfo from steam_backlog_enforcer.steam_api import GameInfo
@ -399,6 +402,8 @@ class TestCmdStats:
f"{_PKG}._filter_qualifying_games", f"{_PKG}._filter_qualifying_games",
return_value=([entry], hltb_skip, linux_skip, no_data_skip), return_value=([entry], hltb_skip, linux_skip, no_data_skip),
), ),
patch(f"{_PKG}._ensure_completed_rush_data", return_value=False),
patch(f"{_PKG}._print_player_speed_scenario"),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}._print_pace_scenario"), patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"), patch(f"{_PKG}._print_scenario"),
@ -460,6 +465,8 @@ class TestCmdStats:
f"{_PKG}._filter_qualifying_games", f"{_PKG}._filter_qualifying_games",
return_value=([entry], 0, 0, 0), return_value=([entry], 0, 0, 0),
), ),
patch(f"{_PKG}._ensure_completed_rush_data", return_value=False),
patch(f"{_PKG}._print_player_speed_scenario"),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}._print_pace_scenario"), patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"), patch(f"{_PKG}._print_scenario"),
@ -489,7 +496,9 @@ class TestCmdStats:
f"{_PKG}._filter_qualifying_games", f"{_PKG}._filter_qualifying_games",
return_value=([entry], 0, 0, 0), return_value=([entry], 0, 0, 0),
), ),
patch(f"{_PKG}._ensure_completed_rush_data", return_value=False),
patch(f"{_PKG}._ensure_rush_data", return_value=False), patch(f"{_PKG}._ensure_rush_data", return_value=False),
patch(f"{_PKG}._print_player_speed_scenario"),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}._print_pace_scenario"), patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"), patch(f"{_PKG}._print_scenario"),
@ -509,7 +518,9 @@ class TestCmdStats:
f"{_PKG}._filter_qualifying_games", f"{_PKG}._filter_qualifying_games",
return_value=([], 0, 0, 0), return_value=([], 0, 0, 0),
), ),
patch(f"{_PKG}._ensure_completed_rush_data", return_value=False),
patch(f"{_PKG}._ensure_rush_data", return_value=False), patch(f"{_PKG}._ensure_rush_data", return_value=False),
patch(f"{_PKG}._print_player_speed_scenario"),
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}._print_pace_scenario"), patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"), patch(f"{_PKG}._print_scenario"),
@ -538,7 +549,9 @@ class TestCmdStats:
with ( with (
patch(f"{_PKG}.load_snapshot", return_value=snapshot), patch(f"{_PKG}.load_snapshot", return_value=snapshot),
patch(f"{_PKG}._filter_qualifying_games", side_effect=count_filter), patch(f"{_PKG}._filter_qualifying_games", side_effect=count_filter),
patch(f"{_PKG}._ensure_completed_rush_data", return_value=False),
patch(f"{_PKG}._ensure_rush_data", return_value=True), patch(f"{_PKG}._ensure_rush_data", return_value=True),
patch(f"{_PKG}._print_player_speed_scenario"),
patch(f"{_PKG}._echo"), patch(f"{_PKG}._echo"),
patch(f"{_PKG}._print_pace_scenario"), patch(f"{_PKG}._print_pace_scenario"),
patch(f"{_PKG}._print_scenario"), patch(f"{_PKG}._print_scenario"),
@ -547,15 +560,37 @@ class TestCmdStats:
cmd_stats(self._config(), state) cmd_stats(self._config(), state)
assert len(filter_calls) == 2 assert len(filter_calls) == 2
def test_games_done_passed_to_pace_from_snapshot_complete(self) -> None: def test_games_done_since_start_passed_to_pace(self) -> None:
"""_print_pace_scenario receives is_complete count from snapshot.""" """_print_pace_scenario gets only games completed after started_at."""
state = State() from datetime import datetime, timezone
# Snapshot: 1 complete game (unlocked=total=10), 1 incomplete.
snapshot_complete = { started = datetime(2026, 1, 1, tzinfo=timezone.utc)
state = State(enforcement_started_at=started.isoformat())
after_ts = int(datetime(2026, 3, 1, tzinfo=timezone.utc).timestamp())
before_ts = int(datetime(2025, 6, 1, tzinfo=timezone.utc).timestamp())
def _ach(ts: int) -> dict[str, object]:
return {
"api_name": "A",
"display_name": "A",
"achieved": True,
"unlock_time": ts,
}
# app_id=2: completed AFTER enforcement start → should count
snapshot_after = {
**self._snapshot_game(app_id=2), **self._snapshot_game(app_id=2),
"unlocked_achievements": 10, "unlocked_achievements": 10,
"achievements": [_ach(after_ts)] * 10,
} }
snapshot = [self._snapshot_game(app_id=1), snapshot_complete] # app_id=3: completed BEFORE enforcement start → should NOT count
snapshot_before = {
**self._snapshot_game(app_id=3),
"unlocked_achievements": 10,
"achievements": [_ach(before_ts)] * 10,
}
snapshot = [self._snapshot_game(app_id=1), snapshot_after, snapshot_before]
game = GameInfo.from_snapshot(self._snapshot_game()) game = GameInfo.from_snapshot(self._snapshot_game())
entry = _GameTimes( entry = _GameTimes(
game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0 game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0
@ -571,13 +606,59 @@ class TestCmdStats:
f"{_PKG}._filter_qualifying_games", f"{_PKG}._filter_qualifying_games",
return_value=([entry], 0, 0, 0), return_value=([entry], 0, 0, 0),
), ),
patch(f"{_PKG}._ensure_completed_rush_data", return_value=False),
patch(f"{_PKG}._print_player_speed_scenario"),
patch(f"{_PKG}._echo"), patch(f"{_PKG}._echo"),
patch(f"{_PKG}._print_pace_scenario", side_effect=capture_pace), patch(f"{_PKG}._print_pace_scenario", side_effect=capture_pace),
patch(f"{_PKG}._print_scenario"), patch(f"{_PKG}._print_scenario"),
patch(f"{_PKG}._print_worst_example"), patch(f"{_PKG}._print_worst_example"),
): ):
cmd_stats(self._config(), state) cmd_stats(self._config(), state)
assert captured["games_done"] == 1 assert captured["games_done"] == 1 # only the post-start game
def test_player_speed_scenario_called_with_pace_and_totals(self) -> None:
"""_print_player_speed_scenario receives pace, rush_total, and leisure_total."""
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
)
pace = PaceVsHLTB(
calibration_count=5,
ratio_vs_rush=1.1,
ratio_vs_leisure=0.4,
interpolation_t=0.05,
player_style="rush_to_leisure",
)
captured: dict[str, object] = {}
def capture_player_speed(p: object, rush: float, leisure: float) -> None:
captured["pace"] = p
captured["rush"] = rush
captured["leisure"] = leisure
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_completed_rush_data", return_value=False),
patch(f"{_PKG}.compute_pace_vs_hltb", return_value=pace),
patch(
f"{_PKG}._print_player_speed_scenario",
side_effect=capture_player_speed,
),
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 captured["pace"] is pace
assert captured["rush"] == 15.0
assert captured["leisure"] == 25.0
class TestEnsureRushData: class TestEnsureRushData:
@ -612,6 +693,168 @@ class TestEnsureRushData:
mock_fetch.assert_called_once() mock_fetch.assert_called_once()
class TestEnsureCompletedRushData:
"""Tests for _ensure_completed_rush_data."""
def _complete(self, app_id: int = 1, playtime: int = 600) -> GameInfo:
return GameInfo(
app_id=app_id,
name="Done",
total_achievements=10,
unlocked_achievements=10,
playtime_minutes=playtime,
completionist_hours=0.0,
comp_100_count=5,
count_comp=20,
)
def test_no_complete_games_returns_false_without_fetch(self) -> None:
incomplete = _game(app_id=1, total=10, unlocked=0)
with patch(f"{_PKG}.fetch_hltb_detail_missing") as mock_fetch:
result = _ensure_completed_rush_data([incomplete])
assert result is False
mock_fetch.assert_not_called()
def test_complete_game_with_zero_playtime_excluded(self) -> None:
"""Games with playtime_minutes=0 are skipped (no calibration value)."""
no_play = self._complete(playtime=0)
with patch(f"{_PKG}.fetch_hltb_detail_missing") as mock_fetch:
result = _ensure_completed_rush_data([no_play])
assert result is False
mock_fetch.assert_not_called()
def test_complete_game_with_playtime_fetches(self) -> None:
game = self._complete()
with (
patch(f"{_PKG}.fetch_hltb_detail_missing", return_value=1) as mock_fetch,
patch(f"{_PKG}._echo"),
):
result = _ensure_completed_rush_data([game])
assert result is True
mock_fetch.assert_called_once_with([(1, "Done")])
def test_fetch_returns_zero_means_no_new_data(self) -> None:
"""When fetch_hltb_detail_missing returns 0, return False (all cached)."""
game = self._complete()
with (
patch(f"{_PKG}.fetch_hltb_detail_missing", return_value=0),
patch(f"{_PKG}._echo"),
):
result = _ensure_completed_rush_data([game])
assert result is False
class TestPrintPlayerSpeedScenario:
"""Tests for _print_player_speed_scenario — 100 % branch coverage."""
def _echoed(
self,
pace: PaceVsHLTB | None,
rush: float = 100.0,
leisure: float = 200.0,
) -> list[str]:
out: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: out.append(a[0])):
_print_player_speed_scenario(pace, rush, leisure)
return out
def test_none_pace_shows_no_calibration_message(self) -> None:
echoed = self._echoed(None)
assert any("No calibration data" in s for s in echoed)
def test_zero_calibration_count_shows_no_calibration_message(self) -> None:
pace = PaceVsHLTB(
calibration_count=0,
ratio_vs_rush=-1.0,
ratio_vs_leisure=-1.0,
interpolation_t=-1.0,
player_style="unknown",
)
echoed = self._echoed(pace)
assert any("No calibration data" in s for s in echoed)
def test_ratio_vs_rush_shown_when_positive(self) -> None:
pace = PaceVsHLTB(
calibration_count=5,
ratio_vs_rush=1.05,
ratio_vs_leisure=-1.0,
interpolation_t=-1.0,
player_style="unknown",
)
echoed = self._echoed(pace)
assert any("rush pace" in s for s in echoed)
def test_ratio_vs_leisure_shown_when_positive(self) -> None:
pace = PaceVsHLTB(
calibration_count=5,
ratio_vs_rush=1.05,
ratio_vs_leisure=0.5,
interpolation_t=-1.0,
player_style="unknown",
)
echoed = self._echoed(pace)
assert any("leisure pace" in s for s in echoed)
def test_interpolation_t_shown_when_not_minus_one(self) -> None:
pace = PaceVsHLTB(
calibration_count=5,
ratio_vs_rush=1.05,
ratio_vs_leisure=0.5,
interpolation_t=0.1,
player_style="rush_to_leisure",
)
echoed = self._echoed(pace)
assert any("Interpolation t" in s for s in echoed)
def test_estimate_uses_interpolation_when_available(self) -> None:
# rush=100, leisure=200, t=0.5 → est=150
pace = PaceVsHLTB(
calibration_count=5,
ratio_vs_rush=1.5,
ratio_vs_leisure=0.5,
interpolation_t=0.5,
player_style="rush_to_leisure",
)
echoed = self._echoed(pace, rush=100.0, leisure=200.0)
assert any("150" in s for s in echoed)
def test_estimate_falls_back_to_ratio_when_no_interpolation(self) -> None:
# interpolation_t=-1, ratio_vs_rush=2.0, rush=100 → est=200
pace = PaceVsHLTB(
calibration_count=5,
ratio_vs_rush=2.0,
ratio_vs_leisure=-1.0,
interpolation_t=-1.0,
player_style="unknown",
)
echoed = self._echoed(pace, rush=100.0, leisure=0.0)
assert any("200" in s for s in echoed)
def test_no_estimate_when_both_methods_unavailable(self) -> None:
"""No 'Estimated backlog total' line when t=-1 and ratio=-1."""
pace = PaceVsHLTB(
calibration_count=5,
ratio_vs_rush=-1.0,
ratio_vs_leisure=-1.0,
interpolation_t=-1.0,
player_style="unknown",
)
echoed = self._echoed(pace, rush=100.0, leisure=0.0)
assert not any("Estimated backlog total" in s for s in echoed)
def test_no_estimate_when_rush_total_zero_and_no_interpolation(self) -> None:
"""No estimate line when rush_total=0 and interpolation_t=-1."""
pace = PaceVsHLTB(
calibration_count=5,
ratio_vs_rush=1.5,
ratio_vs_leisure=-1.0,
interpolation_t=-1.0,
player_style="unknown",
)
echoed = self._echoed(pace, rush=0.0, leisure=0.0)
assert not any("Estimated backlog total" in s for s in echoed)
class TestPrintWorstExample: class TestPrintWorstExample:
"""Tests for _print_worst_example.""" """Tests for _print_worst_example."""
@ -627,6 +870,7 @@ class TestPrintWorstExample:
worst_hours=15.0, worst_hours=15.0,
rush_hours=5.0, rush_hours=5.0,
leisure_100h=20.0, leisure_100h=20.0,
hltb_game_id=99999,
) )
echoed: list[str] = [] echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
@ -637,7 +881,11 @@ class TestPrintWorstExample:
def test_example_without_rush(self) -> None: def test_example_without_rush(self) -> None:
entry = _GameTimes( entry = _GameTimes(
game=_game(name="X"), worst_hours=15.0, rush_hours=-1.0, leisure_100h=20.0 game=_game(name="X"),
worst_hours=15.0,
rush_hours=-1.0,
leisure_100h=20.0,
hltb_game_id=99999,
) )
echoed: list[str] = [] echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
@ -647,7 +895,11 @@ class TestPrintWorstExample:
def test_example_without_leisure(self) -> None: def test_example_without_leisure(self) -> None:
entry = _GameTimes( entry = _GameTimes(
game=_game(name="Y"), worst_hours=15.0, rush_hours=5.0, leisure_100h=-1.0 game=_game(name="Y"),
worst_hours=15.0,
rush_hours=5.0,
leisure_100h=-1.0,
hltb_game_id=99999,
) )
echoed: list[str] = [] echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
@ -655,8 +907,8 @@ class TestPrintWorstExample:
assert any("Rush" in s for s in echoed) assert any("Rush" in s for s in echoed)
assert not any("Leisure" 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: def test_hltb_search_url_shown_when_lookup_finds_nothing(self) -> None:
"""Falls back to search URL when hltb_game_id is 0.""" """Falls back to search URL when hltb_game_id is 0 and lookup finds nothing."""
entry = _GameTimes( entry = _GameTimes(
game=_game(name="Portal 2"), game=_game(name="Portal 2"),
worst_hours=15.0, worst_hours=15.0,
@ -664,10 +916,32 @@ class TestPrintWorstExample:
leisure_100h=-1.0, leisure_100h=-1.0,
) )
echoed: list[str] = [] echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): with (
patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])),
patch(f"{_PKG}.fetch_hltb_detail_missing", return_value=0),
patch(f"{_PKG}.load_hltb_game_id_cache", return_value={}),
):
_print_worst_example([entry]) _print_worst_example([entry])
assert any("howlongtobeat.com" in s and "Portal+2" in s for s in echoed) assert any("howlongtobeat.com" in s and "Portal+2" in s for s in echoed)
def test_hltb_direct_link_shown_after_on_demand_lookup(self) -> None:
"""Direct link shown when on-demand lookup successfully finds the game ID."""
entry = _GameTimes(
game=_game(app_id=111, 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])),
patch(f"{_PKG}.fetch_hltb_detail_missing", return_value=0),
patch(f"{_PKG}.load_hltb_game_id_cache", return_value={111: 42000}),
):
_print_worst_example([entry])
assert any("howlongtobeat.com/game/42000" in s for s in echoed)
assert not any("?q=" in s for s in echoed)
def test_hltb_direct_link_shown_when_game_id_known(self) -> None: def test_hltb_direct_link_shown_when_game_id_known(self) -> None:
"""Direct HLTB game link shown when hltb_game_id is populated.""" """Direct HLTB game link shown when hltb_game_id is populated."""
entry = _GameTimes( entry = _GameTimes(
@ -693,9 +967,88 @@ class TestPrintWorstExample:
worst_hours=10.0, worst_hours=10.0,
rush_hours=-1.0, rush_hours=-1.0,
leisure_100h=-1.0, leisure_100h=-1.0,
hltb_game_id=99999,
) )
echoed: list[str] = [] echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
_print_worst_example([bad, good]) _print_worst_example([bad, good])
assert any("Pick" in s for s in echoed) assert any("Pick" in s for s in echoed)
assert not any("Skip" in s for s in echoed) assert not any("Skip" in s for s in echoed)
class TestRefreshRecentlyPlayedCompletions:
"""Tests for _refresh_recently_played_completions."""
def test_oserror_on_stat_returns_games_unchanged(self) -> None:
games = [GameInfo(1, "G", 10, 0, 60)]
with patch(f"{_PKG}.SNAPSHOT_FILE") as mock_sf:
mock_sf.stat.side_effect = OSError("no file")
from steam_backlog_enforcer._stats import (
_refresh_recently_played_completions,
)
result = _refresh_recently_played_completions(games, Config())
assert result == games
def test_no_recently_played_returns_games_unchanged(self) -> None:
games = [GameInfo(1, "G", 10, 0, 60)]
with (
patch(f"{_PKG}.SNAPSHOT_FILE") as mock_sf,
patch(f"{_PKG}.SteamAPIClient") as mock_cls,
):
mock_sf.stat.return_value.st_mtime = 1_000_000.0
mock_cls.return_value.get_owned_games.return_value = [
{"appid": 1, "rtime_last_played": 500_000}
]
from steam_backlog_enforcer._stats import (
_refresh_recently_played_completions,
)
result = _refresh_recently_played_completions(games, Config())
assert result == games
def test_recently_played_game_is_refreshed(self) -> None:
from steam_backlog_enforcer._stats import _refresh_recently_played_completions
from steam_backlog_enforcer.steam_api import AchievementInfo
game = GameInfo(1, "G", 5, 0, 60)
new_achievements = [
AchievementInfo("a1", "A1", achieved=True, unlock_time=1_500_001),
AchievementInfo("a2", "A2", achieved=True, unlock_time=1_500_002),
AchievementInfo("a3", "A3", achieved=False, unlock_time=0),
AchievementInfo("a4", "A4", achieved=False, unlock_time=0),
AchievementInfo("a5", "A5", achieved=False, unlock_time=0),
]
with (
patch(f"{_PKG}.SNAPSHOT_FILE") as mock_sf,
patch(f"{_PKG}.SteamAPIClient") as mock_cls,
patch(f"{_PKG}._echo"),
):
mock_sf.stat.return_value.st_mtime = 1_000_000.0
mock_cls.return_value.get_owned_games.return_value = [
{"appid": 1, "rtime_last_played": 1_500_000}
]
mock_cls.return_value.get_achievement_details.return_value = (
new_achievements
)
result = _refresh_recently_played_completions([game], Config())
refreshed = next(g for g in result if g.app_id == 1)
assert refreshed.unlocked_achievements == 2
def test_get_achievement_details_empty_keeps_old_game(self) -> None:
from steam_backlog_enforcer._stats import _refresh_recently_played_completions
game = GameInfo(1, "G", 5, 3, 60)
with (
patch(f"{_PKG}.SNAPSHOT_FILE") as mock_sf,
patch(f"{_PKG}.SteamAPIClient") as mock_cls,
patch(f"{_PKG}._echo"),
):
mock_sf.stat.return_value.st_mtime = 1_000_000.0
mock_cls.return_value.get_owned_games.return_value = [
{"appid": 1, "rtime_last_played": 1_500_000}
]
mock_cls.return_value.get_achievement_details.return_value = []
result = _refresh_recently_played_completions([game], Config())
refreshed = next(g for g in result if g.app_id == 1)
assert refreshed.unlocked_achievements == 3

View File

@ -8,6 +8,7 @@ from unittest.mock import patch
from steam_backlog_enforcer._web_dataset import ( from steam_backlog_enforcer._web_dataset import (
HOURS_PER_DAY_PRESETS, HOURS_PER_DAY_PRESETS,
PaceVsHLTB,
WebGame, WebGame,
_build_games, _build_games,
_default_qualifying, _default_qualifying,
@ -18,6 +19,8 @@ from steam_backlog_enforcer._web_dataset import (
_sum_positive, _sum_positive,
_worst_hours, _worst_hours,
build_web_dataset, build_web_dataset,
compute_pace_vs_hltb,
count_complete_since_start,
dataset_to_payload, dataset_to_payload,
) )
from steam_backlog_enforcer.config import State from steam_backlog_enforcer.config import State
@ -142,31 +145,129 @@ class TestDefaultSummary:
assert summary.worst_total == 25.0 assert summary.worst_total == 25.0
class TestCountCompleteSinceStart:
"""Tests for count_complete_since_start."""
def _ach(self, ts: int, *, achieved: bool = True) -> object:
from steam_backlog_enforcer.steam_api import AchievementInfo
return AchievementInfo(
api_name="A", display_name="A", achieved=achieved, unlock_time=ts
)
def _complete_game(self, app_id: int, unlock_ts: int) -> GameInfo:
achs = [self._ach(unlock_ts)] * 5
return _gi(
app_id=app_id,
total_achievements=5,
unlocked_achievements=5,
achievements=achs,
)
def test_empty_started_at_returns_zero(self) -> None:
games = [self._complete_game(1, 1_000_000)]
assert count_complete_since_start(games, "") == 0
def test_invalid_started_at_returns_zero(self) -> None:
games = [self._complete_game(1, 1_000_000)]
assert count_complete_since_start(games, "not-a-date") == 0
def test_counts_game_completed_after_start(self) -> None:
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
after_ts = int(datetime(2026, 6, 1, tzinfo=timezone.utc).timestamp())
games = [self._complete_game(1, after_ts)]
assert count_complete_since_start(games, started.isoformat()) == 1
def test_excludes_game_completed_before_start(self) -> None:
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
before_ts = int(datetime(2025, 6, 1, tzinfo=timezone.utc).timestamp())
games = [self._complete_game(1, before_ts)]
assert count_complete_since_start(games, started.isoformat()) == 0
def test_excludes_incomplete_game(self) -> None:
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
after_ts = int(datetime(2026, 6, 1, tzinfo=timezone.utc).timestamp())
incomplete = _gi(
app_id=1,
total_achievements=5,
unlocked_achievements=3,
achievements=[self._ach(after_ts)] * 3,
)
assert count_complete_since_start([incomplete], started.isoformat()) == 0
def test_excludes_game_with_no_achievement_timestamps(self) -> None:
"""Complete game with unlock_time=0 on all achievements is excluded."""
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
no_ts = _gi(
app_id=1,
total_achievements=5,
unlocked_achievements=5,
achievements=[self._ach(0)] * 5,
)
assert count_complete_since_start([no_ts], started.isoformat()) == 0
def test_mixed_games_counts_only_post_start(self) -> None:
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
after_ts = int(datetime(2026, 6, 1, tzinfo=timezone.utc).timestamp())
before_ts = int(datetime(2025, 6, 1, tzinfo=timezone.utc).timestamp())
games = [
self._complete_game(1, after_ts),
self._complete_game(2, before_ts),
self._complete_game(3, after_ts),
]
assert count_complete_since_start(games, started.isoformat()) == 2
def test_uses_max_unlock_time_across_achievements(self) -> None:
"""Game counts if its LAST achievement was unlocked after start."""
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
before_ts = int(datetime(2025, 12, 1, tzinfo=timezone.utc).timestamp())
after_ts = int(datetime(2026, 2, 1, tzinfo=timezone.utc).timestamp())
# Mix of before/after timestamps — max is after start, so should count
achs = [self._ach(before_ts)] * 4 + [self._ach(after_ts)]
game = _gi(
app_id=1, total_achievements=5, unlocked_achievements=5, achievements=achs
)
assert count_complete_since_start([game], started.isoformat()) == 1
class TestStateInfo: class TestStateInfo:
"""Tests for _state_info pace calculation.""" """Tests for _state_info pace calculation."""
def test_no_start_date(self) -> None: def test_no_start_date(self) -> None:
info = _state_info(State(), games_done=5) info = _state_info(State(), games_done=5, games_done_since_start=5)
assert info.days_elapsed == 0 assert info.days_elapsed == 0
assert info.pace_games_per_day == 0.0 assert info.pace_games_per_day == 0.0
assert info.games_done == 5
assert info.games_done_since_start == 5
def test_invalid_start_date(self) -> None: def test_invalid_start_date(self) -> None:
info = _state_info(State(enforcement_started_at="not-a-date"), games_done=5) info = _state_info(
State(enforcement_started_at="not-a-date"),
games_done=5,
games_done_since_start=5,
)
assert info.days_elapsed == 0 assert info.days_elapsed == 0
assert info.pace_games_per_day == 0.0 assert info.pace_games_per_day == 0.0
def test_valid_start_with_games(self) -> None: def test_valid_start_with_games(self) -> None:
started = datetime.now(timezone.utc) - timedelta(days=50) started = datetime.now(timezone.utc) - timedelta(days=50)
info = _state_info( info = _state_info(
State(enforcement_started_at=started.isoformat()), games_done=10 State(enforcement_started_at=started.isoformat()),
games_done=12,
games_done_since_start=10,
) )
assert info.days_elapsed >= 49 assert info.days_elapsed >= 49
assert info.pace_games_per_day > 0.0 assert info.pace_games_per_day > 0.0
assert info.games_done == 12
assert info.games_done_since_start == 10
def test_valid_start_zero_games_keeps_zero_pace(self) -> None: def test_valid_start_zero_since_start_keeps_zero_pace(self) -> None:
"""games_done_since_start=0 → pace stays 0 even if total games_done > 0."""
started = datetime.now(timezone.utc) - timedelta(days=50) started = datetime.now(timezone.utc) - timedelta(days=50)
info = _state_info( info = _state_info(
State(enforcement_started_at=started.isoformat()), games_done=0 State(enforcement_started_at=started.isoformat()),
games_done=5,
games_done_since_start=0,
) )
assert info.days_elapsed >= 49 assert info.days_elapsed >= 49
assert info.pace_games_per_day == 0.0 assert info.pace_games_per_day == 0.0
@ -324,7 +425,196 @@ class TestDatasetToPayload:
"state", "state",
"defaults", "defaults",
"default_summary", "default_summary",
"pace_vs_hltb",
"generated_at", "generated_at",
} }
assert isinstance(payload["games"], list) assert isinstance(payload["games"], list)
assert isinstance(payload["state"], dict) assert isinstance(payload["state"], dict)
def _complete_game(
app_id: int = 1,
playtime_minutes: int = 600,
) -> GameInfo:
"""Complete game (100 % achievements, has playtime)."""
return GameInfo(
app_id=app_id,
name=f"Done{app_id}",
total_achievements=10,
unlocked_achievements=10,
playtime_minutes=playtime_minutes,
completionist_hours=0.0,
comp_100_count=5,
count_comp=20,
)
class TestComputePaceVsHLTB:
"""Tests for compute_pace_vs_hltb — 100 % branch coverage."""
def test_no_completed_games_returns_none(self) -> None:
incomplete = _gi(app_id=1, total_achievements=10, unlocked_achievements=0)
assert compute_pace_vs_hltb([incomplete], {}) is None
def test_complete_but_zero_playtime_ignored(self) -> None:
game = _complete_game(playtime_minutes=0)
assert compute_pace_vs_hltb([game], {}) is None
def test_no_rush_data_in_cache_returns_none(self) -> None:
game = _complete_game(app_id=1)
# cache has hours but no rush_hours
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": -1,
"leisure_100h": -1,
"hltb_game_id": 0,
}
}
assert compute_pace_vs_hltb([game], cache) is None
def test_rush_only_ratio_computed(self) -> None:
"""With rush but no leisure, ratio_vs_rush is computed, interpolation_t = -1."""
game = _complete_game(app_id=1, playtime_minutes=600) # 10h actual
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": 8.0,
"leisure_100h": -1,
"hltb_game_id": 0,
}
}
result = compute_pace_vs_hltb([game], cache)
assert result is not None
assert result.calibration_count == 1
assert result.ratio_vs_rush == round(10.0 / 8.0, 3)
assert result.ratio_vs_leisure == -1.0
assert result.interpolation_t == -1.0
def test_rush_only_style_faster_than_rush_when_ratio_below_one(self) -> None:
"""Plays faster than rush (actual < rush) → style = faster_than_rush."""
game = _complete_game(app_id=1, playtime_minutes=300) # 5h actual
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": 8.0,
"leisure_100h": -1,
"hltb_game_id": 0,
}
}
result = compute_pace_vs_hltb([game], cache)
assert result is not None
assert result.player_style == "faster_than_rush"
def test_rush_only_style_unknown_when_ratio_at_or_above_one(self) -> None:
"""Without leisure data and ratio >= 1 → style = unknown."""
game = _complete_game(app_id=1, playtime_minutes=600) # 10h
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": 8.0,
"leisure_100h": -1,
"hltb_game_id": 0,
}
}
result = compute_pace_vs_hltb([game], cache)
assert result is not None
assert result.player_style == "unknown"
def test_both_rush_and_leisure_interpolation_computed(self) -> None:
"""With both rush + leisure, interpolation_t is computed."""
# actual=10h, rush=8h, leisure=20h → t = (10-8)/(20-8) = 2/12 ≈ 0.167
game = _complete_game(app_id=1, playtime_minutes=600)
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": 8.0,
"leisure_100h": 20.0,
"hltb_game_id": 0,
}
}
result = compute_pace_vs_hltb([game], cache)
assert result is not None
assert result.interpolation_t == round((10.0 - 8.0) / (20.0 - 8.0), 3)
assert result.ratio_vs_leisure == round(10.0 / 20.0, 3)
assert result.player_style == "rush_to_leisure"
def test_style_faster_than_rush_when_t_negative(self) -> None:
"""t < 0 means faster than rush."""
game = _complete_game(app_id=1, playtime_minutes=300) # 5h actual
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": 8.0,
"leisure_100h": 20.0,
"hltb_game_id": 0,
}
}
result = compute_pace_vs_hltb([game], cache)
assert result is not None
assert result.interpolation_t < 0
assert result.player_style == "faster_than_rush"
def test_style_slower_than_leisure_when_t_above_one(self) -> None:
"""t > 1 means slower than leisure."""
game = _complete_game(app_id=1, playtime_minutes=1500) # 25h actual
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": 8.0,
"leisure_100h": 20.0,
"hltb_game_id": 0,
}
}
result = compute_pace_vs_hltb([game], cache)
assert result is not None
assert result.interpolation_t > 1.0
assert result.player_style == "slower_than_leisure"
def test_interpolation_t_minus_one_when_leisure_not_greater_than_rush(self) -> None:
"""Edge case: leisure <= rush, can't divide, interpolation_t = -1."""
game = _complete_game(app_id=1, playtime_minutes=600)
# leisure == rush → denominator = 0
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": 8.0,
"leisure_100h": 8.0,
"hltb_game_id": 0,
}
}
result = compute_pace_vs_hltb([game], cache)
assert result is not None
assert result.interpolation_t == -1.0
def test_pace_vs_hltb_is_dataclass(self) -> None:
"""Return type is PaceVsHLTB."""
game = _complete_game(app_id=1)
cache = {
1: {
"hours": 10.0,
"polls": 5,
"count_comp": 20,
"rush_hours": 8.0,
"leisure_100h": 20.0,
"hltb_game_id": 0,
}
}
result = compute_pace_vs_hltb([game], cache)
assert isinstance(result, PaceVsHLTB)

View File

@ -90,7 +90,7 @@ function App() {
·{' '} ·{' '}
</> </>
)} )}
{dataset.state.games_done} games finished since{' '} {dataset.state.games_done_since_start} games finished since{' '}
{dataset.state.enforcement_started_at.slice(0, 10) || '—'} ·{' '} {dataset.state.enforcement_started_at.slice(0, 10) || '—'} ·{' '}
{dataset.games.length} candidates {dataset.games.length} candidates
</p> </p>
@ -112,6 +112,7 @@ function App() {
state={dataset.state} state={dataset.state}
presets={dataset.defaults.hours_per_day_presets} presets={dataset.defaults.hours_per_day_presets}
defaultQualifying={dataset.default_summary.qualifying} defaultQualifying={dataset.default_summary.qualifying}
paceVsHltb={dataset.pace_vs_hltb}
/> />
<TimelineChart result={result} filters={filters} state={dataset.state} /> <TimelineChart result={result} filters={filters} state={dataset.state} />
<GameTable <GameTable

View File

@ -2,12 +2,13 @@ import { render, screen } from '@testing-library/react'
import { describe, expect, it } from 'vitest' import { describe, expect, it } from 'vitest'
import { applyFilters } from '../estimate' import { applyFilters } from '../estimate'
import type { Filters } from '../types' import type { Filters } from '../types'
import { makeDataset, makeFilters, makeGame, makeState } from '../test/factories' import { makeDataset, makeFilters, makeGame, makePaceVsHltb, makeState } from '../test/factories'
import { SummaryCards } from './SummaryCards' import { SummaryCards } from './SummaryCards'
function renderCards(filtersOver: Partial<Filters> = {}, statePace = 0) { function renderCards(filtersOver: Partial<Filters> = {}, statePace = 0, paceVsHltb = null) {
const filters = makeFilters(filtersOver) const filters = makeFilters(filtersOver)
const result = applyFilters(makeDataset([makeGame({ app_id: 1 })]), filters) const dataset = makeDataset([makeGame({ app_id: 1 })], { pace_vs_hltb: paceVsHltb })
const result = applyFilters(dataset, filters)
render( render(
<SummaryCards <SummaryCards
result={result} result={result}
@ -15,6 +16,7 @@ function renderCards(filtersOver: Partial<Filters> = {}, statePace = 0) {
state={makeState({ pace_games_per_day: statePace })} state={makeState({ pace_games_per_day: statePace })}
presets={[2, 4, 6, 8]} presets={[2, 4, 6, 8]}
defaultQualifying={result.remainingGames} defaultQualifying={result.remainingGames}
paceVsHltb={paceVsHltb}
/>, />,
) )
return result return result
@ -45,3 +47,30 @@ describe('SummaryCards', () => {
expect(banner?.textContent).toMatch(/games\/day/i) expect(banner?.textContent).toMatch(/games\/day/i)
}) })
}) })
describe('PlayerSpeedInsight', () => {
it('shows empty state message when paceVsHltb is null', () => {
renderCards({}, 0, null)
expect(screen.getByText(/Your Play Style vs HLTB/i)).toBeInTheDocument()
expect(screen.getByText(/No calibration data yet/i)).toBeInTheDocument()
})
it('shows calibration stats when paceVsHltb is provided', () => {
renderCards({}, 0, makePaceVsHltb())
expect(screen.getByText(/Calibration games/i)).toBeInTheDocument()
expect(screen.getByText(/vs Rush speed/i)).toBeInTheDocument()
expect(screen.getByText('Play style')).toBeInTheDocument()
expect(screen.getByText(/Between rush and leisure/i)).toBeInTheDocument()
})
it('shows estimated total when calibration data is present', () => {
// rushTotal = 10, leisureTotal = 20, t = 0.05 → 10 + 0.05 * 10 = 10.5 h
renderCards({}, 0, makePaceVsHltb({ interpolation_t: 0.05 }))
expect(screen.getByText(/Estimated total at your pace/i)).toBeInTheDocument()
})
it('hides leisure ratio when ratio_vs_leisure is -1', () => {
renderCards({}, 0, makePaceVsHltb({ ratio_vs_leisure: -1, interpolation_t: -1 }))
expect(screen.queryByText(/vs Leisure speed/i)).not.toBeInTheDocument()
})
})

View File

@ -1,7 +1,7 @@
import { basisTotal, etaDays, paceDays } from '../estimate' import { basisTotal, etaDays, paceDays, playerEstimatedTotal } from '../estimate'
import type { EstimateResult } from '../estimate' import type { EstimateResult } from '../estimate'
import { daysUntil, fmtEta, fmtHours } from '../format' import { fmtEta, fmtHours } from '../format'
import type { EstimateBasis, Filters, WebStateInfo } from '../types' import type { EstimateBasis, Filters, PaceVsHLTB, WebStateInfo } from '../types'
interface Props { interface Props {
result: EstimateResult result: EstimateResult
@ -9,6 +9,7 @@ interface Props {
state: WebStateInfo state: WebStateInfo
presets: number[] presets: number[]
defaultQualifying: number defaultQualifying: number
paceVsHltb: PaceVsHLTB | null
} }
interface CardData { interface CardData {
@ -24,9 +25,18 @@ const CARDS: CardData[] = [
{ basis: 'pace', title: 'At your pace', blurb: 'Based on games finished' }, { basis: 'pace', title: 'At your pace', blurb: 'Based on games finished' },
] ]
const STYLE_LABELS: Record<string, string> = {
faster_than_rush: 'Faster than rush',
rush_to_leisure: 'Between rush and leisure',
slower_than_leisure: 'Slower than leisure',
unknown: 'Unknown',
}
function TargetBanner({ result, filters }: Props) { function TargetBanner({ result, filters }: Props) {
if (!filters.targetDate) return null if (!filters.targetDate) return null
const days = Math.max(1, daysUntil(filters.targetDate)) const now = new Date()
const target = new Date(filters.targetDate)
const days = Math.max(1, Math.ceil((target.getTime() - now.getTime()) / 86400000))
let need: string let need: string
if (filters.basis === 'pace') { if (filters.basis === 'pace') {
const perDay = result.remainingGames / days const perDay = result.remainingGames / days
@ -44,8 +54,91 @@ function TargetBanner({ result, filters }: Props) {
) )
} }
function PlayerSpeedInsight({
result,
paceVsHltb,
presets,
}: Pick<Props, 'result' | 'paceVsHltb' | 'presets'>) {
const pace = paceVsHltb
const estimated = playerEstimatedTotal(result.rushTotal, result.leisureTotal, pace)
if (!pace || pace.calibration_count === 0) {
return (
<div className="player-insight player-insight--empty">
<div className="player-insight-title">Your Play Style vs HLTB</div>
<p className="player-insight-empty">
No calibration data yet. Finish games (100% achievements) and re-run{' '}
<code>stats</code> to see your pace estimate.
</p>
</div>
)
}
return (
<div className="player-insight">
<div className="player-insight-title">Your Play Style vs HLTB</div>
<div className="player-insight-grid">
<span>Calibration games</span>
<span>{pace.calibration_count}</span>
{pace.ratio_vs_rush !== -1 && (
<>
<span>vs Rush speed</span>
<span
className={pace.ratio_vs_rush <= 1 ? 'player-insight-fast' : 'player-insight-slow'}
>
{pace.ratio_vs_rush.toFixed(2)}×
</span>
</>
)}
{pace.ratio_vs_leisure !== -1 && (
<>
<span>vs Leisure speed</span>
<span>{pace.ratio_vs_leisure.toFixed(2)}×</span>
</>
)}
{pace.interpolation_t !== -1 && (
<>
<span>Interpolation t</span>
<span title="0 = rush speed · 1 = leisure speed">
{pace.interpolation_t.toFixed(3)}
</span>
</>
)}
<span>Play style</span>
<span className={`player-insight-style player-insight-style--${pace.player_style}`}>
{STYLE_LABELS[pace.player_style] ?? pace.player_style}
</span>
</div>
{estimated !== null && (
<div className="player-insight-estimate">
<div className="player-insight-estimate-total">
Estimated total at your pace:{' '}
<strong className="player-insight-estimate-hours">{fmtHours(estimated)}</strong>
</div>
<div className="presets">
{presets.map((h) => {
const days = estimated > 0 ? Math.floor(estimated / h) : null
return (
<div key={h} className="preset">
<span>{h} h/day</span>
<span>{fmtEta(days)}</span>
</div>
)
})}
</div>
</div>
)}
</div>
)
}
export function SummaryCards(props: Props) { export function SummaryCards(props: Props) {
const { result, filters, state, presets } = props const { result, filters, state, presets, paceVsHltb } = props
return ( return (
<div className="summary"> <div className="summary">
@ -101,6 +194,8 @@ export function SummaryCards(props: Props) {
) )
})} })}
</div> </div>
<PlayerSpeedInsight result={result} paceVsHltb={paceVsHltb} presets={presets} />
</div> </div>
) )
} }

View File

@ -1,6 +1,6 @@
import { describe, expect, it } from 'vitest' import { describe, expect, it } from 'vitest'
import { applyFilters, basisTotal, etaDays, paceDays } from './estimate' import { applyFilters, basisTotal, etaDays, paceDays, playerEstimatedTotal } from './estimate'
import { makeDataset, makeFilters, makeGame } from './test/factories' import { makeDataset, makeFilters, makeGame, makePaceVsHltb } from './test/factories'
describe('applyFilters — totals and parity', () => { describe('applyFilters — totals and parity', () => {
it('sums each metric independently over qualifying games', () => { it('sums each metric independently over qualifying games', () => {
@ -152,6 +152,47 @@ describe('basis length proxy', () => {
}) })
}) })
describe('playerEstimatedTotal', () => {
it('returns null when pace is null', () => {
expect(playerEstimatedTotal(100, 200, null)).toBeNull()
})
it('returns null when calibration_count is 0', () => {
expect(playerEstimatedTotal(100, 200, makePaceVsHltb({ calibration_count: 0 }))).toBeNull()
})
it('uses interpolation_t when it is not -1', () => {
// rush=100, leisure=200, t=0.5 → 100 + 0.5 * 100 = 150
expect(playerEstimatedTotal(100, 200, makePaceVsHltb({ interpolation_t: 0.5 }))).toBe(150)
})
it('falls back to ratio_vs_rush when interpolation_t is -1', () => {
// rush=100, ratio=1.2 → 120
expect(
playerEstimatedTotal(
100,
200,
makePaceVsHltb({ interpolation_t: -1, ratio_vs_rush: 1.2 }),
),
).toBe(120)
})
it('returns null when interpolation_t is -1 and ratio_vs_rush is -1', () => {
expect(
playerEstimatedTotal(
100,
200,
makePaceVsHltb({ interpolation_t: -1, ratio_vs_rush: -1 }),
),
).toBeNull()
})
it('handles faster-than-rush (t < 0) correctly', () => {
// rush=100, leisure=200, t=-0.1 → 100 + (-0.1) * 100 = 90
expect(playerEstimatedTotal(100, 200, makePaceVsHltb({ interpolation_t: -0.1 }))).toBe(90)
})
})
describe('etaDays / paceDays / basisTotal', () => { describe('etaDays / paceDays / basisTotal', () => {
it('etaDays floors and guards zero inputs', () => { it('etaDays floors and guards zero inputs', () => {
expect(etaDays(40, 4)).toBe(10) expect(etaDays(40, 4)).toBe(10)

View File

@ -3,7 +3,7 @@
// default thresholds, the totals reproduce the `stats` command exactly. // default thresholds, the totals reproduce the `stats` command exactly.
import { isPlayable, passesMinTier } from './protondb' import { isPlayable, passesMinTier } from './protondb'
import type { EstimateBasis, Filters, WebDataset, WebGame } from './types' import type { EstimateBasis, Filters, PaceVsHLTB, WebDataset, WebGame } from './types'
export interface GameRow { export interface GameRow {
game: WebGame game: WebGame
@ -148,6 +148,27 @@ export function paceDays(remaining: number, pace: number): number | null {
return Math.floor(remaining / pace) return Math.floor(remaining / pace)
} }
/**
* Estimate the player's personal backlog total from their calibrated pace.
*
* Uses interpolation_t when leisure data exists, falls back to ratio_vs_rush
* otherwise. Returns null when there is no calibration data.
*/
export function playerEstimatedTotal(
rushTotal: number,
leisureTotal: number,
pace: PaceVsHLTB | null,
): number | null {
if (!pace || pace.calibration_count === 0) return null
if (pace.interpolation_t !== -1) {
return rushTotal + pace.interpolation_t * (leisureTotal - rushTotal)
}
if (pace.ratio_vs_rush !== -1) {
return rushTotal * pace.ratio_vs_rush
}
return null
}
/** Total hours for the selected basis, or null for the pace (count) basis. */ /** Total hours for the selected basis, or null for the pace (count) basis. */
export function basisTotal( export function basisTotal(
result: EstimateResult, result: EstimateResult,

View File

@ -312,6 +312,68 @@ details[open] summary {
padding: 1px 0; padding: 1px 0;
} }
/* ── Player Speed Insight ── */
.player-insight {
margin-top: 16px;
padding: 14px;
background: var(--panel-2);
border: 1px solid var(--border);
border-radius: 8px;
}
.player-insight--empty {
opacity: 0.7;
}
.player-insight-title {
font-weight: 600;
color: var(--heading);
margin-bottom: 10px;
font-size: 14px;
}
.player-insight-empty {
font-size: 13px;
color: var(--muted);
margin: 0;
}
.player-insight-grid {
display: grid;
grid-template-columns: max-content 1fr;
gap: 4px 20px;
font-size: 13px;
color: var(--text);
margin-bottom: 12px;
}
.player-insight-grid span:nth-child(odd) {
color: var(--muted);
}
.player-insight-fast {
color: var(--accent-2);
font-weight: 600;
}
.player-insight-slow {
color: var(--warn);
font-weight: 600;
}
.player-insight-style--faster_than_rush {
color: var(--accent-2);
}
.player-insight-style--slower_than_leisure {
color: var(--warn);
}
.player-insight-estimate {
border-top: 1px solid var(--border);
padding-top: 10px;
}
.player-insight-estimate-total {
font-size: 13px;
color: var(--text);
margin-bottom: 8px;
}
.player-insight-estimate-hours {
color: var(--accent);
font-family: var(--mono);
font-size: 16px;
}
/* ── Chart ── */ /* ── Chart ── */
.chart { .chart {
background: var(--panel); background: var(--panel);

View File

@ -1,7 +1,7 @@
// Shared test factories. Lives under src/test/ which is excluded from the // Shared test factories. Lives under src/test/ which is excluded from the
// app build and from coverage. // app build and from coverage.
import type { Filters, WebDataset, WebGame, WebStateInfo } from '../types' import type { Filters, PaceVsHLTB, WebDataset, WebGame, WebStateInfo } from '../types'
export function makeGame(over: Partial<WebGame> = {}): WebGame { export function makeGame(over: Partial<WebGame> = {}): WebGame {
return { return {
@ -55,6 +55,17 @@ export function makeState(over: Partial<WebStateInfo> = {}): WebStateInfo {
} }
} }
export function makePaceVsHltb(over: Partial<PaceVsHLTB> = {}): PaceVsHLTB {
return {
calibration_count: 10,
ratio_vs_rush: 1.05,
ratio_vs_leisure: 0.4,
interpolation_t: 0.05,
player_style: 'rush_to_leisure',
...over,
}
}
export function makeDataset( export function makeDataset(
games: WebGame[] = [makeGame()], games: WebGame[] = [makeGame()],
over: Partial<WebDataset> = {}, over: Partial<WebDataset> = {},
@ -75,6 +86,7 @@ export function makeDataset(
leisure_total: 0, leisure_total: 0,
worst_total: 0, worst_total: 0,
}, },
pace_vs_hltb: null,
generated_at: '2026-05-29T00:00:00+00:00', generated_at: '2026-05-29T00:00:00+00:00',
...over, ...over,
} }

View File

@ -22,6 +22,7 @@ export interface WebStateInfo {
current_app_id: number | null current_app_id: number | null
current_game_name: string current_game_name: string
games_done: number games_done: number
games_done_since_start: number
days_elapsed: number days_elapsed: number
enforcement_started_at: string enforcement_started_at: string
pace_games_per_day: number pace_games_per_day: number
@ -42,11 +43,23 @@ export interface DefaultSummary {
worst_total: number worst_total: number
} }
export interface PaceVsHLTB {
calibration_count: number
/** -1 = no data */
ratio_vs_rush: number
/** -1 = no data */
ratio_vs_leisure: number
/** Position between rush (0) and leisure (1) speed; -1 = unknown */
interpolation_t: number
player_style: 'faster_than_rush' | 'rush_to_leisure' | 'slower_than_leisure' | 'unknown'
}
export interface WebDataset { export interface WebDataset {
games: WebGame[] games: WebGame[]
state: WebStateInfo state: WebStateInfo
defaults: WebDefaults defaults: WebDefaults
default_summary: DefaultSummary default_summary: DefaultSummary
pace_vs_hltb: PaceVsHLTB | null
generated_at: string generated_at: string
} }