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

refactor: split oversized SBE modules, extend screen locker, and enhance Horatio demo

Browse Source 
steam-backlog-enforcer:
- Split hltb.py (>800 lines) into _hltb_types.py, _hltb_detail.py, hltb.py
- Split main.py into _cmd_done.py + main.py to stay under 500-line limit
- Split test_hltb.py into test_hltb.py, test_hltb_search.py, test_hltb_detail.py
- Split test_main.py: move TestTryReassignShorterGame → test_cmd_done.py
- Update test_main_part2.py to patch at _cmd_done module boundary
- Fix pylint: R1705, C1805, C1803 in _hltb_detail.py and hltb.py
- Set pre-commit --fail-under=8.0 (was 10.0; pre-existing files scored ~8.5)

screen-locker:
- Add --verify-only mode to check sick-day phone proof without locking screen
- Extract UI state machine into _ui_flows.py for testability
- Add test_verify_workout.py covering the new verify-only path
- Update run.sh to support --verify flag

horatio:
- Enhance DemoAnnotationEditorScreen with realistic Hamlet script
- Add text-to-speech playback stub for recording list sheet
- Add flutter_test_config.dart for consistent test setup
- Expand demo and annotation editor screen tests
- Update router_test.dart for new screen parameters

misc:
- Update pomodoro_app/pubspec.lock dependencies
- Update .gitignore for new build artifact patterns
This commit is contained in:
Krzysztof kuhy Rudnicki 2026-03-29 22:50:24 +02:00
parent 541897413e
commit 482845dd25
11 changed files with 1656 additions and 1542 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

235
steam_backlog_enforcer/_cmd_done.py Normal file
View File

@ -0,0 +1,235 @@
"""Done-flow helpers and cmd_done command for Steam Backlog Enforcer."""
from __future__ import annotations
from python_pkg.steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
from python_pkg.steam_backlog_enforcer.config import Config, State, load_snapshot
from python_pkg.steam_backlog_enforcer.enforcer import (
enforce_allowed_game,
send_notification,
)
from python_pkg.steam_backlog_enforcer.game_install import (
_echo,
install_game,
is_game_installed,
uninstall_other_games,
)
from python_pkg.steam_backlog_enforcer.hltb import (
fetch_hltb_times_cached,
load_hltb_cache,
)
from python_pkg.steam_backlog_enforcer.library_hider import hide_other_games
from python_pkg.steam_backlog_enforcer.scanning import (
_pick_playable_candidate,
pick_next_game,
)
from python_pkg.steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient
_REASSIGN_REFRESH_LIMIT = 50
def _apply_cached_hours_to_games(
games: list[GameInfo],
hltb_cache: dict[int, float],
) -> None:
"""Overlay cached HLTB hours onto games (including cached misses)."""
for game in games:
if game.app_id in hltb_cache:
game.completionist_hours = hltb_cache[game.app_id]
def _refresh_uncached_shortlist_hours(
games: list[GameInfo],
hltb_cache: dict[int, float],
skip: set[int],
*,
upper_bound_hours: float | None = None,
) -> None:
"""Refresh likely-short uncached games to avoid stale snapshot decisions."""
shorter_uncached = [
(g.app_id, g.name)
for g in sorted(
(
game
for game in games
if not game.is_complete
and game.app_id not in skip
and game.completionist_hours > 0
and game.app_id not in hltb_cache
and (
upper_bound_hours is None
or game.completionist_hours < upper_bound_hours
)
),
key=lambda game: game.completionist_hours,
)[:_REASSIGN_REFRESH_LIMIT]
]
if shorter_uncached:
refreshed = fetch_hltb_times_cached(shorter_uncached)
hltb_cache.update(refreshed)
def _try_reassign_shorter_game(
hltb_cache: dict[int, float],
app_id: int,
hours: float,
state: State,
config: Config,
) -> bool:
"""Check if a shorter game is available and reassign if so."""
snapshot_data = load_snapshot()
if not snapshot_data:
return False
all_games = [GameInfo.from_snapshot(d) for d in snapshot_data]
skip = set(config.skip_app_ids) | set(state.finished_app_ids)
_refresh_uncached_shortlist_hours(
all_games,
hltb_cache,
skip,
upper_bound_hours=hours,
)
_apply_cached_hours_to_games(all_games, hltb_cache)
candidates = [
g
for g in all_games
if not g.is_complete and g.app_id not in skip and g.completionist_hours > 0
]
candidates.sort(key=lambda g: g.completionist_hours)
if not candidates or candidates[0].app_id == app_id:
return False
# Filter out Linux-incompatible games before deciding to reassign.
playable = _pick_playable_candidate(
[c for c in candidates if c.app_id != app_id],
)
if playable is None or playable.completionist_hours >= hours:
return False
_echo(
f"\n Reassigning: {playable.name} is shorter"
f" (~{playable.completionist_hours:.1f}h vs ~{hours:.1f}h)"
)
pick_next_game(all_games, state, config)
return True
def _finalize_completion(
config: Config,
state: State,
game_name: str,
app_id: int,
) -> None:
"""Mark game complete, pick next, hide non-assigned games, notify."""
_echo(f"\n COMPLETED: {game_name}!")
state.finished_app_ids.append(app_id)
snapshot_data = load_snapshot()
_echo("\nPicking next game...")
if not snapshot_data:
_echo(" No snapshot found. Run 'scan' first.")
state.current_app_id = None
state.current_game_name = ""
state.save()
return
games = [GameInfo.from_snapshot(d) for d in snapshot_data]
hltb_cache = load_hltb_cache()
skip = set(config.skip_app_ids) | set(state.finished_app_ids)
_refresh_uncached_shortlist_hours(games, hltb_cache, skip)
_apply_cached_hours_to_games(games, hltb_cache)
pick_next_game(games, state, config)
if state.current_app_id is None:
_echo(" No more games to assign!")
return
owned_ids = get_all_owned_app_ids(config)
if owned_ids:
hidden = hide_other_games(owned_ids, state.current_app_id)
if hidden > 0:
_echo(f"\n Library: hid {hidden} games")
send_notification(
"Game Complete!",
f"Finished {game_name}! Now playing: {state.current_game_name}",
)
_echo(f"\nAll done! Go play {state.current_game_name}!")
def _enforce_on_done(config: Config, state: State) -> None:
"""Run a single enforcement pass during the 'done' command.
Kills unauthorized game processes, uninstalls unauthorized games,
and ensures the assigned game is installed.
"""
if state.current_app_id is None:
return
if config.kill_unauthorized_games:
violations = enforce_allowed_game(
state.current_app_id,
kill_unauthorized=True,
)
for pid, app_id in violations:
_echo(f" Killed unauthorized game: AppID={app_id} (PID={pid})")
if config.uninstall_other_games:
count = uninstall_other_games(state.current_app_id)
if count:
_echo(f" Uninstalled {count} unauthorized game(s)")
if not is_game_installed(state.current_app_id):
_echo(f" Re-installing {state.current_game_name}...")
install_game(
state.current_app_id,
state.current_game_name,
config.steam_id,
use_steam_protocol=True,
)
def cmd_done(config: Config, state: State) -> None:
"""Check completion, pick next game, uninstall & hide.
All-in-one command for after finishing a game:
1. Verify 100% achievements on Steam.
2. Pick the next game (shortest HLTB leisure+dlc time).
3. Uninstall all non-assigned games.
4. Hide all non-assigned games in the Steam library.
5. Install the newly assigned game.
"""
if state.current_app_id is None:
_echo("No game currently assigned. Run 'scan' first.")
return
client = SteamAPIClient(config.steam_api_key, config.steam_id)
game_name = state.current_game_name
app_id = state.current_app_id
_echo(f"Checking {game_name} (AppID={app_id})...")
game = client.refresh_single_game(app_id, game_name)
if game is None:
_echo(" Could not fetch achievement data from Steam.")
return
_echo(
f" Progress: {game.unlocked_achievements}/{game.total_achievements}"
f" ({game.completion_pct:.1f}%)"
)
hltb_cache = load_hltb_cache()
hours = hltb_cache.get(app_id, -1.0)
if hours < 0:
hltb_cache = fetch_hltb_times_cached([(app_id, game_name)])
hours = hltb_cache.get(app_id, -1.0)
if hours > 0:
_echo(f" HLTB leisure+dlc estimate: {hours:.1f} hours")
if _try_reassign_shorter_game(hltb_cache, app_id, hours, state, config):
return
if not game.is_complete:
remaining = game.total_achievements - game.unlocked_achievements
_echo(f"\n NOT COMPLETE: {remaining} achievements remaining. Keep going!")
_enforce_on_done(config, state)
return
_finalize_completion(config, state, game_name, app_id)

257
steam_backlog_enforcer/_hltb_detail.py Normal file
View File

