diff --git a/.mcp.json b/.mcp.json new file mode 100644 index 0000000..939c950 --- /dev/null +++ b/.mcp.json @@ -0,0 +1,10 @@ +{ + "mcpServers": { + "steam-backlog-enforcer": { + "type": "stdio", + "command": "${HOME}/.venvs/steam-backlog-mcp/bin/python", + "args": ["-m", "steam_backlog_enforcer._mcp"], + "env": {} + } + } +} diff --git a/README.md b/README.md index 2ed4598..fd50a85 100644 --- a/README.md +++ b/README.md @@ -53,3 +53,26 @@ Preserves Proton versions and Steam Linux Runtime. ```bash python -m python_pkg.steam_backlog_enforcer.main uninstall ``` + +## MCP server (Claude Code integration) + +The enforcer exposes an MCP server (`steam_backlog_enforcer._mcp`) so Claude Code +and its subagents can query the backlog and drive it through typed tools. + +- **Read tools:** `get_dataset`, `get_status`, `get_stats`, `list_backlog`. +- **Gated write tools:** `pick_manual`, `block_gaming` — each defaults to a + dry-run **preview**; pass `confirm=true` to actually apply. These write tools + must **never** be added to a permission allowlist (a subagent could otherwise + bypass the human confirmation). + +The `mcp` SDK is an optional dependency (`pip install -e '.[mcp]'`), kept out of +the CLI/systemd system-python path. One-time setup of the dedicated venv that +Claude Code spawns: + +```bash +./scripts/setup_mcp.sh +``` + +Registration lives in the checked-in [`.mcp.json`](./.mcp.json) (project scope). +Restart Claude Code in this repo and approve the project MCP-server prompt for it +to load. Verify with `claude mcp list`. diff --git a/pyproject.toml b/pyproject.toml index 3958c33..61bec1d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,8 +6,30 @@ requires-python = ">=3.10" dependencies = [ "requests>=2.0", "howlongtobeatpy>=1.0", + # Directly imported by the package (aiohttp: _hltb_* async lookups; + # websockets: library_hider). aiohttp was previously satisfied only + # transitively via howlongtobeatpy and websockets was undeclared — both + # are declared here so a clean install imports the package. + "aiohttp>=3.0", + "websockets>=11.0", ] +# Optional extras. The `mcp` group is ONLY needed by the MCP server +# (steam_backlog_enforcer._mcp), which Claude Code spawns via a dedicated +# venv (see scripts/setup_mcp.sh / .mcp.json). It is deliberately NOT a core +# dependency so the systemd/CLI system-python path stays dependency-light and +# never has to import the MCP SDK. +[project.optional-dependencies] +mcp = ["mcp>=1.9"] + +# Explicit package discovery. Without this, setuptools' flat-layout +# auto-discovery aborts because the repo root holds two top-level dirs +# (`steam_backlog_enforcer/` and the React `web/` frontend) and it refuses +# to guess. Scoping discovery to the Python package fixes `pip install -e .` +# for the CLI, systemd, AND the dedicated MCP venv. +[tool.setuptools.packages.find] +include = ["steam_backlog_enforcer*"] + # ============================================================================ # RUFF - Fast Python linter and formatter # ============================================================================ diff --git a/scripts/setup_mcp.sh b/scripts/setup_mcp.sh new file mode 100755 index 0000000..d487fb1 --- /dev/null +++ b/scripts/setup_mcp.sh @@ -0,0 +1,41 @@ +#!/bin/bash + +# ============================================================================ +# Set up the dedicated virtualenv that hosts the Steam Backlog Enforcer MCP +# server. Claude Code spawns this interpreter (see the repo-root .mcp.json) to +# run `python -m steam_backlog_enforcer._mcp` over stdio. +# +# The MCP SDK (`mcp`) lives ONLY in this venv — it is an optional dependency, +# deliberately kept out of the system/CLI/systemd python path. Both `mcp` and +# `steam_backlog_enforcer` must be importable by this one interpreter or the +# MCP server fails to start silently. +# +# Idempotent: safe to re-run to pick up dependency changes. +# ============================================================================ + +set -euo pipefail + +readonly SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +readonly REPO_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)" +readonly VENV_DIR="${HOME}/.venvs/steam-backlog-mcp" + +main() { + echo "== Setting up MCP venv at ${VENV_DIR} ==" + + if [[ ! -d "${VENV_DIR}" ]]; then + python3 -m venv "${VENV_DIR}" + fi + + "${VENV_DIR}/bin/pip" install --quiet --upgrade pip + "${VENV_DIR}/bin/pip" install --quiet -e "${REPO_DIR}[mcp]" + + echo "== Verifying imports ==" + "${VENV_DIR}/bin/python" -c \ + "import mcp, steam_backlog_enforcer; print('mcp + steam_backlog_enforcer import OK')" + + echo + echo "Done. The server is registered via ${REPO_DIR}/.mcp.json" + echo "Restart Claude Code in this repo and approve the project MCP server prompt." +} + +main "$@" diff --git a/steam_backlog_enforcer/_actions.py b/steam_backlog_enforcer/_actions.py new file mode 100644 index 0000000..54c2927 --- /dev/null +++ b/steam_backlog_enforcer/_actions.py @@ -0,0 +1,119 @@ +"""Stdout-free, state-only core operations shared by the CLI and MCP server. + +Every function here is safe to call from a stdio MCP server, where STDOUT +carries the JSON-RPC protocol and any stray write corrupts the stream. That +means: no ``print``/``_echo``/``sys.stdout`` writes, no ``input()``, and no +``sys.exit()``. The interactive CLI (``main.py``) reuses these same functions so +there is a single tested implementation of the underlying behaviour. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Any + +from steam_backlog_enforcer._total_block import get_total_block_status +from steam_backlog_enforcer.game_install import get_installed_games, is_protected_app +from steam_backlog_enforcer.store_blocker import is_store_blocked + +if TYPE_CHECKING: + from steam_backlog_enforcer.config import State + +# Days before the manual-pick lock automatically expires. Single source of +# truth: both ``main.py`` and the MCP server import it from here. +MANUAL_LOCK_DAYS = 14 + + +def is_manual_pick_locked(state: State) -> bool: + """Return ``True`` if the manual-pick lock is currently in force. + + The lock releases once the picked game reaches ``finished_app_ids`` or once + ``MANUAL_LOCK_DAYS`` have elapsed since the pick timestamp. + + Args: + state: The loaded enforcer state. + + Returns: + Whether the manual-pick lock is active right now. + """ + if state.manual_pick_app_id is None: + return False + + # Lock released once the game appears in finished_app_ids. + if state.manual_pick_app_id in state.finished_app_ids: + return False + + # Lock released after MANUAL_LOCK_DAYS days from the pick timestamp. + if state.manual_pick_started_at: + try: + started = datetime.fromisoformat(state.manual_pick_started_at) + deadline = started + timedelta(days=MANUAL_LOCK_DAYS) + if datetime.now(timezone.utc) >= deadline: + return False + except ValueError: + pass + + return True + + +def apply_manual_pick(state: State, app_id: int, game_name: str) -> None: + """Assign *app_id* as the manually-picked game and persist ``state``. + + This is the non-interactive, side-effect-scoped core of the CLI's + ``pick-manual`` command. It mutates and saves ``State`` only; it deliberately + does **not** run the destructive post-assignment cascade (uninstalling other + games, installing the pick, hiding the library) that the CLI performs after + its interactive ``YES`` confirmation. Keeping this state-only means an + automated caller (the MCP server) can never wipe installed games. + + Args: + state: The enforcer state to mutate and save. + app_id: The Steam app id to lock in. + game_name: Human-readable name for the picked game. + """ + now = datetime.now(timezone.utc).isoformat() + state.manual_pick_app_id = app_id + state.manual_pick_game_name = game_name + state.manual_pick_started_at = now + state.current_app_id = app_id + state.current_game_name = game_name + if not state.enforcement_started_at: + state.enforcement_started_at = now + state.save() + + +def status_payload(state: State) -> dict[str, Any]: + """Build the structured status snapshot that ``cmd_status`` renders as text. + + Pure data, no stdout — safe for the MCP server. Reads only ``State`` and the + stdout-free leaf helpers; never constructs ``Config`` and never exposes the + Steam API key. + + Args: + state: The loaded enforcer state. + + Returns: + A JSON-ready dict describing the current enforcement status. + """ + total_block = get_total_block_status() + installed = get_installed_games() + real_games = [(aid, name) for aid, name in installed if not is_protected_app(aid)] + assigned_installed = ( + any(aid == state.current_app_id for aid, _ in installed) + if state.current_app_id + else None + ) + return { + "current_app_id": state.current_app_id, + "current_game_name": state.current_game_name or None, + "finished_count": len(state.finished_app_ids), + "store_blocked": is_store_blocked(), + "installed_count": len(real_games), + "assigned_game_installed": assigned_installed, + "total_block": { + "active": total_block.active, + "days_remaining": round(total_block.days_remaining, 1), + "until": total_block.until.isoformat() if total_block.until else None, + }, + "manual_pick_locked": is_manual_pick_locked(state), + } diff --git a/steam_backlog_enforcer/_mcp.py b/steam_backlog_enforcer/_mcp.py new file mode 100644 index 0000000..739330d --- /dev/null +++ b/steam_backlog_enforcer/_mcp.py @@ -0,0 +1,261 @@ +"""MCP (Model Context Protocol) server for Steam Backlog Enforcer. + +Exposes the enforcer's read surface and two *gated* write actions as typed MCP +tools, so an MCP client (Claude Code and its subagents) can query and — with +explicit confirmation — drive the backlog without shelling out to the CLI. + +Run via the dedicated venv that has the ``mcp`` extra installed:: + + ~/.venvs/steam-backlog-mcp/bin/python -m steam_backlog_enforcer._mcp + +(see ``scripts/setup_mcp.sh`` and the repo-root ``.mcp.json``). + +Safety invariants (do not break when adding tools): + * **stdout is the JSON-RPC channel.** This module and every function a tool + calls must never write to stdout. All logging is routed to STDERR below, and + tools call only stdout-free leaf helpers (never the ``cmd_*`` handlers, which + ``_echo`` to stdout / ``input()`` / ``sys.exit()``). + * **No secret ever leaves.** There is no tool that returns ``Config`` or reads + ``config.json``; read tools load only ``State`` and the secrets-free web + dataset. Game-name lookups use the on-disk snapshot only (no Steam API key). + * **Writes are gated.** Every write tool defaults to a dry-run preview and + mutates only when ``confirm=True``. These write tools must never be added to + a permission allowlist (a subagent could then bypass the human). +""" + +from __future__ import annotations + +import logging +import sys +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from steam_backlog_enforcer._actions import ( + MANUAL_LOCK_DAYS, + apply_manual_pick, + status_payload, +) +from steam_backlog_enforcer._total_block import start_total_block +from steam_backlog_enforcer._web_dataset import build_web_dataset, dataset_to_payload +from steam_backlog_enforcer.config import State, load_snapshot +from steam_backlog_enforcer.steam_api import GameInfo + +# Log to STDERR only — STDOUT carries the MCP JSON-RPC protocol frames, so a +# single stray stdout write would corrupt the stream and kill the session. +logging.basicConfig( + level=logging.INFO, + stream=sys.stderr, + format="%(asctime)s [%(levelname)s] steam-mcp: %(message)s", +) +logger = logging.getLogger(__name__) + +mcp = FastMCP("steam-backlog-enforcer") + +_DEFAULT_LIST_LIMIT = 50 + + +def _resolve_game_name(app_id: int) -> str | None: + """Resolve a game name from the on-disk snapshot only (no Config / API). + + Deliberately snapshot-only: it never constructs ``Config`` and never touches + the Steam API key, so it is safe to call from an MCP tool. A game absent from + the last snapshot simply resolves to ``None``. + + Args: + app_id: The Steam app id to look up. + + Returns: + The game name, or ``None`` if not present in the snapshot. + """ + snapshot = load_snapshot() + if snapshot: + for entry in snapshot: + if entry.get("app_id") == app_id: + return str(entry["name"]) + return None + + +def _backlog_sort_key(game: GameInfo) -> tuple[int, float]: + """Sort incomplete games shortest-completionist-first, unknowns last.""" + if game.completionist_hours > 0: + return (0, game.completionist_hours) + return (1, 0.0) + + +# ────────────────────────────────────────────────────────────── +# Read tools (State-only; never expose Config / the Steam API key) +# ────────────────────────────────────────────────────────────── + + +@mcp.tool() +def get_dataset() -> dict[str, Any]: + """Return the full, secrets-free backlog dataset. + + Includes every incomplete candidate game with HowLongToBeat times, + ProtonDB tiers, pace-vs-HLTB calibration, and the default CLI thresholds — + the same projection the local web UI consumes. Reads on-disk caches only + (no network, no Steam API key). + """ + return dataset_to_payload(build_web_dataset(State.load())) + + +@mcp.tool() +def get_status() -> dict[str, Any]: + """Return the current enforcer status. + + Reports the assigned game, finished count, whether the Steam store is + blocked, installed-game count, any active total block (with days remaining), + and whether a manual-pick lock is in force. + """ + return status_payload(State.load()) + + +@mcp.tool() +def get_stats() -> dict[str, Any]: + """Return backlog completion-time estimates. + + A focused subset of the dataset: the default qualifying-games summary + (rush / leisure / worst-case totals) and the player's measured pace versus + HowLongToBeat. + """ + payload = dataset_to_payload(build_web_dataset(State.load())) + return { + "default_summary": payload["default_summary"], + "pace_vs_hltb": payload["pace_vs_hltb"], + } + + +@mcp.tool() +def list_backlog(limit: int = _DEFAULT_LIST_LIMIT) -> dict[str, Any]: + """List incomplete games, shortest-completionist-first, capped at ``limit``. + + Args: + limit: Maximum number of games to return (non-positive returns none). + """ + snapshot = load_snapshot() + if snapshot is None: + return { + "total": 0, + "returned": 0, + "games": [], + "note": "No snapshot found. Run 'scan' first.", + } + games = [GameInfo.from_snapshot(entry) for entry in snapshot] + incomplete = sorted((g for g in games if not g.is_complete), key=_backlog_sort_key) + capped = incomplete[: max(0, limit)] + return { + "total": len(incomplete), + "returned": len(capped), + "games": [ + { + "app_id": g.app_id, + "name": g.name, + "completion_pct": round(g.completion_pct, 1), + "completionist_hours": g.completionist_hours, + } + for g in capped + ], + } + + +# ────────────────────────────────────────────────────────────── +# Gated write tools (preview unless confirm=True; NEVER allowlist these) +# ────────────────────────────────────────────────────────────── + + +@mcp.tool() +def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]: + """Lock the enforcer to a game by Steam app_id (gated write). + + With ``confirm=False`` (the default) this performs **no** mutation and + returns a preview of what confirming would do. Call again with + ``confirm=True`` to apply. Applying overrides the current assignment and + locks all other commands for ``MANUAL_LOCK_DAYS`` days (or until 100% + achievements). Unlike the CLI's ``pick-manual``, this mutates **state only** + — it never uninstalls, installs, or hides games. + + Args: + app_id: The Steam app id to lock in (must exist in the last snapshot). + confirm: Set ``True`` to actually apply the pick; otherwise preview. + """ + game_name = _resolve_game_name(app_id) + if game_name is None: + return { + "ok": False, + "reason": f"AppID={app_id} not found in the snapshot. Run 'scan' first.", + } + if not confirm: + return { + "ok": True, + "preview": True, + "action": "pick_manual", + "app_id": app_id, + "game_name": game_name, + "effect": ( + "Overrides the current assignment and locks all other commands " + f"for {MANUAL_LOCK_DAYS} days or until 100% achievements." + ), + "confirm_required": True, + } + apply_manual_pick(State.load(), app_id, game_name) + logger.info("pick_manual applied: %s (AppID=%s)", game_name, app_id) + return { + "ok": True, + "applied": True, + "action": "pick_manual", + "app_id": app_id, + "game_name": game_name, + } + + +@mcp.tool() +def block_gaming(days: int, *, confirm: bool = False) -> dict[str, Any]: + """Start a total gaming block for ``days`` days (gated, privileged write). + + With ``confirm=False`` (the default) returns a preview only. With + ``confirm=True`` it attempts the block, which **requires root** (it edits + ``/etc/hosts`` and uninstalls Steam). The MCP server runs unprivileged, so + this will normally return ``{"ok": false, "reason": "requires elevated + privileges"}`` rather than succeeding. There is no in-app undo. + + Args: + days: Whole number of days to block all gaming (must be >= 1). + confirm: Set ``True`` to attempt the block; otherwise preview. + """ + if days < 1: + return {"ok": False, "reason": "days must be at least 1."} + if not confirm: + return { + "ok": True, + "preview": True, + "action": "block_gaming", + "days": days, + "effect": ( + f"For {days} day(s): uninstalls Steam, kills game launchers, and " + "blocks Steam + game-website domains. There is NO in-app undo." + ), + "requires_root": True, + "confirm_required": True, + } + try: + applied = start_total_block(days) + except (OSError, RuntimeError) as exc: # never crash the server on a write + logger.warning("block_gaming failed: %s", exc) + return {"ok": False, "reason": "requires elevated privileges"} + if not applied: + return {"ok": False, "reason": "requires elevated privileges"} + logger.info("block_gaming applied: %d day(s)", days) + return {"ok": True, "applied": True, "action": "block_gaming", "days": days} + + +def main() -> None: + """Run the MCP server over stdio (STDOUT = JSON-RPC, STDERR = logs).""" + logger.info( + "Starting steam-backlog-enforcer MCP server (python=%s)", sys.executable + ) + mcp.run() # pragma: no cover + + +if __name__ == "__main__": + main() diff --git a/steam_backlog_enforcer/main.py b/steam_backlog_enforcer/main.py index 4db532d..26e4221 100644 --- a/steam_backlog_enforcer/main.py +++ b/steam_backlog_enforcer/main.py @@ -8,6 +8,15 @@ import sys import time from typing import TYPE_CHECKING +from steam_backlog_enforcer._actions import ( + MANUAL_LOCK_DAYS as _MANUAL_LOCK_DAYS, +) +from steam_backlog_enforcer._actions import ( + apply_manual_pick, +) +from steam_backlog_enforcer._actions import ( + is_manual_pick_locked as _is_manual_pick_locked, +) from steam_backlog_enforcer._cmd_done import cmd_done from steam_backlog_enforcer._enforce_loop import ( do_enforce, @@ -72,9 +81,6 @@ logger = logging.getLogger(__name__) _LIST_DISPLAY_LIMIT = 50 _MIN_CLI_ARGS = 2 -# Days before the manual-pick lock automatically expires. -_MANUAL_LOCK_DAYS = 14 - # Commands that remain usable while the manual pick lock is active. # Principle: only what is needed to release the lock (done/check) or # that cannot change the game assignment (status, enforce, setup, serve). @@ -124,31 +130,10 @@ def _enforce_total_block_lock(command: str) -> None: # ────────────────────────────────────────────────────────────── # Manual pick lock helpers +# (the predicate itself lives in _actions so the MCP server can reuse it) # ────────────────────────────────────────────────────────────── -def _is_manual_pick_locked(state: State) -> bool: - """Return True if the manual-pick lock is currently in force.""" - if state.manual_pick_app_id is None: - return False - - # Lock released once the game appears in finished_app_ids. - if state.manual_pick_app_id in state.finished_app_ids: - return False - - # Lock released after 14 days from the pick timestamp. - if state.manual_pick_started_at: - try: - started = datetime.fromisoformat(state.manual_pick_started_at) - deadline = started + timedelta(days=_MANUAL_LOCK_DAYS) - if datetime.now(timezone.utc) >= deadline: - return False - except ValueError: - pass - - return True - - def _show_manual_pick_lock_message(state: State) -> None: """Print the aggressive lock-active message to stdout.""" _echo("\n" + "=" * 60) @@ -592,14 +577,9 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None: _echo("Aborted.") return - state.manual_pick_app_id = app_id - state.manual_pick_game_name = game_name - state.manual_pick_started_at = datetime.now(timezone.utc).isoformat() - state.current_app_id = app_id - state.current_game_name = game_name - if not state.enforcement_started_at: - state.enforcement_started_at = datetime.now(timezone.utc).isoformat() - state.save() + # State mutation is the shared, stdout-free core (also used by the MCP + # server); the destructive post-assignment cascade below stays CLI-only. + apply_manual_pick(state, app_id, game_name) _echo(f"\nManual pick confirmed: {game_name} (AppID={app_id})") _echo(f"Lock active from now until 100% achievements or {_MANUAL_LOCK_DAYS} days.") diff --git a/steam_backlog_enforcer/tests/test_actions.py b/steam_backlog_enforcer/tests/test_actions.py new file mode 100644 index 0000000..f276883 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_actions.py @@ -0,0 +1,130 @@ +"""Tests for the stdout-free, state-only cores in ``_actions``.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +from steam_backlog_enforcer import _actions +from steam_backlog_enforcer._actions import ( + apply_manual_pick, + is_manual_pick_locked, + status_payload, +) +from steam_backlog_enforcer._total_block import TotalBlockStatus +from steam_backlog_enforcer.config import State + + +def _iso_days_ago(days: float) -> str: + return (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() + + +class TestIsManualPickLocked: + def test_no_pick_is_unlocked(self) -> None: + assert is_manual_pick_locked(State()) is False + + def test_finished_game_releases_lock(self) -> None: + state = State(manual_pick_app_id=5, finished_app_ids=[5]) + assert is_manual_pick_locked(state) is False + + def test_empty_timestamp_stays_locked(self) -> None: + # manual pick set but no started_at → locked (no expiry to evaluate). + state = State(manual_pick_app_id=5, manual_pick_started_at="") + assert is_manual_pick_locked(state) is True + + def test_recent_pick_is_locked(self) -> None: + state = State(manual_pick_app_id=5, manual_pick_started_at=_iso_days_ago(1)) + assert is_manual_pick_locked(state) is True + + def test_expired_pick_releases_lock(self) -> None: + state = State( + manual_pick_app_id=5, + manual_pick_started_at=_iso_days_ago(_actions.MANUAL_LOCK_DAYS + 1), + ) + assert is_manual_pick_locked(state) is False + + def test_malformed_timestamp_stays_locked(self) -> None: + state = State(manual_pick_app_id=5, manual_pick_started_at="not-a-date") + assert is_manual_pick_locked(state) is True + + +class TestApplyManualPick: + def test_sets_all_fields_and_enforcement_start(self) -> None: + state = State() + apply_manual_pick(state, 440, "Team Fortress 2") + assert state.manual_pick_app_id == 440 + assert state.manual_pick_game_name == "Team Fortress 2" + assert state.current_app_id == 440 + assert state.current_game_name == "Team Fortress 2" + assert state.manual_pick_started_at != "" + # enforcement_started_at was empty, so it is set now. + assert state.enforcement_started_at != "" + + def test_preserves_existing_enforcement_start(self) -> None: + state = State(enforcement_started_at="2020-01-01T00:00:00+00:00") + apply_manual_pick(state, 620, "Portal 2") + assert state.enforcement_started_at == "2020-01-01T00:00:00+00:00" + + def test_persists_to_disk(self) -> None: + state = State() + apply_manual_pick(state, 70, "Half-Life") + assert State.load().current_app_id == 70 + + +class TestStatusPayload: + def _patch_leaves( + self, + *, + total_block: TotalBlockStatus, + installed: list[tuple[int, str]], + store_blocked: bool, + protected_ids: set[int], + ) -> object: + return patch.multiple( + "steam_backlog_enforcer._actions", + get_total_block_status=lambda: total_block, + get_installed_games=lambda: installed, + is_store_blocked=lambda: store_blocked, + is_protected_app=lambda aid: aid in protected_ids, + ) + + def test_assigned_and_installed(self) -> None: + block = TotalBlockStatus( + active=False, started_at=None, until=None, days=0, days_remaining=0.0 + ) + with self._patch_leaves( + total_block=block, + installed=[(440, "TF2"), (70, "HL")], + store_blocked=True, + protected_ids=set(), + ): + payload = status_payload(State(current_app_id=440, current_game_name="TF2")) + assert payload["current_app_id"] == 440 + assert payload["installed_count"] == 2 + assert payload["assigned_game_installed"] is True + assert payload["store_blocked"] is True + assert payload["manual_pick_locked"] is False + + def test_no_assignment_and_protected_filtering(self) -> None: + block = TotalBlockStatus( + active=True, + started_at=None, + until=datetime(2030, 1, 1, tzinfo=timezone.utc), + days=3, + days_remaining=2.55, + ) + with self._patch_leaves( + total_block=block, + installed=[(440, "TF2"), (228980, "Steamworks")], + store_blocked=False, + protected_ids={228980}, + ): + payload = status_payload(State()) + assert payload["current_app_id"] is None + assert payload["current_game_name"] is None + assert payload["assigned_game_installed"] is None + # protected app filtered out of the count. + assert payload["installed_count"] == 1 + assert payload["total_block"]["active"] is True + assert payload["total_block"]["days_remaining"] == 2.5 + assert payload["total_block"]["until"] == "2030-01-01T00:00:00+00:00" diff --git a/steam_backlog_enforcer/tests/test_mcp.py b/steam_backlog_enforcer/tests/test_mcp.py new file mode 100644 index 0000000..5b28682 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_mcp.py @@ -0,0 +1,211 @@ +"""Tests for the MCP server tools in ``_mcp``. + +Tools are patched at the ``_mcp`` module namespace (where the helpers are +imported), keeping each tool's own logic under test while isolating the +already-tested leaf functions. +""" + +from __future__ import annotations + +from unittest.mock import patch + +from steam_backlog_enforcer import _mcp +from steam_backlog_enforcer.steam_api import GameInfo + + +def _game(app_id: int, hours: float) -> GameInfo: + return GameInfo( + app_id=app_id, + name=f"g{app_id}", + total_achievements=10, + unlocked_achievements=1, + playtime_minutes=0, + completionist_hours=hours, + ) + + +class TestReadTools: + def test_get_dataset(self) -> None: + with ( + patch.object(_mcp, "State") as state, + patch.object(_mcp, "build_web_dataset", return_value="DS") as bwd, + patch.object(_mcp, "dataset_to_payload", return_value={"x": 1}) as dtp, + ): + assert _mcp.get_dataset() == {"x": 1} + bwd.assert_called_once_with(state.load.return_value) + dtp.assert_called_once_with("DS") + + def test_get_status(self) -> None: + with ( + patch.object(_mcp, "State") as state, + patch.object(_mcp, "status_payload", return_value={"ok": 1}) as sp, + ): + assert _mcp.get_status() == {"ok": 1} + sp.assert_called_once_with(state.load.return_value) + + def test_get_stats_subsets_dataset(self) -> None: + payload = { + "default_summary": {"qualifying": 3}, + "pace_vs_hltb": None, + "games": ["ignored"], + } + with ( + patch.object(_mcp, "State"), + patch.object(_mcp, "build_web_dataset"), + patch.object(_mcp, "dataset_to_payload", return_value=payload), + ): + out = _mcp.get_stats() + assert out == {"default_summary": {"qualifying": 3}, "pace_vs_hltb": None} + + +class TestListBacklog: + def test_no_snapshot_returns_note(self) -> None: + with patch.object(_mcp, "load_snapshot", return_value=None): + out = _mcp.list_backlog() + assert out["total"] == 0 + assert out["games"] == [] + assert "note" in out + + def test_sorts_shortest_first_excludes_complete_and_caps(self) -> None: + snap = [ + { + "app_id": 440, + "name": "TF2", + "total_achievements": 100, + "unlocked_achievements": 10, + "completionist_hours": 50.0, + }, + # complete → excluded + { + "app_id": 620, + "name": "Portal 2", + "total_achievements": 50, + "unlocked_achievements": 50, + "completionist_hours": 20.0, + }, + { + "app_id": 70, + "name": "HL", + "total_achievements": 10, + "unlocked_achievements": 1, + "completionist_hours": 12.0, + }, + # unknown hours → sorted last + { + "app_id": 30, + "name": "NoHrs", + "total_achievements": 10, + "unlocked_achievements": 0, + "completionist_hours": -1, + }, + ] + with patch.object(_mcp, "load_snapshot", return_value=snap): + out = _mcp.list_backlog(limit=2) + assert out["total"] == 3 + assert out["returned"] == 2 + assert [g["app_id"] for g in out["games"]] == [70, 440] + assert out["games"][0]["completion_pct"] == 10.0 + + def test_negative_limit_returns_none(self) -> None: + snap = [ + { + "app_id": 70, + "name": "HL", + "total_achievements": 10, + "unlocked_achievements": 1, + } + ] + with patch.object(_mcp, "load_snapshot", return_value=snap): + out = _mcp.list_backlog(limit=-5) + assert out["returned"] == 0 + assert out["games"] == [] + + +class TestResolveAndSort: + def test_resolve_found(self) -> None: + with patch.object( + _mcp, "load_snapshot", return_value=[{"app_id": 440, "name": "TF2"}] + ): + assert _mcp._resolve_game_name(440) == "TF2" + + def test_resolve_missing(self) -> None: + with patch.object( + _mcp, "load_snapshot", return_value=[{"app_id": 1, "name": "X"}] + ): + assert _mcp._resolve_game_name(440) is None + + def test_resolve_no_snapshot(self) -> None: + with patch.object(_mcp, "load_snapshot", return_value=None): + assert _mcp._resolve_game_name(440) is None + + def test_sort_key_branches(self) -> None: + assert _mcp._backlog_sort_key(_game(1, 5.0)) == (0, 5.0) + assert _mcp._backlog_sort_key(_game(2, -1)) == (1, 0.0) + + +class TestPickManualGate: + def test_not_found(self) -> None: + with patch.object(_mcp, "load_snapshot", return_value=[]): + out = _mcp.pick_manual(440) + assert out["ok"] is False + assert "not found" in out["reason"] + + def test_preview_does_not_mutate(self) -> None: + with ( + patch.object( + _mcp, "load_snapshot", return_value=[{"app_id": 440, "name": "TF2"}] + ), + patch.object(_mcp, "apply_manual_pick") as amp, + ): + out = _mcp.pick_manual(440) + assert out["preview"] is True + assert out["game_name"] == "TF2" + amp.assert_not_called() + + def test_confirm_applies(self) -> None: + with ( + patch.object( + _mcp, "load_snapshot", return_value=[{"app_id": 440, "name": "TF2"}] + ), + patch.object(_mcp, "State") as state, + patch.object(_mcp, "apply_manual_pick") as amp, + ): + out = _mcp.pick_manual(440, confirm=True) + assert out["applied"] is True + assert out["app_id"] == 440 + amp.assert_called_once_with(state.load.return_value, 440, "TF2") + + +class TestBlockGamingGate: + def test_invalid_days(self) -> None: + out = _mcp.block_gaming(0) + assert out["ok"] is False + + def test_preview(self) -> None: + out = _mcp.block_gaming(3) + assert out["preview"] is True + assert out["requires_root"] is True + + def test_confirm_success(self) -> None: + with patch.object(_mcp, "start_total_block", return_value=True): + out = _mcp.block_gaming(3, confirm=True) + assert out["applied"] is True + assert out["days"] == 3 + + def test_confirm_unprivileged_returns_gracefully(self) -> None: + with patch.object(_mcp, "start_total_block", return_value=False): + out = _mcp.block_gaming(3, confirm=True) + assert out["ok"] is False + assert "privileges" in out["reason"] + + def test_confirm_oserror_returns_gracefully(self) -> None: + with patch.object(_mcp, "start_total_block", side_effect=OSError("boom")): + out = _mcp.block_gaming(3, confirm=True) + assert out["ok"] is False + assert "privileges" in out["reason"] + + +def test_main_runs_stdio_server() -> None: + with patch.object(_mcp.mcp, "run") as run: + _mcp.main() + run.assert_called_once_with()