mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-11-10 19:07:05 +01:00
Remove whitelist exception cooldown; approve immediately
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:
parent
26c4771192
commit
de76c6541e
@ -15,7 +15,6 @@ from steam_backlog_enforcer._total_block import (
|
|||||||
)
|
)
|
||||||
from steam_backlog_enforcer._whitelist import (
|
from steam_backlog_enforcer._whitelist import (
|
||||||
lock_enforcement_files,
|
lock_enforcement_files,
|
||||||
promote_pending_exceptions,
|
|
||||||
)
|
)
|
||||||
from steam_backlog_enforcer.config import (
|
from steam_backlog_enforcer.config import (
|
||||||
CONFIG_DIR,
|
CONFIG_DIR,
|
||||||
@ -296,12 +295,7 @@ def _enforce_loop_iteration(config: Config, state: State) -> None:
|
|||||||
config.steam_id,
|
config.steam_id,
|
||||||
)
|
)
|
||||||
|
|
||||||
# D) Promote any cooldown-elapsed pending exceptions to approved.
|
# D) Re-apply immutable flag so config cannot be edited without root.
|
||||||
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.
|
|
||||||
lock_enforcement_files(CONFIG_FILE)
|
lock_enforcement_files(CONFIG_FILE)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@ -24,7 +24,6 @@ logger = logging.getLogger(__name__)
|
|||||||
# File paths (patched in tests via conftest)
|
# 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"
|
APPROVED_EXCEPTIONS_FILE: Path = CONFIG_DIR / "approved_exceptions.json"
|
||||||
EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log"
|
EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log"
|
||||||
|
|
||||||
@ -32,8 +31,6 @@ EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log"
|
|||||||
# Constants
|
# Constants
|
||||||
# ──────────────────────────────────────────────────────────────
|
# ──────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
WHITELIST_COOLDOWN_SECONDS: int = 86400 # 24 hours
|
|
||||||
|
|
||||||
_MIN_REASON_WORDS: int = 5
|
_MIN_REASON_WORDS: int = 5
|
||||||
_MIN_REASON_LENGTH: int = 25
|
_MIN_REASON_LENGTH: int = 25
|
||||||
_MIN_ENTROPY: float = 3.0
|
_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]]:
|
def _load_approved() -> list[dict[str, object]]:
|
||||||
"""Load approved exception entries from disk."""
|
"""Load approved exception entries from disk."""
|
||||||
if not APPROVED_EXCEPTIONS_FILE.exists():
|
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:
|
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
|
The entry becomes active right away (no cooldown). Returns a
|
||||||
elapsed (24 h by default). Returns a human-readable status message.
|
human-readable status message.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
app_id: Steam application ID to add.
|
app_id: Steam application ID to add.
|
||||||
reason: Validated justification text (must pass :func:`validate_reason`).
|
reason: Validated justification text (must pass :func:`validate_reason`).
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Human-readable confirmation or remaining-cooldown message.
|
Human-readable confirmation message.
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
ValueError: If the reason fails validation or the ID is already approved.
|
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."
|
msg = f"AppID {app_id} is already in the approved exceptions list."
|
||||||
raise ValueError(msg)
|
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()
|
now = time.time()
|
||||||
still_pending: list[dict[str, object]] = []
|
approved.append(
|
||||||
newly_approved: list[int] = []
|
{
|
||||||
|
"app_id": app_id,
|
||||||
|
"reason": reason,
|
||||||
|
"approved_at": now,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
_save_approved(approved)
|
||||||
|
_append_audit_log(app_id, reason, "APPROVED")
|
||||||
|
|
||||||
for entry in pending:
|
return f"Exception approved for AppID {app_id}. Active immediately. Reason logged."
|
||||||
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
|
|
||||||
|
|
||||||
|
|
||||||
def get_approved_exception_ids() -> frozenset[int]:
|
def get_approved_exception_ids() -> frozenset[int]:
|
||||||
"""Return the frozenset of currently approved exception app IDs.
|
"""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:
|
Returns:
|
||||||
Frozenset of approved app IDs.
|
Frozenset of approved app IDs.
|
||||||
"""
|
"""
|
||||||
approved = _load_approved()
|
approved = _load_approved()
|
||||||
return frozenset(int(e["app_id"]) for e in 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())
|
|
||||||
|
|||||||
@ -5,7 +5,6 @@ from __future__ import annotations
|
|||||||
from datetime import datetime, timedelta, timezone
|
from datetime import datetime, timedelta, timezone
|
||||||
import logging
|
import logging
|
||||||
import sys
|
import sys
|
||||||
import time
|
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
from steam_backlog_enforcer._actions import (
|
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._web_server import serve
|
||||||
from steam_backlog_enforcer._whitelist import (
|
from steam_backlog_enforcer._whitelist import (
|
||||||
WHITELIST_COOLDOWN_SECONDS,
|
|
||||||
add_pending_exception,
|
add_pending_exception,
|
||||||
list_pending_exceptions,
|
|
||||||
validate_reason,
|
validate_reason,
|
||||||
)
|
)
|
||||||
from steam_backlog_enforcer.config import (
|
from steam_backlog_enforcer.config import (
|
||||||
@ -364,17 +361,16 @@ _ADD_EXCEPTION_USAGE = (
|
|||||||
"Example:\n"
|
"Example:\n"
|
||||||
" add-exception 440 --reason "
|
" add-exception 440 --reason "
|
||||||
'"TF2 is needed for a community event this weekend"\n\n'
|
'"TF2 is needed for a community event this weekend"\n\n'
|
||||||
f"Exceptions become active after a {WHITELIST_COOLDOWN_SECONDS // 3600}h "
|
"Exceptions become active immediately."
|
||||||
"cooldown."
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def cmd_add_exception(args: list[str]) -> None:
|
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>"
|
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.
|
a genuine justification of at least 5 words with sufficient entropy.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
@ -411,20 +407,6 @@ def cmd_add_exception(args: list[str]) -> None:
|
|||||||
|
|
||||||
_echo(msg)
|
_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:
|
def cmd_install(config: Config, state: State) -> None:
|
||||||
"""Manually trigger install of the assigned game."""
|
"""Manually trigger install of the assigned game."""
|
||||||
|
|||||||
@ -113,10 +113,6 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]:
|
|||||||
),
|
),
|
||||||
),
|
),
|
||||||
# Whitelist exception files (_whitelist module-level constants)
|
# Whitelist exception files (_whitelist module-level constants)
|
||||||
patch(
|
|
||||||
"steam_backlog_enforcer._whitelist.PENDING_EXCEPTIONS_FILE",
|
|
||||||
fake_config / "pending_exceptions.json",
|
|
||||||
),
|
|
||||||
patch(
|
patch(
|
||||||
"steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE",
|
"steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE",
|
||||||
fake_config / "approved_exceptions.json",
|
fake_config / "approved_exceptions.json",
|
||||||
|
|||||||
@ -100,22 +100,6 @@ class TestEnforceLoopIteration:
|
|||||||
mock_guard.assert_not_called()
|
mock_guard.assert_not_called()
|
||||||
mock_installed.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:
|
class TestDoEnforce:
|
||||||
"""Tests for do_enforce."""
|
"""Tests for do_enforce."""
|
||||||
|
|||||||
@ -3,13 +3,11 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import sys
|
import sys
|
||||||
import time
|
|
||||||
from typing import Any
|
from typing import Any
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from steam_backlog_enforcer._whitelist import WHITELIST_COOLDOWN_SECONDS
|
|
||||||
from steam_backlog_enforcer.config import Config, State
|
from steam_backlog_enforcer.config import Config, State
|
||||||
from steam_backlog_enforcer.main import (
|
from steam_backlog_enforcer.main import (
|
||||||
cmd_add_exception,
|
cmd_add_exception,
|
||||||
@ -448,37 +446,17 @@ class TestCmdAddException:
|
|||||||
):
|
):
|
||||||
cmd_add_exception(["440", "--reason", _VALID_REASON])
|
cmd_add_exception(["440", "--reason", _VALID_REASON])
|
||||||
|
|
||||||
def test_happy_path_no_pending(self) -> None:
|
def test_happy_path(self) -> None:
|
||||||
with (
|
with (
|
||||||
patch(f"{PKG}._echo") as mock_echo,
|
patch(f"{PKG}._echo") as mock_echo,
|
||||||
patch(
|
patch(
|
||||||
f"{PKG}.add_pending_exception",
|
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])
|
cmd_add_exception(["440", "--reason", _VALID_REASON])
|
||||||
mock_echo.assert_called()
|
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
|
# main() dispatch to add-exception
|
||||||
|
|||||||
@ -2,26 +2,20 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import time
|
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from steam_backlog_enforcer._whitelist import (
|
from steam_backlog_enforcer._whitelist import (
|
||||||
WHITELIST_COOLDOWN_SECONDS,
|
|
||||||
_append_audit_log,
|
_append_audit_log,
|
||||||
_load_approved,
|
_load_approved,
|
||||||
_load_pending,
|
|
||||||
_save_approved,
|
_save_approved,
|
||||||
_save_pending,
|
|
||||||
_shannon_entropy,
|
_shannon_entropy,
|
||||||
_try_set_immutable,
|
_try_set_immutable,
|
||||||
add_pending_exception,
|
add_pending_exception,
|
||||||
get_approved_exception_ids,
|
get_approved_exception_ids,
|
||||||
list_pending_exceptions,
|
|
||||||
lock_enforcement_files,
|
lock_enforcement_files,
|
||||||
promote_pending_exceptions,
|
|
||||||
unlock_for_write,
|
unlock_for_write,
|
||||||
validate_reason,
|
validate_reason,
|
||||||
)
|
)
|
||||||
@ -194,39 +188,11 @@ class TestLockAndUnlock:
|
|||||||
|
|
||||||
|
|
||||||
# ──────────────────────────────────────────────────────────────
|
# ──────────────────────────────────────────────────────────────
|
||||||
# Persistence helpers (_load_pending, _save_pending, etc.)
|
# Persistence helpers (_load_approved, _save_approved)
|
||||||
# ──────────────────────────────────────────────────────────────
|
# ──────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
class TestPersistence:
|
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:
|
def test_load_approved_missing_file(self) -> None:
|
||||||
assert _load_approved() == []
|
assert _load_approved() == []
|
||||||
|
|
||||||
@ -300,10 +266,12 @@ class TestAddPendingException:
|
|||||||
with patch("shutil.which", return_value=None):
|
with patch("shutil.which", return_value=None):
|
||||||
msg = add_pending_exception(440, _VALID_REASON)
|
msg = add_pending_exception(440, _VALID_REASON)
|
||||||
assert "440" in msg
|
assert "440" in msg
|
||||||
assert "24h" in msg or "hours" in msg or "active" in msg.lower()
|
assert "immediately" in msg.lower()
|
||||||
pending = _load_pending()
|
approved = _load_approved()
|
||||||
assert len(pending) == 1
|
assert len(approved) == 1
|
||||||
assert int(pending[0]["app_id"]) == 440
|
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:
|
def test_invalid_reason_raises(self) -> None:
|
||||||
with pytest.raises(
|
with pytest.raises(
|
||||||
@ -322,100 +290,6 @@ class TestAddPendingException:
|
|||||||
):
|
):
|
||||||
add_pending_exception(440, _VALID_REASON)
|
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
|
# get_approved_exception_ids
|
||||||
@ -439,27 +313,6 @@ class TestGetApprovedExceptionIds:
|
|||||||
assert result == frozenset({440, 730})
|
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
|
# Extra coverage for validate_reason branches 94 & 106
|
||||||
# ──────────────────────────────────────────────────────────────
|
# ──────────────────────────────────────────────────────────────
|
||||||
@ -481,16 +334,3 @@ class TestValidateReasonExtraBranches:
|
|||||||
err = validate_reason(reason)
|
err = validate_reason(reason)
|
||||||
assert err is not None
|
assert err is not None
|
||||||
assert "repetitive" in err
|
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
|
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user