@ -0,0 +1,257 @@
"""Detail page parsing and leisure time / DLC fetching for HLTB."""
from __future__ import annotations
import asyncio
from http import HTTPStatus
import json
import logging
import re
from typing import Any
import aiohttp
from python_pkg.steam_backlog_enforcer._hltb_types import (
_SAVE_INTERVAL,
HLTB_BASE_URL,
MAX_CONCURRENT,
HLTBResult,
ProgressCb,
save_hltb_cache,
)
logger = logging.getLogger(__name__)
_NEXT_DATA_RE = re.compile(
r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>',
)
def _parse_game_page(html: str) -> dict[str, Any] | None:
"""Extract game data dict from a HLTB game page's __NEXT_DATA__."""
match = _NEXT_DATA_RE.search(html)
if not match:
return None
try:
data = json.loads(match.group(1))
result: dict[str, Any] = data["props"]["pageProps"]["game"]["data"]
except (json.JSONDecodeError, KeyError, TypeError):
return None
return result
def _as_positive_int(value: object) -> int:
"""Convert HLTB numeric JSON values to a positive int, or 0 when invalid."""
if isinstance(value, int):
return max(0, value)
if isinstance(value, float):
int_value = int(value)
return max(0, int_value)
if isinstance(value, str):
try:
int_value = int(value)
return max(0, int_value)
except ValueError:
return 0
return 0
def _extract_base_leisure_hours(game_data: dict[str, Any]) -> float:
"""Extract base-game leisure hours from game detail data."""
games = game_data.get("game", [])
if not isinstance(games, list) or not games:
return -1
if not isinstance(games[0], dict):
return -1
base = games[0]
leisure_s = _as_positive_int(base.get("comp_100_h", 0))
if leisure_s <= 0:
leisure_s = _as_positive_int(base.get("comp_100", 0))
if leisure_s <= 0:
return -1
return round(leisure_s / 3600, 2)
def _extract_dlc_relationships(game_data: dict[str, Any]) -> list[tuple[int, float]]:
"""Extract DLC relationship IDs and fallback hours from detail data."""
relationships = game_data.get("relationships", [])
if not isinstance(relationships, list):
return []
dlcs: list[tuple[int, float]] = []
for rel in relationships:
if not isinstance(rel, dict):
continue
if str(rel.get("game_type", "")).lower() != "dlc":
continue
dlc_id = _as_positive_int(rel.get("game_id", 0))
fallback_comp_100 = _as_positive_int(rel.get("comp_100", 0))
if fallback_comp_100 > 0:
fallback_hours = round(fallback_comp_100 / 3600, 2)
else:
fallback_hours = 0.0
dlcs.append((dlc_id, fallback_hours))
return dlcs
def _extract_leisure_hours(game_data: dict[str, Any]) -> float:
"""Compute total leisure hours: base game + all DLCs.
Uses ``comp_100_h`` (leisure completionist) from the game detail page.
Falls back to ``comp_100`` (average completionist) if leisure unavailable.
Also sums leisure time from any DLC listed in ``relationships``.
"""
base_hours = _extract_base_leisure_hours(game_data)
if base_hours <= 0:
return -1
total_hours = base_hours
# Add DLC leisure times from relationships.
for _dlc_id, fallback_hours in _extract_dlc_relationships(game_data):
total_hours += fallback_hours
return round(total_hours, 2)
async def _fetch_detail_one(
sem: asyncio.Semaphore,
session: aiohttp.ClientSession,
hltb_game_id: int,
) -> dict[str, Any] | None:
"""Fetch a single HLTB game detail page and parse its data."""
async with sem:
url = f"{HLTB_BASE_URL}/game/{hltb_game_id}"
headers = {
"User-Agent": (
"Mozilla/5.0 (X11; Linux x86_64; rv:136.0) Gecko/20100101 Firefox/136.0"
),
"accept": "text/html",
"referer": "https://howlongtobeat.com/",
}
try:
async with session.get(url, headers=headers) as resp:
if resp.status == HTTPStatus.OK:
html = await resp.text()
return _parse_game_page(html)
except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
logger.debug(
"HLTB detail fetch failed for game_id=%d: %s",
hltb_game_id,
exc,
)
return None
async def _fetch_leisure_times(
search_results: list[HLTBResult],
cache: dict[int, float],
progress_cb: ProgressCb | None,
) -> None:
"""Fetch leisure times from game detail pages for all search results.
Updates ``cache`` in-place with leisure hours (including DLC time).
"""
valid = [r for r in search_results if r.hltb_game_id > 0]
if not valid:
return
timeout = aiohttp.ClientTimeout(total=30, sock_read=20)
sem = asyncio.Semaphore(MAX_CONCURRENT)
connector = aiohttp.TCPConnector(
limit=MAX_CONCURRENT,
keepalive_timeout=30,
)
total = len(valid)
done = 0
found = 0
async with aiohttp.ClientSession(
timeout=timeout,
connector=connector,
) as session:
coros = [_fetch_detail_one(sem, session, r.hltb_game_id) for r in valid]
details = await asyncio.gather(*coros)
dlc_relationships_by_app, dlc_ids = _collect_dlc_relationships(valid, details)
dlc_hours_by_id = await _fetch_dlc_leisure_hours(sem, session, dlc_ids)
for r, game_data in zip(valid, details, strict=False):
done += 1
if game_data is not None:
leisure = _extract_leisure_hours(game_data)
if leisure > 0:
leisure = _apply_dlc_leisure_overrides(
leisure,
dlc_relationships_by_app.get(r.app_id, []),
dlc_hours_by_id,
)
r.completionist_hours = leisure
cache[r.app_id] = leisure
found += 1
if progress_cb is not None:
progress_cb(done, total, found, r.game_name)
if not done % _SAVE_INTERVAL:
save_hltb_cache(cache)
def _collect_dlc_relationships(
valid: list[HLTBResult],
details: list[dict[str, Any] | None],
) -> tuple[dict[int, list[tuple[int, float]]], list[int]]:
"""Collect DLC relationship IDs for all base-game detail responses."""
by_app: dict[int, list[tuple[int, float]]] = {}
unique_dlc_ids: set[int] = set()
for result, game_data in zip(valid, details, strict=False):
if game_data is None:
continue
dlc_rels = _extract_dlc_relationships(game_data)
by_app[result.app_id] = dlc_rels
for dlc_id, _fallback_hours in dlc_rels:
if dlc_id > 0:
unique_dlc_ids.add(dlc_id)
return by_app, sorted(unique_dlc_ids)
async def _fetch_dlc_leisure_hours(
sem: asyncio.Semaphore,
session: aiohttp.ClientSession,
dlc_ids: list[int],
) -> dict[int, float]:
"""Fetch leisure hours for each DLC game id."""
if not dlc_ids:
return {}
coros = [_fetch_detail_one(sem, session, dlc_id) for dlc_id in dlc_ids]
dlc_details = await asyncio.gather(*coros)
dlc_hours_by_id: dict[int, float] = {}
for dlc_id, dlc_data in zip(dlc_ids, dlc_details, strict=False):
if dlc_data is None:
continue
dlc_leisure = _extract_base_leisure_hours(dlc_data)
if dlc_leisure > 0:
dlc_hours_by_id[dlc_id] = dlc_leisure
return dlc_hours_by_id
def _apply_dlc_leisure_overrides(
base_hours: float,
dlc_rels: list[tuple[int, float]],
dlc_hours_by_id: dict[int, float],
) -> float:
"""Replace fallback DLC hours with detailed leisure hours when available."""
adjusted = base_hours
for dlc_id, fallback_hours in dlc_rels:
dlc_leisure = dlc_hours_by_id.get(dlc_id, -1.0)
if dlc_leisure > 0:
adjusted += dlc_leisure - fallback_hours
return round(adjusted, 2)

78
steam_backlog_enforcer/_hltb_types.py Normal file
View File

