mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-11-10 19:27:04 +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 (
|
||||
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)
|
||||
|
||||
|
||||
|
||||
@ -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] = []
|
||||
|
||||
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"],
|
||||
"reason": 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)
|
||||
_append_audit_log(app_id, reason, "APPROVED")
|
||||
|
||||
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())
|
||||
|
||||
@ -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."""
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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."""
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user