mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-11-10 19:27:04 +01:00
Grade against 3.14 too, not just run on it
`requires-python` and the CI pin already said 3.14, but ruff, mypy and pylint were still grading against 3.10 -- three tools judging an interpreter nothing runs any more, and ruff withholding the pyupgrade fixes that only become safe at the newer target. `pin_ci_python.py` rewrites exactly these three knobs. What the stale target was hiding: `datetime.timezone.utc` where `datetime.UTC` is now correct (UP017), `socket.timeout` where `TimeoutError` is (UP041), deprecated `typing` imports (UP035), PEP 695 `type` aliases (UP040), and ruff's stable formatter applying PEP 758 -- `except (A, B):` becomes `except A, B:`. That last one makes the source 3.14-only at the syntax level, which is what `requires-python = ">=3.14"` already promised; kuhy chose it deliberately over holding ruff a release back (2026-08-29). ruff, ruff-format, mypy --strict and pylint all clean at the new target. The two modified web/ TypeScript files are another session's in-flight work and are deliberately left unstaged. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SfnJmxxsZ534AKk16rC2Gr
This commit is contained in:
parent
08aa5c42f8
commit
356530a01c
@ -36,7 +36,7 @@ include = ["steam_backlog_enforcer*"]
|
||||
# RUFF - Fast Python linter and formatter
|
||||
# ============================================================================
|
||||
[tool.ruff]
|
||||
target-version = "py310"
|
||||
target-version = "py314"
|
||||
include = ["*.py", "**/*.py"]
|
||||
# scripts/file_length is a VENDORED copy of ~/utils/file_length and must stay
|
||||
# byte-identical to upstream; ruff --unsafe-fixes otherwise strips its print()
|
||||
@ -85,7 +85,7 @@ docstring-code-format = true
|
||||
# MYPY - Static type checker
|
||||
# ============================================================================
|
||||
[tool.mypy]
|
||||
python_version = "3.10"
|
||||
python_version = "3.14"
|
||||
strict = true
|
||||
warn_return_any = true
|
||||
warn_unused_configs = true
|
||||
@ -116,7 +116,7 @@ exclude = [".venv/"]
|
||||
analyse-fallback-blocks = true
|
||||
persistent = true
|
||||
jobs = 0
|
||||
py-version = "3.10"
|
||||
py-version = "3.14"
|
||||
ignore = [".venv", "__pycache__"]
|
||||
ignore-patterns = [".*\\.pyi$"]
|
||||
|
||||
|
||||
@ -9,9 +9,9 @@ there is a single tested implementation of the underlying behaviour.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import importlib
|
||||
from typing import TYPE_CHECKING, Any, TypeAlias
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from steam_backlog_enforcer._allowed_games import (
|
||||
active_manual_picks,
|
||||
@ -135,7 +135,7 @@ def apply_manual_pick(
|
||||
f"(cap is {max_picks}): {names}."
|
||||
)
|
||||
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
now = datetime.now(UTC).isoformat()
|
||||
# Rewriting from `active` also prunes finished/expired entries.
|
||||
state.manual_picks = [
|
||||
*active,
|
||||
@ -160,7 +160,7 @@ _MOVED_TO_MANUAL_PICK_LIFECYCLE = frozenset(
|
||||
|
||||
# Whatever the re-exported name turns out to be -- a function, a class or
|
||||
# a constant. Aliased so the annotation is a name rather than a bare Any.
|
||||
_Reexport: TypeAlias = Any
|
||||
type _Reexport = Any
|
||||
|
||||
|
||||
def __getattr__(name: str) -> _Reexport:
|
||||
|
||||
@ -9,7 +9,7 @@ deletion safety net without creating a circular import.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -45,7 +45,7 @@ def _pick_is_active(state: State, pick: dict[str, Any]) -> bool:
|
||||
started = datetime.fromisoformat(started_at)
|
||||
except ValueError:
|
||||
return True
|
||||
if datetime.now(timezone.utc) >= started + timedelta(days=MANUAL_LOCK_DAYS):
|
||||
if datetime.now(UTC) >= started + timedelta(days=MANUAL_LOCK_DAYS):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
@ -37,7 +37,7 @@ def _get_shared_js_ws_url() -> str | None:
|
||||
try:
|
||||
resp = requests.get(f"http://127.0.0.1:{_CDP_PORT}/json", timeout=5)
|
||||
targets = resp.json()
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
return None
|
||||
|
||||
for target in targets:
|
||||
|
||||
@ -5,7 +5,7 @@ from __future__ import annotations
|
||||
import importlib
|
||||
import logging
|
||||
import sys
|
||||
from typing import TYPE_CHECKING, Any, TypeAlias
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from steam_backlog_enforcer.config import State, load_snapshot
|
||||
from steam_backlog_enforcer.game_install import (
|
||||
@ -185,7 +185,7 @@ _MOVED_TO_CMD_DONE_FINALIZE = frozenset(
|
||||
|
||||
# Whatever the re-exported name turns out to be -- a function, a class or
|
||||
# a constant. Aliased so the annotation is a name rather than a bare Any.
|
||||
_Reexport: TypeAlias = Any
|
||||
type _Reexport = Any
|
||||
|
||||
|
||||
def __getattr__(name: str) -> _Reexport:
|
||||
|
||||
@ -9,7 +9,7 @@ the JSON-RPC channel.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import os
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
@ -112,7 +112,7 @@ def cmd_gaming_reset(config: Config, _state: State) -> int:
|
||||
return 1
|
||||
|
||||
released = release_block()
|
||||
now = datetime.now(timezone.utc).astimezone()
|
||||
now = datetime.now(UTC).astimezone()
|
||||
save_state(
|
||||
PlaytimeState(day_key=gaming_day_key(now), last_tick_at=now.timestamp()),
|
||||
demo=rules_for(config, demo=False).demo,
|
||||
|
||||
@ -6,11 +6,10 @@ them introduces no cycle. Split to keep both files under the 250-line cap.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
import re
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import aiohttp
|
||||
|
||||
@ -22,6 +21,9 @@ from steam_backlog_enforcer._hltb_types import (
|
||||
HLTB_BASE_URL,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import asyncio
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_NEXT_DATA_RE = re.compile(
|
||||
@ -108,7 +110,7 @@ async def _fetch_detail_one(
|
||||
if resp.status == HTTPStatus.OK:
|
||||
html = await resp.text()
|
||||
return _parse_game_page(html)
|
||||
except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
|
||||
except (TimeoutError, aiohttp.ClientError) as exc:
|
||||
logger.debug(
|
||||
"HLTB detail fetch failed for game_id=%d: %s",
|
||||
hltb_game_id,
|
||||
|
||||
@ -26,7 +26,7 @@ def _parse_game_page(html: str) -> dict[str, Any] | None:
|
||||
try:
|
||||
data = json.loads(match.group(1))
|
||||
result: dict[str, Any] = data["props"]["pageProps"]["game"]["data"]
|
||||
except (json.JSONDecodeError, KeyError, TypeError):
|
||||
except json.JSONDecodeError, KeyError, TypeError:
|
||||
return None
|
||||
return result
|
||||
|
||||
|
||||
@ -116,7 +116,7 @@ async def _search_one(
|
||||
continue
|
||||
result = _build_result_from_best(app_id, name, query_name, best)
|
||||
break
|
||||
except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
|
||||
except (TimeoutError, aiohttp.ClientError) as exc:
|
||||
logger.debug("HLTB search failed for '%s': %s", query_name, exc)
|
||||
|
||||
# Update cache immediately (miss = -1).
|
||||
|
||||
@ -6,7 +6,6 @@ them introduces no cycle. Split to keep both files under the 250-line cap.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from http import HTTPStatus
|
||||
import json
|
||||
import logging
|
||||
@ -60,7 +59,7 @@ def _get_hltb_search_url() -> str:
|
||||
if search_info and search_info.search_url:
|
||||
url: str = HTMLRequests.BASE_URL + search_info.search_url
|
||||
return url
|
||||
except (OSError, RuntimeError, ValueError, TypeError):
|
||||
except OSError, RuntimeError, ValueError, TypeError:
|
||||
logger.debug("Failed to discover HLTB search URL, using default")
|
||||
return "https://howlongtobeat.com/api/finder"
|
||||
|
||||
@ -94,7 +93,7 @@ async def _get_auth_info(
|
||||
hp_key=data.get("hpKey", ""),
|
||||
hp_val=data.get("hpVal", ""),
|
||||
)
|
||||
except (aiohttp.ClientError, asyncio.TimeoutError):
|
||||
except TimeoutError, aiohttp.ClientError:
|
||||
logger.warning("Failed to get HLTB auth token")
|
||||
return None
|
||||
|
||||
|
||||
@ -103,14 +103,14 @@ def _read_raw_cache() -> dict[int, dict[str, Any]]:
|
||||
return {}
|
||||
try:
|
||||
data = json.loads(HLTB_CACHE_FILE.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, OSError):
|
||||
except json.JSONDecodeError, OSError:
|
||||
logger.warning("Corrupt HLTB cache, starting fresh.")
|
||||
return {}
|
||||
out: dict[int, dict[str, Any]] = {}
|
||||
for k, v in data.items():
|
||||
try:
|
||||
aid = int(k)
|
||||
except (TypeError, ValueError):
|
||||
except TypeError, ValueError:
|
||||
continue
|
||||
if isinstance(v, dict):
|
||||
out[aid] = {
|
||||
@ -131,7 +131,7 @@ def _read_raw_cache() -> dict[int, dict[str, Any]]:
|
||||
"leisure_100h": -1,
|
||||
"hltb_game_id": 0,
|
||||
}
|
||||
except (TypeError, ValueError):
|
||||
except TypeError, ValueError:
|
||||
continue
|
||||
return out
|
||||
|
||||
|
||||
@ -6,7 +6,7 @@ under the 250-line cap.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from steam_backlog_enforcer._actions import (
|
||||
@ -61,7 +61,7 @@ def manual_pick_age_days(state: State, app_id: int) -> float | None:
|
||||
started = datetime.fromisoformat(pick["started_at"])
|
||||
except ValueError:
|
||||
return None
|
||||
return (datetime.now(timezone.utc) - started).total_seconds() / 86400
|
||||
return (datetime.now(UTC) - started).total_seconds() / 86400
|
||||
|
||||
|
||||
def abandon_manual_pick(state: State, app_id: int) -> bool:
|
||||
|
||||
@ -7,7 +7,7 @@ assigned.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
@ -93,7 +93,7 @@ def reset_gaming_time(*, confirm: bool = False) -> dict[str, Any]:
|
||||
return {"ok": False, "reason": "requires elevated privileges"}
|
||||
try:
|
||||
released = release_block()
|
||||
now = datetime.now(timezone.utc).astimezone()
|
||||
now = datetime.now(UTC).astimezone()
|
||||
save_state(
|
||||
PlaytimeState(day_key=gaming_day_key(now), last_tick_at=now.timestamp()),
|
||||
demo=False,
|
||||
|
||||
@ -36,7 +36,7 @@ def _load_owned_app_ids_cache(steam_id: str) -> list[int] | None:
|
||||
data: dict[str, Any] = json.loads(
|
||||
_OWNED_IDS_CACHE_FILE.read_text(encoding="utf-8")
|
||||
)
|
||||
except (json.JSONDecodeError, OSError, ValueError):
|
||||
except json.JSONDecodeError, OSError, ValueError:
|
||||
return None
|
||||
|
||||
cached_steam_id = str(data.get("steam_id", ""))
|
||||
@ -105,7 +105,7 @@ def get_all_owned_app_ids(config: Config) -> list[int]:
|
||||
continue
|
||||
seen.add(app_id)
|
||||
merged_ids.append(app_id)
|
||||
except (OSError, RuntimeError, ValueError):
|
||||
except OSError, RuntimeError, ValueError:
|
||||
if snapshot_ids:
|
||||
return snapshot_ids
|
||||
logger.warning("Could not fetch owned game list for hiding.")
|
||||
|
||||
@ -62,7 +62,7 @@ def uninstall_package(package: str) -> bool:
|
||||
timeout=120,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Failed to run pacman -R %s", package)
|
||||
return False
|
||||
if result.returncode == 0:
|
||||
|
||||
@ -28,7 +28,7 @@ is why :func:`get_pids_by_cmdline_names` exists alongside
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import replace
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import logging
|
||||
import time
|
||||
from typing import TYPE_CHECKING
|
||||
@ -91,7 +91,7 @@ def playtime_tick(
|
||||
session: Cross-tick engagement and logging state.
|
||||
demo: Whether this is a short-budget demo run.
|
||||
"""
|
||||
now = datetime.now(timezone.utc).astimezone()
|
||||
now = datetime.now(UTC).astimezone()
|
||||
rules = rules_for(config, demo=demo)
|
||||
state = roll_over(_state_or_recover(rules, now=now), day_key=gaming_day_key(now))
|
||||
|
||||
|
||||
@ -75,7 +75,7 @@ def _read_days() -> dict[str, float]:
|
||||
raw = json.loads(HISTORY_FILE.read_text(encoding="utf-8"))
|
||||
except FileNotFoundError:
|
||||
return {}
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
logger.warning(
|
||||
"Playtime history at %s is unreadable; starting over.", HISTORY_FILE
|
||||
)
|
||||
|
||||
@ -69,7 +69,7 @@ def _read_ppid(pid: int) -> int | None:
|
||||
"""
|
||||
try:
|
||||
raw = (_PROC / str(pid) / "stat").read_text(encoding="utf-8")
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
return None
|
||||
_, _, rest = raw.rpartition(") ")
|
||||
fields = rest.split()
|
||||
|
||||
@ -20,7 +20,7 @@ the sink degrades to a warning on the daemon's own logger and carries on.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import json
|
||||
import logging
|
||||
import logging.handlers
|
||||
@ -133,7 +133,7 @@ class BudgetLog:
|
||||
if sink is None:
|
||||
return
|
||||
payload = {
|
||||
"timestamp": datetime.now(timezone.utc).astimezone().isoformat(),
|
||||
"timestamp": datetime.now(UTC).astimezone().isoformat(),
|
||||
"event": event,
|
||||
**fields,
|
||||
}
|
||||
|
||||
@ -46,7 +46,7 @@ def _mountpoints(path: Path) -> set[Path]:
|
||||
"""
|
||||
try:
|
||||
raw = path.read_text(encoding="utf-8")
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
logger.warning("Could not read mountinfo at %s.", path)
|
||||
return set()
|
||||
|
||||
|
||||
@ -32,7 +32,7 @@ def notify_desktop_user(title: str, body: str) -> None:
|
||||
user = _resolve_desktop_user()
|
||||
try:
|
||||
_run_as_user(["notify-send", title, body, "--icon=dialog-warning"], user)
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
logger.debug("Could not send desktop notification.")
|
||||
|
||||
|
||||
|
||||
@ -73,7 +73,7 @@ def _match_cmdline(entry: Path, names: frozenset[str]) -> str | None:
|
||||
"""
|
||||
try:
|
||||
raw = (entry / "cmdline").read_bytes()
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
return None
|
||||
|
||||
argv = [
|
||||
@ -106,7 +106,7 @@ def process_name(pid: int) -> str | None:
|
||||
"""
|
||||
try:
|
||||
return (_PROC / str(pid) / "comm").read_text(encoding="utf-8").strip() or None
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@ -26,7 +26,7 @@ def _run(cmd: list[str]) -> bool:
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(cmd, capture_output=True, check=False, timeout=15)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Playtime block command failed: %s", cmd)
|
||||
return False
|
||||
if result.returncode != 0:
|
||||
|
||||
@ -181,7 +181,7 @@ def load_state(*, demo: bool) -> PlaytimeState | None:
|
||||
return None
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
logger.warning("Playtime state at %s is unreadable; assuming tampered.", path)
|
||||
return None
|
||||
if not isinstance(data, dict) or data.get("schema_version") != _SCHEMA_VERSION:
|
||||
|
||||
@ -6,7 +6,7 @@ the 250-line cap. Leaf helpers: nothing here calls back into ``scanning``.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
@ -70,7 +70,7 @@ def _assign_chosen_game(
|
||||
state.current_app_id = chosen.app_id
|
||||
state.current_game_name = chosen.name
|
||||
if not state.enforcement_started_at:
|
||||
state.enforcement_started_at = datetime.now(timezone.utc).isoformat()
|
||||
state.enforcement_started_at = datetime.now(UTC).isoformat()
|
||||
state.save()
|
||||
hours_str = (
|
||||
f" (~{chosen.completionist_hours:.1f}h leisure+dlc)"
|
||||
|
||||
@ -6,7 +6,7 @@ the 250-line cap. Leaf helpers: nothing here calls back into ``_stats``.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import secrets
|
||||
from typing import TYPE_CHECKING
|
||||
from urllib.parse import quote_plus
|
||||
@ -68,7 +68,7 @@ def _format_completion_date(hours: float, daily_hours: float) -> str:
|
||||
if hours <= 0 or daily_hours <= 0:
|
||||
return "N/A"
|
||||
days = int(hours / daily_hours)
|
||||
target = datetime.now(timezone.utc) + timedelta(days=days)
|
||||
target = datetime.now(UTC) + timedelta(days=days)
|
||||
return f"{days} days ({target.strftime('%Y-%m-%d')})"
|
||||
|
||||
|
||||
@ -92,7 +92,7 @@ def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None:
|
||||
_echo(f" Invalid enforcement_started_at: {state.enforcement_started_at!r}")
|
||||
return
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
now = datetime.now(UTC)
|
||||
days_elapsed = max(1, (now - started).days)
|
||||
|
||||
if games_done == 0:
|
||||
|
||||
@ -58,7 +58,7 @@ def _is_iptables_blocked() -> bool:
|
||||
timeout=5,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
return False
|
||||
else:
|
||||
return result.returncode == 0 and "DROP" in result.stdout
|
||||
@ -121,7 +121,7 @@ def _block_store_iptables() -> bool:
|
||||
timeout=5,
|
||||
check=True,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Failed to block store via iptables")
|
||||
return False
|
||||
else:
|
||||
@ -150,7 +150,7 @@ def _unblock_store_iptables() -> bool:
|
||||
timeout=5,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Failed to unblock iptables")
|
||||
return False
|
||||
else:
|
||||
|
||||
@ -18,7 +18,7 @@ active (package-block).
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import json
|
||||
import logging
|
||||
import shutil
|
||||
@ -70,7 +70,7 @@ def _read_lock() -> dict[str, object] | None:
|
||||
return None
|
||||
try:
|
||||
data = json.loads(TOTAL_BLOCK_LOCK_FILE.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, OSError, ValueError):
|
||||
except json.JSONDecodeError, OSError, ValueError:
|
||||
return None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
@ -85,7 +85,7 @@ def is_total_block_active() -> bool:
|
||||
until = data.get("until")
|
||||
if not isinstance(until, int | float):
|
||||
return False
|
||||
return datetime.now(timezone.utc).timestamp() < until
|
||||
return datetime.now(UTC).timestamp() < until
|
||||
|
||||
|
||||
def total_block_needs_cleanup() -> bool:
|
||||
@ -113,17 +113,17 @@ def get_total_block_status() -> TotalBlockStatus:
|
||||
days = data.get("days")
|
||||
|
||||
started_dt = (
|
||||
datetime.fromtimestamp(started_at, tz=timezone.utc)
|
||||
datetime.fromtimestamp(started_at, tz=UTC)
|
||||
if isinstance(started_at, int | float)
|
||||
else None
|
||||
)
|
||||
until_dt = (
|
||||
datetime.fromtimestamp(until, tz=timezone.utc)
|
||||
datetime.fromtimestamp(until, tz=UTC)
|
||||
if isinstance(until, int | float)
|
||||
else None
|
||||
)
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
now = datetime.now(UTC)
|
||||
active = until_dt is not None and now < until_dt
|
||||
days_remaining = (
|
||||
(until_dt - now).total_seconds() / 86400 if active and until_dt else 0.0
|
||||
|
||||
@ -51,7 +51,7 @@ def apply_total_block_hosts() -> bool:
|
||||
try:
|
||||
_disable_hosts_protection()
|
||||
_sudo_write_hosts(new_content)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Failed to write total-block hosts entries")
|
||||
return False
|
||||
finally:
|
||||
@ -78,7 +78,7 @@ def remove_total_block_hosts() -> bool:
|
||||
try:
|
||||
_disable_hosts_protection()
|
||||
_sudo_write_hosts(new_content)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Failed to remove total-block hosts entries")
|
||||
return False
|
||||
finally:
|
||||
|
||||
@ -38,7 +38,7 @@ def _load_cached_ips() -> set[str]:
|
||||
return set()
|
||||
try:
|
||||
data = json.loads(_IPTABLES_IP_CACHE_FILE.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, OSError, ValueError):
|
||||
except json.JSONDecodeError, OSError, ValueError:
|
||||
return set()
|
||||
if not isinstance(data, list):
|
||||
return set()
|
||||
@ -156,7 +156,7 @@ def apply_total_block_iptables() -> bool:
|
||||
timeout=5,
|
||||
check=True,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Failed to apply total-block iptables rules")
|
||||
return False
|
||||
else:
|
||||
@ -187,7 +187,7 @@ def remove_total_block_iptables() -> bool:
|
||||
timeout=5,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Failed to remove total-block iptables rules")
|
||||
return False
|
||||
else:
|
||||
|
||||
@ -9,7 +9,7 @@ in :mod:`steam_backlog_enforcer._total_block_net`; the two are independent.
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
@ -188,11 +188,11 @@ def _log_steam_purge(removed_paths: list[str], removed_packages: list[str]) -> N
|
||||
existing = json.loads(_STEAM_PURGE_LOG_FILE.read_text(encoding="utf-8"))
|
||||
if not isinstance(existing, list):
|
||||
existing = []
|
||||
except (OSError, json.JSONDecodeError, ValueError):
|
||||
except OSError, json.JSONDecodeError, ValueError:
|
||||
existing = []
|
||||
existing.append(
|
||||
{
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
"removed_paths": removed_paths,
|
||||
"removed_packages": removed_packages,
|
||||
}
|
||||
|
||||
@ -37,7 +37,7 @@ __all__ = [
|
||||
]
|
||||
|
||||
from dataclasses import asdict
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from steam_backlog_enforcer._actions import allowed_app_ids
|
||||
@ -114,7 +114,7 @@ def _state_info(
|
||||
except ValueError:
|
||||
started = None
|
||||
if started is not None:
|
||||
now = datetime.now(timezone.utc)
|
||||
now = datetime.now(UTC)
|
||||
days_elapsed = max(1, (now - started).days)
|
||||
if games_done_since_start > 0:
|
||||
pace = round(games_done_since_start / days_elapsed, 4)
|
||||
@ -169,7 +169,7 @@ def build_web_dataset(state: State) -> WebDataset:
|
||||
),
|
||||
default_summary=_default_summary(rows),
|
||||
pace_vs_hltb=pace_vs_hltb,
|
||||
generated_at=datetime.now(timezone.utc).isoformat(),
|
||||
generated_at=datetime.now(UTC).isoformat(),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@ -65,7 +65,7 @@ class _Handler(BaseHTTPRequestHandler):
|
||||
"""Build and send the gaming-budget snapshot as JSON."""
|
||||
try:
|
||||
body = json.dumps(build_budget_snapshot(demo=demo)).encode("utf-8")
|
||||
except (OSError, ValueError, KeyError):
|
||||
except OSError, ValueError, KeyError:
|
||||
logger.exception("Failed to build budget snapshot")
|
||||
self._send(HTTPStatus.INTERNAL_SERVER_ERROR, b"budget error", "text/plain")
|
||||
return
|
||||
@ -76,7 +76,7 @@ class _Handler(BaseHTTPRequestHandler):
|
||||
try:
|
||||
payload = dataset_to_payload(build_web_dataset(State.load()))
|
||||
body = json.dumps(payload).encode("utf-8")
|
||||
except (OSError, ValueError, KeyError):
|
||||
except OSError, ValueError, KeyError:
|
||||
logger.exception("Failed to build web dataset")
|
||||
self._send(HTTPStatus.INTERNAL_SERVER_ERROR, b"dataset error", "text/plain")
|
||||
return
|
||||
|
||||
@ -7,7 +7,7 @@ import importlib
|
||||
import logging
|
||||
import math
|
||||
import re
|
||||
from typing import TYPE_CHECKING, Any, TypeAlias
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from steam_backlog_enforcer.config import CONFIG_DIR
|
||||
|
||||
@ -130,7 +130,7 @@ _MOVED_TO_WHITELIST_LOCKING = frozenset(
|
||||
|
||||
# Whatever the re-exported name turns out to be -- a function, a class or
|
||||
# a constant. Aliased so the annotation is a name rather than a bare Any.
|
||||
_Reexport: TypeAlias = Any
|
||||
type _Reexport = Any
|
||||
|
||||
|
||||
def __getattr__(name: str) -> _Reexport:
|
||||
|
||||
@ -75,7 +75,7 @@ def _immutable_flag_is(path: Path, *, immutable: bool) -> bool:
|
||||
with path.open("rb") as handle:
|
||||
# mutate_flag defaults to True, so buf receives the flags in place.
|
||||
fcntl.ioctl(handle.fileno(), _FS_IOC_GETFLAGS, buf)
|
||||
except (OSError, ValueError):
|
||||
except OSError, ValueError:
|
||||
return False
|
||||
return bool(buf[0] & _FS_IMMUTABLE_FL) is immutable
|
||||
|
||||
@ -144,7 +144,7 @@ def _load_approved() -> list[dict[str, object]]:
|
||||
data: object = json.loads(APPROVED_EXCEPTIONS_FILE.read_text(encoding="utf-8"))
|
||||
if isinstance(data, list):
|
||||
return cast("list[dict[str, object]]", data)
|
||||
except (json.JSONDecodeError, OSError, ValueError):
|
||||
except json.JSONDecodeError, OSError, ValueError:
|
||||
pass
|
||||
return []
|
||||
|
||||
|
||||
@ -4,7 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
@ -158,7 +158,7 @@ class State:
|
||||
|
||||
def skip_for_days(self, app_id: int, days: int) -> None:
|
||||
"""Mark ``app_id`` as skipped for ``days`` days from now (UTC)."""
|
||||
expires = datetime.now(timezone.utc) + timedelta(days=days)
|
||||
expires = datetime.now(UTC) + timedelta(days=days)
|
||||
self.skipped_until[str(app_id)] = expires.isoformat()
|
||||
|
||||
def active_skipped_ids(self) -> set[int]:
|
||||
@ -167,7 +167,7 @@ class State:
|
||||
Mutates ``self.skipped_until`` to drop expired or malformed entries.
|
||||
Callers should ``save()`` if they want the prune persisted.
|
||||
"""
|
||||
now = datetime.now(timezone.utc)
|
||||
now = datetime.now(UTC)
|
||||
active: set[int] = set()
|
||||
to_remove: list[str] = []
|
||||
for aid_str, ts in self.skipped_until.items():
|
||||
@ -197,7 +197,7 @@ class State:
|
||||
if STATE_FILE.exists():
|
||||
try:
|
||||
data = json.loads(STATE_FILE.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, OSError, ValueError):
|
||||
except json.JSONDecodeError, OSError, ValueError:
|
||||
logger.warning("Corrupt state file, using defaults.")
|
||||
return cls()
|
||||
state = cls(
|
||||
|
||||
@ -38,7 +38,7 @@ def get_running_steam_game_pids() -> dict[int, int]:
|
||||
if value.isdigit():
|
||||
running[int(entry.name)] = int(value)
|
||||
break
|
||||
except (PermissionError, OSError, ValueError):
|
||||
except PermissionError, OSError, ValueError:
|
||||
continue
|
||||
|
||||
return running
|
||||
@ -64,7 +64,7 @@ def get_pids_by_process_names(names: frozenset[str]) -> dict[int, str]:
|
||||
continue
|
||||
try:
|
||||
comm = (entry / "comm").read_text(encoding="utf-8").strip()
|
||||
except (PermissionError, OSError, ValueError):
|
||||
except PermissionError, OSError, ValueError:
|
||||
continue
|
||||
if comm in truncated:
|
||||
running[int(entry.name)] = truncated[comm]
|
||||
@ -170,7 +170,7 @@ def send_notification(title: str, body: str) -> None:
|
||||
timeout=5,
|
||||
check=False,
|
||||
)
|
||||
except (FileNotFoundError, OSError):
|
||||
except FileNotFoundError, OSError:
|
||||
logger.debug("notify-send not available.")
|
||||
except subprocess.TimeoutExpired:
|
||||
# sudo can block where a bare notify-send would not; a dropped
|
||||
|
||||
@ -87,7 +87,7 @@ def _trigger_steam_install(app_id: int, label: str) -> bool:
|
||||
timeout=15,
|
||||
check=False,
|
||||
)
|
||||
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
|
||||
except FileNotFoundError, OSError, subprocess.TimeoutExpired:
|
||||
return False
|
||||
else:
|
||||
logger.info("Triggered Steam install for %s via protocol handler", label)
|
||||
|
||||
@ -15,7 +15,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import importlib
|
||||
import logging
|
||||
from typing import Any, TypeAlias
|
||||
from typing import Any
|
||||
|
||||
import aiohttp
|
||||
|
||||
@ -191,7 +191,7 @@ _REEXPORTED = {
|
||||
|
||||
# Whatever the re-exported name turns out to be -- a function, a class or
|
||||
# a constant. Aliased so the annotation is a name rather than a bare Any.
|
||||
_Reexport: TypeAlias = Any
|
||||
type _Reexport = Any
|
||||
|
||||
|
||||
def __getattr__(name: str) -> _Reexport:
|
||||
|
||||
@ -7,7 +7,7 @@ import cycle.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import sys
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
@ -128,7 +128,7 @@ def _describe_pick(state: State, pick: dict[str, object]) -> bool:
|
||||
try:
|
||||
started = datetime.fromisoformat(started_at)
|
||||
deadline = started + timedelta(days=_MANUAL_LOCK_DAYS)
|
||||
days_left = (deadline - datetime.now(timezone.utc)).days
|
||||
days_left = (deadline - datetime.now(UTC)).days
|
||||
_echo(f" Locked since: {started.strftime('%Y-%m-%d')}")
|
||||
_echo(
|
||||
f" Deadline: {deadline.strftime('%Y-%m-%d')}"
|
||||
|
||||
@ -81,7 +81,7 @@ def _resolve_game_name(config: Config, app_id: int) -> str | None:
|
||||
for g in owned:
|
||||
if g.get("appid") == app_id:
|
||||
return str(g.get("name", f"Unknown ({app_id})"))
|
||||
except (SteamAPIError, OSError, RuntimeError, ValueError):
|
||||
except SteamAPIError, OSError, RuntimeError, ValueError:
|
||||
return None
|
||||
|
||||
return None
|
||||
|
||||
@ -146,7 +146,7 @@ async def _fetch_one(
|
||||
confidence=data.get("confidence", ""),
|
||||
total_reports=data.get("total", 0),
|
||||
)
|
||||
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as e:
|
||||
except (TimeoutError, aiohttp.ClientError, OSError) as e:
|
||||
logger.warning("ProtonDB fetch failed for AppID=%d: %s", app_id, e)
|
||||
return None # Don't cache transient failures — retry next run.
|
||||
|
||||
|
||||
@ -134,7 +134,7 @@ def _block_via_hosts_install() -> bool:
|
||||
timeout=120,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
except OSError, subprocess.SubprocessError:
|
||||
logger.exception("Failed to run hosts install script")
|
||||
return False
|
||||
else:
|
||||
|
||||
@ -6,24 +6,22 @@ tests. ``name-tests-test`` exempts ``tests/_*.py`` for exactly this.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
from steam_backlog_enforcer._allowed_games import MANUAL_LOCK_DAYS
|
||||
from steam_backlog_enforcer._total_block import TotalBlockStatus
|
||||
from steam_backlog_enforcer.config import State
|
||||
|
||||
STARTED_AT = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
||||
STARTED_AT = (datetime.now(UTC) - timedelta(days=1)).isoformat()
|
||||
|
||||
# A start timestamp that is always past the 14-day deadline.
|
||||
EXPIRED_AT = (
|
||||
datetime.now(timezone.utc) - timedelta(days=MANUAL_LOCK_DAYS + 1)
|
||||
).isoformat()
|
||||
EXPIRED_AT = (datetime.now(UTC) - timedelta(days=MANUAL_LOCK_DAYS + 1)).isoformat()
|
||||
|
||||
ACTIVE_STATUS = TotalBlockStatus(
|
||||
active=True,
|
||||
started_at=datetime.now(timezone.utc) - timedelta(hours=1),
|
||||
until=datetime.now(timezone.utc) + timedelta(hours=23),
|
||||
started_at=datetime.now(UTC) - timedelta(hours=1),
|
||||
until=datetime.now(UTC) + timedelta(hours=23),
|
||||
days=1,
|
||||
days_remaining=0.96,
|
||||
)
|
||||
@ -106,8 +104,8 @@ def locked_state(
|
||||
)
|
||||
|
||||
|
||||
RECENT_PICK = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
||||
OLD_PICK = (datetime.now(timezone.utc) - timedelta(days=8)).isoformat()
|
||||
RECENT_PICK = (datetime.now(UTC) - timedelta(days=1)).isoformat()
|
||||
OLD_PICK = (datetime.now(UTC) - timedelta(days=8)).isoformat()
|
||||
|
||||
VALID_REASON = "I need this game installed for a work presentation this week."
|
||||
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
@ -16,7 +16,7 @@ from steam_backlog_enforcer.config import State
|
||||
|
||||
|
||||
def _iso_days_ago(days: float) -> str:
|
||||
return (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
|
||||
return (datetime.now(UTC) - timedelta(days=days)).isoformat()
|
||||
|
||||
|
||||
def _pick(app_id: int = 440, name: str = "TF2", days_ago: float = 1.0) -> dict:
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from steam_backlog_enforcer import _actions, _allowed_games
|
||||
from steam_backlog_enforcer._actions import (
|
||||
@ -19,7 +19,7 @@ from steam_backlog_enforcer.config import State
|
||||
|
||||
|
||||
def _iso_days_ago(days: float) -> str:
|
||||
return (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
|
||||
return (datetime.now(UTC) - timedelta(days=days)).isoformat()
|
||||
|
||||
|
||||
def _pick(app_id: int = 440, name: str = "TF2", days_ago: float = 1.0) -> dict:
|
||||
@ -56,9 +56,7 @@ class TestAbandonManualPick:
|
||||
state = self._state()
|
||||
abandon_manual_pick(state, 440)
|
||||
expiry = datetime.fromisoformat(state.skipped_until["440"])
|
||||
expected = datetime.now(timezone.utc) + timedelta(
|
||||
days=_actions.ABANDON_COOLDOWN_DAYS
|
||||
)
|
||||
expected = datetime.now(UTC) + timedelta(days=_actions.ABANDON_COOLDOWN_DAYS)
|
||||
assert abs((expiry - expected).total_seconds()) < 60
|
||||
|
||||
def test_persists_to_disk(self) -> None:
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer._actions import (
|
||||
@ -13,7 +13,7 @@ from steam_backlog_enforcer.config import State
|
||||
|
||||
|
||||
def _iso_days_ago(days: float) -> str:
|
||||
return (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
|
||||
return (datetime.now(UTC) - timedelta(days=days)).isoformat()
|
||||
|
||||
|
||||
def _pick(app_id: int = 440, name: str = "TF2", days_ago: float = 1.0) -> dict:
|
||||
@ -67,7 +67,7 @@ class TestStatusPayload:
|
||||
block = TotalBlockStatus(
|
||||
active=True,
|
||||
started_at=None,
|
||||
until=datetime(2030, 1, 1, tzinfo=timezone.utc),
|
||||
until=datetime(2030, 1, 1, tzinfo=UTC),
|
||||
days=3,
|
||||
days_remaining=2.55,
|
||||
)
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest.mock import patch
|
||||
@ -140,18 +141,18 @@ class TestState:
|
||||
state.skip_for_days(42, 7)
|
||||
assert "42" in state.skipped_until
|
||||
# Round-trip parse and check ~7 days in the future.
|
||||
from datetime import datetime, timezone
|
||||
from datetime import datetime
|
||||
|
||||
expiry = datetime.fromisoformat(state.skipped_until["42"])
|
||||
delta = (expiry - datetime.now(timezone.utc)).total_seconds()
|
||||
delta = (expiry - datetime.now(UTC)).total_seconds()
|
||||
assert 6 * 86400 < delta <= 7 * 86400 + 1
|
||||
|
||||
def test_active_skipped_ids_returns_active(self) -> None:
|
||||
"""Test active skipped ids returns active."""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
state = State()
|
||||
future = datetime.now(timezone.utc) + timedelta(days=3)
|
||||
future = datetime.now(UTC) + timedelta(days=3)
|
||||
state.skipped_until["100"] = future.isoformat()
|
||||
assert state.active_skipped_ids() == {100}
|
||||
# Active entry retained.
|
||||
@ -159,10 +160,10 @@ class TestState:
|
||||
|
||||
def test_active_skipped_ids_prunes_expired(self) -> None:
|
||||
"""Test active skipped ids prunes expired."""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
state = State()
|
||||
past = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
past = datetime.now(UTC) - timedelta(days=1)
|
||||
state.skipped_until["50"] = past.isoformat()
|
||||
assert state.active_skipped_ids() == set()
|
||||
assert "50" not in state.skipped_until
|
||||
|
||||
@ -3,10 +3,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Self
|
||||
from unittest.mock import patch
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_search import _AuthInfo
|
||||
from steam_backlog_enforcer.hltb import (
|
||||
_fetch_batch_confidence_only,
|
||||
|
||||
@ -3,11 +3,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Any
|
||||
from typing import Any, Self
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import aiohttp
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_comp_extract import _extract_dlc_relationships
|
||||
from steam_backlog_enforcer._hltb_detail import (
|
||||
|
||||
@ -2,9 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from typing_extensions import Self
|
||||
from typing import Any, Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_comp_extract import _extract_comp_100_avg_and_high
|
||||
|
||||
|
||||
@ -3,11 +3,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Any
|
||||
from typing import Any, Self
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_detail import _fetch_leisure_times
|
||||
from steam_backlog_enforcer._hltb_types import (
|
||||
HLTBResult,
|
||||
|
||||
@ -3,11 +3,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Any
|
||||
from typing import Any, Self
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_detail import _fetch_leisure_times
|
||||
from steam_backlog_enforcer._hltb_types import (
|
||||
_SAVE_INTERVAL,
|
||||
|
||||
@ -4,11 +4,10 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from typing import Any
|
||||
from typing import Any, Self
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import aiohttp
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_comp_extract import _fetch_detail_one
|
||||
from steam_backlog_enforcer._hltb_detail import _process_game_detail
|
||||
|
||||
@ -2,11 +2,9 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Self
|
||||
from unittest.mock import patch
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer.hltb import (
|
||||
fetch_hltb_detail_missing,
|
||||
)
|
||||
|
||||
@ -2,11 +2,9 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Self
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer.hltb import (
|
||||
HLTB_BASE_URL,
|
||||
HLTBResult,
|
||||
|
||||
@ -3,11 +3,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import TYPE_CHECKING, Any, Self
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import aiohttp
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_search import (
|
||||
_search_one,
|
||||
|
||||
@ -2,11 +2,9 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from typing import Any, Self
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_detail import _extract_leisure_hours
|
||||
from steam_backlog_enforcer._hltb_matching import _build_search_variants
|
||||
|
||||
|
||||
@ -3,11 +3,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import TYPE_CHECKING, Any, Self
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_search import (
|
||||
_fetch_batch,
|
||||
_SearchCtx,
|
||||
|
||||
@ -4,11 +4,9 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from typing import Any
|
||||
from typing import Any, Self
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_page_parse import _parse_game_page
|
||||
from steam_backlog_enforcer._hltb_search import _fetch_batch
|
||||
from steam_backlog_enforcer._hltb_search_api import _pick_best_hltb_entry
|
||||
|
||||
@ -3,11 +3,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import TYPE_CHECKING, Any, Self
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from steam_backlog_enforcer._hltb_search import (
|
||||
_search_one,
|
||||
_SearchCtx,
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
"""Tests for finalize_completion: snapshot refresh and install retry."""
|
||||
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest.mock import (
|
||||
patch,
|
||||
)
|
||||
@ -9,7 +9,9 @@ from steam_backlog_enforcer._cmd_done import (
|
||||
_finalize_completion,
|
||||
)
|
||||
from steam_backlog_enforcer.config import Config, State
|
||||
from steam_backlog_enforcer.steam_api import GameInfo
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from steam_backlog_enforcer.steam_api import GameInfo
|
||||
|
||||
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize"
|
||||
# _refresh_uncached_shortlist_hours stayed in _cmd_done, so its
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer._total_block import TotalBlockStatus
|
||||
@ -104,7 +104,7 @@ class TestCmdStatusLockHint:
|
||||
# ──────────────────────────────────────────────────────────────
|
||||
|
||||
# A pick made yesterday / one made over a week ago. Both are abandonable.
|
||||
_RECENT_PICK = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
||||
_RECENT_PICK = (datetime.now(UTC) - timedelta(days=1)).isoformat()
|
||||
|
||||
|
||||
def _abandonable_state(app_id: int = 100, started_at: str = _RECENT_PICK) -> State:
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer import _mcp_actions as mcp_actions
|
||||
@ -58,7 +58,7 @@ class TestAbandonPickGate:
|
||||
"""The MCP escape hatch mirrors the CLI grace rules, state-only."""
|
||||
|
||||
def _state(self, *, days_ago: float = 1.0, app_id: int = 440) -> State:
|
||||
started = (datetime.now(timezone.utc) - timedelta(days=days_ago)).isoformat()
|
||||
started = (datetime.now(UTC) - timedelta(days=days_ago)).isoformat()
|
||||
return State(
|
||||
manual_picks=[
|
||||
{"app_id": app_id, "game_name": "TF2", "started_at": started}
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
"""Tests for applying/releasing the playtime block and the kill walk."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer._playtime_block import (
|
||||
@ -10,6 +10,9 @@ from steam_backlog_enforcer._playtime_block import (
|
||||
release_block,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
PKG = "steam_backlog_enforcer._playtime_block"
|
||||
|
||||
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer._playtime_notify import (
|
||||
@ -20,6 +20,9 @@ from steam_backlog_enforcer._playtime_state import (
|
||||
)
|
||||
from steam_backlog_enforcer.config import Config
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
PKG = "steam_backlog_enforcer._playtime"
|
||||
LOCAL = timezone(timedelta(hours=2))
|
||||
NOW = datetime(2026, 7, 27, 20, 0, tzinfo=LOCAL)
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer._stats import (
|
||||
@ -123,11 +123,11 @@ class TestCmdStatsGroup2:
|
||||
def test_games_done_since_start_passed_to_pace(self) -> None:
|
||||
"""_print_pace_scenario gets only games completed after started_at."""
|
||||
|
||||
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
started = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
state = State(enforcement_started_at=started.isoformat())
|
||||
|
||||
after_ts = int(datetime(2026, 3, 1, tzinfo=timezone.utc).timestamp())
|
||||
before_ts = int(datetime(2025, 6, 1, tzinfo=timezone.utc).timestamp())
|
||||
after_ts = int(datetime(2026, 3, 1, tzinfo=UTC).timestamp())
|
||||
before_ts = int(datetime(2025, 6, 1, tzinfo=UTC).timestamp())
|
||||
|
||||
def _ach(ts: int) -> dict[str, object]:
|
||||
"""Test ach."""
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer._stats import (
|
||||
@ -167,7 +167,7 @@ class TestPrintPaceScenario:
|
||||
|
||||
def test_no_games_finished(self) -> None:
|
||||
"""Test no games finished."""
|
||||
started = datetime.now(timezone.utc) - timedelta(days=30)
|
||||
started = datetime.now(UTC) - timedelta(days=30)
|
||||
state = State(enforcement_started_at=started.isoformat())
|
||||
echoed: list[str] = []
|
||||
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
|
||||
@ -176,7 +176,7 @@ class TestPrintPaceScenario:
|
||||
|
||||
def test_normal_pace(self) -> None:
|
||||
"""Test normal pace."""
|
||||
started = datetime.now(timezone.utc) - timedelta(days=60)
|
||||
started = datetime.now(UTC) - timedelta(days=60)
|
||||
state = State(enforcement_started_at=started.isoformat())
|
||||
echoed: list[str] = []
|
||||
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
import json
|
||||
|
||||
from steam_backlog_enforcer._total_block import (
|
||||
@ -17,7 +17,7 @@ from steam_backlog_enforcer.tests._total_block_paths import (
|
||||
|
||||
PKG = "steam_backlog_enforcer._total_block"
|
||||
|
||||
_NOW = datetime.now(timezone.utc).timestamp()
|
||||
_NOW = datetime.now(UTC).timestamp()
|
||||
|
||||
|
||||
class TestIsTotalBlockActive:
|
||||
|
||||
@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import replace
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from steam_backlog_enforcer._web_dataset import (
|
||||
WebGame,
|
||||
@ -101,22 +101,22 @@ class TestCountCompleteSinceStart:
|
||||
|
||||
def test_counts_game_completed_after_start(self) -> None:
|
||||
"""Test counts game completed after start."""
|
||||
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
after_ts = int(datetime(2026, 6, 1, tzinfo=timezone.utc).timestamp())
|
||||
started = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
after_ts = int(datetime(2026, 6, 1, tzinfo=UTC).timestamp())
|
||||
games = [self._complete_game(1, after_ts)]
|
||||
assert count_complete_since_start(games, started.isoformat()) == 1
|
||||
|
||||
def test_excludes_game_completed_before_start(self) -> None:
|
||||
"""Test excludes game completed before start."""
|
||||
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
before_ts = int(datetime(2025, 6, 1, tzinfo=timezone.utc).timestamp())
|
||||
started = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
before_ts = int(datetime(2025, 6, 1, tzinfo=UTC).timestamp())
|
||||
games = [self._complete_game(1, before_ts)]
|
||||
assert count_complete_since_start(games, started.isoformat()) == 0
|
||||
|
||||
def test_excludes_incomplete_game(self) -> None:
|
||||
"""Test excludes incomplete game."""
|
||||
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
after_ts = int(datetime(2026, 6, 1, tzinfo=timezone.utc).timestamp())
|
||||
started = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
after_ts = int(datetime(2026, 6, 1, tzinfo=UTC).timestamp())
|
||||
incomplete = _gi(
|
||||
app_id=1,
|
||||
total_achievements=5,
|
||||
@ -127,7 +127,7 @@ class TestCountCompleteSinceStart:
|
||||
|
||||
def test_excludes_game_with_no_achievement_timestamps(self) -> None:
|
||||
"""Complete game with unlock_time=0 on all achievements is excluded."""
|
||||
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
started = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
no_ts = _gi(
|
||||
app_id=1,
|
||||
total_achievements=5,
|
||||
@ -138,9 +138,9 @@ class TestCountCompleteSinceStart:
|
||||
|
||||
def test_mixed_games_counts_only_post_start(self) -> None:
|
||||
"""Test mixed games counts only post start."""
|
||||
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
after_ts = int(datetime(2026, 6, 1, tzinfo=timezone.utc).timestamp())
|
||||
before_ts = int(datetime(2025, 6, 1, tzinfo=timezone.utc).timestamp())
|
||||
started = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
after_ts = int(datetime(2026, 6, 1, tzinfo=UTC).timestamp())
|
||||
before_ts = int(datetime(2025, 6, 1, tzinfo=UTC).timestamp())
|
||||
games = [
|
||||
self._complete_game(1, after_ts),
|
||||
self._complete_game(2, before_ts),
|
||||
@ -150,9 +150,9 @@ class TestCountCompleteSinceStart:
|
||||
|
||||
def test_uses_max_unlock_time_across_achievements(self) -> None:
|
||||
"""Game counts if its LAST achievement was unlocked after start."""
|
||||
started = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
before_ts = int(datetime(2025, 12, 1, tzinfo=timezone.utc).timestamp())
|
||||
after_ts = int(datetime(2026, 2, 1, tzinfo=timezone.utc).timestamp())
|
||||
started = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
before_ts = int(datetime(2025, 12, 1, tzinfo=UTC).timestamp())
|
||||
after_ts = int(datetime(2026, 2, 1, tzinfo=UTC).timestamp())
|
||||
# Mix of before/after timestamps — max is after start, so should count
|
||||
achs = [self._ach(before_ts)] * 4 + [self._ach(after_ts)]
|
||||
game = _gi(
|
||||
@ -184,7 +184,7 @@ class TestStateInfo:
|
||||
|
||||
def test_valid_start_with_games(self) -> None:
|
||||
"""Test valid start with games."""
|
||||
started = datetime.now(timezone.utc) - timedelta(days=50)
|
||||
started = datetime.now(UTC) - timedelta(days=50)
|
||||
info = _state_info(
|
||||
State(enforcement_started_at=started.isoformat()),
|
||||
games_done=12,
|
||||
@ -197,7 +197,7 @@ class TestStateInfo:
|
||||
|
||||
def test_valid_start_zero_since_start_keeps_zero_pace(self) -> None:
|
||||
"""games_done_since_start=0 → pace stays 0 even if total games_done > 0."""
|
||||
started = datetime.now(timezone.utc) - timedelta(days=50)
|
||||
started = datetime.now(UTC) - timedelta(days=50)
|
||||
info = _state_info(
|
||||
State(enforcement_started_at=started.isoformat()),
|
||||
games_done=5,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user