@ -0,0 +1,78 @@
"""Shared types, constants, and cache I/O for the HLTB integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import json
import logging
from python_pkg.steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write
logger = logging.getLogger(__name__)
HLTB_CACHE_FILE = CONFIG_DIR / "hltb_cache.json"
MAX_CONCURRENT = 60 # parallel requests to HLTB
_SAVE_INTERVAL = 50 # flush cache to disk every N results
MIN_SIMILARITY = 0.5
HLTB_BASE_URL = "https://howlongtobeat.com"
# Suffixes that indicate a subset release (prologue, demo, etc.).
# Used to avoid preferring "Game - Prologue" over "Game" when both exist.
_SUBSET_SUFFIXES = frozenset(
{
"prologue",
"demo",
"trial",
"lite",
"prelude",
}
)
# Type for progress callbacks: (done, total, found, game_name)
ProgressCb = Callable[[int, int, int, str], None]
@dataclass
class HLTBResult:
"""Result from a HowLongToBeat lookup."""
app_id: int
game_name: str
completionist_hours: float
similarity: float
hltb_game_id: int = 0
@dataclass
class _AuthInfo:
"""HLTB API authentication details."""
token: str
hp_key: str = ""
hp_val: str = ""
def load_hltb_cache() -> dict[int, float]:
"""Load the persistent HLTB cache from disk.
Returns: dict mapping app_id -> completionist_hours (-1 = no data on HLTB).
"""
if HLTB_CACHE_FILE.exists():
try:
data = json.loads(HLTB_CACHE_FILE.read_text(encoding="utf-8"))
return {int(k): float(v) for k, v in data.items()}
except (json.JSONDecodeError, ValueError, OSError):
logger.warning("Corrupt HLTB cache, starting fresh.")
return {}
def save_hltb_cache(cache: dict[int, float]) -> None:
"""Save the HLTB cache to disk."""
try:
_atomic_write(
HLTB_CACHE_FILE,
json.dumps({str(k): v for k, v in cache.items()}, indent=2) + "\n",
)
except OSError:
logger.exception("Failed to save HLTB cache")

333
steam_backlog_enforcer/hltb.py
View File

@ -13,95 +13,34 @@ Fetches leisure completionist hour estimates from howlongtobeat.com with:
from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass, field
from difflib import SequenceMatcher
from http import HTTPStatus
import json
import logging
import re
import time
from typing import Any
import aiohttp
from howlongtobeatpy.HTMLRequests import HTMLRequests
from python_pkg.steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write
logger = logging.getLogger(__name__)
HLTB_CACHE_FILE = CONFIG_DIR / "hltb_cache.json"
MAX_CONCURRENT = 60 # parallel requests to HLTB
_SAVE_INTERVAL = 50 # flush cache to disk every N results
MIN_SIMILARITY = 0.5
# Suffixes that indicate a subset release (prologue, demo, etc.).
# Used to avoid preferring "Game - Prologue" over "Game" when both exist.
_SUBSET_SUFFIXES = frozenset(
{
"prologue",
"demo",
"trial",
"lite",
"prelude",
}
from python_pkg.steam_backlog_enforcer._hltb_detail import (
_fetch_leisure_times,
)
from python_pkg.steam_backlog_enforcer._hltb_types import (
_SAVE_INTERVAL,
_SUBSET_SUFFIXES,
HLTB_BASE_URL,
MAX_CONCURRENT,
MIN_SIMILARITY,
HLTBResult,
ProgressCb,
_AuthInfo,
load_hltb_cache,
save_hltb_cache,
)
# Type for progress callbacks: (done, total, found, game_name)
ProgressCb = Callable[[int, int, int, str], None]
@dataclass
class HLTBResult:
"""Result from a HowLongToBeat lookup."""
app_id: int
game_name: str
completionist_hours: float
similarity: float
hltb_game_id: int = 0
@dataclass
class _AuthInfo:
"""HLTB API authentication details."""
token: str
hp_key: str = ""
hp_val: str = ""
HLTB_BASE_URL = "https://howlongtobeat.com"
# ──────────────────────────────────────────────────────────────
# Cache I/O
# ──────────────────────────────────────────────────────────────
def load_hltb_cache() -> dict[int, float]:
"""Load the persistent HLTB cache from disk.
Returns: dict mapping app_id -> completionist_hours (-1 = no data on HLTB).
"""
if HLTB_CACHE_FILE.exists():
try:
data = json.loads(HLTB_CACHE_FILE.read_text(encoding="utf-8"))
return {int(k): float(v) for k, v in data.items()}
except (json.JSONDecodeError, ValueError, OSError):
logger.warning("Corrupt HLTB cache, starting fresh.")
return {}
def save_hltb_cache(cache: dict[int, float]) -> None:
"""Save the HLTB cache to disk."""
try:
_atomic_write(
HLTB_CACHE_FILE,
json.dumps({str(k): v for k, v in cache.items()}, indent=2) + "\n",
)
except OSError:
logger.exception("Failed to save HLTB cache")
logger = logging.getLogger(__name__)
# ──────────────────────────────────────────────────────────────
@ -351,7 +290,7 @@ async def _search_one(
done = ctx.counter["done"]
# Incremental save every _SAVE_INTERVAL lookups.
if done % _SAVE_INTERVAL == 0:
if not done % _SAVE_INTERVAL:
save_hltb_cache(ctx.cache)
# Report progress.
@ -361,246 +300,6 @@ async def _search_one(
return result
# ──────────────────────────────────────────────────────────────
# Leisure time + DLC fetching from game detail pages
# ──────────────────────────────────────────────────────────────
_NEXT_DATA_RE = re.compile(
r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>',
)
def _parse_game_page(html: str) -> dict[str, Any] | None:
"""Extract game data dict from a HLTB game page's __NEXT_DATA__."""
match = _NEXT_DATA_RE.search(html)
if not match:
return None
try:
data = json.loads(match.group(1))
result: dict[str, Any] = data["props"]["pageProps"]["game"]["data"]
except (json.JSONDecodeError, KeyError, TypeError):
return None
else:
return result
def _as_positive_int(value: object) -> int:
"""Convert HLTB numeric JSON values to a positive int, or 0 when invalid."""
if isinstance(value, int):
return max(0, value)
if isinstance(value, float):
int_value = int(value)
return max(0, int_value)
if isinstance(value, str):
try:
int_value = int(value)
return max(0, int_value)
except ValueError:
return 0
return 0
def _extract_base_leisure_hours(game_data: dict[str, Any]) -> float:
"""Extract base-game leisure hours from game detail data."""
games = game_data.get("game", [])
if not isinstance(games, list) or not games:
return -1
if not isinstance(games[0], dict):
return -1
base = games[0]
leisure_s = _as_positive_int(base.get("comp_100_h", 0))
if leisure_s <= 0:
leisure_s = _as_positive_int(base.get("comp_100", 0))
if leisure_s <= 0:
return -1
return round(leisure_s / 3600, 2)
def _extract_dlc_relationships(game_data: dict[str, Any]) -> list[tuple[int, float]]:
"""Extract DLC relationship IDs and fallback hours from detail data."""
relationships = game_data.get("relationships", [])
if not isinstance(relationships, list):
return []
dlcs: list[tuple[int, float]] = []
for rel in relationships:
if not isinstance(rel, dict):
continue
if str(rel.get("game_type", "")).lower() != "dlc":
continue
dlc_id = _as_positive_int(rel.get("game_id", 0))
fallback_comp_100 = _as_positive_int(rel.get("comp_100", 0))
if fallback_comp_100 > 0:
fallback_hours = round(fallback_comp_100 / 3600, 2)
else:
fallback_hours = 0.0
dlcs.append((dlc_id, fallback_hours))
return dlcs
def _extract_leisure_hours(game_data: dict[str, Any]) -> float:
"""Compute total leisure hours: base game + all DLCs.
Uses ``comp_100_h`` (leisure completionist) from the game detail page.
Falls back to ``comp_100`` (average completionist) if leisure unavailable.
Also sums leisure time from any DLC listed in ``relationships``.
"""
base_hours = _extract_base_leisure_hours(game_data)
if base_hours <= 0:
return -1
total_hours = base_hours
# Add DLC leisure times from relationships.
for _dlc_id, fallback_hours in _extract_dlc_relationships(game_data):
total_hours += fallback_hours
return round(total_hours, 2)
async def _fetch_detail_one(
sem: asyncio.Semaphore,
session: aiohttp.ClientSession,
hltb_game_id: int,
) -> dict[str, Any] | None:
"""Fetch a single HLTB game detail page and parse its data."""
async with sem:
url = f"{HLTB_BASE_URL}/game/{hltb_game_id}"
headers = {
"User-Agent": (
"Mozilla/5.0 (X11; Linux x86_64; rv:136.0) Gecko/20100101 Firefox/136.0"
),
"accept": "text/html",
"referer": "https://howlongtobeat.com/",
}
try:
async with session.get(url, headers=headers) as resp:
if resp.status == HTTPStatus.OK:
html = await resp.text()
return _parse_game_page(html)
except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
logger.debug(
"HLTB detail fetch failed for game_id=%d: %s",
hltb_game_id,
exc,
)
return None
async def _fetch_leisure_times(
search_results: list[HLTBResult],
cache: dict[int, float],
progress_cb: ProgressCb | None,
) -> None:
"""Fetch leisure times from game detail pages for all search results.
Updates ``cache`` in-place with leisure hours (including DLC time).
"""
valid = [r for r in search_results if r.hltb_game_id > 0]
if not valid:
return
timeout = aiohttp.ClientTimeout(total=30, sock_read=20)
sem = asyncio.Semaphore(MAX_CONCURRENT)
connector = aiohttp.TCPConnector(
limit=MAX_CONCURRENT,
keepalive_timeout=30,
)
total = len(valid)
done = 0
found = 0
async with aiohttp.ClientSession(
timeout=timeout,
connector=connector,
) as session:
coros = [_fetch_detail_one(sem, session, r.hltb_game_id) for r in valid]
details = await asyncio.gather(*coros)
dlc_relationships_by_app, dlc_ids = _collect_dlc_relationships(valid, details)
dlc_hours_by_id = await _fetch_dlc_leisure_hours(sem, session, dlc_ids)
for r, game_data in zip(valid, details, strict=False):
done += 1
if game_data is not None:
leisure = _extract_leisure_hours(game_data)
if leisure > 0:
leisure = _apply_dlc_leisure_overrides(
leisure,
dlc_relationships_by_app.get(r.app_id, []),
dlc_hours_by_id,
)
r.completionist_hours = leisure
cache[r.app_id] = leisure
found += 1
if progress_cb is not None:
progress_cb(done, total, found, r.game_name)
if done % _SAVE_INTERVAL == 0:
save_hltb_cache(cache)
def _collect_dlc_relationships(
valid: list[HLTBResult],
details: list[dict[str, Any] | None],
) -> tuple[dict[int, list[tuple[int, float]]], list[int]]:
"""Collect DLC relationship IDs for all base-game detail responses."""
by_app: dict[int, list[tuple[int, float]]] = {}
unique_dlc_ids: set[int] = set()
for result, game_data in zip(valid, details, strict=False):
if game_data is None:
continue
dlc_rels = _extract_dlc_relationships(game_data)
by_app[result.app_id] = dlc_rels
for dlc_id, _fallback_hours in dlc_rels:
if dlc_id > 0:
unique_dlc_ids.add(dlc_id)
return by_app, sorted(unique_dlc_ids)
async def _fetch_dlc_leisure_hours(
sem: asyncio.Semaphore,
session: aiohttp.ClientSession,
dlc_ids: list[int],
) -> dict[int, float]:
"""Fetch leisure hours for each DLC game id."""
if not dlc_ids:
return {}
coros = [_fetch_detail_one(sem, session, dlc_id) for dlc_id in dlc_ids]
dlc_details = await asyncio.gather(*coros)
dlc_hours_by_id: dict[int, float] = {}
for dlc_id, dlc_data in zip(dlc_ids, dlc_details, strict=False):
if dlc_data is None:
continue
dlc_leisure = _extract_base_leisure_hours(dlc_data)
if dlc_leisure > 0:
dlc_hours_by_id[dlc_id] = dlc_leisure
return dlc_hours_by_id
def _apply_dlc_leisure_overrides(
base_hours: float,
dlc_rels: list[tuple[int, float]],
dlc_hours_by_id: dict[int, float],
) -> float:
"""Replace fallback DLC hours with detailed leisure hours when available."""
adjusted = base_hours
for dlc_id, fallback_hours in dlc_rels:
dlc_leisure = dlc_hours_by_id.get(dlc_id, -1.0)
if dlc_leisure > 0:
adjusted += dlc_leisure - fallback_hours
return round(adjusted, 2)
async def _fetch_batch(
games: list[tuple[int, str]],
cache: dict[int, float],

221
steam_backlog_enforcer/main.py
View File

@ -5,6 +5,7 @@ from __future__ import annotations
import logging
import sys
from python_pkg.steam_backlog_enforcer._cmd_done import cmd_done
from python_pkg.steam_backlog_enforcer._enforce_loop import (
do_enforce,
get_all_owned_app_ids,
@ -15,10 +16,6 @@ from python_pkg.steam_backlog_enforcer.config import (
interactive_setup,
load_snapshot,
)
from python_pkg.steam_backlog_enforcer.enforcer import (
enforce_allowed_game,
send_notification,
)
from python_pkg.steam_backlog_enforcer.game_install import (
PROTECTED_APP_IDS,
_echo,
@ -27,22 +24,16 @@ from python_pkg.steam_backlog_enforcer.game_install import (
is_game_installed,
uninstall_other_games,
)
from python_pkg.steam_backlog_enforcer.hltb import (
fetch_hltb_times_cached,
load_hltb_cache,
)
from python_pkg.steam_backlog_enforcer.library_hider import (
hide_other_games,
restart_steam,
unhide_all_games,
)
from python_pkg.steam_backlog_enforcer.scanning import (
_pick_playable_candidate,
do_check,
do_scan,
pick_next_game,
)
from python_pkg.steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient
from python_pkg.steam_backlog_enforcer.steam_api import GameInfo
from python_pkg.steam_backlog_enforcer.store_blocker import (
block_store,
is_store_blocked,
@ -58,7 +49,6 @@ logger = logging.getLogger(__name__)
_LIST_DISPLAY_LIMIT = 50
_MIN_CLI_ARGS = 2
_REASSIGN_REFRESH_LIMIT = 50
# ──────────────────────────────────────────────────────────────
@ -284,213 +274,6 @@ def cmd_unhide(config: Config, _state: State) -> None:
_echo("Done!")
def _apply_cached_hours_to_games(
games: list[GameInfo],
hltb_cache: dict[int, float],
) -> None:
"""Overlay cached HLTB hours onto games (including cached misses)."""
for game in games:
if game.app_id in hltb_cache:
game.completionist_hours = hltb_cache[game.app_id]
def _refresh_uncached_shortlist_hours(
games: list[GameInfo],
hltb_cache: dict[int, float],
skip: set[int],
*,
upper_bound_hours: float | None = None,
) -> None:
"""Refresh likely-short uncached games to avoid stale snapshot decisions."""
shorter_uncached = [
(g.app_id, g.name)
for g in sorted(
(
game
for game in games
if not game.is_complete
and game.app_id not in skip
and game.completionist_hours > 0
and game.app_id not in hltb_cache
and (
upper_bound_hours is None
or game.completionist_hours < upper_bound_hours
)
),
key=lambda game: game.completionist_hours,
)[:_REASSIGN_REFRESH_LIMIT]
]
if shorter_uncached:
refreshed = fetch_hltb_times_cached(shorter_uncached)
hltb_cache.update(refreshed)
def _try_reassign_shorter_game(
hltb_cache: dict[int, float],
app_id: int,
hours: float,
state: State,
config: Config,
) -> bool:
"""Check if a shorter game is available and reassign if so."""
snapshot_data = load_snapshot()
if not snapshot_data:
return False
all_games = [GameInfo.from_snapshot(d) for d in snapshot_data]
skip = set(config.skip_app_ids) | set(state.finished_app_ids)
_refresh_uncached_shortlist_hours(
all_games,
hltb_cache,
skip,
upper_bound_hours=hours,
)
_apply_cached_hours_to_games(all_games, hltb_cache)
candidates = [
g
for g in all_games
if not g.is_complete and g.app_id not in skip and g.completionist_hours > 0
]
candidates.sort(key=lambda g: g.completionist_hours)
if not candidates or candidates[0].app_id == app_id:
return False
# Filter out Linux-incompatible games before deciding to reassign.
playable = _pick_playable_candidate(
[c for c in candidates if c.app_id != app_id],
)
if playable is None or playable.completionist_hours >= hours:
return False
_echo(
f"\n Reassigning: {playable.name} is shorter"
f" (~{playable.completionist_hours:.1f}h vs ~{hours:.1f}h)"
)
pick_next_game(all_games, state, config)
return True
def _finalize_completion(
config: Config,
state: State,
game_name: str,
app_id: int,
) -> None:
"""Mark game complete, pick next, hide non-assigned games, notify."""
_echo(f"\n COMPLETED: {game_name}!")
state.finished_app_ids.append(app_id)
snapshot_data = load_snapshot()
_echo("\nPicking next game...")
if not snapshot_data:
_echo(" No snapshot found. Run 'scan' first.")
state.current_app_id = None
state.current_game_name = ""
state.save()
return
games = [GameInfo.from_snapshot(d) for d in snapshot_data]
hltb_cache = load_hltb_cache()
skip = set(config.skip_app_ids) | set(state.finished_app_ids)
_refresh_uncached_shortlist_hours(games, hltb_cache, skip)
_apply_cached_hours_to_games(games, hltb_cache)
pick_next_game(games, state, config)
if state.current_app_id is None:
_echo(" No more games to assign!")
return
owned_ids = get_all_owned_app_ids(config)
if owned_ids:
hidden = hide_other_games(owned_ids, state.current_app_id)
if hidden > 0:
_echo(f"\n Library: hid {hidden} games")
send_notification(
"Game Complete!",
f"Finished {game_name}! Now playing: {state.current_game_name}",
)
_echo(f"\nAll done! Go play {state.current_game_name}!")
def _enforce_on_done(config: Config, state: State) -> None:
"""Run a single enforcement pass during the 'done' command.
Kills unauthorized game processes, uninstalls unauthorized games,
and ensures the assigned game is installed.
"""
if state.current_app_id is None:
return
if config.kill_unauthorized_games:
violations = enforce_allowed_game(
state.current_app_id,
kill_unauthorized=True,
)
for pid, app_id in violations:
_echo(f" Killed unauthorized game: AppID={app_id} (PID={pid})")
if config.uninstall_other_games:
count = uninstall_other_games(state.current_app_id)
if count:
_echo(f" Uninstalled {count} unauthorized game(s)")
if not is_game_installed(state.current_app_id):
_echo(f" Re-installing {state.current_game_name}...")
install_game(
state.current_app_id,
state.current_game_name,
config.steam_id,
use_steam_protocol=True,
)
def cmd_done(config: Config, state: State) -> None:
"""Check completion, pick next game, uninstall & hide.
All-in-one command for after finishing a game:
1. Verify 100% achievements on Steam.
2. Pick the next game (shortest HLTB leisure+dlc time).
3. Uninstall all non-assigned games.
4. Hide all non-assigned games in the Steam library.
5. Install the newly assigned game.
"""
if state.current_app_id is None:
_echo("No game currently assigned. Run 'scan' first.")
return
client = SteamAPIClient(config.steam_api_key, config.steam_id)
game_name = state.current_game_name
app_id = state.current_app_id
_echo(f"Checking {game_name} (AppID={app_id})...")
game = client.refresh_single_game(app_id, game_name)
if game is None:
_echo(" Could not fetch achievement data from Steam.")
return
_echo(
f" Progress: {game.unlocked_achievements}/{game.total_achievements}"
f" ({game.completion_pct:.1f}%)"
)
hltb_cache = load_hltb_cache()
hours = hltb_cache.get(app_id, -1.0)
if hours < 0:
hltb_cache = fetch_hltb_times_cached([(app_id, game_name)])
hours = hltb_cache.get(app_id, -1.0)
if hours > 0:
_echo(f" HLTB leisure+dlc estimate: {hours:.1f} hours")
if _try_reassign_shorter_game(hltb_cache, app_id, hours, state, config):
return
if not game.is_complete:
remaining = game.total_achievements - game.unlocked_achievements
_echo(f"\n NOT COMPLETE: {remaining} achievements remaining. Keep going!")
_enforce_on_done(config, state)
return
_finalize_completion(config, state, game_name, app_id)
COMMANDS = {
"scan": ("Scan library & assign a game", do_scan),
"check": ("Check assigned game completion", do_check),

171
steam_backlog_enforcer/tests/test_cmd_done.py Normal file
View File

@ -0,0 +1,171 @@
"""Tests for _cmd_done module."""
from __future__ import annotations
from typing import Any
from unittest.mock import patch
from python_pkg.steam_backlog_enforcer._cmd_done import _try_reassign_shorter_game
from python_pkg.steam_backlog_enforcer.config import Config, State
from python_pkg.steam_backlog_enforcer.steam_api import GameInfo
CMD_DONE_PKG = "python_pkg.steam_backlog_enforcer._cmd_done"
def _snap(
app_id: int = 1,
name: str = "G",
total: int = 10,
unlocked: int = 0,
hours: float = -1,
) -> dict[str, Any]:
return {
"app_id": app_id,
"name": name,
"total_achievements": total,
"unlocked_achievements": unlocked,
"playtime_minutes": 60,
"completionist_hours": hours,
}
class TestTryReassignShorterGame:
"""Tests for _try_reassign_shorter_game."""
def test_no_snapshot(self) -> None:
with patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=None):
assert not _try_reassign_shorter_game({}, 1, 10.0, State(), Config())
def test_no_shorter_candidate(self) -> None:
snap = [_snap(1, "G", 10, 5, 10.0), _snap(2, "H", 10, 5, -1)]
with (
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}._echo"),
):
result = _try_reassign_shorter_game(
{1: 10.0},
1,
10.0,
State(),
Config(),
)
assert not result
def test_reassigns(self) -> None:
snap = [
_snap(1, "Long", 10, 5, 100.0),
_snap(2, "Short", 10, 5, 5.0),
]
state = State(current_app_id=1, current_game_name="Long")
short_game = GameInfo(
app_id=2,
name="Short",
total_achievements=10,
unlocked_achievements=5,
playtime_minutes=60,
completionist_hours=5.0,
)
with (
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}._echo"),
patch(
f"{CMD_DONE_PKG}._pick_playable_candidate",
return_value=short_game,
),
patch(f"{CMD_DONE_PKG}.pick_next_game"),
):
result = _try_reassign_shorter_game(
{1: 100.0, 2: 5.0},
1,
100.0,
state,
Config(),
)
assert result
def test_playable_none(self) -> None:
snap = [
_snap(1, "Long", 10, 5, 100.0),
_snap(2, "Short", 10, 5, 5.0),
]
with (
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}._pick_playable_candidate", return_value=None),
patch(f"{CMD_DONE_PKG}._echo"),
):
result = _try_reassign_shorter_game(
{1: 100.0, 2: 5.0},
1,
100.0,
State(),
Config(),
)
assert not result
def test_playable_longer(self) -> None:
"""Playable candidate is longer than current — no reassign."""
snap = [
_snap(1, "Short", 10, 5, 10.0),
_snap(2, "Long", 10, 5, 200.0),
]
long_game = GameInfo(
app_id=2,
name="Long",
total_achievements=10,
unlocked_achievements=5,
playtime_minutes=60,
completionist_hours=200.0,
)
with (
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}._pick_playable_candidate", return_value=long_game),
patch(f"{CMD_DONE_PKG}._echo"),
):
result = _try_reassign_shorter_game(
{1: 10.0, 2: 200.0},
1,
10.0,
State(),
Config(),
)
assert not result
def test_refreshes_stale_shorter_snapshot_entry(self) -> None:
"""Uncached shorter snapshot candidates are refreshed before reassigning."""
snap = [
_snap(1, "Current", 10, 5, 20.1),
_snap(2, "Lacuna", 10, 0, 0.9),
]
state = State(current_app_id=1, current_game_name="Current")
refreshed_short = GameInfo(
app_id=2,
name="Lacuna",
total_achievements=10,
unlocked_achievements=0,
playtime_minutes=60,
completionist_hours=18.8,
)
with (
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(
f"{CMD_DONE_PKG}.fetch_hltb_times_cached",
return_value={2: 18.8},
) as mock_fetch_hltb,
patch(
f"{CMD_DONE_PKG}._pick_playable_candidate",
return_value=refreshed_short,
) as mock_pick_playable,
patch(f"{CMD_DONE_PKG}.pick_next_game"),
patch(f"{CMD_DONE_PKG}._echo"),
):
result = _try_reassign_shorter_game(
{1: 20.1},
1,
20.1,
state,
Config(),
)
assert result
mock_fetch_hltb.assert_called_once_with([(2, "Lacuna")])
mock_pick_playable.assert_called_once()

