From 6b80cc781f5ab1044f83e57d062fca46f7707704 Mon Sep 17 00:00:00 2001 From: Samuel Gudi Date: Fri, 8 May 2026 13:18:57 +0200 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20Option=20A=20rewrite=20=E2=80=94?= =?UTF-8?q?=20import=20api.models/api.profiles=20canonically=20(#1616)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per maintainer review, replace duplicated I/O with canonical helpers for locking, profile scoping, index consistency, and validation. Profile scoping (#1614) enforced on all CRUD via _profiles_match matching api/routes.py:75 semantics exactly. AI-authored, human-reviewed. Co-Authored-By: Claude (Opus 4.7) --- mcp_server.py | 570 +++++++++++++++++++++++++++++++++++++++ tests/test_mcp_server.py | 385 ++++++++++++++++++++++++++ 2 files changed, 955 insertions(+) create mode 100644 mcp_server.py create mode 100644 tests/test_mcp_server.py diff --git a/mcp_server.py b/mcp_server.py new file mode 100644 index 00000000..9ef8aec4 --- /dev/null +++ b/mcp_server.py @@ -0,0 +1,570 @@ +#!/usr/bin/env python3 +""" +Hermes WebUI MCP Server — exposes project and session management +as MCP tools for any MCP-compatible agent. + +Option A rewrite (2026-05-08): imports api.models and api.profiles +directly from the webui codebase, using canonical helpers for +locking, profile scoping, index consistency, and validation. + + pip install mcp # one-time setup + python3 mcp_server.py # start via stdio + +MCP config for Hermes Agent (add to config.yaml): + mcp_servers: + hermes-webui: + command: /path/to/venv/bin/python3 + args: [/path/to/hermes-webui/mcp_server.py] + env: + HERMES_WEBUI_PASSWORD: your_password + +Profile override (optional): + args: [/path/to/hermes-webui/mcp_server.py, --profile, myprofile] + +AI-authoring disclosure: this file was rewritten by MILO (Hermes Agent) +under human direction, per maintainer guidelines for #1616. +""" + +import argparse +import json +import os +import re +import sys +import time +import uuid +from pathlib import Path + +from mcp.server import Server +from mcp.server.stdio import stdio_server +from mcp.types import Tool, TextContent + +# ── Ensure the repo root is on sys.path so api.* imports work ───────────── +_REPO_ROOT = Path(__file__).parent.resolve() +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + +# ── CLI: optional --profile override ────────────────────────────────────── +_profile_arg: str | None = None +_parser = argparse.ArgumentParser(add_help=False) +_parser.add_argument("--profile", type=str, default=None) +_args, _unknown = _parser.parse_known_args() +_profile_arg = _args.profile + +# ── Import webui canonical modules (after path setup) ───────────────────── +import api.config as _cfg +from api.config import ( + STATE_DIR, SESSION_DIR, SESSION_INDEX_FILE, PROJECTS_FILE, HOME, +) +from api.models import load_projects, save_projects +from api.profiles import get_active_profile_name, _is_root_profile + +# ── Apply --profile override before any module uses get_active_profile_name +if _profile_arg is not None: + import api.profiles as _profiles + _profiles._active_profile = _profile_arg + +# ── API auth state ───────────────────────────────────────────────────────── +WEBUI_URL = "http://127.0.0.1:8788" +_auth_cookie: str | None = None +_auth_expires: float = 0 # unix timestamp after which we re-auth + +server = Server("hermes-webui") + + +# ═══════════════════════════════════════════════════════════════════════════ +# Helpers — filesystem (project CRUD via canonical api.models) +# ═══════════════════════════════════════════════════════════════════════════ + +def _active_profile() -> str: + """Shorthand for the current profile name (--profile or auto-detected).""" + return get_active_profile_name() or 'default' + + +def _validate_color(color: str | None) -> str | None: + """Return an error string if color is invalid, else None.""" + if color is not None and not re.match(r"^#[0-9a-fA-F]{3,8}$", color): + return "Invalid color format (use #RGB, #RRGGBB, or #RRGGBBAA)" + return None + + +def _profiles_match(row_profile: str | None, active: str | None) -> bool: + """Cross-profile ownership check — mirrors api/routes.py:_profiles_match (#1614). + + A row with no profile (None or empty) is treated as belonging to the root + profile ('default'), per the canonical webapp convention. This keeps the + MCP visibility model identical to the HTTP API: a non-root profile cannot + see legacy untagged projects, only the root profile (or a renamed-root + alias) can. + """ + row = row_profile or 'default' + act = active or 'default' + if row == act: + return True + if _is_root_profile(row) and _is_root_profile(act): + return True + return False + + +def _load_index() -> list: + """Read the session index. Falls back to empty list on failure.""" + if not SESSION_INDEX_FILE.exists(): + return [] + try: + return json.loads(SESSION_INDEX_FILE.read_text(encoding="utf-8")) + except Exception: + return [] + + +def _session_compact(row: dict) -> dict: + """Lightweight compact representation of a session index entry.""" + return { + "session_id": row.get("session_id"), + "title": row.get("title"), + "project_id": row.get("project_id"), + "workspace": row.get("workspace"), + "model": row.get("model"), + "message_count": row.get("message_count", 0), + "source_tag": row.get("source_tag"), + "is_cli_session": row.get("is_cli_session", False), + "profile": row.get("profile"), + } + + +# ═══════════════════════════════════════════════════════════════════════════ +# Helpers — HTTP API (for mutations that need cache sync) +# ═══════════════════════════════════════════════════════════════════════════ + +def _api_password() -> str | None: + """Return the plaintext webui password from HERMES_WEBUI_PASSWORD, or None. + + settings.json stores only the bcrypt hash, which the login endpoint cannot + accept — it calls verify_password(plaintext) against the stored hash. So + there's no usable fallback when the env var is unset; the MCP simply runs + in unauthenticated mode and any auth-protected mutation will fail clearly + with the server's 401 instead of silently sending an unusable hash. + """ + pw = os.environ.get("HERMES_WEBUI_PASSWORD", "").strip() + return pw or None + + +def _api_auth() -> str | None: + """Authenticate and return cookie value, or None if auth disabled/fails.""" + global _auth_cookie, _auth_expires + + pw = _api_password() + if not pw: + return None # auth not enabled — API calls will fail anyway + + # Reuse cookie if still valid (25 days — server issues 30-day cookies) + if _auth_cookie and time.time() < _auth_expires: + return _auth_cookie + + import urllib.request + + try: + req = urllib.request.Request( + f"{WEBUI_URL}/api/auth/login", + data=json.dumps({"password": pw}).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + resp = urllib.request.urlopen(req, timeout=5) + cookie = resp.headers.get("Set-Cookie", "") + if cookie: + _auth_cookie = cookie.split(";")[0] # "hermes_session=VALUE; ..." + _auth_expires = time.time() + 25 * 86400 # 25 days + return _auth_cookie + except Exception: + _auth_cookie = None + return None + + +def _api_post(endpoint: str, body: dict) -> dict: + """POST to webui API with auth cookie. Returns parsed JSON response.""" + import urllib.request + import urllib.error + + cookie = _api_auth() + headers = {"Content-Type": "application/json"} + if cookie: + headers["Cookie"] = cookie + + try: + req = urllib.request.Request( + f"{WEBUI_URL}{endpoint}", + data=json.dumps(body).encode(), + headers=headers, + method="POST", + ) + resp = urllib.request.urlopen(req, timeout=5) + return json.loads(resp.read()) + except urllib.error.HTTPError as e: + err_body = json.loads(e.read()) + return {"error": f"API {e.code}: {err_body.get('error', 'unknown')}"} + except Exception as e: + return {"error": f"API unreachable: {e}"} + + +# ═══════════════════════════════════════════════════════════════════════════ +# Tool handlers — read-only (filesystem, profile-aware) +# ═══════════════════════════════════════════════════════════════════════════ + +async def handle_list_projects(_arguments: dict) -> list[TextContent]: + """List all projects with session counts, scoped to active profile.""" + projects = load_projects() + active = _active_profile() + index = _load_index() + + # Session counts per project (from index) + counts: dict[str, int] = {} + for s in index: + pid = s.get("project_id") + if pid: + counts[pid] = counts.get(pid, 0) + 1 + + result = [] + for p in projects: + # Profile filter: legacy untagged rows are treated as 'default' by + # _profiles_match, so non-root profiles correctly hide them. + if not _profiles_match(p.get("profile"), active): + continue + entry = dict(p) + entry["session_count"] = counts.get(p["project_id"], 0) + result.append(entry) + + return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False, indent=2))] + + +async def handle_list_sessions(arguments: dict) -> list[TextContent]: + """List sessions, optionally filtered by project or unassigned status.""" + project_id = arguments.get("project_id") + unassigned = arguments.get("unassigned", False) + limit = max(1, min(500, arguments.get("limit", 50))) + active = _active_profile() + + index = _load_index() + sessions = [_session_compact(s) for s in index if s.get("session_id")] + + # Filter by profile: legacy untagged rows are treated as 'default' by + # _profiles_match (canonical convention), so non-root profiles hide them. + sessions = [s for s in sessions if _profiles_match(s.get("profile"), active)] + + if unassigned: + sessions = [s for s in sessions if not s["project_id"]] + elif project_id: + sessions = [s for s in sessions if s["project_id"] == project_id] + + sessions = sessions[:limit] + return [TextContent(type="text", text=json.dumps(sessions, ensure_ascii=False, indent=2))] + + +# ═══════════════════════════════════════════════════════════════════════════ +# Tool handlers — project CRUD (canonical helpers, profile-scoped) +# ═══════════════════════════════════════════════════════════════════════════ + +async def handle_create_project(arguments: dict) -> list[TextContent]: + """Create a new project (profile-scoped, exact-match title collision).""" + name = arguments.get("name", "").strip()[:128] + if not name: + return [TextContent(type="text", text=json.dumps( + {"error": "name is required"}, ensure_ascii=False))] + + color = arguments.get("color") + color_err = _validate_color(color) + if color_err: + return [TextContent(type="text", text=json.dumps( + {"error": color_err}, ensure_ascii=False))] + + active = _active_profile() + projects = load_projects() + + # Title collision: exact match (consistent with ensure_cron_project) + if any(p.get("name") == name and _profiles_match(p.get("profile"), active) + for p in projects): + return [TextContent(type="text", text=json.dumps( + {"error": f"Project '{name}' already exists"}, ensure_ascii=False))] + + proj = { + "project_id": uuid.uuid4().hex[:12], + "name": name, + "color": color, + "profile": active, + "created_at": time.time(), + } + projects.append(proj) + save_projects(projects) + + proj["session_count"] = 0 + return [TextContent(type="text", text=json.dumps(proj, ensure_ascii=False, indent=2))] + + +async def handle_rename_project(arguments: dict) -> list[TextContent]: + """Rename a project and optionally change its color (profile-checked).""" + project_id = arguments.get("project_id") + name = arguments.get("name", "").strip()[:128] + if not project_id or not name: + return [TextContent(type="text", text=json.dumps( + {"error": "project_id and name are required"}, ensure_ascii=False))] + + color = arguments.get("color") + color_err = _validate_color(color) + if color_err: + return [TextContent(type="text", text=json.dumps( + {"error": color_err}, ensure_ascii=False))] + + active = _active_profile() + projects = load_projects() + proj = next((p for p in projects if p["project_id"] == project_id), None) + if not proj: + return [TextContent(type="text", text=json.dumps( + {"error": "Project not found"}, ensure_ascii=False))] + + # #1614: profile ownership check + if not _profiles_match(proj.get("profile"), active): + return [TextContent(type="text", text=json.dumps( + {"error": "Project not found"}, ensure_ascii=False))] + + proj["name"] = name + if color is not None: + proj["color"] = color + save_projects(projects) + return [TextContent(type="text", text=json.dumps(proj, ensure_ascii=False, indent=2))] + + +async def handle_delete_project(arguments: dict) -> list[TextContent]: + """Delete a project and unassign all its sessions (profile-checked).""" + project_id = arguments.get("project_id") + if not project_id: + return [TextContent(type="text", text=json.dumps( + {"error": "project_id is required"}, ensure_ascii=False))] + + active = _active_profile() + projects = load_projects() + proj = next((p for p in projects if p["project_id"] == project_id), None) + if not proj: + return [TextContent(type="text", text=json.dumps( + {"error": "Project not found"}, ensure_ascii=False))] + + # #1614: profile ownership check + if not _profiles_match(proj.get("profile"), active): + return [TextContent(type="text", text=json.dumps( + {"error": "Project not found"}, ensure_ascii=False))] + + projects = [p for p in projects if p["project_id"] != project_id] + save_projects(projects) + + # Unassign sessions — use API if auth available, filesystem otherwise + unassigned = 0 + has_auth = bool(_api_password()) + if SESSION_DIR.exists(): + for p in SESSION_DIR.glob("*.json"): + if p.name.startswith("_"): + continue + try: + session_data = json.loads(p.read_text(encoding="utf-8")) + if session_data.get("project_id") == project_id: + sid = p.stem + if has_auth: + result = _api_post("/api/session/move", + {"session_id": sid, "project_id": None}) + if "ok" in result or "session" in result: + unassigned += 1 + else: + # Filesystem fallback (may be overwritten by server cache) + session_data["project_id"] = None + tmp = p.with_suffix(".tmp") + tmp.write_text( + json.dumps(session_data, ensure_ascii=False, indent=2), + encoding="utf-8") + os.replace(tmp, p) + unassigned += 1 + except Exception: + pass + + return [TextContent(type="text", text=json.dumps({ + "ok": True, + "deleted": proj["name"], + "unassigned_sessions": unassigned, + }, ensure_ascii=False))] + + +# ═══════════════════════════════════════════════════════════════════════════ +# Tool handlers — mutations (HTTP API with auth, cache-safe) +# ═══════════════════════════════════════════════════════════════════════════ + +async def handle_rename_session(arguments: dict) -> list[TextContent]: + """Rename a session via the authenticated webui API (cache-safe).""" + session_id = arguments.get("session_id") + title = arguments.get("title", "").strip()[:80] + if not session_id or not title: + return [TextContent(type="text", text=json.dumps( + {"error": "session_id and title are required"}, ensure_ascii=False))] + + result = _api_post("/api/session/rename", + {"session_id": session_id, "title": title}) + if "error" in result: + return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))] + + session = result.get("session", {}) + return [TextContent(type="text", text=json.dumps({ + "ok": True, + "session_id": session_id, + "title": session.get("title", title), + "method": "api", + }, ensure_ascii=False, indent=2))] + + +async def handle_move_session(arguments: dict) -> list[TextContent]: + """Assign a session to a project via the authenticated webui API (cache-safe).""" + session_id = arguments.get("session_id") + project_id = arguments.get("project_id") # None/null = unassign + if not session_id: + return [TextContent(type="text", text=json.dumps( + {"error": "session_id is required"}, ensure_ascii=False))] + + # If project_id is provided, verify it exists and is profile-accessible + if project_id is not None: + projects = load_projects() + active = _active_profile() + target = next((p for p in projects if p["project_id"] == project_id), None) + if not target: + return [TextContent(type="text", text=json.dumps( + {"error": "Project not found"}, ensure_ascii=False))] + # #1614: refuse moves into projects owned by another profile + if not _profiles_match(target.get("profile"), active): + return [TextContent(type="text", text=json.dumps( + {"error": "Project not found"}, ensure_ascii=False))] + + result = _api_post("/api/session/move", + {"session_id": session_id, "project_id": project_id}) + if "error" in result: + return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))] + + session = result.get("session", {}) + return [TextContent(type="text", text=json.dumps({ + "ok": True, + "session_id": session_id, + "project_id": project_id, + "title": session.get("title"), + "method": "api", + }, ensure_ascii=False, indent=2))] + + +# ═══════════════════════════════════════════════════════════════════════════ +# MCP Server wiring +# ═══════════════════════════════════════════════════════════════════════════ + +TOOLS = [ + Tool( + name="list_projects", + description="List all session projects with their IDs, names, colors, and session counts (scoped to active profile).", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="create_project", + description="Create a new project for organizing sessions (profile-scoped).", + inputSchema={ + "type": "object", + "properties": { + "name": {"type": "string", "description": "Project name (max 128 chars)"}, + "color": {"type": "string", "description": "Optional hex color (#RGB, #RRGGBB, or #RRGGBBAA)"}, + }, + "required": ["name"], + }, + ), + Tool( + name="rename_project", + description="Rename a project and optionally change its color (profile-checked).", + inputSchema={ + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "12-char project ID"}, + "name": {"type": "string", "description": "New name (max 128 chars)"}, + "color": {"type": "string", "description": "Optional new hex color"}, + }, + "required": ["project_id", "name"], + }, + ), + Tool( + name="delete_project", + description="Delete a project and unassign all its sessions (profile-checked).", + inputSchema={ + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "12-char project ID to delete"}, + }, + "required": ["project_id"], + }, + ), + Tool( + name="rename_session", + description="Rename a session (updates sidebar via authenticated API, cache-safe).", + inputSchema={ + "type": "object", + "properties": { + "session_id": {"type": "string", "description": "Session ID"}, + "title": {"type": "string", "description": "New title (max 80 chars)"}, + }, + "required": ["session_id", "title"], + }, + ), + Tool( + name="move_session", + description="Assign a session to a project. Pass project_id=null to unassign. Uses authenticated API for cache safety (profile-checked).", + inputSchema={ + "type": "object", + "properties": { + "session_id": {"type": "string", "description": "Session ID"}, + "project_id": {"type": ["string", "null"], "description": "Project ID (or null to unassign)"}, + }, + "required": ["session_id", "project_id"], + }, + ), + Tool( + name="list_sessions", + description="List sessions, optionally filtered by project or unassigned status (profile-scoped).", + inputSchema={ + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Filter sessions by project ID"}, + "unassigned": {"type": "boolean", "description": "Show only sessions with no project"}, + "limit": {"type": "integer", "description": "Max results (default: 50, max: 500)"}, + }, + "required": [], + }, + ), +] + +HANDLERS = { + "list_projects": handle_list_projects, + "create_project": handle_create_project, + "rename_project": handle_rename_project, + "delete_project": handle_delete_project, + "rename_session": handle_rename_session, + "move_session": handle_move_session, + "list_sessions": handle_list_sessions, +} + + +@server.list_tools() +async def list_tools() -> list[Tool]: + return TOOLS + + +@server.call_tool() +async def call_tool(name: str, arguments: dict) -> list[TextContent]: + handler = HANDLERS.get(name) + if not handler: + return [TextContent(type="text", text=json.dumps( + {"error": f"Unknown tool: {name}"}, ensure_ascii=False))] + return await handler(arguments) + + +async def main(): + async with stdio_server() as (read, write): + await server.run(read, write, server.create_initialization_options()) + + +if __name__ == "__main__": + import asyncio + asyncio.run(main()) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py new file mode 100644 index 00000000..f83fa7df --- /dev/null +++ b/tests/test_mcp_server.py @@ -0,0 +1,385 @@ +"""Tests for mcp_server.py — Option A rewrite (Issue #1616). + +Covers: project CRUD, profile scoping, title collision, color validation, +session listing, cross-profile isolation. + +Uses HERMES_WEBUI_STATE_DIR env var to point to a temp directory, +so tests don't touch the real webui state. Module is re-imported +per test class to ensure clean state. +""" + +import json +import os +import sys +import tempfile +import uuid +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.asyncio + +# ── Ensure repo root on path ────────────────────────────────────────────── +_REPO = Path(__file__).parent.parent.resolve() +if str(_REPO) not in sys.path: + sys.path.insert(0, str(_REPO)) + + +# ═══════════════════════════════════════════════════════════════════════════ +# Helpers +# ═══════════════════════════════════════════════════════════════════════════ + +def _fresh_state_dir(): + """Create a clean temp state dir and set HERMES_WEBUI_STATE_DIR.""" + td = tempfile.mkdtemp() + state_dir = Path(td) + sessions_dir = state_dir / "sessions" + sessions_dir.mkdir(parents=True) + (state_dir / "projects.json").write_text("[]", encoding="utf-8") + (sessions_dir / "_index.json").write_text("[]", encoding="utf-8") + os.environ["HERMES_WEBUI_STATE_DIR"] = str(state_dir) + return state_dir + + +def _cleanup_state_dir(state_dir: Path): + """Remove temp state dir and clear env var.""" + import shutil + shutil.rmtree(state_dir, ignore_errors=True) + os.environ.pop("HERMES_WEBUI_STATE_DIR", None) + + +def _reimport_mcp(): + """Re-import mcp_server with current env vars and profile. + + Returns (mcp_module, profiles_module) — profiles_module is the + live api.profiles reference that the re-imported mcp_server uses. + """ + # Clear cached module and api submodules that cache paths + for key in list(sys.modules.keys()): + if key == 'mcp_server' or key.startswith('mcp_server.') or \ + key == 'api.config' or key == 'api.models' or key == 'api.profiles': + del sys.modules[key] + + import importlib + import api.config as cfg + importlib.reload(cfg) + + # Re-acquire api.profiles reference (old one is stale after sys.modules clear) + import api.profiles as fresh_profiles + fresh_profiles._active_profile = 'default' + + import mcp_server as mod + return mod, fresh_profiles + + +async def _call(mod, tool_name, **kwargs): + """Call a tool handler and return parsed JSON.""" + handler = mod.HANDLERS[tool_name] + result = await handler(kwargs) + return json.loads(result[0].text) + + +# ═══════════════════════════════════════════════════════════════════════════ +# Project CRUD +# ═══════════════════════════════════════════════════════════════════════════ + +class TestCreateProject: + @pytest.fixture(autouse=True) + def setup(self): + self.state_dir = _fresh_state_dir() + self.mod, self.profiles = _reimport_mcp() + yield + _cleanup_state_dir(self.state_dir) + + async def test_create_basic(self): + result = await _call(self.mod, "create_project", name="Test Project") + assert "project_id" in result + assert result["name"] == "Test Project" + assert result["profile"] == "default" + assert result["session_count"] == 0 + + async def test_create_with_color(self): + result = await _call(self.mod, "create_project", + name="Colored", color="#ff6600") + assert result["color"] == "#ff6600" + + async def test_create_duplicate_exact_match(self): + await _call(self.mod, "create_project", name="My Project") + result = await _call(self.mod, "create_project", name="My Project") + assert "error" in result + assert "already exists" in result["error"] + + async def test_create_case_sensitive_no_collision(self): + """Exact match: 'MY project' and 'My Project' are different.""" + await _call(self.mod, "create_project", name="My Project") + result = await _call(self.mod, "create_project", name="MY project") + assert "project_id" in result + + async def test_create_empty_name(self): + result = await _call(self.mod, "create_project", name="") + assert "error" in result + + async def test_create_invalid_color(self): + result = await _call(self.mod, "create_project", + name="Bad", color="not-a-color") + assert "error" in result + assert "Invalid color" in result["error"] + + async def test_create_valid_color_formats(self): + for color in ["#fff", "#ff6600", "#ff6600aa"]: + result = await _call(self.mod, "create_project", + name=f"Color-{color}", color=color) + assert result["color"] == color + + +class TestRenameProject: + @pytest.fixture(autouse=True) + def setup(self): + self.state_dir = _fresh_state_dir() + self.mod, self.profiles = _reimport_mcp() + yield + _cleanup_state_dir(self.state_dir) + + async def test_rename_basic(self): + created = await _call(self.mod, "create_project", name="Old") + pid = created["project_id"] + result = await _call(self.mod, "rename_project", + project_id=pid, name="New") + assert result["name"] == "New" + assert result["project_id"] == pid + + async def test_rename_with_color(self): + created = await _call(self.mod, "create_project", name="X") + result = await _call(self.mod, "rename_project", + project_id=created["project_id"], + name="X", color="#000") + assert result["color"] == "#000" + + async def test_rename_not_found(self): + result = await _call(self.mod, "rename_project", + project_id="nonexistent", name="Nope") + assert "error" in result + + async def test_rename_wrong_profile(self): + created = await _call(self.mod, "create_project", name="DefaultOwned") + pid = created["project_id"] + self.profiles._active_profile = 'other' + result = await _call(self.mod, "rename_project", + project_id=pid, name="Stolen") + assert "error" in result + assert "not found" in result["error"].lower() + + +class TestDeleteProject: + @pytest.fixture(autouse=True) + def setup(self): + self.state_dir = _fresh_state_dir() + self.mod, self.profiles = _reimport_mcp() + yield + _cleanup_state_dir(self.state_dir) + + async def test_delete_basic(self): + created = await _call(self.mod, "create_project", name="ToDelete") + pid = created["project_id"] + result = await _call(self.mod, "delete_project", project_id=pid) + assert result["ok"] is True + assert result["deleted"] == "ToDelete" + + async def test_delete_not_found(self): + result = await _call(self.mod, "delete_project", + project_id="nonexistent") + assert "error" in result + + async def test_delete_wrong_profile(self): + created = await _call(self.mod, "create_project", name="Owned") + pid = created["project_id"] + self.profiles._active_profile = 'other' + result = await _call(self.mod, "delete_project", project_id=pid) + assert "error" in result + + +# ═══════════════════════════════════════════════════════════════════════════ +# Profile Scoping +# ═══════════════════════════════════════════════════════════════════════════ + +class TestProfileScoping: + @pytest.fixture(autouse=True) + def setup(self): + self.state_dir = _fresh_state_dir() + self.mod, self.profiles = _reimport_mcp() + yield + _cleanup_state_dir(self.state_dir) + + async def test_projects_tagged_with_profile(self): + result = await _call(self.mod, "create_project", name="Tagged") + assert result["profile"] == "default" + + async def test_list_projects_respects_profile(self): + # Create under default + await _call(self.mod, "create_project", name="DefaultProject") + + # Switch to other + self.profiles._active_profile = 'other' + await _call(self.mod, "create_project", name="OtherProject") + + # List should only show current profile's projects + projects = await _call(self.mod, "list_projects") + names = [p["name"] for p in projects] + assert "OtherProject" in names + assert "DefaultProject" not in names + + # Switch back + self.profiles._active_profile = 'default' + projects = await _call(self.mod, "list_projects") + names = [p["name"] for p in projects] + assert "DefaultProject" in names + assert "OtherProject" not in names + + async def test_cross_profile_isolation_create(self): + """Same name in different profiles should be allowed.""" + await _call(self.mod, "create_project", name="Shared") + self.profiles._active_profile = 'other' + result = await _call(self.mod, "create_project", name="Shared") + assert "project_id" in result + + async def test_legacy_untagged_hidden_from_non_root_profile(self): + """Untagged projects (no `profile` field) belong to the root profile. + + Mirrors api/routes.py:_profiles_match where a missing profile coerces + to 'default'. A non-root profile must NOT see legacy untagged rows. + """ + # Manually write a legacy untagged project (pre-#1614 schema) + from api.config import PROJECTS_FILE + legacy = [{ + "project_id": "legacy000001", + "name": "LegacyUntagged", + "color": None, + "created_at": 1700000000.0, + # No "profile" field on purpose + }] + PROJECTS_FILE.write_text(json.dumps(legacy), encoding="utf-8") + + # Non-root profile must NOT see it + self.profiles._active_profile = 'other' + projects = await _call(self.mod, "list_projects") + names = [p["name"] for p in projects] + assert "LegacyUntagged" not in names + + # Root profile still sees it (load_projects backfills `profile` + # to 'default', so visibility is preserved for the root). + self.profiles._active_profile = 'default' + projects = await _call(self.mod, "list_projects") + names = [p["name"] for p in projects] + assert "LegacyUntagged" in names + + async def test_legacy_untagged_rename_blocked_from_non_root(self): + """Non-root profile cannot rename a legacy untagged project.""" + from api.config import PROJECTS_FILE + legacy = [{ + "project_id": "legacy000002", + "name": "Legacy", + "color": None, + "created_at": 1700000000.0, + }] + PROJECTS_FILE.write_text(json.dumps(legacy), encoding="utf-8") + self.profiles._active_profile = 'other' + result = await _call(self.mod, "rename_project", + project_id="legacy000002", name="Stolen") + assert "error" in result + + +# ═══════════════════════════════════════════════════════════════════════════ +# Session listing +# ═══════════════════════════════════════════════════════════════════════════ + +class TestListSessions: + @pytest.fixture(autouse=True) + def setup(self): + self.state_dir = _fresh_state_dir() + self.mod, self.profiles = _reimport_mcp() + yield + _cleanup_state_dir(self.state_dir) + + async def test_list_empty(self): + result = await _call(self.mod, "list_sessions") + assert result == [] + + async def test_list_with_limit(self): + result = await _call(self.mod, "list_sessions", limit=10) + assert isinstance(result, list) + + async def test_list_unassigned(self): + result = await _call(self.mod, "list_sessions", unassigned=True) + assert isinstance(result, list) + + +# ═══════════════════════════════════════════════════════════════════════════ +# Session mutations (HTTP API — basic validation only) +# ═══════════════════════════════════════════════════════════════════════════ + +class TestSessionMutations: + @pytest.fixture(autouse=True) + def setup(self): + self.state_dir = _fresh_state_dir() + self.mod, self.profiles = _reimport_mcp() + yield + _cleanup_state_dir(self.state_dir) + + async def test_rename_missing_args(self): + result = await _call(self.mod, "rename_session", + session_id="", title="") + assert "error" in result + + async def test_move_missing_args(self): + result = await _call(self.mod, "move_session", + session_id="", project_id="x") + assert "error" in result + + async def test_move_project_not_found(self): + result = await _call(self.mod, "move_session", + session_id="s1", project_id="nonexistent") + assert "error" in result + + async def test_move_target_owned_by_other_profile_rejected(self): + """A project owned by profile A is invisible to profile B (#1614).""" + created = await _call(self.mod, "create_project", name="ATarget") + pid = created["project_id"] + self.profiles._active_profile = 'other' + result = await _call(self.mod, "move_session", + session_id="any", project_id=pid) + assert "error" in result + assert "not found" in result["error"].lower() + + +# ═══════════════════════════════════════════════════════════════════════════ +# Auth helper +# ═══════════════════════════════════════════════════════════════════════════ + +class TestApiPassword: + @pytest.fixture(autouse=True) + def setup(self): + self.state_dir = _fresh_state_dir() + # Ensure env var is unset for the test + os.environ.pop("HERMES_WEBUI_PASSWORD", None) + self.mod, self.profiles = _reimport_mcp() + yield + _cleanup_state_dir(self.state_dir) + + async def test_no_env_no_settings_returns_none(self): + assert self.mod._api_password() is None + + async def test_password_hash_in_settings_is_ignored(self): + """settings.json holds a hash, not a plaintext password — must NOT + be returned as if it were a usable password.""" + from api.config import STATE_DIR as _SD + (_SD / "settings.json").write_text( + json.dumps({"password_hash": "$2b$12$abcdefghijk"}), + encoding="utf-8") + assert self.mod._api_password() is None + + async def test_env_var_returned(self): + os.environ["HERMES_WEBUI_PASSWORD"] = "secret123" + try: + assert self.mod._api_password() == "secret123" + finally: + os.environ.pop("HERMES_WEBUI_PASSWORD", None)