diff --git a/steam_backlog_enforcer/_cmd_done.py b/steam_backlog_enforcer/_cmd_done.py
new file mode 100644
index 0000000..eada9b0
--- /dev/null
+++ b/steam_backlog_enforcer/_cmd_done.py
@@ -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)
diff --git a/steam_backlog_enforcer/_hltb_detail.py b/steam_backlog_enforcer/_hltb_detail.py
new file mode 100644
index 0000000..03a425e
--- /dev/null
+++ b/steam_backlog_enforcer/_hltb_detail.py
@@ -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'',
+)
+
+
+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)
diff --git a/steam_backlog_enforcer/_hltb_types.py b/steam_backlog_enforcer/_hltb_types.py
new file mode 100644
index 0000000..ddadec3
--- /dev/null
+++ b/steam_backlog_enforcer/_hltb_types.py
@@ -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")
diff --git a/steam_backlog_enforcer/hltb.py b/steam_backlog_enforcer/hltb.py
index ec57ea4..ce05768 100644
--- a/steam_backlog_enforcer/hltb.py
+++ b/steam_backlog_enforcer/hltb.py
@@ -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'',
-)
-
-
-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],
diff --git a/steam_backlog_enforcer/main.py b/steam_backlog_enforcer/main.py
index 555b447..df8a3e0 100644
--- a/steam_backlog_enforcer/main.py
+++ b/steam_backlog_enforcer/main.py
@@ -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),
diff --git a/steam_backlog_enforcer/tests/test_cmd_done.py b/steam_backlog_enforcer/tests/test_cmd_done.py
new file mode 100644
index 0000000..3458fcc
--- /dev/null
+++ b/steam_backlog_enforcer/tests/test_cmd_done.py
@@ -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()
diff --git a/steam_backlog_enforcer/tests/test_hltb.py b/steam_backlog_enforcer/tests/test_hltb.py
index bf402eb..2cb7be0 100644
--- a/steam_backlog_enforcer/tests/test_hltb.py
+++ b/steam_backlog_enforcer/tests/test_hltb.py
@@ -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 = (
- '"
- )
- assert _parse_game_page(html) == game_data
-
- def test_no_script_tag(self) -> None:
- assert _parse_game_page("") is None
-
- def test_bad_json(self) -> None:
- html = ''
- assert _parse_game_page(html) is None
-
- def test_missing_keys(self) -> None:
- html = (
- ''
- )
- 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 = (
- '"
- )
- 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, "no script")
- 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
diff --git a/steam_backlog_enforcer/tests/test_hltb_detail.py b/steam_backlog_enforcer/tests/test_hltb_detail.py
new file mode 100644
index 0000000..7d28c4e
--- /dev/null
+++ b/steam_backlog_enforcer/tests/test_hltb_detail.py
@@ -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 = (
+ '"
+ )
+ 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, "no script")
+ 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
diff --git a/steam_backlog_enforcer/tests/test_hltb_search.py b/steam_backlog_enforcer/tests/test_hltb_search.py
new file mode 100644
index 0000000..1d24037
--- /dev/null
+++ b/steam_backlog_enforcer/tests/test_hltb_search.py
@@ -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 = (
+ '"
+ )
+ assert _parse_game_page(html) == game_data
+
+ def test_no_script_tag(self) -> None:
+ assert _parse_game_page("") is None
+
+ def test_bad_json(self) -> None:
+ html = ''
+ assert _parse_game_page(html) is None
+
+ def test_missing_keys(self) -> None:
+ html = (
+ ''
+ )
+ 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
diff --git a/steam_backlog_enforcer/tests/test_main.py b/steam_backlog_enforcer/tests/test_main.py
index 86aea70..78cf24f 100644
--- a/steam_backlog_enforcer/tests/test_main.py
+++ b/steam_backlog_enforcer/tests/test_main.py
@@ -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()
diff --git a/steam_backlog_enforcer/tests/test_main_part2.py b/steam_backlog_enforcer/tests/test_main_part2.py
index 1252b7d..0b5583f 100644
--- a/steam_backlog_enforcer/tests/test_main_part2.py
+++ b/steam_backlog_enforcer/tests/test_main_part2.py
@@ -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)