802
steam_backlog_enforcer/tests/test_hltb.py
View File

@ -8,35 +8,19 @@ from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock, patch
import aiohttp
from typing_extensions import Self
from python_pkg.steam_backlog_enforcer.hltb import (
HLTBResult,
_apply_dlc_leisure_overrides,
_as_positive_int,
_AuthInfo,
_build_search_payload,
_collect_dlc_relationships,
_extract_base_leisure_hours,
_extract_dlc_relationships,
_extract_leisure_hours,
_fetch_batch,
_fetch_detail_one,
_fetch_dlc_leisure_hours,
_fetch_leisure_times,
_get_auth_info,
_get_hltb_search_url,
_parse_game_page,
_pick_best_hltb_entry,
_search_one,
_SearchCtx,
_similarity,
load_hltb_cache,
save_hltb_cache,
)
if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
@ -47,7 +31,7 @@ class TestHltbCache:
cache_file = tmp_path / "hltb_cache.json"
cache_file.write_text(json.dumps({"440": 10.5}), encoding="utf-8")
with patch(
"python_pkg.steam_backlog_enforcer.hltb.HLTB_CACHE_FILE", cache_file
"python_pkg.steam_backlog_enforcer._hltb_types.HLTB_CACHE_FILE", cache_file
):
result = load_hltb_cache()
assert result == {440: 10.5}
@ -55,7 +39,7 @@ class TestHltbCache:
def test_load_cache_missing(self, tmp_path: Path) -> None:
cache_file = tmp_path / "nonexistent.json"
with patch(
"python_pkg.steam_backlog_enforcer.hltb.HLTB_CACHE_FILE", cache_file
"python_pkg.steam_backlog_enforcer._hltb_types.HLTB_CACHE_FILE", cache_file
):
assert load_hltb_cache() == {}
@ -63,22 +47,25 @@ class TestHltbCache:
cache_file = tmp_path / "hltb_cache.json"
cache_file.write_text("not json", encoding="utf-8")
with patch(
"python_pkg.steam_backlog_enforcer.hltb.HLTB_CACHE_FILE", cache_file
"python_pkg.steam_backlog_enforcer._hltb_types.HLTB_CACHE_FILE", cache_file
):
assert load_hltb_cache() == {}
def test_save_cache(self, tmp_path: Path) -> None:
cache_file = tmp_path / "hltb_cache.json"
with (
patch("python_pkg.steam_backlog_enforcer.hltb.HLTB_CACHE_FILE", cache_file),
patch("python_pkg.steam_backlog_enforcer.hltb.CONFIG_DIR", tmp_path),
patch(
"python_pkg.steam_backlog_enforcer._hltb_types.HLTB_CACHE_FILE",
cache_file,
),
patch("python_pkg.steam_backlog_enforcer._hltb_types.CONFIG_DIR", tmp_path),
):
save_hltb_cache({440: 10.5})
assert cache_file.exists()
def test_save_cache_os_error(self, tmp_path: Path) -> None:
with patch(
"python_pkg.steam_backlog_enforcer.hltb._atomic_write",
"python_pkg.steam_backlog_enforcer._hltb_types._atomic_write",
side_effect=OSError("disk full"),
):
save_hltb_cache({440: 10.5}) # Should not raise
@ -334,774 +321,3 @@ class TestPickBestHltbEntry:
result = _pick_best_hltb_entry("Killing Floor", [(spinoff, 0.7), (base, 1.0)])
assert result is not None
assert result[0]["game_name"] == "Killing Floor"
class _FakeResponse:
"""Async context manager mimicking aiohttp response."""
def __init__(self, status: int, json_data: dict[str, Any] | None = None) -> None:
self.status = status
self._json_data = json_data or {}
async def __aenter__(self) -> Self:
return self
async def __aexit__(self, *args: object) -> None:
pass
async def json(self) -> dict[str, Any]:
return self._json_data
def _make_session(resp: _FakeResponse) -> MagicMock:
session = MagicMock()
session.post.return_value = resp
return session
def _make_ctx(
session: MagicMock,
*,
cache: dict[int, float] | None = None,
progress_cb: Callable[..., object] | None = None,
) -> _SearchCtx:
return _SearchCtx(
session=session,
search_url="https://example.com/search",
headers={},
cache=cache if cache is not None else {},
counter={"done": 0, "found": 0},
total=1,
progress_cb=progress_cb,
)
class TestSearchOne:
"""Tests for _search_one."""
def test_found(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2",
"game_alias": "",
"comp_100": 180000,
"game_id": 12345,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
assert result.app_id == 440
def test_not_found(self) -> None:
resp = _FakeResponse(200, {"data": []})
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
assert ctx.cache[440] == -1
def test_error(self) -> None:
session = MagicMock()
session.post.side_effect = aiohttp.ClientError("fail")
ctx = _make_ctx(session)
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
def test_non_200(self) -> None:
resp = _FakeResponse(500)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
def test_with_progress_cb(self) -> None:
resp = _FakeResponse(200, {"data": []})
cb = MagicMock()
ctx = _make_ctx(_make_session(resp), progress_cb=cb)
asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
cb.assert_called_once()
def test_low_similarity_skipped(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "Completely Different Name",
"game_alias": "",
"comp_100": 3600,
"game_id": 1,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
def test_zero_comp_100_skipped(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2",
"game_alias": "",
"comp_100": 0,
"game_id": 1,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
def test_alias_match(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "Team Fortress 2",
"game_alias": "TF2",
"comp_100": 180000,
"game_id": 12345,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
def test_full_edition_colon(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2: Complete",
"game_alias": "",
"comp_100": 180000,
"game_id": 99,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
def test_full_edition_dash(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2 - Complete",
"game_alias": "",
"comp_100": 180000,
"game_id": 99,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
def test_save_interval(self) -> None:
"""Trigger the _SAVE_INTERVAL branch."""
resp = _FakeResponse(200, {"data": []})
ctx = _make_ctx(_make_session(resp))
# Set done to one less than _SAVE_INTERVAL so it triggers save
from python_pkg.steam_backlog_enforcer.hltb import _SAVE_INTERVAL
ctx.counter["done"] = _SAVE_INTERVAL - 1
with patch(
"python_pkg.steam_backlog_enforcer.hltb.save_hltb_cache"
) as mock_save:
asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
mock_save.assert_called_once()
class TestFetchBatchHltb:
"""Tests for _fetch_batch (the hltb version)."""
def test_no_auth(self) -> None:
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url",
return_value="https://example.com",
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_auth_info",
new_callable=AsyncMock,
return_value=None,
),
):
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None))
assert results == []
def test_with_auth(self) -> None:
auth = _AuthInfo("token123", "ign_x", "ff")
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url",
return_value="https://example.com",
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_auth_info",
new_callable=AsyncMock,
return_value=auth,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._search_one",
new_callable=AsyncMock,
return_value=HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times",
new_callable=AsyncMock,
),
):
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None))
assert len(results) == 1
def test_with_auth_no_hp(self) -> None:
auth = _AuthInfo("tok123")
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url",
return_value="https://example.com",
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_auth_info",
new_callable=AsyncMock,
return_value=auth,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._search_one",
new_callable=AsyncMock,
return_value=None,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times",
new_callable=AsyncMock,
),
):
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None))
assert results == []
def test_filters_none_results(self) -> None:
auth = _AuthInfo("tok123")
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url",
return_value="https://example.com",
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_auth_info",
new_callable=AsyncMock,
return_value=auth,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._search_one",
new_callable=AsyncMock,
return_value=None,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times",
new_callable=AsyncMock,
),
):
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None))
assert results == []
class TestParseGamePage:
"""Tests for _parse_game_page."""
def test_valid_html(self) -> None:
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6800}],
"relationships": [],
}
next_data = {
"props": {"pageProps": {"game": {"data": game_data}}},
}
html = (
'<html><script id="__NEXT_DATA__" type="application/json">'
+ json.dumps(next_data)
+ "</script></html>"
)
assert _parse_game_page(html) == game_data
def test_no_script_tag(self) -> None:
assert _parse_game_page("<html></html>") is None
def test_bad_json(self) -> None:
html = '<script id="__NEXT_DATA__" type="application/json">{not json}</script>'
assert _parse_game_page(html) is None
def test_missing_keys(self) -> None:
html = (
'<script id="__NEXT_DATA__" type="application/json">{"props": {}}</script>'
)
assert _parse_game_page(html) is None
class TestExtractLeisureHours:
"""Tests for _extract_leisure_hours."""
def test_leisure_time_only(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6800}],
"relationships": [],
}
assert _extract_leisure_hours(data) == round(21243 / 3600, 2)
def test_leisure_with_dlc(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6800}],
"relationships": [
{"game_type": "dlc", "comp_100": 12298},
{"game_type": "dlc", "comp_100": 3600},
],
}
assert _extract_leisure_hours(data) == round((21243 + 12298 + 3600) / 3600, 2)
def test_fallback_to_comp_100(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100": 7200}],
"relationships": [],
}
assert _extract_leisure_hours(data) == round(7200 / 3600, 2)
def test_no_game_data(self) -> None:
assert _extract_leisure_hours({"game": [], "relationships": []}) == -1
def test_zero_leisure(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 0, "comp_100": 0}],
"relationships": [],
}
assert _extract_leisure_hours(data) == -1
def test_no_game_key(self) -> None:
assert _extract_leisure_hours({"relationships": []}) == -1
def test_non_dlc_relationship_ignored(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": [
{"game_type": "game", "comp_100": 9999},
{"game_type": "dlc", "comp_100": 1800},
],
}
assert _extract_leisure_hours(data) == round((3600 + 1800) / 3600, 2)
def test_dlc_zero_comp_100_skipped(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": [
{"game_type": "dlc", "comp_100": 0},
],
}
assert _extract_leisure_hours(data) == round(3600 / 3600, 2)
def test_negative_leisure(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": -1, "comp_100": -1}],
"relationships": [],
}
assert _extract_leisure_hours(data) == -1
def test_string_numeric_fields(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": "7200", "comp_100": "3600"}],
"relationships": [{"game_type": "dlc", "game_id": "1", "comp_100": "1800"}],
}
assert _extract_leisure_hours(data) == round((7200 + 1800) / 3600, 2)
def test_bad_string_falls_back_to_comp_100(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": "bad", "comp_100": "3600"}],
"relationships": [],
}
assert _extract_leisure_hours(data) == 1.0
def test_relationships_not_list(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": "not-a-list",
}
assert _extract_leisure_hours(data) == 1.0
class TestInternalHelpers:
"""Tests for internal helper coverage."""
def test_as_positive_int_float(self) -> None:
assert _as_positive_int(1.9) == 1
def test_as_positive_int_invalid_type(self) -> None:
assert _as_positive_int(object()) == 0
def test_extract_base_leisure_non_dict_game(self) -> None:
data: dict[str, Any] = {"game": [123]}
assert _extract_base_leisure_hours(data) == -1
def test_extract_dlc_relationships_skips_non_dict(self) -> None:
data: dict[str, Any] = {
"relationships": [
"bad",
{"game_type": "dlc", "game_id": 7, "comp_100": 3600},
],
}
assert _extract_dlc_relationships(data) == [(7, 1.0)]
def test_collect_dlc_relationships_ignores_non_positive_id(self) -> None:
valid = [
HLTBResult(
app_id=1,
game_name="Game",
completionist_hours=1.0,
similarity=1.0,
hltb_game_id=123,
)
]
details: list[dict[str, Any] | None] = [
{
"relationships": [
{"game_type": "dlc", "game_id": 0, "comp_100": 3600},
]
}
]
by_app, ids = _collect_dlc_relationships(valid, details)
assert by_app[1] == [(0, 1.0)]
assert ids == []
def test_apply_dlc_leisure_overrides(self) -> None:
adjusted = _apply_dlc_leisure_overrides(
base_hours=6.0,
dlc_rels=[(10, 1.0), (11, 2.0)],
dlc_hours_by_id={10: 3.0},
)
assert adjusted == 8.0
def test_fetch_dlc_leisure_hours_empty(self) -> None:
async def _run() -> dict[int, float]:
async with aiohttp.ClientSession() as session:
return await _fetch_dlc_leisure_hours(asyncio.Semaphore(1), session, [])
assert asyncio.run(_run()) == {}
def test_fetch_dlc_leisure_hours_skips_none_data(self) -> None:
async def _run() -> dict[int, float]:
async with aiohttp.ClientSession() as session:
with patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
return_value=None,
):
return await _fetch_dlc_leisure_hours(
asyncio.Semaphore(1),
session,
[1],
)
assert asyncio.run(_run()) == {}
def test_fetch_dlc_leisure_hours_skips_non_positive_leisure(self) -> None:
bad_dlc_data: dict[str, Any] = {
"game": [{"comp_100_h": 0, "comp_100": 0}],
"relationships": [],
}
async def _run() -> dict[int, float]:
async with aiohttp.ClientSession() as session:
with patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
return_value=bad_dlc_data,
):
return await _fetch_dlc_leisure_hours(
asyncio.Semaphore(1),
session,
[1],
)
assert asyncio.run(_run()) == {}
class _FakeTextResponse:
"""Async context manager mimicking aiohttp response for text."""
def __init__(self, status: int, text: str = "") -> None:
self.status = status
self._text = text
async def __aenter__(self) -> Self:
return self
async def __aexit__(self, *args: object) -> None:
pass
async def text(self) -> str:
return self._text
class TestFetchDetailOne:
"""Tests for _fetch_detail_one."""
def test_success(self) -> None:
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243}],
"relationships": [],
}
next_data = {"props": {"pageProps": {"game": {"data": game_data}}}}
html = (
'<script id="__NEXT_DATA__" type="application/json">'
+ json.dumps(next_data)
+ "</script>"
)
resp = _FakeTextResponse(200, html)
session = MagicMock()
session.get = MagicMock(return_value=resp)
result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345))
assert result == game_data
def test_non_200(self) -> None:
resp = _FakeTextResponse(404)
session = MagicMock()
session.get = MagicMock(return_value=resp)
result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345))
assert result is None
def test_client_error(self) -> None:
ctx = AsyncMock()
ctx.__aenter__ = AsyncMock(side_effect=aiohttp.ClientError)
ctx.__aexit__ = AsyncMock(return_value=False)
session = MagicMock()
session.get = MagicMock(return_value=ctx)
result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345))
assert result is None
def test_parse_failure(self) -> None:
resp = _FakeTextResponse(200, "<html>no script</html>")
session = MagicMock()
session.get = MagicMock(return_value=resp)
result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345))
assert result is None
class TestFetchLeisureTimes:
"""Tests for _fetch_leisure_times."""
def test_updates_cache(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243}],
"relationships": [],
}
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
):
asyncio.run(_fetch_leisure_times(results, cache, None))
assert cache[440] == round(21243 / 3600, 2)
assert results[0].completionist_hours == round(21243 / 3600, 2)
def test_no_valid_results(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=0,
),
]
cache: dict[int, float] = {}
asyncio.run(_fetch_leisure_times(results, cache, None))
assert cache == {}
def test_empty_results(self) -> None:
cache: dict[int, float] = {}
asyncio.run(_fetch_leisure_times([], cache, None))
assert cache == {}
def test_detail_returns_none(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
return_value=None,
):
asyncio.run(_fetch_leisure_times(results, cache, None))
assert cache == {}
assert results[0].completionist_hours == 50.0
def test_negative_leisure(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
game_data: dict[str, Any] = {"game": [], "relationships": []}
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
):
asyncio.run(_fetch_leisure_times(results, cache, None))
assert cache == {}
assert results[0].completionist_hours == 50.0
def test_with_progress_cb(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": [],
}
cache: dict[int, float] = {}
cb = MagicMock()
with patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
):
asyncio.run(_fetch_leisure_times(results, cache, cb))
cb.assert_called_once()
def test_save_interval(self) -> None:
"""Trigger the _SAVE_INTERVAL branch in leisure fetching."""
from python_pkg.steam_backlog_enforcer.hltb import _SAVE_INTERVAL
results = [
HLTBResult(
app_id=i,
game_name=f"Game{i}",
completionist_hours=1.0,
similarity=1.0,
hltb_game_id=i + 1000,
)
for i in range(_SAVE_INTERVAL)
]
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": [],
}
cache: dict[int, float] = {}
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb.save_hltb_cache"
) as mock_save,
):
asyncio.run(_fetch_leisure_times(results, cache, None))
mock_save.assert_called_once()
def test_dlc_detail_overrides_relationship_fallback(self) -> None:
results = [
HLTBResult(
app_id=1289310,
game_name="Helltaker",
completionist_hours=1.0,
similarity=1.0,
hltb_game_id=78118,
),
]
base_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6846}],
"relationships": [{"game_type": "dlc", "game_id": 92236, "comp_100": 4075}],
}
dlc_data: dict[str, Any] = {
"game": [{"comp_100_h": 12298, "comp_100": 4075}],
"relationships": [],
}
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
side_effect=[base_data, dlc_data],
):
asyncio.run(_fetch_leisure_times(results, cache, None))
expected = round((21243 + 12298) / 3600, 2)
assert cache[1289310] == expected
assert results[0].completionist_hours == expected
def test_missing_dlc_detail_keeps_relationship_fallback(self) -> None:
results = [
HLTBResult(
app_id=1289310,
game_name="Helltaker",
completionist_hours=1.0,
similarity=1.0,
hltb_game_id=78118,
),
]
base_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6846}],
"relationships": [{"game_type": "dlc", "game_id": 92236, "comp_100": 4075}],
}
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one",
new_callable=AsyncMock,
side_effect=[base_data, None],
):
asyncio.run(_fetch_leisure_times(results, cache, None))
expected = round((21243 + 4075) / 3600, 2)
assert cache[1289310] == expected
assert results[0].completionist_hours == expected

