diff --git a/steam_backlog_enforcer/_enforce_loop.py b/steam_backlog_enforcer/_enforce_loop.py index e2f328f..a3ba805 100644 --- a/steam_backlog_enforcer/_enforce_loop.py +++ b/steam_backlog_enforcer/_enforce_loop.py @@ -15,7 +15,6 @@ from steam_backlog_enforcer._total_block import ( ) from steam_backlog_enforcer._whitelist import ( lock_enforcement_files, - promote_pending_exceptions, ) from steam_backlog_enforcer.config import ( CONFIG_DIR, @@ -296,12 +295,7 @@ def _enforce_loop_iteration(config: Config, state: State) -> None: config.steam_id, ) - # D) Promote any cooldown-elapsed pending exceptions to approved. - newly_approved = promote_pending_exceptions() - for aid in newly_approved: - logger.info("Exception approved: AppID=%d", aid) - - # E) Re-apply immutable flag so config cannot be edited without root. + # D) Re-apply immutable flag so config cannot be edited without root. lock_enforcement_files(CONFIG_FILE) diff --git a/steam_backlog_enforcer/_whitelist.py b/steam_backlog_enforcer/_whitelist.py index 7b4519a..b308b6a 100644 --- a/steam_backlog_enforcer/_whitelist.py +++ b/steam_backlog_enforcer/_whitelist.py @@ -24,7 +24,6 @@ logger = logging.getLogger(__name__) # File paths (patched in tests via conftest) # ────────────────────────────────────────────────────────────── -PENDING_EXCEPTIONS_FILE: Path = CONFIG_DIR / "pending_exceptions.json" APPROVED_EXCEPTIONS_FILE: Path = CONFIG_DIR / "approved_exceptions.json" EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log" @@ -32,8 +31,6 @@ EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log" # Constants # ────────────────────────────────────────────────────────────── -WHITELIST_COOLDOWN_SECONDS: int = 86400 # 24 hours - _MIN_REASON_WORDS: int = 5 _MIN_REASON_LENGTH: int = 25 _MIN_ENTROPY: float = 3.0 @@ -169,24 +166,6 @@ def unlock_for_write(path: Path) -> None: # ────────────────────────────────────────────────────────────── -def _load_pending() -> list[dict[str, object]]: - """Load pending exception entries from disk.""" - if not PENDING_EXCEPTIONS_FILE.exists(): - return [] - try: - data: object = json.loads(PENDING_EXCEPTIONS_FILE.read_text(encoding="utf-8")) - if isinstance(data, list): - return cast("list[dict[str, object]]", data) - except (json.JSONDecodeError, OSError, ValueError): - pass - return [] - - -def _save_pending(entries: list[dict[str, object]]) -> None: - """Persist pending exception entries to disk.""" - _atomic_write(PENDING_EXCEPTIONS_FILE, json.dumps(entries, indent=2) + "\n") - - def _load_approved() -> list[dict[str, object]]: """Load approved exception entries from disk.""" if not APPROVED_EXCEPTIONS_FILE.exists(): @@ -232,17 +211,17 @@ def _append_audit_log(app_id: int, reason: str, event: str) -> None: def add_pending_exception(app_id: int, reason: str) -> str: - """Request a new whitelist exception for *app_id*. + """Add a whitelist exception for *app_id* immediately. - The entry becomes active only after ``WHITELIST_COOLDOWN_SECONDS`` have - elapsed (24 h by default). Returns a human-readable status message. + The entry becomes active right away (no cooldown). Returns a + human-readable status message. Args: app_id: Steam application ID to add. reason: Validated justification text (must pass :func:`validate_reason`). Returns: - Human-readable confirmation or remaining-cooldown message. + Human-readable confirmation message. Raises: ValueError: If the reason fails validation or the ID is already approved. @@ -256,92 +235,25 @@ def add_pending_exception(app_id: int, reason: str) -> str: msg = f"AppID {app_id} is already in the approved exceptions list." raise ValueError(msg) - pending = _load_pending() - for entry in pending: - if int(entry["app_id"]) == app_id: - elapsed = time.time() - float(entry["requested_at"]) - remaining = WHITELIST_COOLDOWN_SECONDS - elapsed - if remaining > 0: - hours = int(remaining // 3600) - mins = int((remaining % 3600) // 60) - return ( - f"AppID {app_id} is already pending; approves in {hours}h {mins}m." - ) - # Cooldown already elapsed for this pending entry — promote now. - break - - entry_new: dict[str, object] = { - "app_id": app_id, - "reason": reason, - "requested_at": time.time(), - } - pending.append(entry_new) - _save_pending(pending) - _append_audit_log(app_id, reason, "REQUESTED") - - hours = WHITELIST_COOLDOWN_SECONDS // 3600 - return ( - f"Exception requested for AppID {app_id}. " - f"Will become active in {hours}h. Reason logged." - ) - - -def promote_pending_exceptions() -> list[int]: - """Move cooldown-elapsed pending entries to the approved list. - - Called by the enforce daemon on each loop iteration. Returns the list - of app IDs that were promoted this call. - - Returns: - List of newly approved app IDs (may be empty). - """ - pending = _load_pending() now = time.time() - still_pending: list[dict[str, object]] = [] - newly_approved: list[int] = [] + approved.append( + { + "app_id": app_id, + "reason": reason, + "approved_at": now, + } + ) + _save_approved(approved) + _append_audit_log(app_id, reason, "APPROVED") - for entry in pending: - elapsed = now - float(entry["requested_at"]) - if elapsed >= WHITELIST_COOLDOWN_SECONDS: - app_id = int(entry["app_id"]) - approved = _load_approved() - if not any(int(e["app_id"]) == app_id for e in approved): - approved.append( - { - "app_id": app_id, - "reason": entry["reason"], - "approved_at": now, - } - ) - _save_approved(approved) - _append_audit_log(app_id, str(entry["reason"]), "APPROVED") - newly_approved.append(app_id) - else: - still_pending.append(entry) - - if len(still_pending) != len(pending): - _save_pending(still_pending) - - return newly_approved + return f"Exception approved for AppID {app_id}. Active immediately. Reason logged." def get_approved_exception_ids() -> frozenset[int]: """Return the frozenset of currently approved exception app IDs. - Does NOT trigger promotion — call :func:`promote_pending_exceptions` - explicitly when timely promotion is required (e.g. the enforce loop). - Returns: Frozenset of approved app IDs. """ approved = _load_approved() return frozenset(int(e["app_id"]) for e in approved) - - -def list_pending_exceptions() -> list[dict[str, object]]: - """Return a copy of the current pending exception list. - - Returns: - List of pending exception dicts with keys app_id, reason, requested_at. - """ - return list(_load_pending()) diff --git a/steam_backlog_enforcer/main.py b/steam_backlog_enforcer/main.py index 26e4221..72438cb 100644 --- a/steam_backlog_enforcer/main.py +++ b/steam_backlog_enforcer/main.py @@ -5,7 +5,6 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone import logging import sys -import time from typing import TYPE_CHECKING from steam_backlog_enforcer._actions import ( @@ -32,9 +31,7 @@ from steam_backlog_enforcer._total_block import ( ) from steam_backlog_enforcer._web_server import serve from steam_backlog_enforcer._whitelist import ( - WHITELIST_COOLDOWN_SECONDS, add_pending_exception, - list_pending_exceptions, validate_reason, ) from steam_backlog_enforcer.config import ( @@ -364,17 +361,16 @@ _ADD_EXCEPTION_USAGE = ( "Example:\n" " add-exception 440 --reason " '"TF2 is needed for a community event this weekend"\n\n' - f"Exceptions become active after a {WHITELIST_COOLDOWN_SECONDS // 3600}h " - "cooldown." + "Exceptions become active immediately." ) def cmd_add_exception(args: list[str]) -> None: - """Request a time-locked whitelist exception. + """Add a whitelist exception, active immediately. Usage: add-exception --reason "" - The exception becomes active after a 24-hour cooldown. The reason must be + The exception becomes active right away (no cooldown). The reason must be a genuine justification of at least 5 words with sufficient entropy. Args: @@ -411,20 +407,6 @@ def cmd_add_exception(args: list[str]) -> None: _echo(msg) - # Show current pending list. - pending = list_pending_exceptions() - if pending: - _echo(f"\nPending exceptions ({len(pending)}):") - now = time.time() - for entry in pending: - aid = int(entry["app_id"]) - elapsed = now - float(entry["requested_at"]) - remaining = max(0.0, WHITELIST_COOLDOWN_SECONDS - elapsed) - hrs = int(remaining // 3600) - mins = int((remaining % 3600) // 60) - status = "ready" if remaining == 0.0 else f"approves in {hrs}h {mins}m" - _echo(f" AppID={aid} [{status}]") - def cmd_install(config: Config, state: State) -> None: """Manually trigger install of the assigned game.""" diff --git a/steam_backlog_enforcer/tests/conftest.py b/steam_backlog_enforcer/tests/conftest.py index e66974c..ff71000 100644 --- a/steam_backlog_enforcer/tests/conftest.py +++ b/steam_backlog_enforcer/tests/conftest.py @@ -113,10 +113,6 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]: ), ), # Whitelist exception files (_whitelist module-level constants) - patch( - "steam_backlog_enforcer._whitelist.PENDING_EXCEPTIONS_FILE", - fake_config / "pending_exceptions.json", - ), patch( "steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE", fake_config / "approved_exceptions.json", diff --git a/steam_backlog_enforcer/tests/test_enforce_loop_part2.py b/steam_backlog_enforcer/tests/test_enforce_loop_part2.py index 9e12b16..919723a 100644 --- a/steam_backlog_enforcer/tests/test_enforce_loop_part2.py +++ b/steam_backlog_enforcer/tests/test_enforce_loop_part2.py @@ -100,22 +100,6 @@ class TestEnforceLoopIteration: mock_guard.assert_not_called() mock_installed.assert_not_called() - def test_promotes_newly_approved_exceptions(self) -> None: - """Loop body at line 286 executes when promote returns non-empty list.""" - config = Config( - kill_unauthorized_games=False, - uninstall_other_games=False, - ) - state = State(current_app_id=1, current_game_name="G") - with ( - patch(f"{PKG}.is_game_installed", return_value=True), - patch( - f"{PKG}.promote_pending_exceptions", - return_value=[440], - ), - ): - _enforce_loop_iteration(config, state) - class TestDoEnforce: """Tests for do_enforce.""" diff --git a/steam_backlog_enforcer/tests/test_main.py b/steam_backlog_enforcer/tests/test_main.py index 81792e8..6b3706c 100644 --- a/steam_backlog_enforcer/tests/test_main.py +++ b/steam_backlog_enforcer/tests/test_main.py @@ -3,13 +3,11 @@ from __future__ import annotations import sys -import time from typing import Any from unittest.mock import patch import pytest -from steam_backlog_enforcer._whitelist import WHITELIST_COOLDOWN_SECONDS from steam_backlog_enforcer.config import Config, State from steam_backlog_enforcer.main import ( cmd_add_exception, @@ -448,37 +446,17 @@ class TestCmdAddException: ): cmd_add_exception(["440", "--reason", _VALID_REASON]) - def test_happy_path_no_pending(self) -> None: + def test_happy_path(self) -> None: with ( patch(f"{PKG}._echo") as mock_echo, patch( f"{PKG}.add_pending_exception", - return_value="Exception requested for AppID 440.", + return_value="Exception approved for AppID 440. Active immediately.", ), - patch(f"{PKG}.list_pending_exceptions", return_value=[]), ): cmd_add_exception(["440", "--reason", _VALID_REASON]) mock_echo.assert_called() - def test_happy_path_with_pending_list(self) -> None: - now = time.time() - pending = [ - {"app_id": 440, "requested_at": now - WHITELIST_COOLDOWN_SECONDS - 1}, - {"app_id": 730, "requested_at": now}, - ] - with ( - patch(f"{PKG}._echo") as mock_echo, - patch( - f"{PKG}.add_pending_exception", - return_value="Exception requested for AppID 440.", - ), - patch(f"{PKG}.list_pending_exceptions", return_value=pending), - ): - cmd_add_exception(["440", "--reason", _VALID_REASON]) - # At least the "Pending exceptions" line should be echoed - calls = [str(c) for c in mock_echo.call_args_list] - assert any("Pending" in s for s in calls) - # ────────────────────────────────────────────────────────────── # main() dispatch to add-exception diff --git a/steam_backlog_enforcer/tests/test_whitelist.py b/steam_backlog_enforcer/tests/test_whitelist.py index 63e6ab8..91b2eb1 100644 --- a/steam_backlog_enforcer/tests/test_whitelist.py +++ b/steam_backlog_enforcer/tests/test_whitelist.py @@ -2,26 +2,20 @@ from __future__ import annotations -import time from typing import TYPE_CHECKING from unittest.mock import patch import pytest from steam_backlog_enforcer._whitelist import ( - WHITELIST_COOLDOWN_SECONDS, _append_audit_log, _load_approved, - _load_pending, _save_approved, - _save_pending, _shannon_entropy, _try_set_immutable, add_pending_exception, get_approved_exception_ids, - list_pending_exceptions, lock_enforcement_files, - promote_pending_exceptions, unlock_for_write, validate_reason, ) @@ -194,39 +188,11 @@ class TestLockAndUnlock: # ────────────────────────────────────────────────────────────── -# Persistence helpers (_load_pending, _save_pending, etc.) +# Persistence helpers (_load_approved, _save_approved) # ────────────────────────────────────────────────────────────── class TestPersistence: - def test_load_pending_missing_file(self) -> None: - assert _load_pending() == [] - - def test_load_pending_corrupt_file(self, tmp_path: Path) -> None: - bad = tmp_path / "pending.json" - bad.write_text("not json{{", encoding="utf-8") - with patch( - "steam_backlog_enforcer._whitelist.PENDING_EXCEPTIONS_FILE", - bad, - ): - assert _load_pending() == [] - - def test_load_pending_non_list(self, tmp_path: Path) -> None: - bad = tmp_path / "pending.json" - bad.write_text('{"key": "value"}', encoding="utf-8") - with patch( - "steam_backlog_enforcer._whitelist.PENDING_EXCEPTIONS_FILE", - bad, - ): - assert _load_pending() == [] - - def test_save_and_load_pending_roundtrip(self) -> None: - entries: list[dict[str, object]] = [ - {"app_id": 440, "reason": "test", "requested_at": 12345.0} - ] - _save_pending(entries) - assert _load_pending() == entries - def test_load_approved_missing_file(self) -> None: assert _load_approved() == [] @@ -300,10 +266,12 @@ class TestAddPendingException: with patch("shutil.which", return_value=None): msg = add_pending_exception(440, _VALID_REASON) assert "440" in msg - assert "24h" in msg or "hours" in msg or "active" in msg.lower() - pending = _load_pending() - assert len(pending) == 1 - assert int(pending[0]["app_id"]) == 440 + assert "immediately" in msg.lower() + approved = _load_approved() + assert len(approved) == 1 + assert int(approved[0]["app_id"]) == 440 + # Now active right away, no cooldown. + assert 440 in get_approved_exception_ids() def test_invalid_reason_raises(self) -> None: with pytest.raises( @@ -322,100 +290,6 @@ class TestAddPendingException: ): add_pending_exception(440, _VALID_REASON) - def test_already_pending_cooldown_remaining(self) -> None: - existing: list[dict[str, object]] = [ - { - "app_id": 440, - "reason": _VALID_REASON, - "requested_at": time.time(), # just now → full cooldown remaining - } - ] - _save_pending(existing) - with patch("shutil.which", return_value=None): - msg = add_pending_exception(440, _VALID_REASON) - assert "already pending" in msg - - def test_already_pending_cooldown_elapsed_promotes(self) -> None: - past = time.time() - WHITELIST_COOLDOWN_SECONDS - 1 - existing: list[dict[str, object]] = [ - {"app_id": 440, "reason": _VALID_REASON, "requested_at": past} - ] - _save_pending(existing) - with patch("shutil.which", return_value=None): - msg = add_pending_exception(440, _VALID_REASON) - # The elapsed entry is broken out of the pending-check loop via break, - # then a new entry is appended → still gets the "Will become active" msg. - assert "440" in msg - - -# ────────────────────────────────────────────────────────────── -# promote_pending_exceptions -# ────────────────────────────────────────────────────────────── - - -class TestPromotePendingExceptions: - def test_no_entries(self) -> None: - result = promote_pending_exceptions() - assert result == [] - - def test_cooldown_not_elapsed(self) -> None: - entries: list[dict[str, object]] = [ - {"app_id": 440, "reason": _VALID_REASON, "requested_at": time.time()} - ] - _save_pending(entries) - with patch("shutil.which", return_value=None): - result = promote_pending_exceptions() - assert result == [] - # Still pending - assert len(_load_pending()) == 1 - - def test_cooldown_elapsed_promotes(self) -> None: - past = time.time() - WHITELIST_COOLDOWN_SECONDS - 1 - entries: list[dict[str, object]] = [ - {"app_id": 440, "reason": _VALID_REASON, "requested_at": past} - ] - _save_pending(entries) - with patch("shutil.which", return_value=None): - result = promote_pending_exceptions() - assert 440 in result - assert _load_pending() == [] - approved_ids = get_approved_exception_ids() - assert 440 in approved_ids - - def test_already_in_approved_not_duplicated(self) -> None: - """If somehow already approved, skip duplicating it.""" - past = time.time() - WHITELIST_COOLDOWN_SECONDS - 1 - entries: list[dict[str, object]] = [ - {"app_id": 440, "reason": _VALID_REASON, "requested_at": past} - ] - _save_pending(entries) - existing_approved: list[dict[str, object]] = [ - {"app_id": 440, "reason": _VALID_REASON, "approved_at": 0.0} - ] - _save_approved(existing_approved) - with patch("shutil.which", return_value=None): - result = promote_pending_exceptions() - # Not added again - assert 440 not in result - approved = _load_approved() - assert sum(1 for e in approved if int(e["app_id"]) == 440) == 1 - - def test_pending_list_saved_when_entries_removed(self) -> None: - """_save_pending is called when the pending list shrinks.""" - past = time.time() - WHITELIST_COOLDOWN_SECONDS - 1 - future = time.time() - entries: list[dict[str, object]] = [ - {"app_id": 440, "reason": _VALID_REASON, "requested_at": past}, - {"app_id": 730, "reason": _VALID_REASON, "requested_at": future}, - ] - _save_pending(entries) - with patch("shutil.which", return_value=None): - result = promote_pending_exceptions() - assert 440 in result - remaining = _load_pending() - assert len(remaining) == 1 - assert int(remaining[0]["app_id"]) == 730 - # ────────────────────────────────────────────────────────────── # get_approved_exception_ids @@ -439,27 +313,6 @@ class TestGetApprovedExceptionIds: assert result == frozenset({440, 730}) -# ────────────────────────────────────────────────────────────── -# list_pending_exceptions -# ────────────────────────────────────────────────────────────── - - -class TestListPendingExceptions: - def test_returns_copy(self) -> None: - """Mutating the result must not affect stored state.""" - entries: list[dict[str, object]] = [ - {"app_id": 440, "reason": "r", "requested_at": 1.0} - ] - _save_pending(entries) - result = list_pending_exceptions() - result.clear() - # Still present on next load - assert len(_load_pending()) == 1 - - def test_empty_when_no_pending(self) -> None: - assert list_pending_exceptions() == [] - - # ────────────────────────────────────────────────────────────── # Extra coverage for validate_reason branches 94 & 106 # ────────────────────────────────────────────────────────────── @@ -481,16 +334,3 @@ class TestValidateReasonExtraBranches: err = validate_reason(reason) assert err is not None assert "repetitive" in err - - def test_different_app_id_in_pending_does_not_block(self) -> None: - """Pending entry with a different app_id must not block a new add (covers - the 264→263 loop-continue branch).""" - other: list[dict[str, object]] = [ - {"app_id": 440, "reason": _VALID_REASON, "requested_at": 1.0} - ] - _save_pending(other) - with patch("shutil.which", return_value=None): - msg = add_pending_exception(730, _VALID_REASON) - assert "730" in msg - pending = _load_pending() - assert len(pending) == 2 # original + new