Remove whitelist exception cooldown; approve immediately
Some checks are pending
pre-commit / pre-commit (push) Waiting to run
Tests / test (3.10) (push) Waiting to run
Tests / test (3.11) (push) Waiting to run
Tests / test (3.12) (push) Waiting to run

Whitelist exceptions previously entered a 24h pending queue and were
promoted to approved by the enforce loop once the cooldown elapsed.
add_pending_exception now approves immediately, so the entire pending/
promote/cooldown mechanism is dead: drop WHITELIST_COOLDOWN_SECONDS,
PENDING_EXCEPTIONS_FILE, _load_pending/_save_pending,
promote_pending_exceptions, and list_pending_exceptions, plus the
enforce-loop promotion step and the CLI's now-empty pending display.
Usage text and docstrings updated to state exceptions are active
immediately. Tests for the removed machinery deleted; add-exception
tests assert immediate approval. Coverage remains 100%.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017RRCdR4D8SaVrN2xraMQxM
This commit is contained in:
Krzysztof Rudnicki 2026-07-10 22:57:39 +02:00
parent 26c4771192
commit de76c6541e
7 changed files with 27 additions and 341 deletions

View File

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

View File

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

View File

@ -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 <app_id> --reason "<text>"
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."""

View File

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

View File

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

View File

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

View File

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