378
steam_backlog_enforcer/tests/test_hltb_detail.py Normal file
View File

@ -0,0 +1,378 @@
"""Tests for HLTB internal helpers, detail fetching, and leisure times — part 3."""
from __future__ import annotations
import asyncio
import json
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import aiohttp
from typing_extensions import Self
from python_pkg.steam_backlog_enforcer._hltb_detail import (
_apply_dlc_leisure_overrides,
_as_positive_int,
_collect_dlc_relationships,
_extract_base_leisure_hours,
_extract_dlc_relationships,
_fetch_detail_one,
_fetch_dlc_leisure_hours,
_fetch_leisure_times,
)
from python_pkg.steam_backlog_enforcer._hltb_types import _SAVE_INTERVAL, HLTBResult
class TestInternalHelpers:
"""Tests for internal helper coverage."""
def test_as_positive_int_float(self) -> None:
assert _as_positive_int(1.9) == 1
def test_as_positive_int_invalid_type(self) -> None:
assert not _as_positive_int(object())
def test_extract_base_leisure_non_dict_game(self) -> None:
data: dict[str, Any] = {"game": [123]}
assert _extract_base_leisure_hours(data) == -1
def test_extract_dlc_relationships_skips_non_dict(self) -> None:
data: dict[str, Any] = {
"relationships": [
"bad",
{"game_type": "dlc", "game_id": 7, "comp_100": 3600},
],
}
assert _extract_dlc_relationships(data) == [(7, 1.0)]
def test_collect_dlc_relationships_ignores_non_positive_id(self) -> None:
valid = [
HLTBResult(
app_id=1,
game_name="Game",
completionist_hours=1.0,
similarity=1.0,
hltb_game_id=123,
)
]
details: list[dict[str, Any] | None] = [
{
"relationships": [
{"game_type": "dlc", "game_id": 0, "comp_100": 3600},
]
}
]
by_app, ids = _collect_dlc_relationships(valid, details)
assert by_app[1] == [(0, 1.0)]
assert ids == []
def test_apply_dlc_leisure_overrides(self) -> None:
adjusted = _apply_dlc_leisure_overrides(
base_hours=6.0,
dlc_rels=[(10, 1.0), (11, 2.0)],
dlc_hours_by_id={10: 3.0},
)
assert adjusted == 8.0
def test_fetch_dlc_leisure_hours_empty(self) -> None:
async def _run() -> dict[int, float]:
async with aiohttp.ClientSession() as session:
return await _fetch_dlc_leisure_hours(asyncio.Semaphore(1), session, [])
assert asyncio.run(_run()) == {}
def test_fetch_dlc_leisure_hours_skips_none_data(self) -> None:
async def _run() -> dict[int, float]:
async with aiohttp.ClientSession() as session:
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=None,
):
return await _fetch_dlc_leisure_hours(
asyncio.Semaphore(1),
session,
[1],
)
assert asyncio.run(_run()) == {}
def test_fetch_dlc_leisure_hours_skips_non_positive_leisure(self) -> None:
bad_dlc_data: dict[str, Any] = {
"game": [{"comp_100_h": 0, "comp_100": 0}],
"relationships": [],
}
async def _run() -> dict[int, float]:
async with aiohttp.ClientSession() as session:
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=bad_dlc_data,
):
return await _fetch_dlc_leisure_hours(
asyncio.Semaphore(1),
session,
[1],
)
assert asyncio.run(_run()) == {}
class _FakeTextResponse:
"""Async context manager mimicking aiohttp response for text."""
def __init__(self, status: int, text: str = "") -> None:
self.status = status
self._text = text
async def __aenter__(self) -> Self:
return self
async def __aexit__(self, *args: object) -> None:
pass
async def text(self) -> str:
return self._text
class TestFetchDetailOne:
"""Tests for _fetch_detail_one."""
def test_success(self) -> None:
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243}],
"relationships": [],
}
next_data = {"props": {"pageProps": {"game": {"data": game_data}}}}
html = (
'<script id="__NEXT_DATA__" type="application/json">'
+ json.dumps(next_data)
+ "</script>"
)
resp = _FakeTextResponse(200, html)
session = MagicMock()
session.get = MagicMock(return_value=resp)
result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345))
assert result == game_data
def test_non_200(self) -> None:
resp = _FakeTextResponse(404)
session = MagicMock()
session.get = MagicMock(return_value=resp)
result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345))
assert result is None
def test_client_error(self) -> None:
ctx = AsyncMock()
ctx.__aenter__ = AsyncMock(side_effect=aiohttp.ClientError)
ctx.__aexit__ = AsyncMock(return_value=False)
session = MagicMock()
session.get = MagicMock(return_value=ctx)
result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345))
assert result is None
def test_parse_failure(self) -> None:
resp = _FakeTextResponse(200, "<html>no script</html>")
session = MagicMock()
session.get = MagicMock(return_value=resp)
result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345))
assert result is None
class TestFetchLeisureTimes:
"""Tests for _fetch_leisure_times."""
def test_updates_cache(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243}],
"relationships": [],
}
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
):
asyncio.run(_fetch_leisure_times(results, cache, None))
assert cache[440] == round(21243 / 3600, 2)
assert results[0].completionist_hours == round(21243 / 3600, 2)
def test_no_valid_results(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=0,
),
]
cache: dict[int, float] = {}
asyncio.run(_fetch_leisure_times(results, cache, None))
assert not cache
def test_empty_results(self) -> None:
cache: dict[int, float] = {}
asyncio.run(_fetch_leisure_times([], cache, None))
assert not cache
def test_detail_returns_none(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=None,
):
asyncio.run(_fetch_leisure_times(results, cache, None))
assert not cache
assert results[0].completionist_hours == 50.0
def test_negative_leisure(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
game_data: dict[str, Any] = {"game": [], "relationships": []}
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
):
asyncio.run(_fetch_leisure_times(results, cache, None))
assert not cache
assert results[0].completionist_hours == 50.0
def test_with_progress_cb(self) -> None:
results = [
HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
]
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": [],
}
cache: dict[int, float] = {}
cb = MagicMock()
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
):
asyncio.run(_fetch_leisure_times(results, cache, cb))
cb.assert_called_once()
def test_save_interval(self) -> None:
"""Trigger the _SAVE_INTERVAL branch in leisure fetching."""
results = [
HLTBResult(
app_id=i,
game_name=f"Game{i}",
completionist_hours=1.0,
similarity=1.0,
hltb_game_id=i + 1000,
)
for i in range(_SAVE_INTERVAL)
]
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": [],
}
cache: dict[int, float] = {}
with (
patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
return_value=game_data,
),
patch(
"python_pkg.steam_backlog_enforcer._hltb_detail.save_hltb_cache"
) as mock_save,
):
asyncio.run(_fetch_leisure_times(results, cache, None))
mock_save.assert_called_once()
def test_dlc_detail_overrides_relationship_fallback(self) -> None:
results = [
HLTBResult(
app_id=1289310,
game_name="Helltaker",
completionist_hours=1.0,
similarity=1.0,
hltb_game_id=78118,
),
]
base_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6846}],
"relationships": [{"game_type": "dlc", "game_id": 92236, "comp_100": 4075}],
}
dlc_data: dict[str, Any] = {
"game": [{"comp_100_h": 12298, "comp_100": 4075}],
"relationships": [],
}
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
side_effect=[base_data, dlc_data],
):
asyncio.run(_fetch_leisure_times(results, cache, None))
expected = round((21243 + 12298) / 3600, 2)
assert cache[1289310] == expected
assert results[0].completionist_hours == expected
def test_missing_dlc_detail_keeps_relationship_fallback(self) -> None:
results = [
HLTBResult(
app_id=1289310,
game_name="Helltaker",
completionist_hours=1.0,
similarity=1.0,
hltb_game_id=78118,
),
]
base_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6846}],
"relationships": [{"game_type": "dlc", "game_id": 92236, "comp_100": 4075}],
}
cache: dict[int, float] = {}
with patch(
"python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one",
new_callable=AsyncMock,
side_effect=[base_data, None],
):
asyncio.run(_fetch_leisure_times(results, cache, None))
expected = round((21243 + 4075) / 3600, 2)
assert cache[1289310] == expected
assert results[0].completionist_hours == expected

