diff --git a/steam_backlog_enforcer/_polls_reporting.py b/steam_backlog_enforcer/_polls_reporting.py new file mode 100644 index 0000000..20e04ef --- /dev/null +++ b/steam_backlog_enforcer/_polls_reporting.py @@ -0,0 +1,108 @@ +"""Backfilling and reporting HLTB poll confidence for finished games. + +Split to keep both files under the 250-line cap. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._echo import _echo +from steam_backlog_enforcer._hltb_types import ( + load_hltb_cache, + load_hltb_polls_cache, + save_hltb_cache, +) +from steam_backlog_enforcer.hltb import fetch_hltb_confidence_cached + +if TYPE_CHECKING: + from steam_backlog_enforcer._steam_models import GameInfo + from steam_backlog_enforcer.config import State + +logger = logging.getLogger(__name__) + + +def _backfill_polls_for_finished( + state: State, + games: list[GameInfo], +) -> dict[int, int]: + """Lazily fetch poll counts for already-finished games missing them. + + Reads the polls cache, identifies finished games whose poll count is + still ``0`` (typically because the cache predates the polls schema), + and triggers a one-shot HLTB search to backfill them. Returns the + refreshed polls cache. + """ + polls_cache = load_hltb_polls_cache() + name_by_id = {g.app_id: g.name for g in games} + missing = [ + (aid, name_by_id[aid]) + for aid in state.finished_app_ids + if aid in name_by_id and polls_cache.get(aid, 0) == 0 + ] + if not missing: + return polls_cache + + logger.info( + "Backfilling HLTB poll counts for %d already-finished games...", + len(missing), + ) + # Force a fresh search by removing the hours entries we want to refetch. + # (fetch_hltb_times_cached skips entries already in the hours cache.) + cache = load_hltb_cache() + preserved_hours = {aid: cache[aid] for aid, _ in missing if aid in cache} + for aid, _name in missing: + cache.pop(aid, None) + save_hltb_cache(cache, polls_cache) + + fetch_hltb_confidence_cached(missing) + + # Restore any previously-known hours that the refetch may have replaced + # with a worse match (we trust prior leisure+dlc estimates). + refreshed_hours = load_hltb_cache() + refreshed_polls = load_hltb_polls_cache() + for aid, prior_hours in preserved_hours.items(): + if prior_hours > 0 and refreshed_hours.get(aid, -1) <= 0: + refreshed_hours[aid] = prior_hours + save_hltb_cache(refreshed_hours, refreshed_polls) + return refreshed_polls + + +def _report_poll_confidence( + chosen: GameInfo, + games: list[GameInfo], + state: State, +) -> None: + """Print HLTB poll-count confidence info for the just-assigned game. + + Shows the chosen game's ``comp_100_count`` (number of polled + completionist times on HowLongToBeat) and the historical minimum + among the user's previously-finished games. Marks a new historical + low so the user can be skeptical of unreliable estimates. + """ + polls_cache = _backfill_polls_for_finished(state, games) + chosen_polls = polls_cache.get(chosen.app_id, chosen.comp_100_count) + chosen.comp_100_count = chosen_polls + + finished_polls = [ + (polls_cache[aid], aid) + for aid in state.finished_app_ids + if polls_cache.get(aid, 0) > 0 + ] + if not finished_polls: + _echo(f" HLTB confidence: {chosen_polls} polled completionist times") + return + + min_polls, min_aid = min(finished_polls) + name_by_id = {g.app_id: g.name for g in games} + min_name = name_by_id.get(min_aid, f"AppID={min_aid}") + + warning = "" + if 0 < chosen_polls < min_polls: + warning = " ⚠ NEW LOW — estimate may be unreliable" + elif chosen_polls == 0: + warning = " ⚠ no polls recorded — estimate may be unreliable" + + _echo(f" HLTB confidence: {chosen_polls} polled completionist times{warning}") + _echo(f" Historical min among finished: {min_polls} ({min_name})") diff --git a/steam_backlog_enforcer/_scanning_confidence.py b/steam_backlog_enforcer/_scanning_confidence.py index 8502881..9eeef0d 100644 --- a/steam_backlog_enforcer/_scanning_confidence.py +++ b/steam_backlog_enforcer/_scanning_confidence.py @@ -16,11 +16,20 @@ from steam_backlog_enforcer.game_install import _echo from steam_backlog_enforcer.hltb import fetch_hltb_confidence_cached if TYPE_CHECKING: - from steam_backlog_enforcer.config import State from steam_backlog_enforcer.steam_api import GameInfo +from steam_backlog_enforcer._polls_reporting import ( + _backfill_polls_for_finished, + _report_poll_confidence, +) + logger = logging.getLogger(__name__) +__all__ = [ + "_backfill_polls_for_finished", + "_report_poll_confidence", +] + _MIN_COMP_100_POLLS = 3 _MIN_COUNT_COMP = 15 _MIN_CONFIDENCE_SUM = 18 @@ -166,88 +175,3 @@ def _candidate_passes_hltb_confidence(game: GameInfo) -> bool: ) return False return True - - -def _backfill_polls_for_finished( - state: State, - games: list[GameInfo], -) -> dict[int, int]: - """Lazily fetch poll counts for already-finished games missing them. - - Reads the polls cache, identifies finished games whose poll count is - still ``0`` (typically because the cache predates the polls schema), - and triggers a one-shot HLTB search to backfill them. Returns the - refreshed polls cache. - """ - polls_cache = load_hltb_polls_cache() - name_by_id = {g.app_id: g.name for g in games} - missing = [ - (aid, name_by_id[aid]) - for aid in state.finished_app_ids - if aid in name_by_id and polls_cache.get(aid, 0) == 0 - ] - if not missing: - return polls_cache - - logger.info( - "Backfilling HLTB poll counts for %d already-finished games...", - len(missing), - ) - # Force a fresh search by removing the hours entries we want to refetch. - # (fetch_hltb_times_cached skips entries already in the hours cache.) - cache = load_hltb_cache() - preserved_hours = {aid: cache[aid] for aid, _ in missing if aid in cache} - for aid, _name in missing: - cache.pop(aid, None) - save_hltb_cache(cache, polls_cache) - - fetch_hltb_confidence_cached(missing) - - # Restore any previously-known hours that the refetch may have replaced - # with a worse match (we trust prior leisure+dlc estimates). - refreshed_hours = load_hltb_cache() - refreshed_polls = load_hltb_polls_cache() - for aid, prior_hours in preserved_hours.items(): - if prior_hours > 0 and refreshed_hours.get(aid, -1) <= 0: - refreshed_hours[aid] = prior_hours - save_hltb_cache(refreshed_hours, refreshed_polls) - return refreshed_polls - - -def _report_poll_confidence( - chosen: GameInfo, - games: list[GameInfo], - state: State, -) -> None: - """Print HLTB poll-count confidence info for the just-assigned game. - - Shows the chosen game's ``comp_100_count`` (number of polled - completionist times on HowLongToBeat) and the historical minimum - among the user's previously-finished games. Marks a new historical - low so the user can be skeptical of unreliable estimates. - """ - polls_cache = _backfill_polls_for_finished(state, games) - chosen_polls = polls_cache.get(chosen.app_id, chosen.comp_100_count) - chosen.comp_100_count = chosen_polls - - finished_polls = [ - (polls_cache[aid], aid) - for aid in state.finished_app_ids - if polls_cache.get(aid, 0) > 0 - ] - if not finished_polls: - _echo(f" HLTB confidence: {chosen_polls} polled completionist times") - return - - min_polls, min_aid = min(finished_polls) - name_by_id = {g.app_id: g.name for g in games} - min_name = name_by_id.get(min_aid, f"AppID={min_aid}") - - warning = "" - if 0 < chosen_polls < min_polls: - warning = " ⚠ NEW LOW — estimate may be unreliable" - elif chosen_polls == 0: - warning = " ⚠ no polls recorded — estimate may be unreliable" - - _echo(f" HLTB confidence: {chosen_polls} polled completionist times{warning}") - _echo(f" Historical min among finished: {min_polls} ({min_name})") diff --git a/steam_backlog_enforcer/_steam_api_client.py b/steam_backlog_enforcer/_steam_api_client.py new file mode 100644 index 0000000..b850453 --- /dev/null +++ b/steam_backlog_enforcer/_steam_api_client.py @@ -0,0 +1,201 @@ +"""HTTP client for the Steam Web API. + +Split to keep both files under the 250-line cap. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor, as_completed +import logging +import threading +import time +from typing import TYPE_CHECKING, Any + +import requests + +from steam_backlog_enforcer._steam_models import ( + MAX_WORKERS, + STEAM_API_BASE, + AchievementInfo, + GameInfo, +) + +if TYPE_CHECKING: + from collections.abc import Callable + +logger = logging.getLogger(__name__) + + +class SteamAPIError(Exception): + """Raised when the Steam API returns an error.""" + + +class SteamAPIClient: + """Client for interacting with the Steam Web API.""" + + def __init__(self, api_key: str, steam_id: str) -> None: + """Initialize the Steam API client. + + Args: + api_key: Steam Web API key. + steam_id: Steam64 ID of the user. + """ + self.api_key = api_key + self.steam_id = steam_id + self.session = requests.Session() + adapter = requests.adapters.HTTPAdapter( + pool_maxsize=MAX_WORKERS, + pool_connections=MAX_WORKERS, + ) + self.session.mount("https://", adapter) + self.session.headers["Accept"] = "application/json" + self._rate_lock = threading.Lock() + self._request_times: list[float] = [] + self._max_rps = 18 + + def _rate_limit(self) -> None: + """Enforce rate limit across threads.""" + while True: + with self._rate_lock: + now = time.time() + self._request_times = [t for t in self._request_times if now - t < 1.0] + if len(self._request_times) < self._max_rps: + self._request_times.append(now) + return + time.sleep(0.06) + + def _get(self, url: str, params: dict[str, Any] | None = None) -> dict[str, Any]: + """Rate-limited GET request.""" + self._rate_limit() + if params is None: + params = {} + params["key"] = self.api_key + try: + resp = self.session.get(url, params=params, timeout=30) + resp.raise_for_status() + result: dict[str, Any] = resp.json() + except requests.RequestException as e: + msg = f"Steam API request failed: {e}" + raise SteamAPIError(msg) from e + else: + return result + + def get_owned_games(self) -> list[dict[str, Any]]: + """Fetch all games owned by the user.""" + url = f"{STEAM_API_BASE}/IPlayerService/GetOwnedGames/v1/" + data = self._get( + url, + { + "steamid": self.steam_id, + "include_appinfo": "true", + "include_played_free_games": "true", + "format": "json", + }, + ) + games: list[dict[str, Any]] = data.get("response", {}).get("games", []) + logger.info("Found %d owned games.", len(games)) + return games + + def get_achievement_details(self, app_id: int) -> list[AchievementInfo]: + """Fetch per-achievement detail for a game.""" + url = f"{STEAM_API_BASE}/ISteamUserStats/GetPlayerAchievements/v1/" + try: + data = self._get( + url, + { + "steamid": self.steam_id, + "appid": str(app_id), + "l": "english", + "format": "json", + }, + ) + except SteamAPIError: + return [] + + stats = data.get("playerstats", {}) + if not stats.get("success", False): + return [] + + raw: list[dict[str, Any]] = stats.get("achievements", []) + return [ + AchievementInfo( + api_name=a.get("apiname", ""), + display_name=a.get("name", a.get("apiname", "")), + achieved=bool(a.get("achieved", 0)), + unlock_time=a.get("unlocktime", 0), + ) + for a in raw + ] + + def _fetch_one_game(self, game_dict: dict[str, Any]) -> GameInfo | None: + """Fetch achievement data for one game. Thread-safe.""" + app_id = game_dict["appid"] + + achievements = self.get_achievement_details(app_id) + if not achievements: + return None + + name = game_dict.get("name", f"Unknown ({app_id})") + total = len(achievements) + unlocked = sum(1 for a in achievements if a.achieved) + + return GameInfo( + app_id=app_id, + name=name, + total_achievements=total, + unlocked_achievements=unlocked, + playtime_minutes=game_dict.get("playtime_forever", 0), + achievements=achievements, + ) + + def build_game_list( + self, + progress_callback: Callable[[int, int], None] | None = None, + ) -> list[GameInfo]: + """Build full game list with achievement data (parallel).""" + owned = self.get_owned_games() + games: list[GameInfo] = [] + done_count = 0 + total = len(owned) + lock = threading.Lock() + + with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: + futures = {pool.submit(self._fetch_one_game, g): g for g in owned} + for future in as_completed(futures): + try: + result = future.result() + except ( + KeyError, + TypeError, + ValueError, + SteamAPIError, + requests.RequestException, + ): + result = None + with lock: + done_count += 1 + if progress_callback: + progress_callback(done_count, total) + if result is not None: + games.append(result) + + games.sort(key=lambda g: g.name.lower()) + return games + + def refresh_single_game( + self, app_id: int, name: str, playtime: int = 0 + ) -> GameInfo | None: + """Re-fetch achievement data for one game.""" + achievements = self.get_achievement_details(app_id) + if not achievements: + return None + total = len(achievements) + unlocked = sum(1 for a in achievements if a.achieved) + return GameInfo( + app_id=app_id, + name=name, + total_achievements=total, + unlocked_achievements=unlocked, + playtime_minutes=playtime, + achievements=achievements, + ) diff --git a/steam_backlog_enforcer/_steam_models.py b/steam_backlog_enforcer/_steam_models.py new file mode 100644 index 0000000..fcf14c6 --- /dev/null +++ b/steam_backlog_enforcer/_steam_models.py @@ -0,0 +1,98 @@ +"""Steam Web API client for fetching games and achievement data.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +STEAM_API_BASE = "https://api.steampowered.com" +MAX_WORKERS = 20 + + +@dataclass +class AchievementInfo: + """Single achievement state.""" + + api_name: str + display_name: str + achieved: bool + unlock_time: int + + +@dataclass +class GameInfo: + """Info about an owned Steam game.""" + + app_id: int + name: str + total_achievements: int + unlocked_achievements: int + playtime_minutes: int + achievements: list[AchievementInfo] = field(default_factory=list) + completionist_hours: float = -1 + comp_100_count: int = 0 + count_comp: int = 0 + + @property + def completion_pct(self) -> float: + """Achievement completion percentage.""" + if self.total_achievements == 0: + return 100.0 + return (self.unlocked_achievements / self.total_achievements) * 100.0 + + @property + def is_complete(self) -> bool: + """True if all achievements are unlocked.""" + return ( + self.total_achievements > 0 + and self.unlocked_achievements >= self.total_achievements + ) + + def to_snapshot(self) -> dict[str, Any]: + """Serialize to JSON-safe dict.""" + return { + "app_id": self.app_id, + "name": self.name, + "total_achievements": self.total_achievements, + "unlocked_achievements": self.unlocked_achievements, + "playtime_minutes": self.playtime_minutes, + "completionist_hours": self.completionist_hours, + "comp_100_count": self.comp_100_count, + "count_comp": self.count_comp, + "achievements": [ + { + "api_name": a.api_name, + "display_name": a.display_name, + "achieved": a.achieved, + "unlock_time": a.unlock_time, + } + for a in self.achievements + ], + } + + @classmethod + def from_snapshot(cls, data: dict[str, Any]) -> GameInfo: + """Deserialize from a cached snapshot dict.""" + achievements = [ + AchievementInfo( + api_name=a["api_name"], + display_name=a.get("display_name", a["api_name"]), + achieved=a["achieved"], + unlock_time=a.get("unlock_time", 0), + ) + for a in data.get("achievements", []) + ] + return cls( + app_id=data["app_id"], + name=data["name"], + total_achievements=data["total_achievements"], + unlocked_achievements=data["unlocked_achievements"], + playtime_minutes=data.get("playtime_minutes", 0), + completionist_hours=data.get("completionist_hours", -1), + comp_100_count=data.get("comp_100_count", 0), + count_comp=data.get("count_comp", 0), + achievements=achievements, + ) diff --git a/steam_backlog_enforcer/steam_api.py b/steam_backlog_enforcer/steam_api.py index 187f149..06f2457 100644 --- a/steam_backlog_enforcer/steam_api.py +++ b/steam_backlog_enforcer/steam_api.py @@ -1,281 +1,19 @@ -"""Steam Web API client for fetching games and achievement data.""" +"""Steam Web API surface: data models and the HTTP client. + +The implementation lives in :mod:`_steam_models` (dataclasses) and +:mod:`_steam_api_client` (HTTP); this module re-exports both so the 47 +call sites that import from ``steam_api`` keep working. Split to keep every +file under the 250-line cap. +""" from __future__ import annotations -from concurrent.futures import ThreadPoolExecutor, as_completed -from dataclasses import dataclass, field -import logging -import threading -import time -from typing import TYPE_CHECKING, Any +from steam_backlog_enforcer._steam_api_client import SteamAPIClient, SteamAPIError +from steam_backlog_enforcer._steam_models import AchievementInfo, GameInfo -import requests - -if TYPE_CHECKING: - from collections.abc import Callable - -logger = logging.getLogger(__name__) - -STEAM_API_BASE = "https://api.steampowered.com" -MAX_WORKERS = 20 - - -@dataclass -class AchievementInfo: - """Single achievement state.""" - - api_name: str - display_name: str - achieved: bool - unlock_time: int - - -@dataclass -class GameInfo: - """Info about an owned Steam game.""" - - app_id: int - name: str - total_achievements: int - unlocked_achievements: int - playtime_minutes: int - achievements: list[AchievementInfo] = field(default_factory=list) - completionist_hours: float = -1 - comp_100_count: int = 0 - count_comp: int = 0 - - @property - def completion_pct(self) -> float: - """Achievement completion percentage.""" - if self.total_achievements == 0: - return 100.0 - return (self.unlocked_achievements / self.total_achievements) * 100.0 - - @property - def is_complete(self) -> bool: - """True if all achievements are unlocked.""" - return ( - self.total_achievements > 0 - and self.unlocked_achievements >= self.total_achievements - ) - - def to_snapshot(self) -> dict[str, Any]: - """Serialize to JSON-safe dict.""" - return { - "app_id": self.app_id, - "name": self.name, - "total_achievements": self.total_achievements, - "unlocked_achievements": self.unlocked_achievements, - "playtime_minutes": self.playtime_minutes, - "completionist_hours": self.completionist_hours, - "comp_100_count": self.comp_100_count, - "count_comp": self.count_comp, - "achievements": [ - { - "api_name": a.api_name, - "display_name": a.display_name, - "achieved": a.achieved, - "unlock_time": a.unlock_time, - } - for a in self.achievements - ], - } - - @classmethod - def from_snapshot(cls, data: dict[str, Any]) -> GameInfo: - """Deserialize from a cached snapshot dict.""" - achievements = [ - AchievementInfo( - api_name=a["api_name"], - display_name=a.get("display_name", a["api_name"]), - achieved=a["achieved"], - unlock_time=a.get("unlock_time", 0), - ) - for a in data.get("achievements", []) - ] - return cls( - app_id=data["app_id"], - name=data["name"], - total_achievements=data["total_achievements"], - unlocked_achievements=data["unlocked_achievements"], - playtime_minutes=data.get("playtime_minutes", 0), - completionist_hours=data.get("completionist_hours", -1), - comp_100_count=data.get("comp_100_count", 0), - count_comp=data.get("count_comp", 0), - achievements=achievements, - ) - - -class SteamAPIError(Exception): - """Raised when the Steam API returns an error.""" - - -class SteamAPIClient: - """Client for interacting with the Steam Web API.""" - - def __init__(self, api_key: str, steam_id: str) -> None: - """Initialize the Steam API client. - - Args: - api_key: Steam Web API key. - steam_id: Steam64 ID of the user. - """ - self.api_key = api_key - self.steam_id = steam_id - self.session = requests.Session() - adapter = requests.adapters.HTTPAdapter( - pool_maxsize=MAX_WORKERS, - pool_connections=MAX_WORKERS, - ) - self.session.mount("https://", adapter) - self.session.headers["Accept"] = "application/json" - self._rate_lock = threading.Lock() - self._request_times: list[float] = [] - self._max_rps = 18 - - def _rate_limit(self) -> None: - """Enforce rate limit across threads.""" - while True: - with self._rate_lock: - now = time.time() - self._request_times = [t for t in self._request_times if now - t < 1.0] - if len(self._request_times) < self._max_rps: - self._request_times.append(now) - return - time.sleep(0.06) - - def _get(self, url: str, params: dict[str, Any] | None = None) -> dict[str, Any]: - """Rate-limited GET request.""" - self._rate_limit() - if params is None: - params = {} - params["key"] = self.api_key - try: - resp = self.session.get(url, params=params, timeout=30) - resp.raise_for_status() - result: dict[str, Any] = resp.json() - except requests.RequestException as e: - msg = f"Steam API request failed: {e}" - raise SteamAPIError(msg) from e - else: - return result - - def get_owned_games(self) -> list[dict[str, Any]]: - """Fetch all games owned by the user.""" - url = f"{STEAM_API_BASE}/IPlayerService/GetOwnedGames/v1/" - data = self._get( - url, - { - "steamid": self.steam_id, - "include_appinfo": "true", - "include_played_free_games": "true", - "format": "json", - }, - ) - games: list[dict[str, Any]] = data.get("response", {}).get("games", []) - logger.info("Found %d owned games.", len(games)) - return games - - def get_achievement_details(self, app_id: int) -> list[AchievementInfo]: - """Fetch per-achievement detail for a game.""" - url = f"{STEAM_API_BASE}/ISteamUserStats/GetPlayerAchievements/v1/" - try: - data = self._get( - url, - { - "steamid": self.steam_id, - "appid": str(app_id), - "l": "english", - "format": "json", - }, - ) - except SteamAPIError: - return [] - - stats = data.get("playerstats", {}) - if not stats.get("success", False): - return [] - - raw: list[dict[str, Any]] = stats.get("achievements", []) - return [ - AchievementInfo( - api_name=a.get("apiname", ""), - display_name=a.get("name", a.get("apiname", "")), - achieved=bool(a.get("achieved", 0)), - unlock_time=a.get("unlocktime", 0), - ) - for a in raw - ] - - def _fetch_one_game(self, game_dict: dict[str, Any]) -> GameInfo | None: - """Fetch achievement data for one game. Thread-safe.""" - app_id = game_dict["appid"] - - achievements = self.get_achievement_details(app_id) - if not achievements: - return None - - name = game_dict.get("name", f"Unknown ({app_id})") - total = len(achievements) - unlocked = sum(1 for a in achievements if a.achieved) - - return GameInfo( - app_id=app_id, - name=name, - total_achievements=total, - unlocked_achievements=unlocked, - playtime_minutes=game_dict.get("playtime_forever", 0), - achievements=achievements, - ) - - def build_game_list( - self, - progress_callback: Callable[[int, int], None] | None = None, - ) -> list[GameInfo]: - """Build full game list with achievement data (parallel).""" - owned = self.get_owned_games() - games: list[GameInfo] = [] - done_count = 0 - total = len(owned) - lock = threading.Lock() - - with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: - futures = {pool.submit(self._fetch_one_game, g): g for g in owned} - for future in as_completed(futures): - try: - result = future.result() - except ( - KeyError, - TypeError, - ValueError, - SteamAPIError, - requests.RequestException, - ): - result = None - with lock: - done_count += 1 - if progress_callback: - progress_callback(done_count, total) - if result is not None: - games.append(result) - - games.sort(key=lambda g: g.name.lower()) - return games - - def refresh_single_game( - self, app_id: int, name: str, playtime: int = 0 - ) -> GameInfo | None: - """Re-fetch achievement data for one game.""" - achievements = self.get_achievement_details(app_id) - if not achievements: - return None - total = len(achievements) - unlocked = sum(1 for a in achievements if a.achieved) - return GameInfo( - app_id=app_id, - name=name, - total_achievements=total, - unlocked_achievements=unlocked, - playtime_minutes=playtime, - achievements=achievements, - ) +__all__ = [ + "AchievementInfo", + "GameInfo", + "SteamAPIClient", + "SteamAPIError", +] diff --git a/steam_backlog_enforcer/tests/conftest.py b/steam_backlog_enforcer/tests/conftest.py index e31bc96..9106585 100644 --- a/steam_backlog_enforcer/tests/conftest.py +++ b/steam_backlog_enforcer/tests/conftest.py @@ -166,7 +166,7 @@ def _no_real_sleep() -> Iterator[None]: with ( patch("steam_backlog_enforcer._steam_client.time.sleep", noop), patch("steam_backlog_enforcer._steam_launch.time.sleep", noop), - patch("steam_backlog_enforcer.steam_api.time.sleep", noop), + patch("steam_backlog_enforcer._steam_api_client.time.sleep", noop), patch("steam_backlog_enforcer._enforce_loop.time.sleep", noop), ): yield diff --git a/steam_backlog_enforcer/tests/test_polls_tracking_part2.py b/steam_backlog_enforcer/tests/test_polls_tracking_part2.py index cf6e5a3..9fc3b32 100644 --- a/steam_backlog_enforcer/tests/test_polls_tracking_part2.py +++ b/steam_backlog_enforcer/tests/test_polls_tracking_part2.py @@ -17,6 +17,7 @@ _TYPES = "steam_backlog_enforcer._hltb_types" _CMD = "steam_backlog_enforcer._cmd_done" _SCAN = "steam_backlog_enforcer.scanning" _SCANCONF = "steam_backlog_enforcer._scanning_confidence" +_POLLS = "steam_backlog_enforcer._polls_reporting" def _state(finished: list[int], current: int | None = None) -> State: @@ -81,11 +82,11 @@ class TestScanningPollsIntegration: ] with ( patch( - f"{_SCANCONF}._backfill_polls_for_finished", + f"{_POLLS}._backfill_polls_for_finished", return_value={1: 1, 2: 5}, ), patch( - f"{_SCANCONF}._echo", + f"{_POLLS}._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), ): @@ -105,9 +106,12 @@ class TestScanningPollsIntegration: comp_100_count=4, ) with ( - patch(f"{_SCANCONF}._backfill_polls_for_finished", return_value={1: 4}), patch( - f"{_SCANCONF}._echo", + f"{_POLLS}._backfill_polls_for_finished", + return_value={1: 4}, + ), + patch( + f"{_POLLS}._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), ): @@ -155,7 +159,10 @@ class TestScanningPollsIntegration: with ( patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), - patch(f"{_SCANCONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch( + f"{_POLLS}.fetch_hltb_confidence_cached", + side_effect=fake_fetch, + ), ): result = _scanning_confidence._backfill_polls_for_finished( _state([2]), @@ -189,7 +196,10 @@ class TestScanningPollsIntegration: with ( patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), - patch(f"{_SCANCONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch( + f"{_POLLS}.fetch_hltb_confidence_cached", + side_effect=fake_fetch, + ), ): _scanning_confidence._backfill_polls_for_finished( _state([2]), diff --git a/steam_backlog_enforcer/tests/test_polls_tracking_report_poll_confidence_chosen_zero_polls.py b/steam_backlog_enforcer/tests/test_polls_tracking_report_poll_confidence_chosen_zero_polls.py index 7b0f820..861ad9d 100644 --- a/steam_backlog_enforcer/tests/test_polls_tracking_report_poll_confidence_chosen_zero_polls.py +++ b/steam_backlog_enforcer/tests/test_polls_tracking_report_poll_confidence_chosen_zero_polls.py @@ -12,6 +12,7 @@ _TYPES = "steam_backlog_enforcer._hltb_types" _CMD = "steam_backlog_enforcer._cmd_done" _SCAN = "steam_backlog_enforcer.scanning" _SCANCONF = "steam_backlog_enforcer._scanning_confidence" +_POLLS = "steam_backlog_enforcer._polls_reporting" def _state(finished: list[int], current: int | None = None) -> State: @@ -45,11 +46,11 @@ class TestScanningPollsIntegrationGroup2: ) with ( patch( - f"{_SCANCONF}._backfill_polls_for_finished", + f"{_POLLS}._backfill_polls_for_finished", return_value={1: 0, 2: 5}, ), patch( - f"{_SCANCONF}._echo", + f"{_POLLS}._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), ): diff --git a/steam_backlog_enforcer/tests/test_polls_tracking_report_poll_confidence_chosen_zero_polls_report_poll_confidence_chosen_equals_min.py b/steam_backlog_enforcer/tests/test_polls_tracking_report_poll_confidence_chosen_zero_polls_report_poll_confidence_chosen_equals_min.py index 673b183..ff5b635 100644 --- a/steam_backlog_enforcer/tests/test_polls_tracking_report_poll_confidence_chosen_zero_polls_report_poll_confidence_chosen_equals_min.py +++ b/steam_backlog_enforcer/tests/test_polls_tracking_report_poll_confidence_chosen_zero_polls_report_poll_confidence_chosen_equals_min.py @@ -17,6 +17,7 @@ _TYPES = "steam_backlog_enforcer._hltb_types" _CMD = "steam_backlog_enforcer._cmd_done" _SCAN = "steam_backlog_enforcer.scanning" _SCANCONF = "steam_backlog_enforcer._scanning_confidence" +_POLLS = "steam_backlog_enforcer._polls_reporting" def _state(finished: list[int], current: int | None = None) -> State: @@ -50,11 +51,11 @@ class TestScanningPollsIntegrationGroup2Group2: ) with ( patch( - f"{_SCANCONF}._backfill_polls_for_finished", + f"{_POLLS}._backfill_polls_for_finished", return_value={1: 5, 2: 5}, ), patch( - f"{_SCANCONF}._echo", + f"{_POLLS}._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), ): @@ -108,7 +109,10 @@ class TestScanningPollsIntegrationGroup2Group2: with ( patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), - patch(f"{_SCANCONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch( + f"{_SCANCONF}.fetch_hltb_confidence_cached", + side_effect=fake_fetch, + ), patch(f"{_SCANCONF}._echo"), ): _scanning_confidence._refresh_candidate_confidence(game) @@ -160,7 +164,8 @@ class TestScanningPollsIntegrationGroup2Group2: patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), patch( - f"{_SCANCONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch + f"{_SCANCONF}.fetch_hltb_confidence_cached", + side_effect=fake_fetch, ) as mock_fetch, patch(f"{_SCANCONF}._echo"), ): diff --git a/steam_backlog_enforcer/tests/test_steam_api_steam_a_p_i_client.py b/steam_backlog_enforcer/tests/test_steam_api_steam_a_p_i_client.py index f616c61..cfdd576 100644 --- a/steam_backlog_enforcer/tests/test_steam_api_steam_a_p_i_client.py +++ b/steam_backlog_enforcer/tests/test_steam_api_steam_a_p_i_client.py @@ -35,7 +35,7 @@ class TestSteamAPIClient: # Fill up the rate limit window client._request_times = [__import__("time").time()] * client._max_rps with patch( - "steam_backlog_enforcer.steam_api.time.sleep", + "steam_backlog_enforcer._steam_api_client.time.sleep", ) as mock_sleep: # Next call should trigger sleep then succeed client._rate_limit()