diff --git a/README.md b/README.md index 49f029e..8e845b0 100644 --- a/README.md +++ b/README.md @@ -33,7 +33,8 @@ python -m python_pkg.steam_backlog_enforcer.main setup | `list` | List incomplete games from snapshot | | `skip` | Skip the currently assigned game | | `enforce` | Run enforcer (block, uninstall, kill) | -| `unblock` | Remove store blocking | +| `unblock [minutes]` | Unblock the store for a window (default 15, max 30); the daemon keeps it open and re-blocks after | +| `buy-dlc` | Same 15-minute store window, to buy a game or DLC | | `reset` | Reset all state | | `installed` | List currently installed Steam games | | `uninstall` | Interactively uninstall non-assigned games | diff --git a/steam_backlog_enforcer/_enforce_loop.py b/steam_backlog_enforcer/_enforce_loop.py index 9670ee2..6d942cd 100644 --- a/steam_backlog_enforcer/_enforce_loop.py +++ b/steam_backlog_enforcer/_enforce_loop.py @@ -22,6 +22,7 @@ from steam_backlog_enforcer._pick_completion import ( from steam_backlog_enforcer._playtime import playtime_tick from steam_backlog_enforcer._playtime_session import PlaytimeSession, new_session from steam_backlog_enforcer._steam_launch import steam_is_installed +from steam_backlog_enforcer._store_window import store_window_tick from steam_backlog_enforcer._total_block import ( end_total_block_cleanup, enforce_total_block_tick, @@ -132,6 +133,10 @@ def _enforce_loop_iteration( if total_block_needs_cleanup(): end_total_block_cleanup() + # Before the Steam-installed guard: an expired window must re-block even + # when Steam is gone. + store_window_tick(config, state) + # With no Steam client there is no library, no installs and no game # processes, so every branch below is a no-op at best - and at worst a # 3s-interval error loop trying to write manifests into a steamapps diff --git a/steam_backlog_enforcer/_scanning_tampering.py b/steam_backlog_enforcer/_scanning_tampering.py index 59df8c4..18b707d 100644 --- a/steam_backlog_enforcer/_scanning_tampering.py +++ b/steam_backlog_enforcer/_scanning_tampering.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any from steam_backlog_enforcer._allowed_games import allowed_games from steam_backlog_enforcer._assignment_progress import iso_to_epoch from steam_backlog_enforcer._snapshot import load_snapshot +from steam_backlog_enforcer._whitelist_locking import get_approved_exception_ids from steam_backlog_enforcer.enforcer import send_notification from steam_backlog_enforcer.game_install import _echo from steam_backlog_enforcer.steam_api import SteamAPIClient @@ -32,7 +33,8 @@ def _legitimately_played(state: State) -> set[int]: lands in ``finished_app_ids`` -- but the achievements that finished it were earned legitimately, so comparing it against a stale snapshot would report the user's own completion as tampering. Every game ever manually picked, - plus every finished game, is therefore exempt. + plus every finished game and every approved whitelist exception, is therefore + exempt. Args: state: Current enforcer state. @@ -42,6 +44,7 @@ def _legitimately_played(state: State) -> set[int]: """ exempt = {app_id for app_id, _ in allowed_games(state)} exempt.update(state.finished_app_ids) + exempt.update(get_approved_exception_ids()) exempt.update( pick["app_id"] for pick in state.manual_picks if pick.get("app_id") is not None ) diff --git a/steam_backlog_enforcer/_store_tools.py b/steam_backlog_enforcer/_store_tools.py index cddb863..5de8e9e 100644 --- a/steam_backlog_enforcer/_store_tools.py +++ b/steam_backlog_enforcer/_store_tools.py @@ -11,19 +11,11 @@ from pathlib import Path import shutil import subprocess -# Path to the hosts install script. _REPO_ROOT resolves to $HOME (this -# module lives two levels below it); the script itself is in the -# linux_configuration checkout under testsAndMisc, not directly under $HOME. -_REPO_ROOT = Path(__file__).resolve().parents[2] -HOSTS_INSTALL_SCRIPT = ( - _REPO_ROOT - / "testsAndMisc" - / "linux_configuration" - / "scripts" - / "periodic_background" - / "hosts" - / "install.sh" -) +# Path to the hosts install script. _SRC_ROOT is ~/src (this module lives two +# levels below it); the generator moved out of testsAndMisc into its own +# hosts-blocker repo, whose custom_entries.hosts carries the Steam Store block. +_SRC_ROOT = Path(__file__).resolve().parents[2] +HOSTS_INSTALL_SCRIPT = _SRC_ROOT / "hosts-blocker" / "install.sh" # iptables chain name for our blocking rules. IPTABLES_CHAIN = "STEAM_ENFORCER" diff --git a/steam_backlog_enforcer/_store_window.py b/steam_backlog_enforcer/_store_window.py new file mode 100644 index 0000000..b929528 --- /dev/null +++ b/steam_backlog_enforcer/_store_window.py @@ -0,0 +1,103 @@ +"""A bounded window in which the Steam store stays unblocked. + +``run.sh unblock`` used to comment the store lines out of ``/etc/hosts`` once +and walk away. Anything that reinstalls the hosts file -- the hourly +``periodic-system-maintenance`` job, any other ``hosts-blocker/install.sh`` run +-- put the block straight back, sometimes within seconds, so a purchase could +not be finished. The window moves that job into the daemon: until the +deadline it re-applies the unblock whenever the block reappears, and after it +re-blocks exactly once. + +Only the CLI writes the deadline (``State.store_unblocked_until``). The daemon +runs as root and never saves state for this: a root-written ``state.json`` +would lock the unprivileged CLI out of its own file. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import logging +from typing import TYPE_CHECKING + +from steam_backlog_enforcer.store_blocker import ( + block_store, + hosts_blocks_store, + unblock_store, +) + +if TYPE_CHECKING: + from steam_backlog_enforcer.config import Config, State + +logger = logging.getLogger(__name__) + +DEFAULT_WINDOW_MINUTES = 15 +MAX_WINDOW_MINUTES = 30 + +# Deadlines this daemon process has already re-blocked for. In memory on +# purpose (see module docstring); a restarted daemon re-blocks at startup anyway. +_reblocked_for: set[str] = set() + + +def open_store_window(state: State, minutes: int) -> datetime: + """Record a store window ending *minutes* from now and unblock at once. + + Args: + state: Enforcer state; saved with the new deadline. + minutes: Window length, 1 to ``MAX_WINDOW_MINUTES``. + + Returns: + The UTC deadline. + + Raises: + ValueError: If *minutes* is out of range. + RuntimeError: If the store could not be unblocked right now. + """ + if not 1 <= minutes <= MAX_WINDOW_MINUTES: + msg = f"minutes must be between 1 and {MAX_WINDOW_MINUTES}, got {minutes}" + raise ValueError(msg) + until = datetime.now(UTC) + timedelta(minutes=minutes) + state.store_unblocked_until = until.isoformat() + state.save() + if not unblock_store(): + msg = "could not unblock the store (sudo?)" + raise RuntimeError(msg) + return until + + +def _deadline(state: State) -> datetime | None: + """The window's deadline, or ``None`` when unset or unparsable.""" + raw = state.store_unblocked_until + if not raw: + return None + try: + return datetime.fromisoformat(raw) + except ValueError: + logger.warning("Ignoring malformed store_unblocked_until %r", raw) + return None + + +def store_window_tick(config: Config, state: State) -> None: + """Keep the store open inside the window; re-block once it has passed. + + Runs every daemon iteration. Outside a window it only parses one string, + so it costs nothing; inside one it reads ``/etc/hosts`` (no subprocess) + and shells out only when the block has actually come back. + + Args: + config: Enforcer configuration (``block_store`` gates the re-block). + state: Freshly reloaded enforcer state. + """ + until = _deadline(state) + if until is None: + return + if datetime.now(UTC) < until: + if hosts_blocks_store(): + logger.info("Store window open until %s: block came back, undoing", until) + unblock_store() + return + if state.store_unblocked_until in _reblocked_for: + return + _reblocked_for.add(state.store_unblocked_until) + if config.block_store: + logger.info("Store window ended at %s: re-blocking the store", until) + block_store() diff --git a/steam_backlog_enforcer/config.py b/steam_backlog_enforcer/config.py index 2cf1021..7ceed01 100644 --- a/steam_backlog_enforcer/config.py +++ b/steam_backlog_enforcer/config.py @@ -191,6 +191,8 @@ class State: """``str(app_id)`` → ISO time of its latest assignment (pick ordering).""" released_at: dict[str, str] = field(default_factory=dict) """``str(app_id)`` → ISO time it was released below 100% (tampering).""" + store_unblocked_until: str = "" + """ISO deadline of the open store window (``unblock``); CLI-written only.""" def skip_for_days(self, app_id: int, days: int) -> None: """Mark ``app_id`` as skipped for ``days`` days from now (UTC).""" diff --git a/steam_backlog_enforcer/library_hider.py b/steam_backlog_enforcer/library_hider.py index a6d7e66..7d9debc 100644 --- a/steam_backlog_enforcer/library_hider.py +++ b/steam_backlog_enforcer/library_hider.py @@ -32,6 +32,7 @@ from steam_backlog_enforcer._steam_launch import ( restart_steam, steam_is_installed, ) +from steam_backlog_enforcer._whitelist_locking import get_approved_exception_ids if TYPE_CHECKING: from steam_backlog_enforcer.config import Config, State @@ -81,10 +82,15 @@ def hide_other_games( cover games that might not yet appear in ``visibleApps`` due to stale MobX state. + Approved whitelist exceptions (``add-exception``) are always treated as + allowed, so a permanently whitelisted game such as a co-op title stays + visible next to the assignment instead of being hidden with the rest. + Returns the total number of games hidden across all passes. """ ensure_steam_debug_port() + allowed_app_ids = allowed_app_ids | get_approved_exception_ids() allowed_json = json.dumps(sorted(allowed_app_ids)) extra_ids = sorted(aid for aid in owned_app_ids if aid not in allowed_app_ids) extra_json = json.dumps(extra_ids) diff --git a/steam_backlog_enforcer/main/__init__.py b/steam_backlog_enforcer/main/__init__.py index 694860e..87f4d88 100644 --- a/steam_backlog_enforcer/main/__init__.py +++ b/steam_backlog_enforcer/main/__init__.py @@ -125,8 +125,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = { "install": ("Install the assigned game", cmd_install), "hide": ("Hide all non-assigned games in library", cmd_hide), "unhide": ("Unhide all games in library", cmd_unhide), - "unblock": ("Remove store blocking", cmd_unblock), - "buy-dlc": ("Temporarily unblock store to buy DLC", cmd_buy_dlc), + "buy-dlc": ("Unblock the store for 15 min to buy a game/DLC", cmd_buy_dlc), "reset": ("Reset all state", cmd_reset), "installed": ("List installed games", cmd_installed), "uninstall": ("Uninstall all non-assigned games", cmd_uninstall), @@ -141,6 +140,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = { # Extra commands with non-standard arg handling (shown in help but not in COMMANDS). _EXTRA_COMMAND_DESCRIPTIONS: dict[str, str] = { "add-exception": "Request 24h-locked whitelist exception (use --reason)", + "unblock": "Unblock the store for [minutes] (default 15, max 30)", "serve": "Start the web UI (--port N; replaces a stale server)", "pick-manual": f"Pick a game by app_id, lock enforcer for {_MANUAL_LOCK_DAYS} days", "abandon-pick": "Undo a manual pick at any time (needs app_id)", @@ -179,26 +179,24 @@ def _dispatch_extra_command(command: str, config: Config, state: State) -> bool: Returns: True if *command* was handled here. """ - if command == "add-exception": - cmd_add_exception(sys.argv[2:]) - return True - if command == "block-gaming": - cmd_block_gaming(sys.argv[2:]) - return True - if command == "serve": - cmd_serve(sys.argv[2:]) - return True - if command == "pick-manual": - cmd_pick_manual(config, state, sys.argv[2:]) - return True - if command == "abandon-pick": - cmd_abandon_pick(config, state, sys.argv[2:]) - return True + args = sys.argv[2:] if command == "enforce": - sys.exit(cmd_enforce(config, state, sys.argv[2:])) + sys.exit(cmd_enforce(config, state, args)) if command == "gaming-unblock": - sys.exit(cmd_gaming_unblock(sys.argv[2:])) - return False + sys.exit(cmd_gaming_unblock(args)) + handlers: dict[str, Callable[[], object]] = { + "add-exception": lambda: cmd_add_exception(args), + "unblock": lambda: cmd_unblock(config, state, args), + "block-gaming": lambda: cmd_block_gaming(args), + "serve": lambda: cmd_serve(args), + "pick-manual": lambda: cmd_pick_manual(config, state, args), + "abandon-pick": lambda: cmd_abandon_pick(config, state, args), + } + handler = handlers.get(command) + if handler is None: + return False + handler() + return True def main() -> None: diff --git a/steam_backlog_enforcer/main/misc.py b/steam_backlog_enforcer/main/misc.py index cd1a0ab..bb8ae18 100644 --- a/steam_backlog_enforcer/main/misc.py +++ b/steam_backlog_enforcer/main/misc.py @@ -16,14 +16,19 @@ from steam_backlog_enforcer._serve_startup import ( ensure_port_available, parse_serve_args, ) +from steam_backlog_enforcer._store_window import ( + DEFAULT_WINDOW_MINUTES, + MAX_WINDOW_MINUTES, + open_store_window, +) from steam_backlog_enforcer._total_block import start_total_block from steam_backlog_enforcer._web_build import build_frontend, frontend_is_stale from steam_backlog_enforcer._web_server import serve from steam_backlog_enforcer._whitelist import validate_reason from steam_backlog_enforcer._whitelist_locking import add_pending_exception from steam_backlog_enforcer.game_install import _echo -from steam_backlog_enforcer.library_hider import restart_steam, unhide_all_games -from steam_backlog_enforcer.store_blocker import block_store, unblock_store +from steam_backlog_enforcer.library_hider import unhide_all_games +from steam_backlog_enforcer.store_blocker import unblock_store if TYPE_CHECKING: from steam_backlog_enforcer.config import Config, State @@ -48,38 +53,34 @@ _BLOCK_GAMING_USAGE = ( ) -def cmd_unblock(_config: Config, _state: State) -> None: - """Remove store blocking.""" - if unblock_store(): - _echo("Steam store unblocked.") - else: - _echo("Failed to unblock. Run with sudo.") +def cmd_unblock(_config: Config, state: State, args: list[str] | None = None) -> None: + """Open a timed store window: unblocked now, re-blocked by the daemon after. + + Usage: unblock [minutes] (default 15, max 30) + + Args: + _config: Unused; kept for the shared command signature. + state: Enforcer state; receives the window deadline. + args: CLI argument list after the command name. + """ + raw = args[0] if args else str(DEFAULT_WINDOW_MINUTES) + try: + until = open_store_window(state, int(raw)) + except ValueError as exc: + _echo(f"Usage: unblock [minutes] (1-{MAX_WINDOW_MINUTES}): {exc}") + sys.exit(1) + except RuntimeError as exc: + _echo(f"Failed to unblock: {exc}") + sys.exit(1) + local = until.astimezone().strftime("%H:%M") + _echo(f"Steam store UNBLOCKED until {local}.") + _echo("The enforcer daemon keeps it open until then (even if a hosts") + _echo("reinstall re-blocks it) and re-blocks it afterwards.") def cmd_buy_dlc(config: Config, state: State) -> None: - """Temporarily unblock the store so the user can buy DLC.""" - if state.current_app_id is None: - _echo("No game currently assigned.") - return - - _echo(f"Current game: {state.current_game_name} (AppID={state.current_app_id})") - _echo("Unblocking Steam store for DLC purchase...") - - if not unblock_store(): - _echo("Failed to unblock store. Run with sudo.") - return - - _echo("\nStore UNBLOCKED — buy your DLC now.") - _echo("Press Enter when you're done to re-block the store...") - input() - - if config.block_store: - if block_store(): - _echo("Store re-blocked. Restarting Steam to clear DNS cache...") - restart_steam() - _echo("Done.") - else: - _echo("Warning: failed to re-block store.") + """Open the default-length store window to buy a game or DLC.""" + cmd_unblock(config, state, []) def cmd_reset(config: Config, state: State) -> None: diff --git a/steam_backlog_enforcer/store_blocker.py b/steam_backlog_enforcer/store_blocker.py index e4ae399..9a78257 100644 --- a/steam_backlog_enforcer/store_blocker.py +++ b/steam_backlog_enforcer/store_blocker.py @@ -39,7 +39,12 @@ logger = logging.getLogger(__name__) def is_store_blocked() -> bool: - """Check if Steam Store domains are blocked in /etc/hosts.""" + """Check if Steam Store domains are blocked in /etc/hosts or iptables.""" + return hosts_blocks_store() or _is_iptables_blocked() + + +def hosts_blocks_store() -> bool: + """Check /etc/hosts alone: a file read, no subprocess, safe to poll.""" try: content = HOSTS_FILE.read_text(encoding="utf-8") # Check for at least the primary store domain. @@ -55,8 +60,7 @@ def is_store_blocked() -> bool: return True except OSError: pass - - return _is_iptables_blocked() + return False def block_store() -> bool: diff --git a/steam_backlog_enforcer/tests/test_budget_view.py b/steam_backlog_enforcer/tests/test_budget_view.py index 292a61c..9cbd4b2 100644 --- a/steam_backlog_enforcer/tests/test_budget_view.py +++ b/steam_backlog_enforcer/tests/test_budget_view.py @@ -28,18 +28,24 @@ from steam_backlog_enforcer.config import Config # effect yet -- without this patch, collecting this module makes a real HTTP # call to the screen locker and pins RULES to whatever budget today happens to # have earned. Stub it explicitly so the constant is the earned budget always. -with patch( - "steam_backlog_enforcer._playtime_state.resolve_budget", - side_effect=lambda config: BudgetResolution( - seconds=float(config.base_gaming_seconds) - + float(config.workout_bonus_seconds) - + float(config.leetcode_bonus_seconds) - + float(config.reading_bonus_seconds), - base_seconds=float(config.base_gaming_seconds), - workout_seconds=float(config.workout_bonus_seconds), - leetcode_seconds=float(config.leetcode_bonus_seconds), - reason="stubbed: fully earned", - reading_seconds=float(config.reading_bonus_seconds), +# The carry is pinned for the same reason: conftest's redirects are not active +# at import either, so rules_for would read the LIVE playtime_state.json and +# fold whatever Fri-Mon carry it holds today into RULES. +with ( + patch("steam_backlog_enforcer._playtime_state.carry_into", return_value=0.0), + patch( + "steam_backlog_enforcer._playtime_state.resolve_budget", + side_effect=lambda config: BudgetResolution( + seconds=float(config.base_gaming_seconds) + + float(config.workout_bonus_seconds) + + float(config.leetcode_bonus_seconds) + + float(config.reading_bonus_seconds), + base_seconds=float(config.base_gaming_seconds), + workout_seconds=float(config.workout_bonus_seconds), + leetcode_seconds=float(config.leetcode_bonus_seconds), + reason="stubbed: fully earned", + reading_seconds=float(config.reading_bonus_seconds), + ), ), ): RULES = rules_for(Config(), demo=False) diff --git a/steam_backlog_enforcer/tests/test_exception_visibility.py b/steam_backlog_enforcer/tests/test_exception_visibility.py new file mode 100644 index 0000000..f36f344 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_exception_visibility.py @@ -0,0 +1,62 @@ +"""Approved whitelist exceptions stay visible and never count as tampering.""" + +from __future__ import annotations + +import json +import re +from unittest.mock import patch + +from steam_backlog_enforcer._scanning_tampering import _legitimately_played +from steam_backlog_enforcer.config import State +from steam_backlog_enforcer.library_hider import hide_other_games + +PKG = "steam_backlog_enforcer.library_hider" +_EXCEPTION = 3126150 +_ASSIGNED = 1091500 +_OTHER = 10 + + +def _hide_and_capture(allowed: set[int]) -> tuple[list[int], list[int]]: + """Run hide_other_games against a fake Steam; return (visible, to_hide).""" + captured: dict[str, str] = {} + + def fake_eval(js: str) -> dict[str, object]: + captured["js"] = js + return {} + + with ( + patch(f"{PKG}.ensure_steam_debug_port"), + patch(f"{PKG}._evaluate_js", side_effect=fake_eval), + patch(f"{PKG}._cdp_result_value", return_value='{"totalHidden": 0}'), + patch( + f"{PKG}.get_approved_exception_ids", + return_value=frozenset({_EXCEPTION}), + ), + ): + hide_other_games([_EXCEPTION, _ASSIGNED, _OTHER], allowed) + visible = re.search(r"new Set\((\[.*?\])\)", captured["js"]) + to_hide = re.search(r"const extraIds = (\[.*?\]);", captured["js"]) + assert visible is not None + assert to_hide is not None + return json.loads(visible.group(1)), json.loads(to_hide.group(1)) + + +def test_exception_is_kept_visible() -> None: + visible, to_hide = _hide_and_capture({_ASSIGNED}) + assert visible == [_ASSIGNED, _EXCEPTION] + assert to_hide == [_OTHER] + + +def test_caller_set_is_not_mutated() -> None: + allowed = {_ASSIGNED} + _hide_and_capture(allowed) + assert allowed == {_ASSIGNED} + + +def test_exception_is_exempt_from_tampering() -> None: + with patch( + "steam_backlog_enforcer._scanning_tampering.get_approved_exception_ids", + return_value=frozenset({_EXCEPTION}), + ): + exempt = _legitimately_played(State(current_app_id=_ASSIGNED)) + assert {_EXCEPTION, _ASSIGNED} <= exempt diff --git a/steam_backlog_enforcer/tests/test_main_dispatch_extra.py b/steam_backlog_enforcer/tests/test_main_dispatch_extra.py index 9634270..fe26a41 100644 --- a/steam_backlog_enforcer/tests/test_main_dispatch_extra.py +++ b/steam_backlog_enforcer/tests/test_main_dispatch_extra.py @@ -47,6 +47,19 @@ class TestMainDispatchServe: mock_cmd.assert_called_once_with(["--port", "8123"]) +class TestMainDispatchUnblock: + def test_dispatches_unblock_with_minutes(self) -> None: + argv = ["prog", "unblock", "10"] + with ( + patch.object(sys, "argv", argv), + patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")), + patch(f"{PKG}.State.load", return_value=State()), + patch(f"{PKG}.cmd_unblock") as mock_cmd, + ): + main() + assert mock_cmd.call_args.args[2] == ["10"] + + class TestMainDispatchPickManual: def test_dispatches_pick_manual(self) -> None: argv = ["prog", "pick-manual", "489830"] diff --git a/steam_backlog_enforcer/tests/test_main_misc.py b/steam_backlog_enforcer/tests/test_main_misc.py index ef0a74c..b47c7ab 100644 --- a/steam_backlog_enforcer/tests/test_main_misc.py +++ b/steam_backlog_enforcer/tests/test_main_misc.py @@ -1,7 +1,11 @@ """Tests for the main CLI: store, reset, setup, exception and block-gaming commands.""" +from datetime import UTC, datetime from unittest.mock import patch +import pytest + +from steam_backlog_enforcer._store_window import DEFAULT_WINDOW_MINUTES from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.main import ( cmd_buy_dlc, @@ -11,76 +15,59 @@ from steam_backlog_enforcer.main import ( ) PKG = "steam_backlog_enforcer.main.misc" +_UNTIL = datetime(2026, 10, 3, 12, 0, tzinfo=UTC) +_FIVE_MINUTES = 5 class TestCmdUnblock: - """Tests for cmd_unblock.""" + """Tests for cmd_unblock: it opens a timed store window.""" - def test_success(self) -> None: + def test_default_window(self) -> None: with ( - patch(f"{PKG}.unblock_store", return_value=True), - patch(f"{PKG}._echo"), - ): - cmd_unblock(Config(), State()) - - def test_fail(self) -> None: - with ( - patch(f"{PKG}.unblock_store", return_value=False), + patch(f"{PKG}.open_store_window", return_value=_UNTIL) as mock_open, patch(f"{PKG}._echo") as mock_echo, ): cmd_unblock(Config(), State()) - assert any("Failed" in str(c) for c in mock_echo.call_args_list) + assert mock_open.call_args.args[1] == DEFAULT_WINDOW_MINUTES + assert any("UNBLOCKED until" in str(c) for c in mock_echo.call_args_list) + + def test_explicit_minutes(self) -> None: + with ( + patch(f"{PKG}.open_store_window", return_value=_UNTIL) as mock_open, + patch(f"{PKG}._echo"), + ): + cmd_unblock(Config(), State(), ["5"]) + assert mock_open.call_args.args[1] == _FIVE_MINUTES + + @pytest.mark.parametrize("raw", ["abc", "0", "31"]) + def test_bad_minutes_exits(self, raw: str) -> None: + with ( + patch(f"{PKG}._echo") as mock_echo, + pytest.raises(SystemExit), + ): + cmd_unblock(Config(), State(), [raw]) + assert any("Usage" in str(c) for c in mock_echo.call_args_list) + + def test_unblock_failure_exits(self) -> None: + with ( + patch(f"{PKG}.open_store_window", side_effect=RuntimeError("sudo?")), + patch(f"{PKG}._echo") as mock_echo, + pytest.raises(SystemExit), + ): + cmd_unblock(Config(), State()) + assert any("Failed" in str(c) for c in mock_echo.call_args_list) class TestCmdBuyDlc: - """Tests for cmd_buy_dlc.""" + """cmd_buy_dlc is the default-length window, assigned game or not.""" - def test_no_game(self) -> None: - with patch(f"{PKG}._echo") as mock_echo: + def test_opens_default_window(self) -> None: + with ( + patch(f"{PKG}.open_store_window", return_value=_UNTIL) as mock_open, + patch(f"{PKG}._echo"), + ): cmd_buy_dlc(Config(), State()) - assert any("No game" in str(c) for c in mock_echo.call_args_list) - - def test_unblock_fails(self) -> None: - state = State(current_app_id=1, current_game_name="G") - with ( - patch(f"{PKG}.unblock_store", return_value=False), - patch(f"{PKG}._echo"), - ): - cmd_buy_dlc(Config(), state) - - def test_success_reblock(self) -> None: - state = State(current_app_id=1, current_game_name="G") - config = Config(block_store=True) - with ( - patch(f"{PKG}.unblock_store", return_value=True), - patch(f"{PKG}.block_store", return_value=True), - patch(f"{PKG}.restart_steam"), - patch(f"{PKG}._echo"), - patch("builtins.input", return_value=""), - ): - cmd_buy_dlc(config, state) - - def test_reblock_fails(self) -> None: - state = State(current_app_id=1, current_game_name="G") - config = Config(block_store=True) - with ( - patch(f"{PKG}.unblock_store", return_value=True), - patch(f"{PKG}.block_store", return_value=False), - patch(f"{PKG}._echo") as mock_echo, - patch("builtins.input", return_value=""), - ): - cmd_buy_dlc(config, state) - assert any("Warning" in str(c) for c in mock_echo.call_args_list) - - def test_no_reblock(self) -> None: - state = State(current_app_id=1, current_game_name="G") - config = Config(block_store=False) - with ( - patch(f"{PKG}.unblock_store", return_value=True), - patch(f"{PKG}._echo"), - patch("builtins.input", return_value=""), - ): - cmd_buy_dlc(config, state) + assert mock_open.call_args.args[1] == DEFAULT_WINDOW_MINUTES class TestCmdReset: diff --git a/steam_backlog_enforcer/tests/test_store_window.py b/steam_backlog_enforcer/tests/test_store_window.py new file mode 100644 index 0000000..64378f1 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_store_window.py @@ -0,0 +1,110 @@ +"""Tests for the timed store window (``unblock [minutes]`` + daemon tick).""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from steam_backlog_enforcer import _store_window +from steam_backlog_enforcer._store_window import ( + MAX_WINDOW_MINUTES, + open_store_window, + store_window_tick, +) +from steam_backlog_enforcer.config import Config, State + +if TYPE_CHECKING: + from collections.abc import Iterator + +PKG = "steam_backlog_enforcer._store_window" + + +@pytest.fixture(autouse=True) +def _fresh_reblock_memory() -> Iterator[None]: + """Each test starts with a daemon that has re-blocked for nothing yet.""" + with patch.object(_store_window, "_reblocked_for", set()): + yield + + +def _state_ending_in(delta: timedelta) -> State: + return State(store_unblocked_until=(datetime.now(UTC) + delta).isoformat()) + + +class TestOpenStoreWindow: + @pytest.mark.parametrize("minutes", [0, MAX_WINDOW_MINUTES + 1]) + def test_rejects_out_of_range(self, minutes: int) -> None: + state = State() + with ( + patch(f"{PKG}.unblock_store") as mock_unblock, + pytest.raises(ValueError, match="between 1 and"), + ): + open_store_window(state, minutes) + mock_unblock.assert_not_called() + assert not state.store_unblocked_until + + def test_saves_deadline_and_unblocks(self) -> None: + state = State() + before = datetime.now(UTC) + with patch(f"{PKG}.unblock_store", return_value=True) as mock_unblock: + until = open_store_window(state, 10) + mock_unblock.assert_called_once() + assert before + timedelta(minutes=10) <= until + assert until <= datetime.now(UTC) + timedelta(minutes=10) + assert State.load().store_unblocked_until == until.isoformat() + + def test_failed_unblock_raises(self) -> None: + with ( + patch(f"{PKG}.unblock_store", return_value=False), + pytest.raises(RuntimeError, match="could not unblock"), + ): + open_store_window(State(), 5) + + +class TestStoreWindowTick: + def test_no_window_does_nothing(self) -> None: + with ( + patch(f"{PKG}.hosts_blocks_store") as mock_check, + patch(f"{PKG}.block_store") as mock_block, + ): + store_window_tick(Config(), State()) + mock_check.assert_not_called() + mock_block.assert_not_called() + + def test_malformed_deadline_is_ignored(self) -> None: + with patch(f"{PKG}.block_store") as mock_block: + store_window_tick(Config(), State(store_unblocked_until="garbage")) + mock_block.assert_not_called() + + def test_open_window_undoes_a_reblock(self) -> None: + state = _state_ending_in(timedelta(minutes=5)) + with ( + patch(f"{PKG}.hosts_blocks_store", return_value=True), + patch(f"{PKG}.unblock_store") as mock_unblock, + ): + store_window_tick(Config(), state) + mock_unblock.assert_called_once() + + def test_open_window_leaves_open_store_alone(self) -> None: + state = _state_ending_in(timedelta(minutes=5)) + with ( + patch(f"{PKG}.hosts_blocks_store", return_value=False), + patch(f"{PKG}.unblock_store") as mock_unblock, + ): + store_window_tick(Config(), state) + mock_unblock.assert_not_called() + + def test_expired_window_reblocks_exactly_once(self) -> None: + state = _state_ending_in(timedelta(seconds=-1)) + with patch(f"{PKG}.block_store") as mock_block: + store_window_tick(Config(block_store=True), state) + store_window_tick(Config(block_store=True), state) + mock_block.assert_called_once() + + def test_expired_window_respects_block_store_off(self) -> None: + state = _state_ending_in(timedelta(seconds=-1)) + with patch(f"{PKG}.block_store") as mock_block: + store_window_tick(Config(block_store=False), state) + mock_block.assert_not_called()