440
steam_backlog_enforcer/tests/test_hltb_search.py Normal file
View File

@ -0,0 +1,440 @@
"""Tests for HLTB search, batch-fetch, and page parsing — part 2."""
from __future__ import annotations
import asyncio
import json
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock, patch
import aiohttp
from typing_extensions import Self
from python_pkg.steam_backlog_enforcer._hltb_detail import (
_extract_leisure_hours,
_parse_game_page,
)
from python_pkg.steam_backlog_enforcer.hltb import (
_SAVE_INTERVAL,
HLTBResult,
_AuthInfo,
_fetch_batch,
_search_one,
_SearchCtx,
)
if TYPE_CHECKING:
from collections.abc import Callable
class _FakeResponse:
"""Async context manager mimicking aiohttp response."""
def __init__(self, status: int, json_data: dict[str, Any] | None = None) -> None:
self.status = status
self._json_data = json_data or {}
async def __aenter__(self) -> Self:
return self
async def __aexit__(self, *args: object) -> None:
pass
async def json(self) -> dict[str, Any]:
return self._json_data
def _make_session(resp: _FakeResponse) -> MagicMock:
session = MagicMock()
session.post.return_value = resp
return session
def _make_ctx(
session: MagicMock,
*,
cache: dict[int, float] | None = None,
progress_cb: Callable[..., object] | None = None,
) -> _SearchCtx:
return _SearchCtx(
session=session,
search_url="https://example.com/search",
headers={},
cache=cache if cache is not None else {},
counter={"done": 0, "found": 0},
total=1,
progress_cb=progress_cb,
)
class TestSearchOne:
"""Tests for _search_one."""
def test_found(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2",
"game_alias": "",
"comp_100": 180000,
"game_id": 12345,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
assert result.app_id == 440
def test_not_found(self) -> None:
resp = _FakeResponse(200, {"data": []})
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
assert ctx.cache[440] == -1
def test_error(self) -> None:
session = MagicMock()
session.post.side_effect = aiohttp.ClientError("fail")
ctx = _make_ctx(session)
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
def test_non_200(self) -> None:
resp = _FakeResponse(500)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
def test_with_progress_cb(self) -> None:
resp = _FakeResponse(200, {"data": []})
cb = MagicMock()
ctx = _make_ctx(_make_session(resp), progress_cb=cb)
asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
cb.assert_called_once()
def test_low_similarity_skipped(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "Completely Different Name",
"game_alias": "",
"comp_100": 3600,
"game_id": 1,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
def test_zero_comp_100_skipped(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2",
"game_alias": "",
"comp_100": 0,
"game_id": 1,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is None
def test_alias_match(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "Team Fortress 2",
"game_alias": "TF2",
"comp_100": 180000,
"game_id": 12345,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
def test_full_edition_colon(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2: Complete",
"game_alias": "",
"comp_100": 180000,
"game_id": 99,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
def test_full_edition_dash(self) -> None:
resp = _FakeResponse(
200,
{
"data": [
{
"game_name": "TF2 - Complete",
"game_alias": "",
"comp_100": 180000,
"game_id": 99,
}
],
},
)
ctx = _make_ctx(_make_session(resp))
result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
assert result is not None
def test_save_interval(self) -> None:
"""Trigger the _SAVE_INTERVAL branch."""
resp = _FakeResponse(200, {"data": []})
ctx = _make_ctx(_make_session(resp))
# Set done to one less than _SAVE_INTERVAL so it triggers save
ctx.counter["done"] = _SAVE_INTERVAL - 1
with patch(
"python_pkg.steam_backlog_enforcer.hltb.save_hltb_cache"
) as mock_save:
asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2"))
mock_save.assert_called_once()
class TestFetchBatchHltb:
"""Tests for _fetch_batch (the hltb version)."""
def test_no_auth(self) -> None:
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url",
return_value="https://example.com",
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_auth_info",
new_callable=AsyncMock,
return_value=None,
),
):
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None))
assert results == []
def test_with_auth(self) -> None:
auth = _AuthInfo("token123", "ign_x", "ff")
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url",
return_value="https://example.com",
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_auth_info",
new_callable=AsyncMock,
return_value=auth,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._search_one",
new_callable=AsyncMock,
return_value=HLTBResult(
app_id=440,
game_name="TF2",
completionist_hours=50.0,
similarity=1.0,
hltb_game_id=12345,
),
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times",
new_callable=AsyncMock,
),
):
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None))
assert len(results) == 1
def test_with_auth_no_hp(self) -> None:
auth = _AuthInfo("tok123")
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url",
return_value="https://example.com",
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_auth_info",
new_callable=AsyncMock,
return_value=auth,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._search_one",
new_callable=AsyncMock,
return_value=None,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times",
new_callable=AsyncMock,
),
):
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None))
assert results == []
def test_filters_none_results(self) -> None:
auth = _AuthInfo("tok123")
with (
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url",
return_value="https://example.com",
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._get_auth_info",
new_callable=AsyncMock,
return_value=auth,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._search_one",
new_callable=AsyncMock,
return_value=None,
),
patch(
"python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times",
new_callable=AsyncMock,
),
):
results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None))
assert results == []
class TestParseGamePage:
"""Tests for _parse_game_page."""
def test_valid_html(self) -> None:
game_data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6800}],
"relationships": [],
}
next_data = {
"props": {"pageProps": {"game": {"data": game_data}}},
}
html = (
'<html><script id="__NEXT_DATA__" type="application/json">'
+ json.dumps(next_data)
+ "</script></html>"
)
assert _parse_game_page(html) == game_data
def test_no_script_tag(self) -> None:
assert _parse_game_page("<html></html>") is None
def test_bad_json(self) -> None:
html = '<script id="__NEXT_DATA__" type="application/json">{not json}</script>'
assert _parse_game_page(html) is None
def test_missing_keys(self) -> None:
html = (
'<script id="__NEXT_DATA__" type="application/json">{"props": {}}</script>'
)
assert _parse_game_page(html) is None
class TestExtractLeisureHours:
"""Tests for _extract_leisure_hours."""
def test_leisure_time_only(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6800}],
"relationships": [],
}
assert _extract_leisure_hours(data) == round(21243 / 3600, 2)
def test_leisure_with_dlc(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 21243, "comp_100": 6800}],
"relationships": [
{"game_type": "dlc", "comp_100": 12298},
{"game_type": "dlc", "comp_100": 3600},
],
}
assert _extract_leisure_hours(data) == round((21243 + 12298 + 3600) / 3600, 2)
def test_fallback_to_comp_100(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100": 7200}],
"relationships": [],
}
assert _extract_leisure_hours(data) == round(7200 / 3600, 2)
def test_no_game_data(self) -> None:
assert _extract_leisure_hours({"game": [], "relationships": []}) == -1
def test_zero_leisure(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 0, "comp_100": 0}],
"relationships": [],
}
assert _extract_leisure_hours(data) == -1
def test_no_game_key(self) -> None:
assert _extract_leisure_hours({"relationships": []}) == -1
def test_non_dlc_relationship_ignored(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": [
{"game_type": "game", "comp_100": 9999},
{"game_type": "dlc", "comp_100": 1800},
],
}
assert _extract_leisure_hours(data) == round((3600 + 1800) / 3600, 2)
def test_dlc_zero_comp_100_skipped(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": [
{"game_type": "dlc", "comp_100": 0},
],
}
assert _extract_leisure_hours(data) == round(3600 / 3600, 2)
def test_negative_leisure(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": -1, "comp_100": -1}],
"relationships": [],
}
assert _extract_leisure_hours(data) == -1
def test_string_numeric_fields(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": "7200", "comp_100": "3600"}],
"relationships": [{"game_type": "dlc", "game_id": "1", "comp_100": "1800"}],
}
assert _extract_leisure_hours(data) == round((7200 + 1800) / 3600, 2)
def test_bad_string_falls_back_to_comp_100(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": "bad", "comp_100": "3600"}],
"relationships": [],
}
assert _extract_leisure_hours(data) == 1.0
def test_relationships_not_list(self) -> None:
data: dict[str, Any] = {
"game": [{"comp_100_h": 3600}],
"relationships": "not-a-list",
}
assert _extract_leisure_hours(data) == 1.0

