diff --git a/steam_backlog_enforcer/_cmd_done.py b/steam_backlog_enforcer/_cmd_done.py index 10867a8..9f11dac 100644 --- a/steam_backlog_enforcer/_cmd_done.py +++ b/steam_backlog_enforcer/_cmd_done.py @@ -7,12 +7,11 @@ import sys from typing import TYPE_CHECKING from steam_backlog_enforcer._hltb_cached import fetch_hltb_times_cached -from steam_backlog_enforcer._hltb_confidence import fetch_hltb_confidence_cached +from steam_backlog_enforcer._hltb_confidence import ( + refetch_poll_counts, +) from steam_backlog_enforcer._hltb_types import ( - load_hltb_cache, load_hltb_polls_cache, - restore_prior_hours, - save_hltb_cache, ) from steam_backlog_enforcer._snapshot import load_snapshot from steam_backlog_enforcer.game_install import ( @@ -84,19 +83,7 @@ def _backfill_polls_for_finished( return polls_cache _echo(f" Backfilling HLTB poll counts for {len(missing)} game(s)...") - 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) - - refreshed_hours = load_hltb_cache() - refreshed_polls = load_hltb_polls_cache() - restore_prior_hours(refreshed_hours, preserved_hours) - save_hltb_cache(refreshed_hours, refreshed_polls) - return refreshed_polls + return refetch_poll_counts(missing) def _report_assigned_confidence( diff --git a/steam_backlog_enforcer/_hltb_cached.py b/steam_backlog_enforcer/_hltb_cached.py index 1d1f24c..a817c61 100644 --- a/steam_backlog_enforcer/_hltb_cached.py +++ b/steam_backlog_enforcer/_hltb_cached.py @@ -7,7 +7,6 @@ under the 250-line cap. from __future__ import annotations import logging -import time from steam_backlog_enforcer._hltb_types import ( ProgressCb, @@ -19,7 +18,7 @@ from steam_backlog_enforcer._hltb_types import ( load_hltb_rush_cache, save_hltb_cache, ) -from steam_backlog_enforcer.hltb import fetch_hltb_times +from steam_backlog_enforcer.hltb import fetch_hltb_times_timed, games_per_second logger = logging.getLogger(__name__) @@ -56,40 +55,20 @@ def fetch_hltb_times_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 + elapsed = fetch_hltb_times_timed(uncached, cache, polls, progress_cb, extras) # 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, + games_per_second(len(uncached), elapsed), ) 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 index 3ae3381..610c42c 100644 --- a/steam_backlog_enforcer/_hltb_confidence.py +++ b/steam_backlog_enforcer/_hltb_confidence.py @@ -25,6 +25,8 @@ from steam_backlog_enforcer._hltb_types import ( from steam_backlog_enforcer.hltb import ( fetch_hltb_confidence, fetch_hltb_times, + fetch_hltb_times_timed, + games_per_second, ) logger = logging.getLogger(__name__) @@ -35,6 +37,30 @@ logger = logging.getLogger(__name__) # ────────────────────────────────────────────────────────────── +def refetch_poll_counts(missing: list[tuple[int, str]]) -> dict[int, int]: + """Re-search ``missing`` for poll counts and return the refreshed polls cache. + + The search-level fetch skips anything already in the hours cache, so the + hours entries are dropped first to force it -- then put back where the + refetch found nothing better, since a prior leisure/DLC estimate is + trusted over a miss. + """ + cache = load_hltb_cache() + polls = load_hltb_polls_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) + + fetch_hltb_confidence_cached(missing) + + refreshed_hours = load_hltb_cache() + refreshed_polls = load_hltb_polls_cache() + restore_prior_hours(refreshed_hours, preserved_hours) + save_hltb_cache(refreshed_hours, refreshed_polls) + return refreshed_polls + + def fetch_hltb_confidence_cached( games: list[tuple[int, str]], progress_cb: ProgressCb | None = None, @@ -143,28 +169,19 @@ def fetch_hltb_detail_missing( ) 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 + elapsed = fetch_hltb_times_timed(missing, cache, polls, progress_cb, extras) restore_prior_hours(cache, prior_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, + games_per_second(len(missing), elapsed), ) return fetched diff --git a/steam_backlog_enforcer/_hltb_search.py b/steam_backlog_enforcer/_hltb_search.py index 5a3e494..204ae4d 100644 --- a/steam_backlog_enforcer/_hltb_search.py +++ b/steam_backlog_enforcer/_hltb_search.py @@ -62,12 +62,11 @@ class _SearchCtx: headers: dict[str, str] cache: dict[int, float] polls: dict[int, int] = field(default_factory=dict) - count_comp: dict[int, int] = field(default_factory=dict) + extras: _HLTBExtras = field(default_factory=_HLTBExtras) auth: _AuthInfo | None = None counter: dict[str, int] = field(default_factory=dict) total: int = 0 progress_cb: ProgressCb | None = None - hltb_game_id: dict[int, int] = field(default_factory=dict) async def _search_one( @@ -123,25 +122,21 @@ async def _search_one( if result is not None: ctx.cache[app_id] = result.completionist_hours ctx.polls[app_id] = result.comp_100_count - ctx.count_comp[app_id] = result.count_comp + ctx.extras.count_comp[app_id] = result.count_comp if result.hltb_game_id > 0: - ctx.hltb_game_id[app_id] = result.hltb_game_id + ctx.extras.hltb_game_id[app_id] = result.hltb_game_id ctx.counter["found"] += 1 else: ctx.cache[app_id] = -1 ctx.polls[app_id] = 0 - ctx.count_comp[app_id] = 0 + ctx.extras.count_comp[app_id] = 0 ctx.counter["done"] += 1 done = ctx.counter["done"] # Incremental save every _SAVE_INTERVAL lookups. if not done % _SAVE_INTERVAL: - save_hltb_cache( - ctx.cache, - ctx.polls, - _HLTBExtras(count_comp=ctx.count_comp, hltb_game_id=ctx.hltb_game_id), - ) + save_hltb_cache(ctx.cache, ctx.polls, ctx.extras) # Report progress. if ctx.progress_cb is not None: @@ -150,17 +145,18 @@ async def _search_one( return result -async def _fetch_batch( +async def _search_batch( games: list[tuple[int, str]], cache: dict[int, float], polls: dict[int, int], progress_cb: ProgressCb | None, - extras: _HLTBExtras | None = None, + extras: _HLTBExtras, ) -> list[HLTBResult]: - """Fetch HLTB data for a batch of games using one shared session.""" - if extras is None: - extras = _HLTBExtras() + """Run the search-level HLTB lookups for a batch through one session. + Hours, poll counts and the extras come from the search results alone; + :func:`_fetch_batch` layers the per-game detail pages on top of this. + """ # 1. Discover the search URL (sync, one-time). search_url = _get_hltb_search_url() logger.info("HLTB search URL: %s", search_url) @@ -191,9 +187,6 @@ async def _fetch_batch( # 4. Fire all searches through a single persistent session. sem = asyncio.Semaphore(MAX_CONCURRENT) - counter = {"done": 0, "found": 0} - total = len(games) - connector = aiohttp.TCPConnector( limit=MAX_CONCURRENT, keepalive_timeout=30, @@ -208,25 +201,29 @@ async def _fetch_batch( headers=headers, cache=cache, polls=polls, - count_comp=extras.count_comp, + extras=extras, auth=auth, - counter=counter, - total=total, + counter={"done": 0, "found": 0}, + total=len(games), progress_cb=progress_cb, - hltb_game_id=extras.hltb_game_id, ) - tasks = [ - _search_one( - sem, - ctx, - app_id, - name, - ) - for app_id, name in games - ] + tasks = [_search_one(sem, ctx, app_id, name) for app_id, name in games] results = await asyncio.gather(*tasks) - search_results = [r for r in results if r is not None] + return [r for r in results if r is not None] + + +async def _fetch_batch( + games: list[tuple[int, str]], + cache: dict[int, float], + polls: dict[int, int], + progress_cb: ProgressCb | None, + extras: _HLTBExtras | None = None, +) -> list[HLTBResult]: + """Fetch HLTB data for a batch of games using one shared session.""" + if extras is None: + extras = _HLTBExtras() + search_results = await _search_batch(games, cache, polls, progress_cb, extras) # 5. Fetch leisure times + DLC from game detail pages. logger.info( diff --git a/steam_backlog_enforcer/_hosts_protection.py b/steam_backlog_enforcer/_hosts_protection.py index 1ba7db8..7165af4 100644 --- a/steam_backlog_enforcer/_hosts_protection.py +++ b/steam_backlog_enforcer/_hosts_protection.py @@ -8,6 +8,7 @@ from __future__ import annotations import logging import subprocess +from typing import TYPE_CHECKING from steam_backlog_enforcer._store_tools import ( GUARDCTL, @@ -19,6 +20,9 @@ from steam_backlog_enforcer.config import ( HOSTS_FILE, ) +if TYPE_CHECKING: + from collections.abc import Callable + logger = logging.getLogger(__name__) @@ -59,31 +63,53 @@ def _sudo_write_hosts(content: str) -> None: ) -def _reblock_hosts() -> bool: - """Uncomment Steam Store entries in /etc/hosts.""" +def rewrite_hosts_lines( + rewrite: Callable[[str], str | None], done_message: str +) -> bool: + """Apply ``rewrite`` to every /etc/hosts line, under the protection dance. + + ``rewrite`` returns the replacement for a line, or ``None`` to keep it. + The file is only written -- and ``done_message`` only logged -- when a + line actually changed; protection is lifted before and restored after + either way. + + Returns: + Whether the edit (or the no-op) completed without an OSError. + """ try: _disable_hosts_protection() content = HOSTS_FILE.read_text(encoding="utf-8") new_lines = [] changed = False for line in content.splitlines(keepends=True): - stripped = line.strip() - if stripped.startswith("# ") and any( - d in stripped for d in BLOCKED_DOMAINS - ): - # Remove the '# ' prefix. - uncommented = line.replace("# ", "", 1) - new_lines.append(uncommented) - changed = True - else: + replacement = rewrite(line) + if replacement is None: new_lines.append(line) + else: + new_lines.append(replacement) + changed = True if changed: _sudo_write_hosts("".join(new_lines)) - logger.info("Re-enabled Steam Store entries in /etc/hosts.") + logger.info(done_message) _enable_hosts_protection() except OSError: logger.exception("Failed to modify /etc/hosts") return False return True + + +def _uncomment_blocked(line: str) -> str | None: + """The line with its '# ' prefix removed, if it is a commented-out block.""" + stripped = line.strip() + if stripped.startswith("# ") and any(d in stripped for d in BLOCKED_DOMAINS): + return line.replace("# ", "", 1) + return None + + +def _reblock_hosts() -> bool: + """Uncomment Steam Store entries in /etc/hosts.""" + return rewrite_hosts_lines( + _uncomment_blocked, "Re-enabled Steam Store entries in /etc/hosts." + ) diff --git a/steam_backlog_enforcer/_leetcode_bonus.py b/steam_backlog_enforcer/_leetcode_bonus.py index 8762113..8dc0f6b 100644 --- a/steam_backlog_enforcer/_leetcode_bonus.py +++ b/steam_backlog_enforcer/_leetcode_bonus.py @@ -32,29 +32,20 @@ transports, the memo, and the incident on total failure. from __future__ import annotations -import http.client -import json import logging from pathlib import Path -import time as time_module -from typing import TYPE_CHECKING, Final -from urllib.parse import urlsplit +from typing import TYPE_CHECKING from steam_backlog_enforcer._bonus_incident import report_leetcode_incident from steam_backlog_enforcer._leetcode_ledger import read_ledger_solved_today +from steam_backlog_enforcer._status_api import AnswerCache, get_status if TYPE_CHECKING: from steam_backlog_enforcer.config import Config logger = logging.getLogger(__name__) -_TIMEOUT_SECONDS: Final = 2.0 -_CACHE_TTL_SECONDS: Final = 60.0 - -# Only successes are cached, matching _workout_budget: a failure is retried on -# the very next tick, so a restarted service takes effect immediately instead -# of costing an hour for another minute. -_cache: dict[str, tuple[float, bool]] = {} +_cache = AnswerCache() def reset_cache() -> None: @@ -77,22 +68,7 @@ def _fetch_leetcode_today(url: str) -> bool: could not check. KeyError: The payload lacked the expected fields. """ - split = urlsplit(url) - conn = http.client.HTTPConnection( - split.hostname or "127.0.0.1", - split.port or 80, - timeout=_TIMEOUT_SECONDS, - ) - try: - conn.request("GET", split.path or "/api/status") - resp = conn.getresponse() - body = resp.read() - if resp.status != http.HTTPStatus.OK: - msg = f"status {resp.status} {resp.reason}" - raise ValueError(msg) - finally: - conn.close() - block = json.loads(body)["leetcode"] + block = get_status(url)["leetcode"] if not block["checked"]: # The server answered, but said it could not look. That is not a "no". msg = f"leetcode-guard could not check: {block.get('reason', 'no reason')}" @@ -113,10 +89,9 @@ def leetcode_solved_today(config: Config) -> bool | None: incident while an honest "not solved yet" does not. """ url = config.leetcode_status_url - now = time_module.monotonic() - cached = _cache.get(url) - if cached is not None and now - cached[0] < _CACHE_TTL_SECONDS: - return cached[1] + cached = _cache.fresh(url) + if cached is not None: + return cached path = Path(config.leetcode_ledger_path).expanduser() answer = read_ledger_solved_today(path) @@ -125,8 +100,7 @@ def leetcode_solved_today(config: Config) -> bool | None: if answer is None: return None - _cache[url] = (now, answer) - return answer + return _cache.store(url, answer=answer) def _fallback_to_endpoint(url: str, path: Path) -> bool | None: diff --git a/steam_backlog_enforcer/_polls_reporting.py b/steam_backlog_enforcer/_polls_reporting.py index b9d4329..f57a02b 100644 --- a/steam_backlog_enforcer/_polls_reporting.py +++ b/steam_backlog_enforcer/_polls_reporting.py @@ -9,12 +9,11 @@ import logging from typing import TYPE_CHECKING from steam_backlog_enforcer._echo import _echo -from steam_backlog_enforcer._hltb_confidence import fetch_hltb_confidence_cached +from steam_backlog_enforcer._hltb_confidence import ( + refetch_poll_counts, +) from steam_backlog_enforcer._hltb_types import ( - load_hltb_cache, load_hltb_polls_cache, - restore_prior_hours, - save_hltb_cache, ) if TYPE_CHECKING: @@ -49,23 +48,7 @@ def _backfill_polls_for_finished( "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() - restore_prior_hours(refreshed_hours, preserved_hours) - save_hltb_cache(refreshed_hours, refreshed_polls) - return refreshed_polls + return refetch_poll_counts(missing) def _report_poll_confidence( diff --git a/steam_backlog_enforcer/_scanning_assign.py b/steam_backlog_enforcer/_scanning_assign.py index 6ecac1b..742ac98 100644 --- a/steam_backlog_enforcer/_scanning_assign.py +++ b/steam_backlog_enforcer/_scanning_assign.py @@ -100,6 +100,31 @@ def _assign_chosen_game( ) +def _clear_assignment(state: State, message: str) -> None: + """Say why nothing could be assigned and leave the slot empty.""" + _echo(message) + state.current_app_id = None + state.current_game_name = "" + state.save() + + +def _no_pick_message(confidence_skipped: int, linux_skipped: int) -> str: + """Why the candidate pass produced nothing: confidence, or ProtonDB.""" + if confidence_skipped > 0 and linux_skipped == 0: + return _NO_CONF_MSG + return "\nNo playable games left (all have poor ProtonDB ratings)!" + + +def _open_candidates(games: list[GameInfo], state: State) -> list[GameInfo]: + """Unfinished, unskipped games, shortest first, with cached confidence.""" + skip = set(state.finished_app_ids) | state.active_skipped_ids() + candidates = [g for g in games if not g.is_complete and g.app_id not in skip] + if candidates: + candidates.sort(key=_sort_key) + _apply_cached_confidence_to_candidates(candidates) + return candidates + + def _pick_next_game_sequential( games: list[GameInfo], state: State, @@ -113,29 +138,18 @@ def _pick_next_game_sequential( ``state`` for that game and the next candidate is evaluated. """ while True: - skip = set(state.finished_app_ids) | state.active_skipped_ids() - candidates = [g for g in games if not g.is_complete and g.app_id not in skip] + candidates = _open_candidates(games, state) if not candidates: - _echo(_NO_CONF_MSG) - state.current_app_id = None - state.current_game_name = "" - state.save() + _clear_assignment(state, _NO_CONF_MSG) return - candidates.sort(key=_sort_key) - _apply_cached_confidence_to_candidates(candidates) chosen, confidence_skipped, linux_skipped = _pick_next_shortest_candidate( candidates ) if chosen is None: - _echo( - _NO_CONF_MSG - if confidence_skipped > 0 and linux_skipped == 0 - else "\nNo playable games left (all have poor ProtonDB ratings)!" + _clear_assignment( + state, _no_pick_message(confidence_skipped, linux_skipped) ) - state.current_app_id = None - state.current_game_name = "" - state.save() return if not on_select(chosen): diff --git a/steam_backlog_enforcer/_status_api.py b/steam_backlog_enforcer/_status_api.py new file mode 100644 index 0000000..4a6d7a6 --- /dev/null +++ b/steam_backlog_enforcer/_status_api.py @@ -0,0 +1,88 @@ +"""One loopback status-API read, shared by the workout and LeetCode bonuses. + +Both bonuses ask a sibling daemon's ``/api/status`` endpoint over loopback and +reuse a successful answer for a minute. The transport and the cache policy +are identical, so they live here once; each bonus keeps only the field it +reads out of the payload. + +Uses ``http.client`` rather than ``urllib.request`` deliberately, matching +``screen_locker._http_workout_fetch``: it takes a host and a path rather than a +URL, so there is no scheme for a bad config value to turn into a ``file://`` +read, and no lint waiver is needed to say so. +""" + +from __future__ import annotations + +import http.client +import json +import time +from typing import Any, Final +from urllib.parse import urlsplit + +# Loopback, so a short timeout is generous. The daemon ticks frequently and +# must never block on this. +_TIMEOUT_SECONDS: Final = 2.0 + +# A successful answer is reused for this long. The daemon would otherwise make +# one HTTP call per tick, and the answer changes at most a few times a day. The +# cost is that a workout logged now raises the budget within a minute rather +# than instantly -- invisible next to a two-hour swing. +_CACHE_TTL_SECONDS: Final = 60.0 + + +def get_status(url: str) -> dict[str, Any]: + """GET a status endpoint and return its JSON body. + + Args: + url: The status endpoint, e.g. ``http://127.0.0.1:8770/api/status``. + + Raises: + OSError: The endpoint could not be reached. + ValueError: The response was not usable JSON, or not a 200. + """ + split = urlsplit(url) + conn = http.client.HTTPConnection( + split.hostname or "127.0.0.1", + split.port or 80, + timeout=_TIMEOUT_SECONDS, + ) + try: + conn.request("GET", split.path or "/api/status") + resp = conn.getresponse() + body = resp.read() + if resp.status != http.HTTPStatus.OK: + msg = f"status {resp.status} {resp.reason}" + raise ValueError(msg) + finally: + conn.close() + payload: dict[str, Any] = json.loads(body) + return payload + + +class AnswerCache: + """Successful answers by URL, each good for :data:`_CACHE_TTL_SECONDS`. + + Only *successes* are stored. A failure is retried on the very next tick, so + a restarted status server takes effect immediately instead of serving the + floor for another minute. + """ + + def __init__(self) -> None: + """Start empty.""" + self._entries: dict[str, tuple[float, bool]] = {} + + def clear(self) -> None: + """Drop every cached answer, forcing the next call to re-ask.""" + self._entries.clear() + + def fresh(self, url: str) -> bool | None: + """The cached answer for ``url`` if it is still within its TTL.""" + cached = self._entries.get(url) + if cached is not None and time.monotonic() - cached[0] < _CACHE_TTL_SECONDS: + return cached[1] + return None + + def store(self, url: str, *, answer: bool) -> bool: + """Remember ``answer`` for ``url`` as of now, and hand it back.""" + self._entries[url] = (time.monotonic(), answer) + return answer diff --git a/steam_backlog_enforcer/_total_block_hosts.py b/steam_backlog_enforcer/_total_block_hosts.py index 2e10893..9cdc594 100644 --- a/steam_backlog_enforcer/_total_block_hosts.py +++ b/steam_backlog_enforcer/_total_block_hosts.py @@ -11,16 +11,16 @@ from __future__ import annotations import logging import subprocess +from steam_backlog_enforcer._hosts_protection import ( + _disable_hosts_protection, + _enable_hosts_protection, + _sudo_write_hosts, +) from steam_backlog_enforcer._total_block_domains import ( _ALL_TOTAL_BLOCK_DOMAINS, NULL_ROUTE_IP, ) from steam_backlog_enforcer.config import HOSTS_FILE -from steam_backlog_enforcer.store_blocker import ( - _disable_hosts_protection, - _enable_hosts_protection, - _sudo_write_hosts, -) logger = logging.getLogger(__name__) diff --git a/steam_backlog_enforcer/_whitelist.py b/steam_backlog_enforcer/_whitelist.py index 33bc57e..da6dfe0 100644 --- a/steam_backlog_enforcer/_whitelist.py +++ b/steam_backlog_enforcer/_whitelist.py @@ -6,22 +6,9 @@ from collections import Counter import logging import math import re -from typing import TYPE_CHECKING - -from steam_backlog_enforcer.config import CONFIG_DIR - -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 # ────────────────────────────────────────────────────────────── diff --git a/steam_backlog_enforcer/_whitelist_locking.py b/steam_backlog_enforcer/_whitelist_locking.py index ca09ad0..d447138 100644 --- a/steam_backlog_enforcer/_whitelist_locking.py +++ b/steam_backlog_enforcer/_whitelist_locking.py @@ -35,16 +35,6 @@ logger = logging.getLogger(__name__) 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 diff --git a/steam_backlog_enforcer/_workout_budget.py b/steam_backlog_enforcer/_workout_budget.py index bf9117b..74d39ef 100644 --- a/steam_backlog_enforcer/_workout_budget.py +++ b/steam_backlog_enforcer/_workout_budget.py @@ -11,42 +11,21 @@ malformed body, non-200 -- yields ``None``, logged at warning with the concrete reason, and the caller grants no workout bonus. Failing open would let the whole coupling be defeated by stopping one user service, and a silent bonus would be indistinguishable from an earned one. - -Uses ``http.client`` rather than ``urllib.request`` deliberately, matching -``screen_locker._http_workout_fetch``: it takes a host and a path rather than a -URL, so there is no scheme for a bad config value to turn into a ``file://`` -read, and no lint waiver is needed to say so. """ from __future__ import annotations -import http.client -import json import logging -import time -from typing import TYPE_CHECKING, Final -from urllib.parse import urlsplit +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._status_api import AnswerCache, get_status if TYPE_CHECKING: from steam_backlog_enforcer.config import Config logger = logging.getLogger(__name__) -# Loopback, so a short timeout is generous. The daemon ticks frequently and -# must never block on this. -_TIMEOUT_SECONDS: Final = 2.0 - -# A successful answer is reused for this long. The daemon would otherwise make -# one HTTP call per tick, and the answer changes at most a few times a day. The -# cost is that a workout logged now raises the budget within a minute rather -# than instantly -- invisible next to a two-hour swing. -_CACHE_TTL_SECONDS: Final = 60.0 - -# Only *successes* are cached. A failure is retried on the very next tick, so a -# restarted status server takes effect immediately instead of serving the floor -# for another minute. A dict rather than a module global so no function here -# needs a `global` statement to write it. -_cache: dict[str, tuple[float, bool]] = {} +_cache = AnswerCache() def reset_cache() -> None: @@ -68,22 +47,7 @@ def _fetch_workout_today(url: str) -> bool: ValueError: The response was not usable JSON, or not a 200. KeyError: The payload lacked the expected fields. """ - split = urlsplit(url) - conn = http.client.HTTPConnection( - split.hostname or "127.0.0.1", - split.port or 80, - timeout=_TIMEOUT_SECONDS, - ) - try: - conn.request("GET", split.path or "/api/status") - resp = conn.getresponse() - body = resp.read() - if resp.status != http.HTTPStatus.OK: - msg = f"status {resp.status} {resp.reason}" - raise ValueError(msg) - finally: - conn.close() - return bool(json.loads(body)["gaming"]["workout_today"]) + return bool(get_status(url)["gaming"]["workout_today"]) def workout_logged_today(config: Config) -> bool | None: @@ -98,10 +62,9 @@ def workout_logged_today(config: Config) -> bool | None: distinct from ``False`` so the caller can log which one happened. """ url = config.workout_status_url - now = time.monotonic() - cached = _cache.get(url) - if cached is not None and now - cached[0] < _CACHE_TTL_SECONDS: - return cached[1] + cached = _cache.fresh(url) + if cached is not None: + return cached try: answer = _fetch_workout_today(url) @@ -123,5 +86,4 @@ def workout_logged_today(config: Config) -> bool | None: ) return None - _cache[url] = (now, answer) - return answer + return _cache.store(url, answer=answer) diff --git a/steam_backlog_enforcer/hltb.py b/steam_backlog_enforcer/hltb.py index e7d29e5..d124f05 100644 --- a/steam_backlog_enforcer/hltb.py +++ b/steam_backlog_enforcer/hltb.py @@ -14,18 +14,10 @@ from __future__ import annotations import asyncio import logging +import time -import aiohttp - -from steam_backlog_enforcer._hltb_search import ( - _fetch_batch, - _get_auth_info, - _get_hltb_search_url, - _search_one, - _SearchCtx, -) +from steam_backlog_enforcer._hltb_search import _fetch_batch, _search_batch from steam_backlog_enforcer._hltb_types import ( - MAX_CONCURRENT, HLTBResult, ProgressCb, _HLTBExtras, @@ -45,74 +37,8 @@ async def _fetch_batch_confidence_only( count_comp: dict[int, int] | None = None, ) -> list[HLTBResult]: """Fetch only search-level HLTB data (hours + confidence), no detail pages.""" - # 1. Discover the search URL (sync, one-time). - search_url = _get_hltb_search_url() - logger.info("HLTB search URL: %s", search_url) - - timeout = aiohttp.ClientTimeout(total=20, sock_read=15) - - # 2. Get auth info (separate session — avoids reuse issues). - async with aiohttp.ClientSession(timeout=timeout) as init_session: - auth = await _get_auth_info(search_url, init_session) - if auth is None: - logger.warning("Could not get HLTB auth info, aborting fetch.") - return [] - logger.info("HLTB auth token acquired.") - - # 3. Build shared headers for all search requests. - headers: dict[str, str] = { - "content-type": "application/json", - "accept": "*/*", - "User-Agent": ( - "Mozilla/5.0 (X11; Linux x86_64; rv:136.0) Gecko/20100101 Firefox/136.0" - ), - "referer": "https://howlongtobeat.com/", - "x-auth-token": auth.token, - } - if auth.hp_key: - headers["x-hp-key"] = auth.hp_key - headers["x-hp-val"] = auth.hp_val - - # 4. Fire all searches through a single persistent session. - sem = asyncio.Semaphore(MAX_CONCURRENT) - counter = {"done": 0, "found": 0} - total = len(games) - - if count_comp is None: - count_comp = {} - - connector = aiohttp.TCPConnector( - limit=MAX_CONCURRENT, - keepalive_timeout=30, - ) - async with aiohttp.ClientSession( - timeout=timeout, - connector=connector, - ) as session: - ctx = _SearchCtx( - session=session, - search_url=search_url, - headers=headers, - cache=cache, - polls=polls, - count_comp=count_comp, - auth=auth, - counter=counter, - total=total, - progress_cb=progress_cb, - ) - tasks = [ - _search_one( - sem, - ctx, - app_id, - name, - ) - for app_id, name in games - ] - results = await asyncio.gather(*tasks) - - return [r for r in results if r is not None] + extras = _HLTBExtras(count_comp=count_comp if count_comp is not None else {}) + return await _search_batch(games, cache, polls, progress_cb, extras) def fetch_hltb_times( @@ -140,6 +66,26 @@ def fetch_hltb_times( ) +def fetch_hltb_times_timed( + games: list[tuple[int, str]], + cache: dict[int, float], + polls: dict[int, int], + progress_cb: ProgressCb | None, + extras: _HLTBExtras, +) -> float: + """Run :func:`fetch_hltb_times` into the given caches; return the seconds taken.""" + t0 = time.monotonic() + fetch_hltb_times( + games, cache=cache, polls=polls, progress_cb=progress_cb, extras=extras + ) + return time.monotonic() - t0 + + +def games_per_second(count: int, elapsed: float) -> float: + """Throughput for a completion log line; 0 when nothing was timed.""" + return count / elapsed if elapsed > 0 else 0 + + def fetch_hltb_confidence( games: list[tuple[int, str]], cache: dict[int, float] | None = None, diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index 56056ad..bdb1fb0 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -15,6 +15,9 @@ from steam_backlog_enforcer._pick_completion import mark_finished, report_comple from steam_backlog_enforcer._scanning_assign import ( _NO_CONF_MSG, _assign_chosen_game, + _clear_assignment, + _no_pick_message, + _open_candidates, _pick_next_game_sequential, _prompt_user_pick, ) @@ -23,10 +26,8 @@ from steam_backlog_enforcer._scanning_candidates import ( _collect_top_candidates, _pick_next_shortest_candidate, _pick_playable_candidate, - _sort_key, ) from steam_backlog_enforcer._scanning_confidence import ( - _apply_cached_confidence_to_candidates, _report_poll_confidence, ) from steam_backlog_enforcer._scanning_tampering import ( @@ -163,31 +164,16 @@ def pick_next_game( _pick_next_game_sequential(games, state, config, on_select) return - skip = set(state.finished_app_ids) | state.active_skipped_ids() - candidates = [g for g in games if not g.is_complete and g.app_id not in skip] - + candidates = _open_candidates(games, state) if not candidates: - _echo(_NO_CONF_MSG) - state.current_app_id = None - state.current_game_name = "" - state.save() + _clear_assignment(state, _NO_CONF_MSG) return - candidates.sort(key=_sort_key) - _apply_cached_confidence_to_candidates(candidates) qualified, confidence_skipped, linux_skipped = _collect_qualified_candidates( candidates ) - if not qualified: - _echo( - _NO_CONF_MSG - if confidence_skipped > 0 and linux_skipped == 0 - else "\nNo playable games left (all have poor ProtonDB ratings)!" - ) - state.current_app_id = None - state.current_game_name = "" - state.save() + _clear_assignment(state, _no_pick_message(confidence_skipped, linux_skipped)) return idx = _prompt_user_pick(qualified) diff --git a/steam_backlog_enforcer/store_blocker.py b/steam_backlog_enforcer/store_blocker.py index 047b174..e4ae399 100644 --- a/steam_backlog_enforcer/store_blocker.py +++ b/steam_backlog_enforcer/store_blocker.py @@ -15,10 +15,8 @@ import logging import subprocess from steam_backlog_enforcer._hosts_protection import ( - _disable_hosts_protection, - _enable_hosts_protection, _reblock_hosts, - _sudo_write_hosts, + rewrite_hosts_lines, ) from steam_backlog_enforcer._store_iptables import ( _block_store_iptables, @@ -161,29 +159,18 @@ def _unblock_hosts() -> bool: logger.info("Steam Store not blocked in /etc/hosts, nothing to do.") return True - try: - _disable_hosts_protection() - content = HOSTS_FILE.read_text(encoding="utf-8") - new_lines = [] - changed = False - for line in content.splitlines(keepends=True): - stripped = line.strip() - if ( - not stripped.startswith("#") - and stripped.startswith(HOSTS_REDIRECT_IP) - and any(d in stripped for d in BLOCKED_DOMAINS) - ): - new_lines.append(f"# {line}" if line.endswith("\n") else f"# {line}\n") - changed = True - else: - new_lines.append(line) + return rewrite_hosts_lines( + _comment_out_blocked, "Commented out Steam Store entries in /etc/hosts." + ) - if changed: - _sudo_write_hosts("".join(new_lines)) - logger.info("Commented out Steam Store entries in /etc/hosts.") - _enable_hosts_protection() - except OSError: - logger.exception("Failed to modify /etc/hosts") - return False - return True +def _comment_out_blocked(line: str) -> str | None: + """The line commented out, if it is a live redirect of a blocked domain.""" + stripped = line.strip() + if ( + not stripped.startswith("#") + and stripped.startswith(HOSTS_REDIRECT_IP) + and any(d in stripped for d in BLOCKED_DOMAINS) + ): + return f"# {line}" if line.endswith("\n") else f"# {line}\n" + return None diff --git a/steam_backlog_enforcer/tests/test_hltb.py b/steam_backlog_enforcer/tests/test_hltb.py index 0741c2c..bcddd0f 100644 --- a/steam_backlog_enforcer/tests/test_hltb.py +++ b/steam_backlog_enforcer/tests/test_hltb.py @@ -11,12 +11,11 @@ import aiohttp from steam_backlog_enforcer._hltb_matching import _similarity from steam_backlog_enforcer._hltb_search import _AuthInfo -from steam_backlog_enforcer._hltb_search_api import _get_hltb_search_url +from steam_backlog_enforcer._hltb_search_api import _get_auth_info, _get_hltb_search_url from steam_backlog_enforcer._hltb_types import ( load_hltb_cache, save_hltb_cache, ) -from steam_backlog_enforcer.hltb import _get_auth_info if TYPE_CHECKING: from pathlib import Path diff --git a/steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py b/steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py index fc8637d..24b7c4c 100644 --- a/steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py +++ b/steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py @@ -13,7 +13,7 @@ from steam_backlog_enforcer.hltb import ( fetch_hltb_confidence, ) -PKG = "steam_backlog_enforcer.hltb" +PKG = "steam_backlog_enforcer._hltb_search" _CONF = "steam_backlog_enforcer._hltb_confidence" 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 ebe72bc..75c3a46 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 @@ -11,6 +11,7 @@ if TYPE_CHECKING: from steam_backlog_enforcer._hltb_types import _HLTBExtras PKG = "steam_backlog_enforcer._hltb_confidence" +_HLTB = "steam_backlog_enforcer.hltb" class _DummySession: @@ -33,7 +34,7 @@ class TestFetchHltbDetailMissing: with ( patch(f"{PKG}.load_hltb_rush_cache", return_value={440: 15.0}), patch(f"{PKG}.load_hltb_game_id_cache", return_value={440: 12345}), - patch(f"{PKG}.fetch_hltb_times") as mock_fetch, + patch(f"{_HLTB}.fetch_hltb_times") as mock_fetch, ): result = fetch_hltb_detail_missing([(440, "TF2")]) assert result == 0 @@ -63,7 +64,7 @@ class TestFetchHltbDetailMissing: patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), - patch(f"{PKG}.fetch_hltb_times", side_effect=add_rush), + patch(f"{_HLTB}.fetch_hltb_times", side_effect=add_rush), patch(f"{PKG}.save_hltb_cache") as mock_save, patch(f"{PKG}.time.monotonic", side_effect=[0.0, 2.0]), ): @@ -90,7 +91,7 @@ class TestFetchHltbDetailMissing: patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), - patch(f"{PKG}.fetch_hltb_times"), # no-op, cache stays empty + patch(f"{_HLTB}.fetch_hltb_times"), # no-op, cache stays empty patch(f"{PKG}.save_hltb_cache", side_effect=capture_save), patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]), ): @@ -131,7 +132,7 @@ class TestFetchHltbDetailMissing: patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), - patch(f"{PKG}.fetch_hltb_times", side_effect=add_hours_and_rush), + patch(f"{_HLTB}.fetch_hltb_times", side_effect=add_hours_and_rush), patch(f"{PKG}.save_hltb_cache", side_effect=capture_save), patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]), ): @@ -148,7 +149,7 @@ class TestFetchHltbDetailMissing: patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), - patch(f"{PKG}.fetch_hltb_times"), + patch(f"{_HLTB}.fetch_hltb_times"), patch(f"{PKG}.save_hltb_cache"), patch(f"{PKG}.time.monotonic", side_effect=[5.0, 5.0]), ): @@ -164,7 +165,7 @@ class TestFetchHltbDetailMissing: patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), - patch(f"{PKG}.fetch_hltb_times"), + patch(f"{_HLTB}.fetch_hltb_times"), patch(f"{PKG}.save_hltb_cache"), patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]), ): diff --git a/steam_backlog_enforcer/tests/test_hltb_part2.py b/steam_backlog_enforcer/tests/test_hltb_part2.py index 402a433..8350343 100644 --- a/steam_backlog_enforcer/tests/test_hltb_part2.py +++ b/steam_backlog_enforcer/tests/test_hltb_part2.py @@ -47,9 +47,9 @@ class TestFetchHltbTimesCached: """Test uncached games fetched.""" with ( patch(f"{_CACHED}.load_hltb_cache", return_value={440: 50.0}), - patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch, + patch(f"{PKG}.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]), + patch(f"{PKG}.time.monotonic", side_effect=[0.0, 2.0]), ): # fetch_hltb_times modifies cache in-place def add_to_cache( @@ -81,9 +81,9 @@ class TestFetchHltbTimesCached: cb = MagicMock() with ( patch(f"{_CACHED}.load_hltb_cache", return_value={}), - patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch, + patch(f"{PKG}.fetch_hltb_times") as mock_fetch, patch(f"{_CACHED}.save_hltb_cache"), - patch(f"{_CACHED}.time.monotonic", side_effect=[0.0, 1.0]), + patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]), ): mock_fetch.return_value = [] result = fetch_hltb_times_cached( @@ -96,9 +96,9 @@ class TestFetchHltbTimesCached: """Covers the elapsed == 0 branch for rate calculation.""" with ( patch(f"{_CACHED}.load_hltb_cache", return_value={}), - patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch, + patch(f"{PKG}.fetch_hltb_times") as mock_fetch, patch(f"{_CACHED}.save_hltb_cache"), - patch(f"{_CACHED}.time.monotonic", side_effect=[5.0, 5.0]), + patch(f"{PKG}.time.monotonic", side_effect=[5.0, 5.0]), ): mock_fetch.return_value = [] fetch_hltb_times_cached([(440, "TF2")]) @@ -107,9 +107,9 @@ class TestFetchHltbTimesCached: """Covers the found count in logging.""" with ( patch(f"{_CACHED}.load_hltb_cache", return_value={}), - patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch, + patch(f"{PKG}.fetch_hltb_times") as mock_fetch, patch(f"{_CACHED}.save_hltb_cache"), - patch(f"{_CACHED}.time.monotonic", side_effect=[0.0, 3.0]), + patch(f"{PKG}.time.monotonic", side_effect=[0.0, 3.0]), ): def add_found( diff --git a/steam_backlog_enforcer/tests/test_hltb_search.py b/steam_backlog_enforcer/tests/test_hltb_search.py index 566f7b9..ed8750f 100644 --- a/steam_backlog_enforcer/tests/test_hltb_search.py +++ b/steam_backlog_enforcer/tests/test_hltb_search.py @@ -101,7 +101,7 @@ class TestSearchOne: ctx = _make_ctx(_make_session(resp)) result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2")) assert result is not None - assert 440 not in ctx.hltb_game_id + assert 440 not in ctx.extras.hltb_game_id def test_not_found(self) -> None: """Test not found.""" diff --git a/steam_backlog_enforcer/tests/test_leetcode_bonus.py b/steam_backlog_enforcer/tests/test_leetcode_bonus.py index c284076..8a33bdf 100644 --- a/steam_backlog_enforcer/tests/test_leetcode_bonus.py +++ b/steam_backlog_enforcer/tests/test_leetcode_bonus.py @@ -22,6 +22,7 @@ if TYPE_CHECKING: from collections.abc import Iterator _PKG = "steam_backlog_enforcer._leetcode_bonus" +_API = "steam_backlog_enforcer._status_api" @pytest.fixture(autouse=True) @@ -104,7 +105,7 @@ class TestFetchLeetcodeToday: {"leetcode": {"checked": False, "solved_today": False, "reason": "dead"}} ).encode() with ( - patch(f"{_PKG}.http.client.HTTPConnection") as conn, + patch(f"{_API}.http.client.HTTPConnection") as conn, pytest.raises(ValueError, match="could not check"), ): resp = conn.return_value.getresponse.return_value @@ -115,7 +116,7 @@ class TestFetchLeetcodeToday: def test_a_non_200_is_an_error(self) -> None: """An error page is not a payload.""" with ( - patch(f"{_PKG}.http.client.HTTPConnection") as conn, + patch(f"{_API}.http.client.HTTPConnection") as conn, pytest.raises(ValueError, match="status 500"), ): resp = conn.return_value.getresponse.return_value @@ -129,7 +130,7 @@ class TestFetchLeetcodeToday: payload = json.dumps( {"leetcode": {"checked": True, "solved_today": True, "reason": "1"}} ).encode() - with patch(f"{_PKG}.http.client.HTTPConnection") as conn: + with patch(f"{_API}.http.client.HTTPConnection") as conn: resp = conn.return_value.getresponse.return_value resp.status = 200 resp.read.return_value = payload diff --git a/steam_backlog_enforcer/tests/test_polls_tracking_backfill_polls_for_finished.py b/steam_backlog_enforcer/tests/test_polls_tracking_backfill_polls_for_finished.py index 8ae5726..2eaef09 100644 --- a/steam_backlog_enforcer/tests/test_polls_tracking_backfill_polls_for_finished.py +++ b/steam_backlog_enforcer/tests/test_polls_tracking_backfill_polls_for_finished.py @@ -14,6 +14,7 @@ if TYPE_CHECKING: _TYPES = "steam_backlog_enforcer._hltb_types" _CMD = "steam_backlog_enforcer._cmd_done" +_CONF = "steam_backlog_enforcer._hltb_confidence" _SCAN = "steam_backlog_enforcer.scanning" @@ -68,7 +69,7 @@ class TestBackfillPollsForFinished: patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), patch(f"{_CMD}.load_snapshot", return_value=[{"app_id": 1, "name": "G"}]), - patch(f"{_CMD}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch(f"{_CONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch), patch(f"{_CMD}._echo"), ): result = _cmd_done._backfill_polls_for_finished(_state([1])) @@ -93,7 +94,7 @@ class TestBackfillPollsForFinished: patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), patch(f"{_CMD}.load_snapshot", return_value=[{"app_id": 7, "name": "G"}]), - patch(f"{_CMD}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch(f"{_CONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch), patch(f"{_CMD}._echo"), ): result = _cmd_done._backfill_polls_for_finished( @@ -121,7 +122,7 @@ class TestBackfillPollsForFinished: patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), patch(f"{_CMD}.load_snapshot", return_value=[{"app_id": 3, "name": "G"}]), - patch(f"{_CMD}.fetch_hltb_confidence_cached", side_effect=fake_fetch), + patch(f"{_CONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch), patch(f"{_CMD}._echo"), ): _cmd_done._backfill_polls_for_finished(_state([3])) diff --git a/steam_backlog_enforcer/tests/test_polls_tracking_part2.py b/steam_backlog_enforcer/tests/test_polls_tracking_part2.py index 9fc3b32..5b37a63 100644 --- a/steam_backlog_enforcer/tests/test_polls_tracking_part2.py +++ b/steam_backlog_enforcer/tests/test_polls_tracking_part2.py @@ -15,6 +15,7 @@ if TYPE_CHECKING: _TYPES = "steam_backlog_enforcer._hltb_types" _CMD = "steam_backlog_enforcer._cmd_done" +_CONF = "steam_backlog_enforcer._hltb_confidence" _SCAN = "steam_backlog_enforcer.scanning" _SCANCONF = "steam_backlog_enforcer._scanning_confidence" _POLLS = "steam_backlog_enforcer._polls_reporting" @@ -160,7 +161,7 @@ class TestScanningPollsIntegration: patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), patch( - f"{_POLLS}.fetch_hltb_confidence_cached", + f"{_CONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch, ), ): @@ -197,7 +198,7 @@ class TestScanningPollsIntegration: patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), patch( - f"{_POLLS}.fetch_hltb_confidence_cached", + f"{_CONF}.fetch_hltb_confidence_cached", side_effect=fake_fetch, ), ): diff --git a/steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py b/steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py index 7f9b712..38730c8 100644 --- a/steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py +++ b/steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py @@ -141,6 +141,11 @@ class TestPickNextGameGroup2: "steam_backlog_enforcer._scanning_confidence._echo", side_effect=lambda *a, **_: echoed.append(a[0]), ), + # The "nothing assignable" line is spoken where the slot is cleared. + patch( + "steam_backlog_enforcer._scanning_assign._echo", + side_effect=lambda *a, **_: echoed.append(a[0]), + ), patch( "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate", return_value=None, diff --git a/steam_backlog_enforcer/tests/test_store_blocker_part2.py b/steam_backlog_enforcer/tests/test_store_blocker_part2.py index f1eeb01..398b6d3 100644 --- a/steam_backlog_enforcer/tests/test_store_blocker_part2.py +++ b/steam_backlog_enforcer/tests/test_store_blocker_part2.py @@ -5,11 +5,13 @@ from __future__ import annotations from typing import TYPE_CHECKING from unittest.mock import patch -from steam_backlog_enforcer.store_blocker import ( +from steam_backlog_enforcer._hosts_protection import ( _disable_hosts_protection, _enable_hosts_protection, _reblock_hosts, _sudo_write_hosts, +) +from steam_backlog_enforcer.store_blocker import ( _unblock_hosts, flush_dns_cache, ) @@ -18,6 +20,7 @@ if TYPE_CHECKING: from pathlib import Path PKG = "steam_backlog_enforcer.store_blocker" +_HP = "steam_backlog_enforcer._hosts_protection" class TestSudoWriteHosts: @@ -70,10 +73,10 @@ class TestUnblockHosts: ) with ( patch(f"{PKG}.is_store_blocked", return_value=True), - patch(f"{PKG}.HOSTS_FILE", hosts_file), - patch(f"{PKG}._disable_hosts_protection"), - patch(f"{PKG}._enable_hosts_protection"), - patch(f"{PKG}._sudo_write_hosts") as mock_write, + patch(f"{_HP}.HOSTS_FILE", hosts_file), + patch(f"{_HP}._disable_hosts_protection"), + patch(f"{_HP}._enable_hosts_protection"), + patch(f"{_HP}._sudo_write_hosts") as mock_write, ): result = _unblock_hosts() assert result is True @@ -88,10 +91,10 @@ class TestUnblockHosts: ) with ( patch(f"{PKG}.is_store_blocked", return_value=True), - patch(f"{PKG}.HOSTS_FILE", hosts_file), - patch(f"{PKG}._disable_hosts_protection"), - patch(f"{PKG}._enable_hosts_protection"), - patch(f"{PKG}._sudo_write_hosts") as mock_write, + patch(f"{_HP}.HOSTS_FILE", hosts_file), + patch(f"{_HP}._disable_hosts_protection"), + patch(f"{_HP}._enable_hosts_protection"), + patch(f"{_HP}._sudo_write_hosts") as mock_write, ): result = _unblock_hosts() assert result is True @@ -100,7 +103,7 @@ class TestUnblockHosts: def test_os_error(self) -> None: with ( patch(f"{PKG}.is_store_blocked", return_value=True), - patch(f"{PKG}._disable_hosts_protection", side_effect=OSError), + patch(f"{_HP}._disable_hosts_protection", side_effect=OSError), ): result = _unblock_hosts() assert result is False diff --git a/steam_backlog_enforcer/tests/test_workout_budget.py b/steam_backlog_enforcer/tests/test_workout_budget.py index 92e93fc..fa64a2c 100644 --- a/steam_backlog_enforcer/tests/test_workout_budget.py +++ b/steam_backlog_enforcer/tests/test_workout_budget.py @@ -25,6 +25,7 @@ if TYPE_CHECKING: from collections.abc import Iterator _PKG = "steam_backlog_enforcer._workout_budget" +_API = "steam_backlog_enforcer._status_api" @pytest.fixture(autouse=True) @@ -112,7 +113,8 @@ class TestWorkoutLoggedToday: """A workout logged later in the day still raises the budget.""" with ( patch(f"{_PKG}._fetch_workout_today", return_value=False) as fetch, - patch(f"{_PKG}.time.monotonic", side_effect=[0.0, 999.0]), + # Stored at 0.0; read back at 999.0 (expired); stored again. + patch(f"{_API}.time.monotonic", side_effect=[0.0, 999.0, 999.0]), ): workout_logged_today(_config()) workout_logged_today(_config())