kuhyx/steam-backlog-enforcer
1
0
Fork 0
mirror of https://github.com/kuhyx/steam-backlog-enforcer.git synced 2026-11-10 19:27:04 +01:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

pylint: fold the remaining split-era duplicates back into single owners

Browse Source 
The 250-line split left function-sized copies behind: the confidence-only
batch fetch was the search half of _fetch_batch verbatim (now
_search_batch, and _SearchCtx carries one _HLTBExtras instead of two of
its dicts); the workout and LeetCode bonuses each had the same loopback
GET and answer cache (now _status_api); the poll-count refetch lived in
both _cmd_done and _polls_reporting (now refetch_poll_counts); the
candidate-selection prelude in both scanning and _scanning_assign; the
/etc/hosts rewrite loop in both block and unblock (now
rewrite_hosts_lines with a per-line rule); the timed HLTB fetch and its
rate line; and the whitelist constants each module only half used.
_hltb_cached's dead facade constants go too. Tests patch where the code
now lives.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHCbkJWXHjtTUaqvnHRAqA
This commit is contained in:
Krzysztof Rudnicki 2026-09-12 20:49:12 +02:00
parent d6d9a2d1a6
commit 2d45515665
27 changed files with 339 additions and 403 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
steam_backlog_enforcer/_cmd_done.py
View File

@ -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(

27
steam_backlog_enforcer/_hltb_cached.py
View File

@ -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",
}
)

39
steam_backlog_enforcer/_hltb_confidence.py
View File

@ -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

61
steam_backlog_enforcer/_hltb_search.py
View File

@ -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(

50
steam_backlog_enforcer/_hosts_protection.py
View File

@ -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."
)

42
steam_backlog_enforcer/_leetcode_bonus.py
View File

@ -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:

25
steam_backlog_enforcer/_polls_reporting.py
View File

@ -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(

44
steam_backlog_enforcer/_scanning_assign.py
View File

@ -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):

88
steam_backlog_enforcer/_status_api.py Normal file
View File

@ -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

10
steam_backlog_enforcer/_total_block_hosts.py
View File

@ -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__)

13
steam_backlog_enforcer/_whitelist.py
View File

@ -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
# ──────────────────────────────────────────────────────────────

10
steam_backlog_enforcer/_whitelist_locking.py
View File

@ -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

56
steam_backlog_enforcer/_workout_budget.py
View File

@ -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)

102
steam_backlog_enforcer/hltb.py
View File

@ -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,

26
steam_backlog_enforcer/scanning.py
View File

@ -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)

33
steam_backlog_enforcer/store_blocker.py
View File

@ -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):
return rewrite_hosts_lines(
_comment_out_blocked, "Commented out Steam Store entries in /etc/hosts."
)
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)
):
new_lines.append(f"# {line}" if line.endswith("\n") else f"# {line}\n")
changed = True
else:
new_lines.append(line)
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
return f"# {line}" if line.endswith("\n") else f"# {line}\n"
return None

3
steam_backlog_enforcer/tests/test_hltb.py
View File

@ -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

2
steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py
View File

@ -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"

13
steam_backlog_enforcer/tests/test_hltb_fetch_hltb_detail_missing.py
View File

@ -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]),
):

16
steam_backlog_enforcer/tests/test_hltb_part2.py
View File

@ -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(

2
steam_backlog_enforcer/tests/test_hltb_search.py
View File

@ -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."""

7
steam_backlog_enforcer/tests/test_leetcode_bonus.py
View File

@ -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

7
steam_backlog_enforcer/tests/test_polls_tracking_backfill_polls_for_finished.py
View File

@ -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]))

5
steam_backlog_enforcer/tests/test_polls_tracking_part2.py
View File

@ -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,
),
):

5
steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py
View File

@ -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,

23
steam_backlog_enforcer/tests/test_store_blocker_part2.py
View File

@ -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

4
steam_backlog_enforcer/tests/test_workout_budget.py
View File

@ -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())