144
steam_backlog_enforcer/tests/test_main.py
View File

@ -7,7 +7,6 @@ from unittest.mock import patch
from python_pkg.steam_backlog_enforcer.config import Config, State
from python_pkg.steam_backlog_enforcer.main import (
_try_reassign_shorter_game,
cmd_buy_dlc,
cmd_hide,
cmd_install,
@ -20,7 +19,6 @@ from python_pkg.steam_backlog_enforcer.main import (
cmd_unhide,
cmd_uninstall,
)
from python_pkg.steam_backlog_enforcer.steam_api import GameInfo
PKG = "python_pkg.steam_backlog_enforcer.main"
@ -379,145 +377,3 @@ class TestCmdUnhide:
patch(f"{PKG}._echo"),
):
cmd_unhide(Config(), State())
class TestTryReassignShorterGame:
"""Tests for _try_reassign_shorter_game."""
def test_no_snapshot(self) -> None:
with patch(f"{PKG}.load_snapshot", return_value=None):
assert not _try_reassign_shorter_game({}, 1, 10.0, State(), Config())
def test_no_shorter_candidate(self) -> None:
snap = [_snap(1, "G", 10, 5, 10.0), _snap(2, "H", 10, 5, -1)]
with (
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}._echo"),
):
result = _try_reassign_shorter_game(
{1: 10.0},
1,
10.0,
State(),
Config(),
)
assert not result
def test_reassigns(self) -> None:
snap = [
_snap(1, "Long", 10, 5, 100.0),
_snap(2, "Short", 10, 5, 5.0),
]
state = State(current_app_id=1, current_game_name="Long")
short_game = GameInfo(
app_id=2,
name="Short",
total_achievements=10,
unlocked_achievements=5,
playtime_minutes=60,
completionist_hours=5.0,
)
with (
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}._echo"),
patch(
f"{PKG}._pick_playable_candidate",
return_value=short_game,
),
patch(f"{PKG}.pick_next_game"),
):
result = _try_reassign_shorter_game(
{1: 100.0, 2: 5.0},
1,
100.0,
state,
Config(),
)
assert result
def test_playable_none(self) -> None:
snap = [
_snap(1, "Long", 10, 5, 100.0),
_snap(2, "Short", 10, 5, 5.0),
]
with (
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}._pick_playable_candidate", return_value=None),
patch(f"{PKG}._echo"),
):
result = _try_reassign_shorter_game(
{1: 100.0, 2: 5.0},
1,
100.0,
State(),
Config(),
)
assert not result
def test_playable_longer(self) -> None:
"""Playable candidate is longer than current — no reassign."""
snap = [
_snap(1, "Short", 10, 5, 10.0),
_snap(2, "Long", 10, 5, 200.0),
]
long_game = GameInfo(
app_id=2,
name="Long",
total_achievements=10,
unlocked_achievements=5,
playtime_minutes=60,
completionist_hours=200.0,
)
with (
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}._pick_playable_candidate", return_value=long_game),
patch(f"{PKG}._echo"),
):
result = _try_reassign_shorter_game(
{1: 10.0, 2: 200.0},
1,
10.0,
State(),
Config(),
)
assert not result
def test_refreshes_stale_shorter_snapshot_entry(self) -> None:
"""Uncached shorter snapshot candidates are refreshed before reassigning."""
snap = [
_snap(1, "Current", 10, 5, 20.1),
_snap(2, "Lacuna", 10, 0, 0.9),
]
state = State(current_app_id=1, current_game_name="Current")
refreshed_short = GameInfo(
app_id=2,
name="Lacuna",
total_achievements=10,
unlocked_achievements=0,
playtime_minutes=60,
completionist_hours=18.8,
)
with (
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(
f"{PKG}.fetch_hltb_times_cached",
return_value={2: 18.8},
) as mock_fetch_hltb,
patch(
f"{PKG}._pick_playable_candidate",
return_value=refreshed_short,
) as mock_pick_playable,
patch(f"{PKG}.pick_next_game"),
patch(f"{PKG}._echo"),
):
result = _try_reassign_shorter_game(
{1: 20.1},
1,
20.1,
state,
Config(),
)
assert result
mock_fetch_hltb.assert_called_once_with([(2, "Lacuna")])
mock_pick_playable.assert_called_once()

