diff --git a/steam_backlog_enforcer/_actions.py b/steam_backlog_enforcer/_actions.py index 0de7737..d139ef0 100644 --- a/steam_backlog_enforcer/_actions.py +++ b/steam_backlog_enforcer/_actions.py @@ -9,26 +9,20 @@ there is a single tested implementation of the underlying behaviour. from __future__ import annotations -from datetime import datetime, timedelta, timezone -from typing import TYPE_CHECKING, Any +from datetime import datetime, timezone +import importlib +from typing import TYPE_CHECKING, Any, TypeAlias from steam_backlog_enforcer._allowed_games import ( active_manual_picks, allowed_games, ) -from steam_backlog_enforcer._total_block import get_total_block_status -from steam_backlog_enforcer.game_install import get_installed_games, is_protected_app # Marks these two as an intentional re-export (they're imported from # _allowed_games above, not defined here) -- without this, mypy's # --no-implicit-reexport flags every downstream `from _actions import # active_manual_picks` as an error. Everything else in this file is # accessible normally; this only needs to list names imported-then-reexported. -__all__ = [ - "active_manual_picks", - "allowed_games", -] -from steam_backlog_enforcer.store_blocker import is_store_blocked if TYPE_CHECKING: from steam_backlog_enforcer.config import State @@ -160,123 +154,31 @@ def apply_manual_pick( return None -def manual_pick_grace_remaining(state: State, app_id: int) -> float | None: - """Return days left in *app_id*'s grace window, or ``None``. - - ``None`` means "no grace window applies": *app_id* is not an active manual - pick, or its timestamp is missing/malformed. A value <= 0 means the window - has closed. - - Args: - state: The loaded enforcer state. - app_id: The manually-picked app id to measure. - - Returns: - Fractional days remaining before the pick can no longer be abandoned, - or ``None`` when the question does not apply. - """ - pick = find_manual_pick(state, app_id) - if pick is None or not pick.get("started_at"): - return None - try: - started = datetime.fromisoformat(pick["started_at"]) - except ValueError: - return None - deadline = started + timedelta(days=MANUAL_GRACE_DAYS) - return (deadline - datetime.now(timezone.utc)).total_seconds() / 86400 - - -def can_abandon_manual_pick(state: State, app_id: int) -> bool: - """Return ``True`` if *app_id* is still inside its grace window. - - Args: - state: The loaded enforcer state. - app_id: The manually-picked app id to check. - - Returns: - Whether ``abandon_manual_pick`` would be accepted right now. - """ - remaining = manual_pick_grace_remaining(state, app_id) - return remaining is not None and remaining > 0 - - -def abandon_manual_pick(state: State, app_id: int) -> bool: - """Drop one manual pick and persist ``state``, if still inside its grace window. - - Only the named pick is dropped: any other active pick keeps its own lock - and deadline. The abandoned app id goes onto the existing skip cooldown so - ``scan`` will not hand the same game straight back. State-only: like - ``apply_manual_pick`` this performs no uninstall/hide cascade, so the MCP - server can call it without touching the filesystem. - - Args: - state: The enforcer state to mutate and save. - app_id: The manually-picked app id to back out of. - - Returns: - ``True`` if the pick was abandoned, ``False`` if its grace window has - closed (or it is not an active pick), in which case ``state`` is - untouched. - """ - if not can_abandon_manual_pick(state, app_id): - return False - - state.skip_for_days(app_id, ABANDON_COOLDOWN_DAYS) - state.manual_picks = [ - p for p in active_manual_picks(state) if p.get("app_id") != app_id - ] - - # The abandoned pick may also have been the current assignment; hand the - # assignment to a surviving pick so the enforcer keeps guarding it, or - # clear it so 'scan' can reassign. - if state.current_app_id == app_id: - survivor = state.manual_picks[-1] if state.manual_picks else None - state.current_app_id = survivor["app_id"] if survivor else None - state.current_game_name = survivor["game_name"] if survivor else "" - - state.save() - return True - - -def status_payload(state: State) -> dict[str, Any]: - """Build the structured status snapshot that ``cmd_status`` renders as text. - - Pure data, no stdout — safe for the MCP server. Reads only ``State`` and the - stdout-free leaf helpers; never constructs ``Config`` and never exposes the - Steam API key. - - Args: - state: The loaded enforcer state. - - Returns: - A JSON-ready dict describing the current enforcement status. - """ - total_block = get_total_block_status() - installed = get_installed_games() - real_games = [(aid, name) for aid, name in installed if not is_protected_app(aid)] - assigned_installed = ( - any(aid == state.current_app_id for aid, _ in installed) - if state.current_app_id - else None - ) - return { - "current_app_id": state.current_app_id, - "current_game_name": state.current_game_name or None, - "finished_count": len(state.finished_app_ids), - "store_blocked": is_store_blocked(), - "installed_count": len(real_games), - "assigned_game_installed": assigned_installed, - "total_block": { - "active": total_block.active, - "days_remaining": round(total_block.days_remaining, 1), - "until": total_block.until.isoformat() if total_block.until else None, - }, - "manual_pick_locked": is_manual_pick_locked(state), - "manual_picks": [ - { - **pick, - "grace_days_left": manual_pick_grace_remaining(state, pick["app_id"]), - } - for pick in active_manual_picks(state) - ], +_MOVED_TO_MANUAL_PICK_LIFECYCLE = frozenset( + { + "abandon_manual_pick", + "can_abandon_manual_pick", + "manual_pick_grace_remaining", + "status_payload", } +) + + +# Whatever the re-exported name turns out to be -- a function, a class or +# a constant. Aliased so the annotation is a name rather than a bare Any. +_Reexport: TypeAlias = Any + + +def __getattr__(name: str) -> _Reexport: + """Re-export the names that moved to :mod:`_manual_pick_lifecycle`. + + Deferred rather than imported at the top because _manual_pick_lifecycle imports + back from this module, so a module-level import would be circular. + """ + if name in _MOVED_TO_MANUAL_PICK_LIFECYCLE: + module = importlib.import_module( + "steam_backlog_enforcer._manual_pick_lifecycle", + ) + return getattr(module, name) + msg = f"module {__name__!r} has no attribute {name!r}" + raise AttributeError(msg) diff --git a/steam_backlog_enforcer/_cmd_done.py b/steam_backlog_enforcer/_cmd_done.py index ac9871c..03df7be 100644 --- a/steam_backlog_enforcer/_cmd_done.py +++ b/steam_backlog_enforcer/_cmd_done.py @@ -2,21 +2,14 @@ from __future__ import annotations +import importlib import logging import sys +from typing import TYPE_CHECKING, Any, TypeAlias -from steam_backlog_enforcer._actions import allowed_app_ids -from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids -from steam_backlog_enforcer.config import Config, State, load_snapshot -from steam_backlog_enforcer.enforcer import ( - enforce_allowed_game, - send_notification, -) +from steam_backlog_enforcer.config import State, load_snapshot from steam_backlog_enforcer.game_install import ( _echo, - install_game, - is_game_installed, - uninstall_other_games, ) from steam_backlog_enforcer.hltb import ( fetch_hltb_confidence_cached, @@ -25,9 +18,9 @@ from steam_backlog_enforcer.hltb import ( load_hltb_polls_cache, save_hltb_cache, ) -from steam_backlog_enforcer.library_hider import try_hide_other_games -from steam_backlog_enforcer.scanning import pick_next_game -from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient + +if TYPE_CHECKING: + from steam_backlog_enforcer.steam_api import GameInfo _REASSIGN_REFRESH_LIMIT = 50 _SKIP_DAYS = 7 @@ -181,153 +174,28 @@ def _refresh_uncached_shortlist_hours( hltb_cache.update(refreshed) -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(state.finished_app_ids) | state.active_skipped_ids() - _refresh_uncached_shortlist_hours(games, hltb_cache, skip) - _apply_cached_hours_to_games(games, hltb_cache) - pick_next_game(games, state, config, on_select=_prompt_keep_or_skip) - - 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, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state)) - if skipped is not None: - _echo(f"\n Library hiding: skipped ({skipped})") - elif hidden > 0: - _echo(f"\n Library: hid {hidden} games") - - if not is_game_installed(state.current_app_id): - logger.info( - "Assigned game still missing after library reconciliation; " - "re-triggering install" - ) - _echo( - "\n Assigned game still missing after library reconciliation; " - "re-triggering install..." - ) - _echo(f"\n Auto-installing {state.current_game_name}...") - install_game( - state.current_app_id, - state.current_game_name, - config.steam_id, - use_steam_protocol=True, - ) - - send_notification( - "Game Complete!", - f"Finished {game_name}! Now playing: {state.current_game_name}", - ) - _echo(f"\nAll done! Go play {state.current_game_name}!") +_MOVED_TO_CMD_DONE_FINALIZE = frozenset( + { + "_enforce_on_done", + "_finalize_completion", + "cmd_done", + } +) -def _enforce_on_done(config: Config, state: State) -> None: - """Run a single enforcement pass during the 'done' command. +# Whatever the re-exported name turns out to be -- a function, a class or +# a constant. Aliased so the annotation is a name rather than a bare Any. +_Reexport: TypeAlias = Any - Kills unauthorized game processes, uninstalls unauthorized games, - and ensures the assigned game is installed. + +def __getattr__(name: str) -> _Reexport: + """Re-export the names that moved to :mod:`_cmd_done_finalize`. + + Deferred rather than imported at the top because _cmd_done_finalize imports + back from this module, so a module-level import would be circular. """ - if state.current_app_id is None: - return - - if config.kill_unauthorized_games: - violations = enforce_allowed_game( - allowed_app_ids(state), - 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(allowed_app_ids(state)) - 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, - ) - - # Reconcile library: hide non-assigned games and unhide the assigned one. - # Without this, an interrupted earlier completion can leave the new - # assigned game hidden and stale games visible. - owned_ids = get_all_owned_app_ids(config) - if owned_ids: - hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state)) - if skipped is not None: - _echo(f" Library hiding: skipped ({skipped})") - elif hidden > 0: - _echo(f" Library: hid {hidden} games") - - -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") - _report_assigned_confidence(app_id, state) - - 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) + if name in _MOVED_TO_CMD_DONE_FINALIZE: + module = importlib.import_module("steam_backlog_enforcer._cmd_done_finalize") + return getattr(module, name) + msg = f"module {__name__!r} has no attribute {name!r}" + raise AttributeError(msg) diff --git a/steam_backlog_enforcer/_cmd_done_finalize.py b/steam_backlog_enforcer/_cmd_done_finalize.py new file mode 100644 index 0000000..475308b --- /dev/null +++ b/steam_backlog_enforcer/_cmd_done_finalize.py @@ -0,0 +1,192 @@ +"""Finalising a completed game and re-enforcing afterwards. + +Split out of :mod:`steam_backlog_enforcer._cmd_done` to keep both files +under the 250-line cap. +""" + +from __future__ import annotations + +import logging + +from steam_backlog_enforcer._actions import allowed_app_ids +from steam_backlog_enforcer._cmd_done import ( + _apply_cached_hours_to_games, + _prompt_keep_or_skip, + _refresh_uncached_shortlist_hours, + _report_assigned_confidence, +) +from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids +from steam_backlog_enforcer.config import Config, State, load_snapshot +from steam_backlog_enforcer.enforcer import ( + enforce_allowed_game, + send_notification, +) +from steam_backlog_enforcer.game_install import ( + _echo, + install_game, + is_game_installed, + uninstall_other_games, +) +from steam_backlog_enforcer.hltb import ( + fetch_hltb_times_cached, + load_hltb_cache, +) +from steam_backlog_enforcer.library_hider import try_hide_other_games +from steam_backlog_enforcer.scanning import pick_next_game +from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient + +_REASSIGN_REFRESH_LIMIT = 50 +_SKIP_DAYS = 7 +logger = logging.getLogger(__name__) + + +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(state.finished_app_ids) | state.active_skipped_ids() + _refresh_uncached_shortlist_hours(games, hltb_cache, skip) + _apply_cached_hours_to_games(games, hltb_cache) + pick_next_game(games, state, config, on_select=_prompt_keep_or_skip) + + 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, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state)) + if skipped is not None: + _echo(f"\n Library hiding: skipped ({skipped})") + elif hidden > 0: + _echo(f"\n Library: hid {hidden} games") + + if not is_game_installed(state.current_app_id): + logger.info( + "Assigned game still missing after library reconciliation; " + "re-triggering install" + ) + _echo( + "\n Assigned game still missing after library reconciliation; " + "re-triggering install..." + ) + _echo(f"\n Auto-installing {state.current_game_name}...") + install_game( + state.current_app_id, + state.current_game_name, + config.steam_id, + use_steam_protocol=True, + ) + + 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( + allowed_app_ids(state), + 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(allowed_app_ids(state)) + 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, + ) + + # Reconcile library: hide non-assigned games and unhide the assigned one. + # Without this, an interrupted earlier completion can leave the new + # assigned game hidden and stale games visible. + owned_ids = get_all_owned_app_ids(config) + if owned_ids: + hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state)) + if skipped is not None: + _echo(f" Library hiding: skipped ({skipped})") + elif hidden > 0: + _echo(f" Library: hid {hidden} games") + + +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") + _report_assigned_confidence(app_id, state) + + 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_cached.py b/steam_backlog_enforcer/_hltb_cached.py new file mode 100644 index 0000000..1d1f24c --- /dev/null +++ b/steam_backlog_enforcer/_hltb_cached.py @@ -0,0 +1,95 @@ +"""Cached batch lookup of HowLongToBeat times. + +Split out of :mod:`steam_backlog_enforcer.hltb` to keep both files +under the 250-line cap. +""" + +from __future__ import annotations + +import logging +import time + +from steam_backlog_enforcer._hltb_types import ( + ProgressCb, + _HLTBExtras, + load_hltb_cache, + load_hltb_count_comp_cache, + load_hltb_leisure_100h_cache, + load_hltb_polls_cache, + load_hltb_rush_cache, + save_hltb_cache, +) +from steam_backlog_enforcer.hltb import fetch_hltb_times + +logger = logging.getLogger(__name__) + + +# ────────────────────────────────────────────────────────────── +# Confidence-only batch fetch (no leisure/DLC detail pages) +# ────────────────────────────────────────────────────────────── + + +def fetch_hltb_times_cached( + games: list[tuple[int, str]], + progress_cb: ProgressCb | None = None, +) -> dict[int, float]: + """Fetch HLTB times, using disk cache for already-known games. + + Args: + games: list of (app_id, name) tuples to look up. + progress_cb: optional callback(done, total, found, game_name). + + Returns: dict mapping app_id -> completionist_hours. + """ + cache = load_hltb_cache() + polls = load_hltb_polls_cache() + extras = _HLTBExtras( + count_comp=load_hltb_count_comp_cache(), + rush=load_hltb_rush_cache(), + leisure_100h=load_hltb_leisure_100h_cache(), + ) + uncached = [(app_id, name) for app_id, name in games if app_id not in cache] + + if uncached: + logger.info( + "Fetching HLTB data for %d uncached games (%d cached)...", + len(uncached), + len(games) - len(uncached), + ) + t0 = time.monotonic() + fetch_hltb_times( + uncached, + cache=cache, + polls=polls, + progress_cb=progress_cb, + extras=extras, + ) + elapsed = time.monotonic() - t0 + + # Final save. + save_hltb_cache(cache, polls, extras) + + found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0) + rate = len(uncached) / elapsed if elapsed > 0 else 0 + logger.info( + "HLTB fetch done: %d/%d found in %.1fs (%.0f games/s)", + found, + len(uncached), + elapsed, + rate, + ) + else: + logger.info("All %d games found in HLTB cache.", len(games)) + + return cache + + +_REEXPORTED_TYPES = frozenset({"HLTB_BASE_URL"}) + +_MOVED_TO_HLTB_CONFIDENCE = frozenset( + { + "fetch_hltb_confidence_cached", + "fetch_hltb_detail_missing", + "get_hltb_submit_url", + } +) diff --git a/steam_backlog_enforcer/_hltb_confidence.py b/steam_backlog_enforcer/_hltb_confidence.py new file mode 100644 index 0000000..dd150ad --- /dev/null +++ b/steam_backlog_enforcer/_hltb_confidence.py @@ -0,0 +1,187 @@ +"""Cached HowLongToBeat confidence lookups. + +Split out of :mod:`steam_backlog_enforcer.hltb` to keep both files +under the 250-line cap. +""" + +from __future__ import annotations + +import logging +import time + +from steam_backlog_enforcer._hltb_types import ( + HLTB_BASE_URL, + ProgressCb, + _HLTBExtras, + load_hltb_cache, + load_hltb_count_comp_cache, + load_hltb_game_id_cache, + load_hltb_leisure_100h_cache, + load_hltb_polls_cache, + load_hltb_rush_cache, + save_hltb_cache, +) +from steam_backlog_enforcer.hltb import ( + fetch_hltb_confidence, + fetch_hltb_times, +) + +logger = logging.getLogger(__name__) + + +# ────────────────────────────────────────────────────────────── +# Confidence-only batch fetch (no leisure/DLC detail pages) +# ────────────────────────────────────────────────────────────── + + +def fetch_hltb_confidence_cached( + games: list[tuple[int, str]], + progress_cb: ProgressCb | None = None, +) -> dict[int, float]: + """Fetch HLTB search-level confidence data, using disk cache for known IDs.""" + cache = load_hltb_cache() + polls = load_hltb_polls_cache() + count_comp = load_hltb_count_comp_cache() + uncached = [(app_id, name) for app_id, name in games if app_id not in cache] + + if uncached: + logger.info( + "Fetching HLTB confidence for %d uncached games (%d cached)...", + len(uncached), + len(games) - len(uncached), + ) + t0 = time.monotonic() + fetch_hltb_confidence( + uncached, + cache=cache, + polls=polls, + progress_cb=progress_cb, + count_comp=count_comp, + ) + elapsed = time.monotonic() - t0 + + save_hltb_cache(cache, polls, _HLTBExtras(count_comp=count_comp)) + + found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0) + rate = len(uncached) / elapsed if elapsed > 0 else 0 + logger.info( + "HLTB confidence fetch done: %d/%d found in %.1fs (%.0f games/s)", + found, + len(uncached), + elapsed, + rate, + ) + else: + logger.info("All %d games found in HLTB cache.", len(games)) + + return cache + + +def fetch_hltb_detail_missing( + games: list[tuple[int, str]], + progress_cb: ProgressCb | None = None, +) -> int: + """Fetch HLTB detail (rush + leisure) for games that are missing it. + + Also backfills ``hltb_game_id`` for any game that already has rush/leisure + data but whose HLTB game ID was never stored (e.g. from an old cache). + Games with both rush data and a game_id are skipped entirely. For the + rest, temporarily removes them from the hours cache so ``fetch_hltb_times`` + will visit their detail pages. Restores prior hours for any game the + re-fetch doesn't find. + + Args: + games: list of (app_id, name) tuples to check. + progress_cb: optional progress callback. + + Returns: + Number of games that now have rush-hour data after the fetch. + """ + rush = load_hltb_rush_cache() + game_id_cache = load_hltb_game_id_cache() + missing_rush = [ + (app_id, name) for app_id, name in games if rush.get(app_id, -1) <= 0 + ] + # Also re-search games that have rush data but no HLTB game ID yet so the + # direct URL can be shown in stats output. + missing_id_only = [ + (app_id, name) + for app_id, name in games + if rush.get(app_id, -1) > 0 and game_id_cache.get(app_id, 0) == 0 + ] + missing = missing_rush + missing_id_only + if not missing: + return 0 + + cache = load_hltb_cache() + polls = load_hltb_polls_cache() + extras = _HLTBExtras( + count_comp=load_hltb_count_comp_cache(), + rush=rush, + leisure_100h=load_hltb_leisure_100h_cache(), + hltb_game_id=game_id_cache, + ) + + # Remove from hours cache so fetch_hltb_times will visit the detail page. + prior_hours: dict[int, float] = {} + for app_id, _ in missing: + prior_hours[app_id] = cache.pop(app_id, -1.0) + + n_rush = len(missing_rush) + n_id = len(missing_id_only) + if n_rush and n_id: + logger.info( + "Fetching HLTB detail for %d games missing rush/leisure data" + " + %d games missing game ID...", + n_rush, + n_id, + ) + elif n_rush: + logger.info( + "Fetching HLTB detail for %d games missing rush/leisure data...", n_rush + ) + else: + logger.info("Backfilling HLTB game ID for %d game(s)...", n_id) + t0 = time.monotonic() + fetch_hltb_times( + missing, + cache=cache, + polls=polls, + progress_cb=progress_cb, + extras=extras, + ) + elapsed = time.monotonic() - t0 + + # Restore prior hours for games the detail fetch didn't re-find. + for app_id, old_hours in prior_hours.items(): + if old_hours > 0 and cache.get(app_id, -1.0) <= 0: + cache[app_id] = old_hours + + save_hltb_cache(cache, polls, extras) + + fetched = sum(1 for app_id, _ in missing_rush if extras.rush.get(app_id, -1) > 0) + rate = len(missing) / elapsed if elapsed > 0 else 0 + logger.info( + "HLTB detail fetch done: %d/%d got rush data in %.1fs (%.0f games/s)", + fetched, + len(missing_rush), + elapsed, + rate, + ) + return fetched + + +def get_hltb_submit_url(game_name: str) -> str | None: + """Look up a game on HLTB and return its submit page URL. + + Args: + game_name: Name of the game to search for. + + Returns: + URL like ``https://howlongtobeat.com/submit/game/12345``, + or ``None`` if the game wasn't found. + """ + results = fetch_hltb_times([(0, game_name)]) + if results and results[0].hltb_game_id: + return f"{HLTB_BASE_URL}/submit/game/{results[0].hltb_game_id}" + return None diff --git a/steam_backlog_enforcer/_manual_pick_lifecycle.py b/steam_backlog_enforcer/_manual_pick_lifecycle.py new file mode 100644 index 0000000..8d89bf6 --- /dev/null +++ b/steam_backlog_enforcer/_manual_pick_lifecycle.py @@ -0,0 +1,166 @@ +"""Grace periods, abandonment and the status payload for manual picks. + +Split out of :mod:`steam_backlog_enforcer._actions` to keep both files +under the 250-line cap. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Any + +from steam_backlog_enforcer._actions import ( + find_manual_pick, + is_manual_pick_locked, +) +from steam_backlog_enforcer._allowed_games import ( + active_manual_picks, + allowed_games, +) +from steam_backlog_enforcer._total_block import get_total_block_status +from steam_backlog_enforcer.game_install import get_installed_games, is_protected_app +from steam_backlog_enforcer.store_blocker import is_store_blocked + +# Marks these two as an intentional re-export (they're imported from +# _allowed_games above, not defined here) -- without this, mypy's +# --no-implicit-reexport flags every downstream `from _actions import +# active_manual_picks` as an error. Everything else in this file is +# accessible normally; this only needs to list names imported-then-reexported. +__all__ = [ + "active_manual_picks", + "allowed_games", +] + +if TYPE_CHECKING: + from steam_backlog_enforcer.config import State + +# Mistake-correction window: for this many days after a manual pick the user +# may still back out via ``abandon-pick``. Deliberately short — it exists to +# undo a wrong pick, not to serve as an escape hatch from the lock. +MANUAL_GRACE_DAYS = 4 + +# How long an abandoned pick stays out of the auto-assignment pool, so that +# ``scan`` does not immediately hand back the game the user just rejected. +ABANDON_COOLDOWN_DAYS = 30 + + +def manual_pick_grace_remaining(state: State, app_id: int) -> float | None: + """Return days left in *app_id*'s grace window, or ``None``. + + ``None`` means "no grace window applies": *app_id* is not an active manual + pick, or its timestamp is missing/malformed. A value <= 0 means the window + has closed. + + Args: + state: The loaded enforcer state. + app_id: The manually-picked app id to measure. + + Returns: + Fractional days remaining before the pick can no longer be abandoned, + or ``None`` when the question does not apply. + """ + pick = find_manual_pick(state, app_id) + if pick is None or not pick.get("started_at"): + return None + try: + started = datetime.fromisoformat(pick["started_at"]) + except ValueError: + return None + deadline = started + timedelta(days=MANUAL_GRACE_DAYS) + return (deadline - datetime.now(timezone.utc)).total_seconds() / 86400 + + +def can_abandon_manual_pick(state: State, app_id: int) -> bool: + """Return ``True`` if *app_id* is still inside its grace window. + + Args: + state: The loaded enforcer state. + app_id: The manually-picked app id to check. + + Returns: + Whether ``abandon_manual_pick`` would be accepted right now. + """ + remaining = manual_pick_grace_remaining(state, app_id) + return remaining is not None and remaining > 0 + + +def abandon_manual_pick(state: State, app_id: int) -> bool: + """Drop one manual pick and persist ``state``, if still inside its grace window. + + Only the named pick is dropped: any other active pick keeps its own lock + and deadline. The abandoned app id goes onto the existing skip cooldown so + ``scan`` will not hand the same game straight back. State-only: like + ``apply_manual_pick`` this performs no uninstall/hide cascade, so the MCP + server can call it without touching the filesystem. + + Args: + state: The enforcer state to mutate and save. + app_id: The manually-picked app id to back out of. + + Returns: + ``True`` if the pick was abandoned, ``False`` if its grace window has + closed (or it is not an active pick), in which case ``state`` is + untouched. + """ + if not can_abandon_manual_pick(state, app_id): + return False + + state.skip_for_days(app_id, ABANDON_COOLDOWN_DAYS) + state.manual_picks = [ + p for p in active_manual_picks(state) if p.get("app_id") != app_id + ] + + # The abandoned pick may also have been the current assignment; hand the + # assignment to a surviving pick so the enforcer keeps guarding it, or + # clear it so 'scan' can reassign. + if state.current_app_id == app_id: + survivor = state.manual_picks[-1] if state.manual_picks else None + state.current_app_id = survivor["app_id"] if survivor else None + state.current_game_name = survivor["game_name"] if survivor else "" + + state.save() + return True + + +def status_payload(state: State) -> dict[str, Any]: + """Build the structured status snapshot that ``cmd_status`` renders as text. + + Pure data, no stdout — safe for the MCP server. Reads only ``State`` and the + stdout-free leaf helpers; never constructs ``Config`` and never exposes the + Steam API key. + + Args: + state: The loaded enforcer state. + + Returns: + A JSON-ready dict describing the current enforcement status. + """ + total_block = get_total_block_status() + installed = get_installed_games() + real_games = [(aid, name) for aid, name in installed if not is_protected_app(aid)] + assigned_installed = ( + any(aid == state.current_app_id for aid, _ in installed) + if state.current_app_id + else None + ) + return { + "current_app_id": state.current_app_id, + "current_game_name": state.current_game_name or None, + "finished_count": len(state.finished_app_ids), + "store_blocked": is_store_blocked(), + "installed_count": len(real_games), + "assigned_game_installed": assigned_installed, + "total_block": { + "active": total_block.active, + "days_remaining": round(total_block.days_remaining, 1), + "until": total_block.until.isoformat() if total_block.until else None, + }, + "manual_pick_locked": is_manual_pick_locked(state), + "manual_picks": [ + { + **pick, + "grace_days_left": manual_pick_grace_remaining(state, pick["app_id"]), + } + for pick in active_manual_picks(state) + ], + } diff --git a/steam_backlog_enforcer/_whitelist.py b/steam_backlog_enforcer/_whitelist.py index b27f132..7aed11d 100644 --- a/steam_backlog_enforcer/_whitelist.py +++ b/steam_backlog_enforcer/_whitelist.py @@ -2,20 +2,14 @@ from __future__ import annotations -import array from collections import Counter -import contextlib -import fcntl -import json +import importlib import logging import math import re -import shutil -import subprocess -import time -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Any, TypeAlias -from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write +from steam_backlog_enforcer.config import CONFIG_DIR if TYPE_CHECKING: from pathlib import Path @@ -119,180 +113,34 @@ def validate_reason(reason: str) -> str | None: _FS_IOC_GETFLAGS = 0x80086601 _FS_IMMUTABLE_FL = 0x00000010 +_MOVED_TO_WHITELIST_LOCKING = frozenset( + { + "_append_audit_log", + "_immutable_flag_is", + "_load_approved", + "_save_approved", + "_try_set_immutable", + "add_pending_exception", + "get_approved_exception_ids", + "lock_enforcement_files", + "unlock_for_write", + } +) -def _immutable_flag_is(path: Path, *, immutable: bool) -> bool: - """Report whether *path*'s immutable bit already equals *immutable*. - Reading the flag is an ioctl, which costs no process; ``chattr`` costs a - fork and an exec. Since ``lock_enforcement_files`` runs on every - enforce-loop iteration, asking first turns a steady ~0.65 execs/second - into none at all for a flag that essentially never changes. +# Whatever the re-exported name turns out to be -- a function, a class or +# a constant. Aliased so the annotation is a name rather than a bare Any. +_Reexport: TypeAlias = Any - Fails closed: any uncertainty at all — filesystem without flag support, - ioctl refused, file unreadable — reports False so the caller still shells - out to chattr. A wrong True would silently stop enforcing immutability. - Args: - path: File to inspect. - immutable: The state to compare against. +def __getattr__(name: str) -> _Reexport: + """Re-export the names that moved to :mod:`_whitelist_locking`. - Returns: - True only if the flag could be read and already matches. + Deferred rather than imported at the top because _whitelist_locking imports + back from this module, so a module-level import would be circular. """ - buf = array.array("L", [0]) - try: - with path.open("rb") as handle: - # mutate_flag defaults to True, so buf receives the flags in place. - fcntl.ioctl(handle.fileno(), _FS_IOC_GETFLAGS, buf) - except (OSError, ValueError): - return False - return bool(buf[0] & _FS_IMMUTABLE_FL) is immutable - - -def _try_set_immutable(path: Path, *, immutable: bool) -> None: - """Silently attempt to set or clear the immutable flag on *path*. - - This is a best-effort operation — it fails silently if chattr is not - available, the process lacks the required capability, or the filesystem - does not support the flag. When the flag already holds the requested - value, no subprocess is spawned at all. - - Args: - path: File to modify. - immutable: True to set +i, False to clear -i. - """ - if not path.exists(): - return - if _immutable_flag_is(path, immutable=immutable): - return - chattr = shutil.which("chattr") - if chattr is None: - return - flag = "+i" if immutable else "-i" - with contextlib.suppress(OSError, subprocess.TimeoutExpired): - subprocess.run( - [chattr, flag, str(path)], - capture_output=True, - check=False, - timeout=5, - ) - - -def lock_enforcement_files(config_file: Path) -> None: - """Apply chattr +i to enforcement-critical config files. - - Called at the end of each enforce-loop iteration. Requires that the - daemon is running as root (or has CAP_LINUX_IMMUTABLE). - - Args: - config_file: Path to the main config.json. - """ - _try_set_immutable(config_file, immutable=True) - _try_set_immutable(APPROVED_EXCEPTIONS_FILE, immutable=True) - - -def unlock_for_write(path: Path) -> None: - """Clear the immutable flag before writing *path*. - - Args: - path: File to unlock. - """ - _try_set_immutable(path, immutable=False) - - -# ────────────────────────────────────────────────────────────── -# Persistence helpers -# ────────────────────────────────────────────────────────────── - - -def _load_approved() -> list[dict[str, object]]: - """Load approved exception entries from disk.""" - if not APPROVED_EXCEPTIONS_FILE.exists(): - return [] - try: - data: object = json.loads(APPROVED_EXCEPTIONS_FILE.read_text(encoding="utf-8")) - if isinstance(data, list): - return cast("list[dict[str, object]]", data) - except (json.JSONDecodeError, OSError, ValueError): - pass - return [] - - -def _save_approved(entries: list[dict[str, object]]) -> None: - """Persist approved exception entries to disk.""" - unlock_for_write(APPROVED_EXCEPTIONS_FILE) - _atomic_write(APPROVED_EXCEPTIONS_FILE, json.dumps(entries, indent=2) + "\n") - _try_set_immutable(APPROVED_EXCEPTIONS_FILE, immutable=True) - - -def _append_audit_log(app_id: int, reason: str, event: str) -> None: - """Append one line to the append-only audit log. - - Each line has the format:: - - ISO-TIMESTAMP | EVENT | app_id=NNN | reason='...' - - Args: - app_id: Steam application ID involved. - reason: Justification text supplied by the user. - event: Short event label such as ``REQUESTED`` or ``APPROVED``. - """ - timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - line = f"{timestamp} | {event} | app_id={app_id} | reason={reason!r}\n" - EXCEPTION_AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True) - with EXCEPTION_AUDIT_LOG.open("a", encoding="utf-8") as fh: - fh.write(line) - - -# ────────────────────────────────────────────────────────────── -# Public API -# ────────────────────────────────────────────────────────────── - - -def add_pending_exception(app_id: int, reason: str) -> str: - """Add a whitelist exception for *app_id* immediately. - - The entry becomes active right away (no cooldown). Returns a - human-readable status message. - - Args: - app_id: Steam application ID to add. - reason: Validated justification text (must pass :func:`validate_reason`). - - Returns: - Human-readable confirmation message. - - Raises: - ValueError: If the reason fails validation or the ID is already approved. - """ - err = validate_reason(reason) - if err is not None: - raise ValueError(err) - - approved = _load_approved() - if any(int(e["app_id"]) == app_id for e in approved): - msg = f"AppID {app_id} is already in the approved exceptions list." - raise ValueError(msg) - - now = time.time() - approved.append( - { - "app_id": app_id, - "reason": reason, - "approved_at": now, - } - ) - _save_approved(approved) - _append_audit_log(app_id, reason, "APPROVED") - - return f"Exception approved for AppID {app_id}. Active immediately. Reason logged." - - -def get_approved_exception_ids() -> frozenset[int]: - """Return the frozenset of currently approved exception app IDs. - - Returns: - Frozenset of approved app IDs. - """ - approved = _load_approved() - return frozenset(int(e["app_id"]) for e in approved) + if name in _MOVED_TO_WHITELIST_LOCKING: + module = importlib.import_module("steam_backlog_enforcer._whitelist_locking") + return getattr(module, name) + msg = f"module {__name__!r} has no attribute {name!r}" + raise AttributeError(msg) diff --git a/steam_backlog_enforcer/_whitelist_locking.py b/steam_backlog_enforcer/_whitelist_locking.py new file mode 100644 index 0000000..e5848d6 --- /dev/null +++ b/steam_backlog_enforcer/_whitelist_locking.py @@ -0,0 +1,229 @@ +"""Making enforcement files immutable so they cannot be edited mid-block. + +Split out of :mod:`steam_backlog_enforcer._whitelist` to keep both files +under the 250-line cap. +""" + +from __future__ import annotations + +import array +import contextlib +import fcntl +import json +import logging +import shutil +import subprocess +import time +from typing import TYPE_CHECKING, cast + +from steam_backlog_enforcer._whitelist import ( + _FS_IMMUTABLE_FL, + _FS_IOC_GETFLAGS, + validate_reason, +) +from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write + +if TYPE_CHECKING: + from pathlib import Path + +logger = logging.getLogger(__name__) + +# ────────────────────────────────────────────────────────────── +# File paths (patched in tests via conftest) +# ────────────────────────────────────────────────────────────── + +APPROVED_EXCEPTIONS_FILE: Path = CONFIG_DIR / "approved_exceptions.json" +EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log" + +# ────────────────────────────────────────────────────────────── +# Constants +# ────────────────────────────────────────────────────────────── + +_MIN_REASON_WORDS: int = 5 +_MIN_REASON_LENGTH: int = 25 +_MIN_ENTROPY: float = 3.0 +# Reject runs of the same character longer than this (e.g. "aaaa"). +_MAX_CHAR_RUN: int = 3 + + +# ────────────────────────────────────────────────────────────── +# Reason validation +# ────────────────────────────────────────────────────────────── + + +def _immutable_flag_is(path: Path, *, immutable: bool) -> bool: + """Report whether *path*'s immutable bit already equals *immutable*. + + Reading the flag is an ioctl, which costs no process; ``chattr`` costs a + fork and an exec. Since ``lock_enforcement_files`` runs on every + enforce-loop iteration, asking first turns a steady ~0.65 execs/second + into none at all for a flag that essentially never changes. + + Fails closed: any uncertainty at all — filesystem without flag support, + ioctl refused, file unreadable — reports False so the caller still shells + out to chattr. A wrong True would silently stop enforcing immutability. + + Args: + path: File to inspect. + immutable: The state to compare against. + + Returns: + True only if the flag could be read and already matches. + """ + buf = array.array("L", [0]) + try: + with path.open("rb") as handle: + # mutate_flag defaults to True, so buf receives the flags in place. + fcntl.ioctl(handle.fileno(), _FS_IOC_GETFLAGS, buf) + except (OSError, ValueError): + return False + return bool(buf[0] & _FS_IMMUTABLE_FL) is immutable + + +def _try_set_immutable(path: Path, *, immutable: bool) -> None: + """Silently attempt to set or clear the immutable flag on *path*. + + This is a best-effort operation — it fails silently if chattr is not + available, the process lacks the required capability, or the filesystem + does not support the flag. When the flag already holds the requested + value, no subprocess is spawned at all. + + Args: + path: File to modify. + immutable: True to set +i, False to clear -i. + """ + if not path.exists(): + return + if _immutable_flag_is(path, immutable=immutable): + return + chattr = shutil.which("chattr") + if chattr is None: + return + flag = "+i" if immutable else "-i" + with contextlib.suppress(OSError, subprocess.TimeoutExpired): + subprocess.run( + [chattr, flag, str(path)], + capture_output=True, + check=False, + timeout=5, + ) + + +def lock_enforcement_files(config_file: Path) -> None: + """Apply chattr +i to enforcement-critical config files. + + Called at the end of each enforce-loop iteration. Requires that the + daemon is running as root (or has CAP_LINUX_IMMUTABLE). + + Args: + config_file: Path to the main config.json. + """ + _try_set_immutable(config_file, immutable=True) + _try_set_immutable(APPROVED_EXCEPTIONS_FILE, immutable=True) + + +def unlock_for_write(path: Path) -> None: + """Clear the immutable flag before writing *path*. + + Args: + path: File to unlock. + """ + _try_set_immutable(path, immutable=False) + + +# ────────────────────────────────────────────────────────────── +# Persistence helpers +# ────────────────────────────────────────────────────────────── + + +def _load_approved() -> list[dict[str, object]]: + """Load approved exception entries from disk.""" + if not APPROVED_EXCEPTIONS_FILE.exists(): + return [] + try: + data: object = json.loads(APPROVED_EXCEPTIONS_FILE.read_text(encoding="utf-8")) + if isinstance(data, list): + return cast("list[dict[str, object]]", data) + except (json.JSONDecodeError, OSError, ValueError): + pass + return [] + + +def _save_approved(entries: list[dict[str, object]]) -> None: + """Persist approved exception entries to disk.""" + unlock_for_write(APPROVED_EXCEPTIONS_FILE) + _atomic_write(APPROVED_EXCEPTIONS_FILE, json.dumps(entries, indent=2) + "\n") + _try_set_immutable(APPROVED_EXCEPTIONS_FILE, immutable=True) + + +def _append_audit_log(app_id: int, reason: str, event: str) -> None: + """Append one line to the append-only audit log. + + Each line has the format:: + + ISO-TIMESTAMP | EVENT | app_id=NNN | reason='...' + + Args: + app_id: Steam application ID involved. + reason: Justification text supplied by the user. + event: Short event label such as ``REQUESTED`` or ``APPROVED``. + """ + timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + line = f"{timestamp} | {event} | app_id={app_id} | reason={reason!r}\n" + EXCEPTION_AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True) + with EXCEPTION_AUDIT_LOG.open("a", encoding="utf-8") as fh: + fh.write(line) + + +# ────────────────────────────────────────────────────────────── +# Public API +# ────────────────────────────────────────────────────────────── + + +def add_pending_exception(app_id: int, reason: str) -> str: + """Add a whitelist exception for *app_id* immediately. + + The entry becomes active right away (no cooldown). Returns a + human-readable status message. + + Args: + app_id: Steam application ID to add. + reason: Validated justification text (must pass :func:`validate_reason`). + + Returns: + Human-readable confirmation message. + + Raises: + ValueError: If the reason fails validation or the ID is already approved. + """ + err = validate_reason(reason) + if err is not None: + raise ValueError(err) + + approved = _load_approved() + if any(int(e["app_id"]) == app_id for e in approved): + msg = f"AppID {app_id} is already in the approved exceptions list." + raise ValueError(msg) + + now = time.time() + approved.append( + { + "app_id": app_id, + "reason": reason, + "approved_at": now, + } + ) + _save_approved(approved) + _append_audit_log(app_id, reason, "APPROVED") + + return f"Exception approved for AppID {app_id}. Active immediately. Reason logged." + + +def get_approved_exception_ids() -> frozenset[int]: + """Return the frozenset of currently approved exception app IDs. + + Returns: + Frozenset of approved app IDs. + """ + approved = _load_approved() + return frozenset(int(e["app_id"]) for e in approved) diff --git a/steam_backlog_enforcer/hltb.py b/steam_backlog_enforcer/hltb.py index 98efd61..abe0f36 100644 --- a/steam_backlog_enforcer/hltb.py +++ b/steam_backlog_enforcer/hltb.py @@ -13,8 +13,9 @@ Fetches leisure completionist hour estimates from howlongtobeat.com with: from __future__ import annotations import asyncio +import importlib import logging -import time +from typing import Any, TypeAlias import aiohttp @@ -26,18 +27,10 @@ from steam_backlog_enforcer._hltb_search import ( _SearchCtx, ) from steam_backlog_enforcer._hltb_types import ( - HLTB_BASE_URL, MAX_CONCURRENT, HLTBResult, ProgressCb, _HLTBExtras, - load_hltb_cache, - load_hltb_count_comp_cache, - load_hltb_game_id_cache, - load_hltb_leisure_100h_cache, - load_hltb_polls_cache, - load_hltb_rush_cache, - save_hltb_cache, ) logger = logging.getLogger(__name__) @@ -176,209 +169,39 @@ def fetch_hltb_confidence( ) -def fetch_hltb_times_cached( - games: list[tuple[int, str]], - progress_cb: ProgressCb | None = None, -) -> dict[int, float]: - """Fetch HLTB times, using disk cache for already-known games. +_REEXPORTED = { + "HLTB_BASE_URL": "steam_backlog_enforcer._hltb_types", + "MAX_CONCURRENT": "steam_backlog_enforcer._hltb_types", + "HLTBResult": "steam_backlog_enforcer._hltb_types", + "ProgressCb": "steam_backlog_enforcer._hltb_types", + "_HLTBExtras": "steam_backlog_enforcer._hltb_types", + "load_hltb_cache": "steam_backlog_enforcer._hltb_types", + "load_hltb_count_comp_cache": "steam_backlog_enforcer._hltb_types", + "load_hltb_game_id_cache": "steam_backlog_enforcer._hltb_types", + "load_hltb_leisure_100h_cache": "steam_backlog_enforcer._hltb_types", + "load_hltb_polls_cache": "steam_backlog_enforcer._hltb_types", + "load_hltb_rush_cache": "steam_backlog_enforcer._hltb_types", + "save_hltb_cache": "steam_backlog_enforcer._hltb_types", + "fetch_hltb_times_cached": "steam_backlog_enforcer._hltb_cached", + "fetch_hltb_confidence_cached": "steam_backlog_enforcer._hltb_confidence", + "fetch_hltb_detail_missing": "steam_backlog_enforcer._hltb_confidence", + "get_hltb_submit_url": "steam_backlog_enforcer._hltb_confidence", +} - Args: - games: list of (app_id, name) tuples to look up. - progress_cb: optional callback(done, total, found, game_name). - Returns: dict mapping app_id -> completionist_hours. +# Whatever the re-exported name turns out to be -- a function, a class or +# a constant. Aliased so the annotation is a name rather than a bare Any. +_Reexport: TypeAlias = Any + + +def __getattr__(name: str) -> _Reexport: + """Serve names that moved out of this module when it was split. + + Deferred via importlib rather than imported at the top: the modules below + import back from here, so a module-level import would be circular. """ - cache = load_hltb_cache() - polls = load_hltb_polls_cache() - extras = _HLTBExtras( - count_comp=load_hltb_count_comp_cache(), - rush=load_hltb_rush_cache(), - leisure_100h=load_hltb_leisure_100h_cache(), - ) - uncached = [(app_id, name) for app_id, name in games if app_id not in cache] - - if uncached: - logger.info( - "Fetching HLTB data for %d uncached games (%d cached)...", - len(uncached), - len(games) - len(uncached), - ) - t0 = time.monotonic() - fetch_hltb_times( - uncached, - cache=cache, - polls=polls, - progress_cb=progress_cb, - extras=extras, - ) - elapsed = time.monotonic() - t0 - - # Final save. - save_hltb_cache(cache, polls, extras) - - found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0) - rate = len(uncached) / elapsed if elapsed > 0 else 0 - logger.info( - "HLTB fetch done: %d/%d found in %.1fs (%.0f games/s)", - found, - len(uncached), - elapsed, - rate, - ) - else: - logger.info("All %d games found in HLTB cache.", len(games)) - - return cache - - -def fetch_hltb_confidence_cached( - games: list[tuple[int, str]], - progress_cb: ProgressCb | None = None, -) -> dict[int, float]: - """Fetch HLTB search-level confidence data, using disk cache for known IDs.""" - cache = load_hltb_cache() - polls = load_hltb_polls_cache() - count_comp = load_hltb_count_comp_cache() - uncached = [(app_id, name) for app_id, name in games if app_id not in cache] - - if uncached: - logger.info( - "Fetching HLTB confidence for %d uncached games (%d cached)...", - len(uncached), - len(games) - len(uncached), - ) - t0 = time.monotonic() - fetch_hltb_confidence( - uncached, - cache=cache, - polls=polls, - progress_cb=progress_cb, - count_comp=count_comp, - ) - elapsed = time.monotonic() - t0 - - save_hltb_cache(cache, polls, _HLTBExtras(count_comp=count_comp)) - - found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0) - rate = len(uncached) / elapsed if elapsed > 0 else 0 - logger.info( - "HLTB confidence fetch done: %d/%d found in %.1fs (%.0f games/s)", - found, - len(uncached), - elapsed, - rate, - ) - else: - logger.info("All %d games found in HLTB cache.", len(games)) - - return cache - - -def fetch_hltb_detail_missing( - games: list[tuple[int, str]], - progress_cb: ProgressCb | None = None, -) -> int: - """Fetch HLTB detail (rush + leisure) for games that are missing it. - - Also backfills ``hltb_game_id`` for any game that already has rush/leisure - data but whose HLTB game ID was never stored (e.g. from an old cache). - Games with both rush data and a game_id are skipped entirely. For the - rest, temporarily removes them from the hours cache so ``fetch_hltb_times`` - will visit their detail pages. Restores prior hours for any game the - re-fetch doesn't find. - - Args: - games: list of (app_id, name) tuples to check. - progress_cb: optional progress callback. - - Returns: - Number of games that now have rush-hour data after the fetch. - """ - rush = load_hltb_rush_cache() - game_id_cache = load_hltb_game_id_cache() - missing_rush = [ - (app_id, name) for app_id, name in games if rush.get(app_id, -1) <= 0 - ] - # Also re-search games that have rush data but no HLTB game ID yet so the - # direct URL can be shown in stats output. - missing_id_only = [ - (app_id, name) - for app_id, name in games - if rush.get(app_id, -1) > 0 and game_id_cache.get(app_id, 0) == 0 - ] - missing = missing_rush + missing_id_only - if not missing: - return 0 - - cache = load_hltb_cache() - polls = load_hltb_polls_cache() - extras = _HLTBExtras( - count_comp=load_hltb_count_comp_cache(), - rush=rush, - leisure_100h=load_hltb_leisure_100h_cache(), - hltb_game_id=game_id_cache, - ) - - # Remove from hours cache so fetch_hltb_times will visit the detail page. - prior_hours: dict[int, float] = {} - for app_id, _ in missing: - prior_hours[app_id] = cache.pop(app_id, -1.0) - - n_rush = len(missing_rush) - n_id = len(missing_id_only) - if n_rush and n_id: - logger.info( - "Fetching HLTB detail for %d games missing rush/leisure data" - " + %d games missing game ID...", - n_rush, - n_id, - ) - elif n_rush: - logger.info( - "Fetching HLTB detail for %d games missing rush/leisure data...", n_rush - ) - else: - logger.info("Backfilling HLTB game ID for %d game(s)...", n_id) - t0 = time.monotonic() - fetch_hltb_times( - missing, - cache=cache, - polls=polls, - progress_cb=progress_cb, - extras=extras, - ) - elapsed = time.monotonic() - t0 - - # Restore prior hours for games the detail fetch didn't re-find. - for app_id, old_hours in prior_hours.items(): - if old_hours > 0 and cache.get(app_id, -1.0) <= 0: - cache[app_id] = old_hours - - save_hltb_cache(cache, polls, extras) - - fetched = sum(1 for app_id, _ in missing_rush if extras.rush.get(app_id, -1) > 0) - rate = len(missing) / elapsed if elapsed > 0 else 0 - logger.info( - "HLTB detail fetch done: %d/%d got rush data in %.1fs (%.0f games/s)", - fetched, - len(missing_rush), - elapsed, - rate, - ) - return fetched - - -def get_hltb_submit_url(game_name: str) -> str | None: - """Look up a game on HLTB and return its submit page URL. - - Args: - game_name: Name of the game to search for. - - Returns: - URL like ``https://howlongtobeat.com/submit/game/12345``, - or ``None`` if the game wasn't found. - """ - results = fetch_hltb_times([(0, game_name)]) - if results and results[0].hltb_game_id: - return f"{HLTB_BASE_URL}/submit/game/{results[0].hltb_game_id}" - return None + home = _REEXPORTED.get(name) + if home is not None: + return getattr(importlib.import_module(home), name) + msg = f"module {__name__!r} has no attribute {name!r}" + raise AttributeError(msg) diff --git a/steam_backlog_enforcer/tests/conftest.py b/steam_backlog_enforcer/tests/conftest.py index 9106585..a769513 100644 --- a/steam_backlog_enforcer/tests/conftest.py +++ b/steam_backlog_enforcer/tests/conftest.py @@ -135,13 +135,21 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]: tmp_path / "fake_home" / ".config" / "CSDSteamBuild", ), ), - # Whitelist exception files (_whitelist module-level constants) + # Derived from CONFIG_DIR at import time, so it captures the REAL path + # before this fixture redirects CONFIG_DIR. Without it the suite + # rewrites the user's owned-games cache. patch( - "steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE", + "steam_backlog_enforcer._owned_apps_cache._OWNED_IDS_CACHE_FILE", + fake_config / "owned_app_ids_cache.json", + ), + # Whitelist exception files (module-level constants; these live in + # _whitelist_locking, which is where the code that reads them is) + patch( + "steam_backlog_enforcer._whitelist_locking.APPROVED_EXCEPTIONS_FILE", fake_config / "approved_exceptions.json", ), patch( - "steam_backlog_enforcer._whitelist.EXCEPTION_AUDIT_LOG", + "steam_backlog_enforcer._whitelist_locking.EXCEPTION_AUDIT_LOG", fake_config / "exception_audit.log", ), # _enforce_loop imports CONFIG_FILE directly; patch the local binding so diff --git a/steam_backlog_enforcer/tests/test_actions_status_payload.py b/steam_backlog_enforcer/tests/test_actions_status_payload.py index a0db66f..919fe0a 100644 --- a/steam_backlog_enforcer/tests/test_actions_status_payload.py +++ b/steam_backlog_enforcer/tests/test_actions_status_payload.py @@ -37,7 +37,7 @@ class TestStatusPayload: ) -> object: """Test patch leaves.""" return patch.multiple( - "steam_backlog_enforcer._actions", + "steam_backlog_enforcer._manual_pick_lifecycle", get_total_block_status=lambda: total_block, get_installed_games=lambda: installed, is_store_blocked=lambda: store_blocked, diff --git a/steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py b/steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py index e004e83..e40d1b1 100644 --- a/steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py +++ b/steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py @@ -15,6 +15,7 @@ from steam_backlog_enforcer.hltb import ( ) PKG = "steam_backlog_enforcer.hltb" +_CONF = "steam_backlog_enforcer._hltb_confidence" class _DummySession: @@ -116,11 +117,11 @@ class TestConfidenceHelpers: def test_fetch_hltb_confidence_cached_all_cached_skips_fetch(self) -> None: """Test fetch hltb confidence cached all cached skips fetch.""" with ( - patch(f"{PKG}.load_hltb_cache", return_value={1: 12.0}), - patch(f"{PKG}.load_hltb_polls_cache", return_value={1: 30}), - patch(f"{PKG}.load_hltb_count_comp_cache", return_value={1: 200}), - patch(f"{PKG}.fetch_hltb_confidence") as mock_fetch, - patch(f"{PKG}.save_hltb_cache") as mock_save, + patch(f"{_CONF}.load_hltb_cache", return_value={1: 12.0}), + patch(f"{_CONF}.load_hltb_polls_cache", return_value={1: 30}), + patch(f"{_CONF}.load_hltb_count_comp_cache", return_value={1: 200}), + patch(f"{_CONF}.fetch_hltb_confidence") as mock_fetch, + patch(f"{_CONF}.save_hltb_cache") as mock_save, ): result = fetch_hltb_confidence_cached([(1, "Game")]) assert result == {1: 12.0} diff --git a/steam_backlog_enforcer/tests/test_hltb_fetch_hltb_detail_missing.py b/steam_backlog_enforcer/tests/test_hltb_fetch_hltb_detail_missing.py index b336127..feda6f7 100644 --- a/steam_backlog_enforcer/tests/test_hltb_fetch_hltb_detail_missing.py +++ b/steam_backlog_enforcer/tests/test_hltb_fetch_hltb_detail_missing.py @@ -14,7 +14,7 @@ from steam_backlog_enforcer.hltb import ( if TYPE_CHECKING: from steam_backlog_enforcer._hltb_types import _HLTBExtras -PKG = "steam_backlog_enforcer.hltb" +PKG = "steam_backlog_enforcer._hltb_confidence" class _DummySession: diff --git a/steam_backlog_enforcer/tests/test_hltb_part2.py b/steam_backlog_enforcer/tests/test_hltb_part2.py index 953156b..c1d5db0 100644 --- a/steam_backlog_enforcer/tests/test_hltb_part2.py +++ b/steam_backlog_enforcer/tests/test_hltb_part2.py @@ -18,6 +18,8 @@ if TYPE_CHECKING: from steam_backlog_enforcer._hltb_types import _HLTBExtras PKG = "steam_backlog_enforcer.hltb" +_CACHED = "steam_backlog_enforcer._hltb_cached" +_CONF = "steam_backlog_enforcer._hltb_confidence" class _DummySession: @@ -38,7 +40,7 @@ class TestFetchHltbTimesCached: def test_all_cached(self) -> None: """Test all cached.""" with ( - patch(f"{PKG}.load_hltb_cache", return_value={440: 50.0}), + patch(f"{_CACHED}.load_hltb_cache", return_value={440: 50.0}), ): result = fetch_hltb_times_cached([(440, "TF2")]) assert result == {440: 50.0} @@ -46,10 +48,10 @@ class TestFetchHltbTimesCached: def test_uncached_games_fetched(self) -> None: """Test uncached games fetched.""" with ( - patch(f"{PKG}.load_hltb_cache", return_value={440: 50.0}), - patch(f"{PKG}.fetch_hltb_times") as mock_fetch, - patch(f"{PKG}.save_hltb_cache") as mock_save, - patch(f"{PKG}.time.monotonic", side_effect=[0.0, 2.0]), + patch(f"{_CACHED}.load_hltb_cache", return_value={440: 50.0}), + patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch, + patch(f"{_CACHED}.save_hltb_cache") as mock_save, + patch(f"{_CACHED}.time.monotonic", side_effect=[0.0, 2.0]), ): # fetch_hltb_times modifies cache in-place def add_to_cache( @@ -80,10 +82,10 @@ class TestFetchHltbTimesCached: """Test uncached with progress cb.""" cb = MagicMock() with ( - patch(f"{PKG}.load_hltb_cache", return_value={}), - patch(f"{PKG}.fetch_hltb_times") as mock_fetch, - patch(f"{PKG}.save_hltb_cache"), - patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]), + patch(f"{_CACHED}.load_hltb_cache", return_value={}), + patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch, + patch(f"{_CACHED}.save_hltb_cache"), + patch(f"{_CACHED}.time.monotonic", side_effect=[0.0, 1.0]), ): mock_fetch.return_value = [] result = fetch_hltb_times_cached( @@ -95,10 +97,10 @@ class TestFetchHltbTimesCached: def test_uncached_zero_elapsed(self) -> None: """Covers the elapsed == 0 branch for rate calculation.""" with ( - patch(f"{PKG}.load_hltb_cache", return_value={}), - patch(f"{PKG}.fetch_hltb_times") as mock_fetch, - patch(f"{PKG}.save_hltb_cache"), - patch(f"{PKG}.time.monotonic", side_effect=[5.0, 5.0]), + patch(f"{_CACHED}.load_hltb_cache", return_value={}), + patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch, + patch(f"{_CACHED}.save_hltb_cache"), + patch(f"{_CACHED}.time.monotonic", side_effect=[5.0, 5.0]), ): mock_fetch.return_value = [] fetch_hltb_times_cached([(440, "TF2")]) @@ -106,10 +108,10 @@ class TestFetchHltbTimesCached: def test_found_count(self) -> None: """Covers the found count in logging.""" with ( - patch(f"{PKG}.load_hltb_cache", return_value={}), - patch(f"{PKG}.fetch_hltb_times") as mock_fetch, - patch(f"{PKG}.save_hltb_cache"), - patch(f"{PKG}.time.monotonic", side_effect=[0.0, 3.0]), + patch(f"{_CACHED}.load_hltb_cache", return_value={}), + patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch, + patch(f"{_CACHED}.save_hltb_cache"), + patch(f"{_CACHED}.time.monotonic", side_effect=[0.0, 3.0]), ): def add_found( @@ -151,13 +153,13 @@ class TestGetHltbSubmitUrl: similarity=1.0, hltb_game_id=12345, ) - with patch(f"{PKG}.fetch_hltb_times", return_value=[mock_result]): + with patch(f"{_CONF}.fetch_hltb_times", return_value=[mock_result]): url = get_hltb_submit_url("TF2") assert url == f"{HLTB_BASE_URL}/submit/game/12345" def test_not_found_empty(self) -> None: """Test not found empty.""" - with patch(f"{PKG}.fetch_hltb_times", return_value=[]): + with patch(f"{_CONF}.fetch_hltb_times", return_value=[]): url = get_hltb_submit_url("Unknown Game") assert url is None @@ -170,6 +172,6 @@ class TestGetHltbSubmitUrl: similarity=1.0, hltb_game_id=0, ) - with patch(f"{PKG}.fetch_hltb_times", return_value=[mock_result]): + with patch(f"{_CONF}.fetch_hltb_times", return_value=[mock_result]): url = get_hltb_submit_url("TF2") assert url is None diff --git a/steam_backlog_enforcer/tests/test_main_done.py b/steam_backlog_enforcer/tests/test_main_done.py index 7724057..d371f3f 100644 --- a/steam_backlog_enforcer/tests/test_main_done.py +++ b/steam_backlog_enforcer/tests/test_main_done.py @@ -9,7 +9,7 @@ from steam_backlog_enforcer.main import ( from steam_backlog_enforcer.steam_api import GameInfo PKG = "steam_backlog_enforcer.main" -CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done" +CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize" class TestCmdDone: diff --git a/steam_backlog_enforcer/tests/test_main_part2.py b/steam_backlog_enforcer/tests/test_main_part2.py index 1f1543d..562af07 100644 --- a/steam_backlog_enforcer/tests/test_main_part2.py +++ b/steam_backlog_enforcer/tests/test_main_part2.py @@ -10,7 +10,7 @@ from steam_backlog_enforcer._cmd_done import ( ) from steam_backlog_enforcer.config import Config, State -CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done" +CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize" PKG = "steam_backlog_enforcer.main" diff --git a/steam_backlog_enforcer/tests/test_main_part2b.py b/steam_backlog_enforcer/tests/test_main_part2b.py index 5ea09b8..e03ee60 100644 --- a/steam_backlog_enforcer/tests/test_main_part2b.py +++ b/steam_backlog_enforcer/tests/test_main_part2b.py @@ -10,7 +10,7 @@ from steam_backlog_enforcer._cmd_done import ( from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.steam_api import GameInfo -CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done" +CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize" PKG = "steam_backlog_enforcer.main" diff --git a/steam_backlog_enforcer/tests/test_main_part2c.py b/steam_backlog_enforcer/tests/test_main_part2c.py index 8cca379..25cbbbe 100644 --- a/steam_backlog_enforcer/tests/test_main_part2c.py +++ b/steam_backlog_enforcer/tests/test_main_part2c.py @@ -11,7 +11,10 @@ from steam_backlog_enforcer._cmd_done import ( from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.steam_api import GameInfo -CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done" +CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize" +# _refresh_uncached_shortlist_hours stayed in _cmd_done, so its +# fetch_hltb_times_cached resolves there, not in the finalize module. +_CMD_DONE = "steam_backlog_enforcer._cmd_done" PKG = "steam_backlog_enforcer.main" @@ -59,7 +62,7 @@ class TestFinalizeCompletionRetries: 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"{CMD_DONE_PKG}.fetch_hltb_times_cached", + f"{_CMD_DONE}.fetch_hltb_times_cached", return_value={3: 18.81}, ) as mock_fetch_hltb, patch(f"{CMD_DONE_PKG}.pick_next_game", side_effect=capture_pick), diff --git a/steam_backlog_enforcer/tests/test_whitelist.py b/steam_backlog_enforcer/tests/test_whitelist.py index 8b50c47..3a04a1f 100644 --- a/steam_backlog_enforcer/tests/test_whitelist.py +++ b/steam_backlog_enforcer/tests/test_whitelist.py @@ -164,7 +164,7 @@ class TestTrySetImmutable: # A fresh tmp file is already mutable, so without this the call # would (correctly) short-circuit before reaching chattr. patch( - "steam_backlog_enforcer._whitelist._immutable_flag_is", + "steam_backlog_enforcer._whitelist_locking._immutable_flag_is", return_value=False, ), patch("subprocess.run") as mock_run, @@ -181,7 +181,7 @@ class TestTrySetImmutable: target.write_text("data", encoding="utf-8") with ( patch( - "steam_backlog_enforcer._whitelist._immutable_flag_is", + "steam_backlog_enforcer._whitelist_locking._immutable_flag_is", return_value=True, ), patch("shutil.which") as mock_which, diff --git a/steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is.py b/steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is.py index 2313e96..e9aaee6 100644 --- a/steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is.py +++ b/steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is.py @@ -148,7 +148,7 @@ class TestLockAndUnlock: with ( patch( - "steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE", + "steam_backlog_enforcer._whitelist_locking.APPROVED_EXCEPTIONS_FILE", approved, ), patch("shutil.which", return_value="/usr/bin/chattr"), @@ -168,7 +168,7 @@ class TestLockAndUnlock: # Pretend the file really is immutable; a fresh tmp file is not, # and unlocking an already-mutable file is now a no-op. patch( - "steam_backlog_enforcer._whitelist._immutable_flag_is", + "steam_backlog_enforcer._whitelist_locking._immutable_flag_is", return_value=False, ), patch("subprocess.run") as mock_run, @@ -198,7 +198,7 @@ class TestPersistence: bad = tmp_path / "approved.json" bad.write_text("{{broken", encoding="utf-8") with patch( - "steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE", + "steam_backlog_enforcer._whitelist_locking.APPROVED_EXCEPTIONS_FILE", bad, ): assert _load_approved() == [] @@ -208,7 +208,7 @@ class TestPersistence: bad = tmp_path / "approved.json" bad.write_text('"just a string"', encoding="utf-8") with patch( - "steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE", + "steam_backlog_enforcer._whitelist_locking.APPROVED_EXCEPTIONS_FILE", bad, ): assert _load_approved() == [] diff --git a/steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is_append_audit_log.py b/steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is_append_audit_log.py index 8c10f7e..ca4c8b6 100644 --- a/steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is_append_audit_log.py +++ b/steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is_append_audit_log.py @@ -66,7 +66,7 @@ class TestAppendAuditLog: """Test audit log written.""" log_file = tmp_path / "audit.log" with patch( - "steam_backlog_enforcer._whitelist.EXCEPTION_AUDIT_LOG", + "steam_backlog_enforcer._whitelist_locking.EXCEPTION_AUDIT_LOG", log_file, ): _append_audit_log(440, "some reason", "REQUESTED") @@ -79,7 +79,7 @@ class TestAppendAuditLog: """Test audit log appends.""" log_file = tmp_path / "audit.log" with patch( - "steam_backlog_enforcer._whitelist.EXCEPTION_AUDIT_LOG", + "steam_backlog_enforcer._whitelist_locking.EXCEPTION_AUDIT_LOG", log_file, ): _append_audit_log(440, "first", "REQUESTED")