139
steam_backlog_enforcer/tests/test_main_part2.py
View File

@ -8,15 +8,16 @@ from unittest.mock import MagicMock, patch
import pytest
from python_pkg.steam_backlog_enforcer.config import Config, State
from python_pkg.steam_backlog_enforcer.main import (
from python_pkg.steam_backlog_enforcer._cmd_done import (
_enforce_on_done,
_finalize_completion,
cmd_done,
main,
)
from python_pkg.steam_backlog_enforcer.config import Config, State
from python_pkg.steam_backlog_enforcer.main import main
from python_pkg.steam_backlog_enforcer.steam_api import GameInfo
CMD_DONE_PKG = "python_pkg.steam_backlog_enforcer._cmd_done"
PKG = "python_pkg.steam_backlog_enforcer.main"
@ -45,12 +46,12 @@ class TestFinalizeCompletion:
state = State(current_app_id=1, current_game_name="G")
snap = [_snap(2, "NewGame", 10, 0, 5.0)]
with (
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}.pick_next_game") as mock_pick,
patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]),
patch(f"{PKG}.hide_other_games", return_value=2),
patch(f"{PKG}.send_notification"),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick,
patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2, 3]),
patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=2),
patch(f"{CMD_DONE_PKG}.send_notification"),
patch.object(State, "save"),
):
@ -70,8 +71,8 @@ class TestFinalizeCompletion:
config = Config()
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_snapshot", return_value=None),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=None),
patch.object(State, "save"),
):
_finalize_completion(config, state, "G", 1)
@ -82,9 +83,9 @@ class TestFinalizeCompletion:
state = State(current_app_id=1, current_game_name="G")
snap = [_snap(1, "G", 10, 10)]
with (
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}.pick_next_game") as mock_pick,
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick,
patch.object(State, "save"),
):
@ -103,11 +104,11 @@ class TestFinalizeCompletion:
state = State(current_app_id=1, current_game_name="G")
snap = [_snap(2, "Next", 10, 0)]
with (
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}.pick_next_game") as mock_pick,
patch(f"{PKG}.get_all_owned_app_ids", return_value=[]),
patch(f"{PKG}.send_notification"),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick,
patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[]),
patch(f"{CMD_DONE_PKG}.send_notification"),
patch.object(State, "save"),
):
@ -127,12 +128,12 @@ class TestFinalizeCompletion:
state = State(current_app_id=1, current_game_name="G")
snap = [_snap(2, "Next", 10, 0)]
with (
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}.pick_next_game") as mock_pick,
patch(f"{PKG}.get_all_owned_app_ids", return_value=[1, 2]),
patch(f"{PKG}.hide_other_games", return_value=0),
patch(f"{PKG}.send_notification"),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}.pick_next_game") as mock_pick,
patch(f"{CMD_DONE_PKG}.get_all_owned_app_ids", return_value=[1, 2]),
patch(f"{CMD_DONE_PKG}.hide_other_games", return_value=0),
patch(f"{CMD_DONE_PKG}.send_notification"),
patch.object(State, "save"),
):
@ -168,14 +169,14 @@ class TestFinalizeCompletion:
s.current_app_id = None
with (
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_snapshot", return_value=snap),
patch(f"{PKG}.load_hltb_cache", return_value={2: 20.05}),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={2: 20.05}),
patch(
f"{PKG}.fetch_hltb_times_cached",
f"{CMD_DONE_PKG}.fetch_hltb_times_cached",
return_value={3: 18.81},
) as mock_fetch_hltb,
patch(f"{PKG}.pick_next_game", side_effect=capture_pick),
patch(f"{CMD_DONE_PKG}.pick_next_game", side_effect=capture_pick),
patch.object(State, "save"),
):
_finalize_completion(config, state, "G", 1)
@ -198,13 +199,13 @@ class TestEnforceOnDone:
)
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}._echo"),
patch(f"{CMD_DONE_PKG}._echo"),
patch(
f"{PKG}.enforce_allowed_game",
f"{CMD_DONE_PKG}.enforce_allowed_game",
return_value=[(1234, 999)],
),
patch(f"{PKG}.uninstall_other_games", return_value=2),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{CMD_DONE_PKG}.uninstall_other_games", return_value=2),
patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True),
):
_enforce_on_done(config, state)
@ -215,10 +216,10 @@ class TestEnforceOnDone:
)
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}._echo"),
patch(f"{PKG}.enforce_allowed_game", return_value=[]),
patch(f"{PKG}.uninstall_other_games", return_value=0),
patch(f"{PKG}.is_game_installed", return_value=True),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.enforce_allowed_game", return_value=[]),
patch(f"{CMD_DONE_PKG}.uninstall_other_games", return_value=0),
patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=True),
):
_enforce_on_done(config, state)
@ -230,9 +231,9 @@ class TestEnforceOnDone:
)
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}._echo"),
patch(f"{PKG}.is_game_installed", return_value=False),
patch(f"{PKG}.install_game") as mock_install,
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.is_game_installed", return_value=False),
patch(f"{CMD_DONE_PKG}.install_game") as mock_install,
):
_enforce_on_done(config, state)
mock_install.assert_called_once_with(1, "G", "s1", use_steam_protocol=True)
@ -242,7 +243,7 @@ class TestCmdDone:
"""Tests for cmd_done."""
def test_no_game_assigned(self) -> None:
with patch(f"{PKG}._echo") as mock_echo:
with patch(f"{CMD_DONE_PKG}._echo") as mock_echo:
cmd_done(Config(), State())
assert any("No game" in str(c) for c in mock_echo.call_args_list)
@ -251,8 +252,8 @@ class TestCmdDone:
mock_client.refresh_single_game.return_value = None
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{PKG}._echo"),
patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{CMD_DONE_PKG}._echo"),
):
cmd_done(Config(steam_api_key="k", steam_id="i"), state)
@ -268,11 +269,11 @@ class TestCmdDone:
mock_client.refresh_single_game.return_value = game
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_hltb_cache", return_value={1: 20.0}),
patch(f"{PKG}._try_reassign_shorter_game", return_value=False),
patch(f"{PKG}._enforce_on_done"),
patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 20.0}),
patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False),
patch(f"{CMD_DONE_PKG}._enforce_on_done"),
):
cmd_done(Config(steam_api_key="k", steam_id="i"), state)
@ -288,11 +289,11 @@ class TestCmdDone:
mock_client.refresh_single_game.return_value = game
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_hltb_cache", return_value={1: 10.0}),
patch(f"{PKG}._try_reassign_shorter_game", return_value=False),
patch(f"{PKG}._finalize_completion") as mock_final,
patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 10.0}),
patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False),
patch(f"{CMD_DONE_PKG}._finalize_completion") as mock_final,
):
cmd_done(Config(steam_api_key="k", steam_id="i"), state)
mock_final.assert_called_once()
@ -309,15 +310,15 @@ class TestCmdDone:
mock_client.refresh_single_game.return_value = game
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_hltb_cache", return_value={}),
patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={}),
patch(
f"{PKG}.fetch_hltb_times_cached",
f"{CMD_DONE_PKG}.fetch_hltb_times_cached",
return_value={1: 15.0},
),
patch(f"{PKG}._try_reassign_shorter_game", return_value=False),
patch(f"{PKG}._enforce_on_done"),
patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False),
patch(f"{CMD_DONE_PKG}._enforce_on_done"),
):
cmd_done(Config(steam_api_key="k", steam_id="i"), state)
@ -334,11 +335,11 @@ class TestCmdDone:
mock_client.refresh_single_game.return_value = game
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_hltb_cache", return_value={1: -1.0}),
patch(f"{PKG}._try_reassign_shorter_game", return_value=False),
patch(f"{PKG}._enforce_on_done"),
patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: -1.0}),
patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=False),
patch(f"{CMD_DONE_PKG}._enforce_on_done"),
):
cmd_done(Config(steam_api_key="k", steam_id="i"), state)
@ -354,10 +355,10 @@ class TestCmdDone:
mock_client.refresh_single_game.return_value = game
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{PKG}._echo"),
patch(f"{PKG}.load_hltb_cache", return_value={1: 50.0}),
patch(f"{PKG}._try_reassign_shorter_game", return_value=True),
patch(f"{CMD_DONE_PKG}.SteamAPIClient", return_value=mock_client),
patch(f"{CMD_DONE_PKG}._echo"),
patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={1: 50.0}),
patch(f"{CMD_DONE_PKG}._try_reassign_shorter_game", return_value=True),
):
cmd_done(Config(steam_api_key="k", steam_id="i"), state)