From 4ec00d8b49b04e025324e3c2a6dfd1e24660a1b2 Mon Sep 17 00:00:00 2001 From: Chris Ahrendt Date: Thu, 2 Jul 2026 06:44:32 -0400 Subject: [PATCH 1/2] Add index management tools (create/drop/build + GSI settings) --- src/cb_mcp/tool_registration.py | 10 +- src/cb_mcp/tools/__init__.py | 53 +++- src/cb_mcp/tools/index_admin.py | 328 ++++++++++++++++++++++ src/cb_mcp/utils/index_ddl.py | 116 ++++++++ src/cb_mcp/utils/index_settings.py | 155 ++++++++++ src/mcp_server.py | 16 +- tests/unit/test_index_admin_tools_unit.py | 211 ++++++++++++++ tests/unit/test_index_ddl_unit.py | 151 ++++++++++ tests/unit/test_index_settings_unit.py | 186 ++++++++++++ tests/unit/test_read_only_mode.py | 8 +- 10 files changed, 1222 insertions(+), 12 deletions(-) create mode 100644 src/cb_mcp/tools/index_admin.py create mode 100644 src/cb_mcp/utils/index_ddl.py create mode 100644 src/cb_mcp/utils/index_settings.py create mode 100644 tests/unit/test_index_admin_tools_unit.py create mode 100644 tests/unit/test_index_ddl_unit.py create mode 100644 tests/unit/test_index_settings_unit.py diff --git a/src/cb_mcp/tool_registration.py b/src/cb_mcp/tool_registration.py index 5730473d..e1b66939 100644 --- a/src/cb_mcp/tool_registration.py +++ b/src/cb_mcp/tool_registration.py @@ -5,7 +5,7 @@ import logging from collections.abc import Callable -from .tools import KV_WRITE_TOOLS, get_tools +from .tools import ADMIN_WRITE_TOOLS, KV_WRITE_TOOLS, get_tools from .utils.config import parse_tool_names from .utils.constants import MCP_SERVER_NAME from .utils.elicitation import wrap_with_confirmation @@ -23,6 +23,7 @@ def prepare_tools_for_registration( disabled_tools: str | None, confirmation_required_tools: str | None, enforce_scopes: bool = False, + admin_write_mode: bool = False, ) -> tuple[list[Callable], set[str], set[str]]: """Prepare final tool list and confirmation configuration for registration. @@ -36,8 +37,9 @@ def prepare_tools_for_registration( (stdio / unauthenticated), so ``enforce_scopes`` only affects whether the wrapper is installed — not whether it does work per call. """ - # When read_only_mode is True, KV write tools are not loaded. - tools = get_tools(read_only_mode=read_only_mode) + # When read_only_mode is True, KV write tools are not loaded. Admin write + # tools (index DDL, GSI settings) additionally require admin_write_mode. + tools = get_tools(read_only_mode=read_only_mode, admin_write_mode=admin_write_mode) loaded_tool_names = {tool.__name__ for tool in tools} disabled_tool_names = parse_tool_names(disabled_tools, loaded_tool_names) @@ -74,7 +76,7 @@ def prepare_tools_for_registration( f"{sorted(skipped_confirmation_tool_names)}" ) - write_tool_names = {fn.__name__ for fn in KV_WRITE_TOOLS} + write_tool_names = {fn.__name__ for fn in (*KV_WRITE_TOOLS, *ADMIN_WRITE_TOOLS)} final_tools: list[Callable] = [] for tool in enabled_tools: diff --git a/src/cb_mcp/tools/__init__.py b/src/cb_mcp/tools/__init__.py index be843280..a9993a3b 100644 --- a/src/cb_mcp/tools/__init__.py +++ b/src/cb_mcp/tools/__init__.py @@ -15,6 +15,15 @@ # Index tools from .index import get_index_advisor_recommendations, list_indexes +# Index management tools (DDL + GSI settings, write) +from .index_admin import ( + admin_index_settings_get, + admin_index_settings_set, + build_deferred_indexes, + create_index, + drop_index, +) + # Key-Value tools from .kv import ( delete_document_by_id, @@ -68,6 +77,8 @@ # Index tools get_index_advisor_recommendations, list_indexes, + # Index settings (read) + admin_index_settings_get, # Query performance analysis tools get_queries_not_selective, get_queries_not_using_covering_index, @@ -86,6 +97,17 @@ delete_document_by_id, ] +# Admin write tools - loaded only when read_only_mode is False AND +# admin_write_mode is True. These mutate cluster structure (index DDL) or +# cluster-wide index settings, a strictly higher privilege than data writes, +# so they gate behind a second, independent flag. +ADMIN_WRITE_TOOLS = [ + create_index, + drop_index, + build_deferred_indexes, + admin_index_settings_set, +] + # List of all tools for easy registration (kept for backward compatibility) ALL_TOOLS = READ_ONLY_TOOLS + KV_WRITE_TOOLS @@ -121,20 +143,39 @@ "insert_document_by_id": ToolAnnotations(idempotentHint=True), "replace_document_by_id": ToolAnnotations(idempotentHint=True), "delete_document_by_id": ToolAnnotations(destructiveHint=True, idempotentHint=True), + # Index management tools (write) + "create_index": ToolAnnotations(idempotentHint=False), + "drop_index": ToolAnnotations(destructiveHint=True, idempotentHint=True), + "build_deferred_indexes": ToolAnnotations(idempotentHint=True), + # Index settings + "admin_index_settings_get": ToolAnnotations(readOnlyHint=True), + "admin_index_settings_set": ToolAnnotations( + destructiveHint=False, idempotentHint=True + ), } -def get_tools(read_only_mode: bool = True) -> list[Callable]: +def get_tools( + read_only_mode: bool = True, + admin_write_mode: bool = False, +) -> list[Callable]: """Get the list of tools based on the mode settings. - This function determines which tools should be loaded based on the - READ_ONLY_MODE setting. When read_only_mode is True, write tools are excluded. + - READ_ONLY_TOOLS are always loaded. + - KV_WRITE_TOOLS load when read_only_mode is False. + - ADMIN_WRITE_TOOLS load only when read_only_mode is False AND + admin_write_mode is True. Cluster-structure mutation (index DDL) and + cluster-wide index settings are a higher privilege than data writes, so + disabling read-only alone does not expose them; an operator must also + opt in to admin writes. """ tools = list(READ_ONLY_TOOLS) if not read_only_mode: # KV write tools are only loaded when READ_ONLY_MODE is False tools.extend(KV_WRITE_TOOLS) + if admin_write_mode: + tools.extend(ADMIN_WRITE_TOOLS) return tools @@ -157,6 +198,11 @@ def get_tools(read_only_mode: bool = True) -> list[Callable]: "explain_sql_plus_plus_query", "get_index_advisor_recommendations", "list_indexes", + "create_index", + "drop_index", + "build_deferred_indexes", + "admin_index_settings_get", + "admin_index_settings_set", "get_cluster_health_and_services", "get_queries_not_selective", "get_queries_not_using_covering_index", @@ -168,6 +214,7 @@ def get_tools(read_only_mode: bool = True) -> list[Callable]: # Tool categories "READ_ONLY_TOOLS", "KV_WRITE_TOOLS", + "ADMIN_WRITE_TOOLS", # Tool annotations "TOOL_ANNOTATIONS", # Convenience diff --git a/src/cb_mcp/tools/index_admin.py b/src/cb_mcp/tools/index_admin.py new file mode 100644 index 00000000..8afb5de2 --- /dev/null +++ b/src/cb_mcp/tools/index_admin.py @@ -0,0 +1,328 @@ +""" +Tools for index management (DDL). + +These tools create, drop, and build GSI indexes and read/update Index +Service settings. They complement the read-only ``index.py`` tools +(``list_indexes`` and ``get_index_advisor_recommendations``): the advisor +recommends indexes, ``list_indexes`` shows them, and these tools act on +them. + +Write gating +------------ +Index DDL mutates cluster structure. These tools are loaded only when the +server is *not* in read-only mode AND admin-write mode is enabled (see +``tools/__init__.py`` ``get_tools`` and ``ADMIN_WRITE_TOOLS``). When an +OAuth token is present, the caller must additionally hold the write scope; +this mirrors the scope enforcement in ``run_sql_plus_plus_query`` so a +read-scoped token cannot mutate indexes even when admin-write mode is on. + +DDL is executed through ``run_cluster_query`` rather than +``run_sql_plus_plus_query``. The latter classifies ``CREATE``/``DROP``/ +``BUILD INDEX`` as structure modifications and blocks them under read-only +mode; routing DDL through the cluster path keeps load-time gating +(``ADMIN_WRITE_TOOLS`` + admin-write mode + scope) as the single, explicit +enforcement point instead of double-gating. + +License: MIT - Copyright (c) 2026 Chris Ahrendt +""" + +import logging +from typing import Any + +from fastmcp import Context +from fastmcp.server.dependencies import get_access_token + +from ..utils.config import get_settings +from ..utils.constants import MCP_SERVER_NAME, SCOPE_WRITE +from ..utils.index_ddl import ( + assert_index_create_ddl, + assert_index_drop_ddl, + safe_ident, +) +from ..utils.index_settings import get_gsi_settings, set_gsi_settings +from .query import run_cluster_query + +logger = logging.getLogger(f"{MCP_SERVER_NAME}.tools.index_admin") + + +def _require_write_scope() -> None: + """Raise ``PermissionError`` if a token is present but lacks the write scope. + + No-op when no token is in context (stdio transport or OAuth disabled), + matching the behavior of ``run_sql_plus_plus_query`` so the same tool + body serves authenticated HTTP and unauthenticated stdio without + branching at registration time. + """ + token = get_access_token() + if token is not None and SCOPE_WRITE not in (token.scopes or []): + held = sorted(set(token.scopes or [])) + msg = f"Index DDL requires the '{SCOPE_WRITE}' scope; token scopes are {held}." + logger.warning(msg) + raise PermissionError(msg) + + +def create_index( + ctx: Context, + statement: str | None = None, + index_name: str | None = None, + bucket_name: str | None = None, + scope_name: str | None = None, + collection_name: str | None = None, + fields: list[str] | None = None, + is_primary: bool = False, + num_replica: int | None = None, + defer_build: bool = False, +) -> dict[str, Any]: + """Create a GSI index. + + Two ways to call this: + + - Raw ``statement``: pass a full SQL++ index-create statement. Only + CREATE INDEX / CREATE PRIMARY INDEX / BUILD INDEX / CREATE + [HYPERSCALE|COMPOSITE] VECTOR INDEX are accepted; any other SQL++ is + rejected. + - Structured fields: provide ``bucket_name`` (and optionally + ``scope_name`` / ``collection_name``, defaulting to ``_default``) with + either ``is_primary=True`` or an ``index_name`` plus ``fields``. + Optional ``num_replica`` and ``defer_build`` become a WITH clause. + + Returns a dict with the executed statement and result rows. + """ + _require_write_scope() + + if statement: + invalid = assert_index_create_ddl(statement) + if invalid: + raise ValueError(invalid) + stmt = statement + else: + if not bucket_name: + raise ValueError("bucket_name is required when statement is not provided") + bucket = safe_ident(bucket_name) + scope = safe_ident(scope_name or "_default") + coll = safe_ident(collection_name or "_default") + + if is_primary: + idx = safe_ident(index_name) if index_name else "" + target = f"{idx} " if idx else "" + stmt = f"CREATE PRIMARY INDEX {target}ON {bucket}.{scope}.{coll}" + else: + if not index_name: + raise ValueError("index_name is required for non-primary indexes") + if not fields: + raise ValueError("fields are required for non-primary indexes") + idx = safe_ident(index_name) + field_list = ", ".join(safe_ident(f) for f in fields) + stmt = f"CREATE INDEX {idx} ON {bucket}.{scope}.{coll} ({field_list})" + + withs = [] + if num_replica is not None: + withs.append(f'"num_replica": {int(num_replica)}') + if defer_build: + withs.append('"defer_build": true') + if withs: + stmt += " WITH {" + ", ".join(withs) + "}" + + logger.info("Executing index create") + rows = run_cluster_query(ctx, stmt) + return {"statement": stmt, "results": rows} + + +def drop_index( + ctx: Context, + statement: str | None = None, + index_name: str | None = None, + bucket_name: str | None = None, + scope_name: str | None = None, + collection_name: str | None = None, + is_primary: bool = False, +) -> dict[str, Any]: + """Drop a GSI index. + + Two ways to call this: + + - Raw ``statement``: a full DROP INDEX / DROP PRIMARY INDEX / DROP VECTOR + INDEX statement. Any other SQL++ is rejected. + - Structured fields: ``bucket_name`` (and optional ``scope_name`` / + ``collection_name``) with either ``is_primary=True`` or ``index_name``. + + Returns a dict with the executed statement and result rows. + """ + _require_write_scope() + + if statement: + invalid = assert_index_drop_ddl(statement) + if invalid: + raise ValueError(invalid) + stmt = statement + else: + if not bucket_name: + raise ValueError("bucket_name is required when statement is not provided") + bucket = safe_ident(bucket_name) + scope = safe_ident(scope_name or "_default") + coll = safe_ident(collection_name or "_default") + + if is_primary: + stmt = f"DROP PRIMARY INDEX ON {bucket}.{scope}.{coll}" + else: + if not index_name: + raise ValueError("index_name is required for non-primary drop") + idx = safe_ident(index_name) + stmt = f"DROP INDEX {idx} ON {bucket}.{scope}.{coll}" + + logger.info("Executing index drop") + rows = run_cluster_query(ctx, stmt) + return {"statement": stmt, "results": rows} + + +def build_deferred_indexes( + ctx: Context, + bucket_name: str, + index_names: list[str], + scope_name: str | None = None, + collection_name: str | None = None, +) -> dict[str, Any]: + """Build one or more deferred GSI indexes. + + ``bucket_name`` and ``index_names`` are required. Provide both + ``scope_name`` and ``collection_name`` to build within a specific + collection (Couchbase 7+); omit both for a bucket-level build. + + Returns a dict with the executed statement and result rows. + """ + _require_write_scope() + + if not index_names: + raise ValueError("index_names must contain at least one index name") + + bucket = safe_ident(bucket_name) + if scope_name and collection_name: + keyspace = f"{bucket}.{safe_ident(scope_name)}.{safe_ident(collection_name)}" + elif scope_name or collection_name: + raise ValueError("scope_name and collection_name must be provided together") + else: + keyspace = bucket + + names = ", ".join(safe_ident(n) for n in index_names) + stmt = f"BUILD INDEX ON {keyspace} ({names})" + + logger.info("Executing deferred index build") + rows = run_cluster_query(ctx, stmt) + return {"statement": stmt, "results": rows} + + +# -------------------------------------------------------------------------- +# GSI settings (cluster-manager /settings/indexes, port 8091/18091) +# -------------------------------------------------------------------------- + +# Named GSI settings exposed as explicit tool parameters. Maps the tool's +# snake_case argument to the endpoint's camelCase form key. These are the +# complete set of keys documented for POST /settings/indexes (Couchbase Server +# 8.0). Verified against docs 2026-07-02. +_GSI_SETTING_KEYS: dict[str, str] = { + "indexer_threads": "indexerThreads", + "log_level": "logLevel", + "max_rollback_points": "maxRollbackPoints", + "storage_mode": "storageMode", + "num_replica": "numReplica", + "redistribute_indexes": "redistributeIndexes", + "enable_page_bloom_filter": "enablePageBloomFilter", + "enable_shard_affinity": "enableShardAffinity", + "memory_snapshot_interval": "memorySnapshotInterval", + "stable_snapshot_interval": "stableSnapshotInterval", +} + +# Allow-list of accepted camelCase form keys for the ``extra`` escape hatch. +# The named parameters already cover the full documented key set; ``extra`` +# exists for forward-compatibility if a future server version adds a key. It +# is validated against this allow-list so it cannot be used to POST arbitrary +# keys to /settings/indexes. When Couchbase documents a new GSI setting, add +# its camelCase key here (and, ideally, a named parameter above). +_VALID_GSI_FORM_KEYS: frozenset[str] = frozenset(_GSI_SETTING_KEYS.values()) + + +def admin_index_settings_get(ctx: Context) -> dict[str, Any]: + """Get the cluster's Global Secondary Index (GSI) settings. + + Reads the GSI settings from the cluster manager (/settings/indexes). + Returns the settings as a dict, e.g. indexerThreads, logLevel, + maxRollbackPoints, storageMode, numReplica, redistributeIndexes. + Read-only. + """ + settings = get_settings(ctx) + return get_gsi_settings( + connection_string=settings["connection_string"], + username=settings["username"], + password=settings["password"], + ca_cert_path=settings.get("ca_cert_path"), + ) + + +def admin_index_settings_set( + ctx: Context, + indexer_threads: int | None = None, + log_level: str | None = None, + max_rollback_points: int | None = None, + storage_mode: str | None = None, + num_replica: int | None = None, + redistribute_indexes: bool | None = None, + enable_page_bloom_filter: bool | None = None, + enable_shard_affinity: bool | None = None, + memory_snapshot_interval: int | None = None, + stable_snapshot_interval: int | None = None, + extra: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Update the cluster's Global Secondary Index (GSI) settings. + + Provide any subset of the named parameters; only supplied values are + sent, and unspecified settings are left unchanged by the server. Use + ``extra`` to pass settings not covered by the named parameters (keys must + be documented GSI setting camelCase names; unknown keys are rejected). + Returns the resulting settings after the update. + + Note: Couchbase advises changing most GSI settings only when directed by + Couchbase Support. This tool loads only when read-only mode is off and + admin-write mode is on, and (when OAuth is active) requires the write + scope. + """ + _require_write_scope() + + named = { + "indexer_threads": indexer_threads, + "log_level": log_level, + "max_rollback_points": max_rollback_points, + "storage_mode": storage_mode, + "num_replica": num_replica, + "redistribute_indexes": redistribute_indexes, + "enable_page_bloom_filter": enable_page_bloom_filter, + "enable_shard_affinity": enable_shard_affinity, + "memory_snapshot_interval": memory_snapshot_interval, + "stable_snapshot_interval": stable_snapshot_interval, + } + params: dict[str, Any] = { + _GSI_SETTING_KEYS[k]: v for k, v in named.items() if v is not None + } + if extra: + unknown = sorted(set(extra) - _VALID_GSI_FORM_KEYS) + if unknown: + raise ValueError( + f"Unknown GSI setting key(s) in 'extra': {unknown}. Allowed " + f"keys are {sorted(_VALID_GSI_FORM_KEYS)}. Use a named " + "parameter where one exists." + ) + params.update(extra) + + if not params: + raise ValueError( + "Provide at least one setting to update (a named parameter or " + "an 'extra' entry)." + ) + + settings = get_settings(ctx) + return set_gsi_settings( + connection_string=settings["connection_string"], + username=settings["username"], + password=settings["password"], + params=params, + ca_cert_path=settings.get("ca_cert_path"), + ) diff --git a/src/cb_mcp/utils/index_ddl.py b/src/cb_mcp/utils/index_ddl.py new file mode 100644 index 00000000..9075c469 --- /dev/null +++ b/src/cb_mcp/utils/index_ddl.py @@ -0,0 +1,116 @@ +""" +Index DDL safety primitives. + +Identifier quoting and statement allow-listing for the index-management +tools. These guard the two paths by which a caller can reach index DDL: + +1. Structured helper fields (bucket/scope/collection/index names, key + fields). Every identifier is backtick-quoted via :func:`safe_ident` + before interpolation, so a name containing a backtick cannot break out + of its quoting and inject arbitrary SQL++. + +2. A raw ``statement`` string. This is convenient but dangerous, so it is + constrained by :func:`assert_index_create_ddl` / + :func:`assert_index_drop_ddl`, which reject anything that is not the + expected index-DDL shape. This prevents the raw path from becoming a + general SQL++ execution channel that bypasses read-only and + admin-write gating. + +License: MIT - Copyright (c) 2026 Chris Ahrendt +""" + +import re + +# Allow-list patterns for the raw ``statement`` path. Anchored at the start so +# a statement must *begin* with the permitted keyword. Note the anchor alone is +# not sufficient: a statement like "CREATE INDEX ... ; DELETE FROM ..." begins +# with a permitted keyword but carries a second statement. ``_is_single_statement`` +# below rejects any embedded statement separator so the raw path cannot be used +# to smuggle a trailing DML/DDL statement past the allow-list. +_INDEX_CREATE_RE = re.compile( + r"""^\s* + ( + CREATE\s+(PRIMARY\s+)?INDEX + | CREATE\s+(?:HYPERSCALE\s+|COMPOSITE\s+)?VECTOR\s+INDEX + ) + \b + """, + re.IGNORECASE | re.VERBOSE, +) + +_INDEX_BUILD_RE = re.compile(r"""^\s*BUILD\s+INDEX\b""", re.IGNORECASE | re.VERBOSE) + +_INDEX_DROP_RE = re.compile( + r"""^\s*DROP\s+((PRIMARY|VECTOR)\s+)?INDEX\b""", + re.IGNORECASE | re.VERBOSE, +) + + +def _is_single_statement(statement: str) -> bool: + """True if *statement* contains no embedded statement separator. + + SQL++ separates statements with ``;``. A trailing semicolon (optionally + followed by whitespace) is allowed; a semicolon with anything non-blank + after it means a second statement is present and the input is rejected. + Backtick-quoted identifiers may legitimately contain a ``;`` inside the + quotes, so those spans are removed before the check to avoid false + positives on names like ``` `weird;name` ```. + """ + # Strip backtick-quoted spans (identifiers) so a ';' inside a quoted name + # is not mistaken for a statement separator. Doubled backticks are escapes. + without_idents = re.sub(r"`(?:[^`]|``)*`", "", statement) + stripped = without_idents.rstrip() + if stripped.endswith(";"): + stripped = stripped[:-1] + return ";" not in stripped + + +def safe_ident(segment: str) -> str: + """Backtick-quote and escape an identifier for SQL++. + + Couchbase SQL++ escapes an embedded backtick by doubling it, so a + caller-supplied name can never terminate its own quoting. Applied to + every bucket / scope / collection / index / field name before it is + interpolated into a statement. + """ + return "`" + (segment or "").replace("`", "``") + "`" + + +def assert_index_create_ddl(statement: str) -> str | None: + """Return an error message if *statement* is not permitted index-create DDL. + + Returns ``None`` when the statement is a single CREATE INDEX / CREATE + PRIMARY INDEX / CREATE [HYPERSCALE|COMPOSITE] VECTOR INDEX statement. Any + other SQL++ - including a permitted statement followed by a second, + smuggled statement - is rejected so the raw ``statement`` path cannot be + used to run arbitrary queries or DML. (BUILD INDEX is handled by + ``build_deferred_indexes``, not this create path.) + """ + stmt = statement or "" + if not _INDEX_CREATE_RE.match(stmt) or not _is_single_statement(stmt): + return ( + "The `statement` parameter only accepts a single index-create DDL " + "statement (CREATE INDEX, CREATE PRIMARY INDEX, or CREATE " + "[HYPERSCALE|COMPOSITE] VECTOR INDEX), with no trailing statement. " + "Use the helper fields (index_name, bucket_name, fields, etc.) for " + "structured creation, or run other SQL++ via run_sql_plus_plus_query." + ) + return None + + +def assert_index_drop_ddl(statement: str) -> str | None: + """Return an error message if *statement* is not permitted index-drop DDL. + + Returns ``None`` when the statement is a single DROP INDEX, DROP PRIMARY + INDEX, or DROP VECTOR INDEX statement. Any other SQL++ - including a + trailing smuggled statement - is rejected. + """ + stmt = statement or "" + if not _INDEX_DROP_RE.match(stmt) or not _is_single_statement(stmt): + return ( + "The `statement` parameter only accepts a single DROP INDEX, DROP " + "PRIMARY INDEX, or DROP VECTOR INDEX statement, with no trailing " + "statement. Use the helper fields for structured drops, or run " + "other SQL++ via run_sql_plus_plus_query." + ) + return None diff --git a/src/cb_mcp/utils/index_settings.py b/src/cb_mcp/utils/index_settings.py new file mode 100644 index 00000000..0812cff7 --- /dev/null +++ b/src/cb_mcp/utils/index_settings.py @@ -0,0 +1,155 @@ +""" +REST client for the Global Secondary Index (GSI) settings endpoint. + +The GSI Settings API is served by the cluster manager on port 8091 (18091 +for TLS) at ``/settings/indexes`` - distinct from the Index Service REST API +(port 9102) used by ``fetch_indexes_from_rest_api`` in ``index_utils``. GET +returns a JSON object of current settings; POST accepts +``application/x-www-form-urlencoded`` key-value pairs and leaves unspecified +parameters unchanged. + +Verified against Couchbase Server docs (rest-api/get-settings-indexes, +post-settings-indexes, rest-index-service) 2026-07-02. + +This helper reuses the connection-string host extraction and SSL +verification logic already present in ``index_utils`` so behavior (Capella +root CA handling, multi-host fallback, TLS detection) stays consistent with +the existing index REST path. + +License: MIT - Copyright (c) 2026 Chris Ahrendt +""" + +import logging +from typing import Any + +import httpx + +from .constants import MCP_SERVER_NAME +from .index_utils import ( + _determine_ssl_verification, + _extract_hosts_from_connection_string, +) + +logger = logging.getLogger(f"{MCP_SERVER_NAME}.utils.index_settings") + +# Cluster-manager ports for the GSI Settings API (NOT the 9102 Index Service +# port used for /getIndexStatus). +_GSI_SETTINGS_PORT = 8091 +_GSI_SETTINGS_TLS_PORT = 18091 +_GSI_SETTINGS_PATH = "/settings/indexes" + + +def _settings_base(connection_string: str) -> tuple[str, int]: + """Return (scheme, port) for the GSI settings endpoint.""" + is_tls = connection_string.lower().startswith("couchbases://") + return ("https", _GSI_SETTINGS_TLS_PORT) if is_tls else ("http", _GSI_SETTINGS_PORT) + + +def get_gsi_settings( + connection_string: str, + username: str, + password: str, + ca_cert_path: str | None = None, + timeout: int = 30, +) -> dict[str, Any]: + """GET the current GSI settings from the cluster. + + Tries each host in the connection string until one responds. Raises + ``RuntimeError`` if all hosts fail. + """ + return _request_gsi_settings( + method="GET", + connection_string=connection_string, + username=username, + password=password, + ca_cert_path=ca_cert_path, + timeout=timeout, + ) + + +def set_gsi_settings( + connection_string: str, + username: str, + password: str, + params: dict[str, Any], + ca_cert_path: str | None = None, + timeout: int = 30, +) -> dict[str, Any]: + """POST GSI settings (form-urlencoded) and return the resulting settings. + + ``params`` values are converted to the string forms the endpoint expects + (booleans as lowercase ``true``/``false``). Only the supplied keys are + sent; unspecified settings are left unchanged by the server. After a + successful POST the current settings are read back and returned. + """ + if not params: + raise ValueError("params must contain at least one setting to update") + + form = {k: _form_value(v) for k, v in params.items() if v is not None} + if not form: + raise ValueError("params contained no non-null settings to update") + + _request_gsi_settings( + method="POST", + connection_string=connection_string, + username=username, + password=password, + ca_cert_path=ca_cert_path, + timeout=timeout, + data=form, + ) + # Read back so the caller sees the applied state. + return get_gsi_settings( + connection_string, username, password, ca_cert_path, timeout + ) + + +def _form_value(value: Any) -> str: + """Render a value for the form-urlencoded body (bools lowercased).""" + if isinstance(value, bool): + return "true" if value else "false" + return str(value) + + +def _request_gsi_settings( + *, + method: str, + connection_string: str, + username: str, + password: str, + ca_cert_path: str | None, + timeout: int, + data: dict[str, str] | None = None, +) -> dict[str, Any]: + """Issue a GET/POST to /settings/indexes, trying each host in turn.""" + hosts = _extract_hosts_from_connection_string(connection_string) + scheme, port = _settings_base(connection_string) + verify_ssl = _determine_ssl_verification(connection_string, ca_cert_path) + + last_error: Exception | None = None + with httpx.Client(verify=verify_ssl, timeout=timeout) as client: + for host in hosts: + url = f"{scheme}://{host}:{port}{_GSI_SETTINGS_PATH}" + try: + logger.info(f"{method} {url}") + if method == "GET": + resp = client.get(url, auth=(username, password)) + else: + resp = client.post(url, data=data, auth=(username, password)) + resp.raise_for_status() + # GET returns settings JSON; POST returns an empty/!json body + # on some versions, so only parse when there is content. + if resp.content and resp.headers.get("content-type", "").startswith( + "application/json" + ): + return resp.json() + return {} + except httpx.HTTPError as e: + logger.warning(f"GSI settings {method} failed on {host}: {e}") + last_error = e + + error_msg = f"GSI settings {method} failed on all hosts: {hosts}" + if last_error: + error_msg += f". Last error: {last_error}" + logger.error(error_msg) + raise RuntimeError(error_msg) diff --git a/src/mcp_server.py b/src/mcp_server.py index f5e2a069..ebcac0ee 100644 --- a/src/mcp_server.py +++ b/src/mcp_server.py @@ -85,6 +85,13 @@ default=DEFAULT_READ_ONLY_MODE, help="Enable read-only mode. When True, all write operations (KV and Query) are disabled and KV write tools are not loaded. Set to False to enable write operations.", ) +@click.option( + "--admin-write-mode", + envvar="CB_MCP_ADMIN_WRITE_MODE", + type=bool, + default=False, + help="Enable admin write tools (index DDL and GSI settings). Requires read-only mode to be False. When False, cluster-structure and index-settings mutation tools are not loaded even if data (KV) writes are enabled.", +) @click.option( "--transport", envvar=[ @@ -212,6 +219,7 @@ def main( client_cert_path, client_key_path, read_only_mode, + admin_write_mode, transport, host, port, @@ -260,6 +268,7 @@ def main( disabled_tools=disabled_tools, confirmation_required_tools=confirmation_required_tools, enforce_scopes=auth is not None, + admin_write_mode=admin_write_mode, ) # CLI-resolved configuration lives on AppContext, not in a module global. @@ -272,6 +281,7 @@ def main( "client_cert_path": client_cert_path, "client_key_path": client_key_path, "read_only_mode": read_only_mode, + "admin_write_mode": admin_write_mode, "transport": transport, "host": host, "port": port, @@ -298,7 +308,8 @@ async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: """Build the lifespan AppContext with settings captured from the CLI.""" logger.info( f"MCP server initialized in lazy mode for tool discovery. " - f"Modes: (read_only_mode={read_only_mode})" + f"Modes: (read_only_mode={read_only_mode}, " + f"admin_write_mode={admin_write_mode})" ) # Diagnostic snapshot for customer support. Filtered at INFO; visible # whenever the user runs with --log-level DEBUG. @@ -329,7 +340,8 @@ async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: mcp = FastMCP(MCP_SERVER_NAME, lifespan=app_lifespan, auth=auth) logger.info( - f"Registering {len(final_tools)} tool(s) with modes (read_only_mode={read_only_mode})" + f"Registering {len(final_tools)} tool(s) with modes " + f"(read_only_mode={read_only_mode}, admin_write_mode={admin_write_mode})" ) # Register tools; FastMCP 3.x add_tool has no annotations kwarg, so wrap first. diff --git a/tests/unit/test_index_admin_tools_unit.py b/tests/unit/test_index_admin_tools_unit.py new file mode 100644 index 00000000..0b17dc2e --- /dev/null +++ b/tests/unit/test_index_admin_tools_unit.py @@ -0,0 +1,211 @@ +""" +Unit tests for the index-management tool bodies (statement construction and +write-scope gating), exercised against stubbed cluster and token modules so +no live Couchbase cluster is required. + +License: MIT - Copyright (c) 2026 Chris Ahrendt +""" + +from contextlib import contextmanager +from unittest.mock import patch + +import pytest + +from cb_mcp.tools.index_admin import ( + build_deferred_indexes, + create_index, + drop_index, +) + + +@contextmanager +def _mock_token(scopes): + """Patch get_access_token at the index_admin call site. + + Pass a list of scopes for an authenticated token, or None for no token + (stdio / OAuth disabled). + """ + + class _Tok: + def __init__(self, scopes_): + self.scopes = scopes_ + + token = _Tok(scopes) if scopes is not None else None + with patch("cb_mcp.tools.index_admin.get_access_token", return_value=token): + yield + + +@pytest.fixture(autouse=True) +def _stub_cluster_query(monkeypatch): + """Patch run_cluster_query at the index_admin call site so statement + construction can be tested without a live cluster. Records nothing; + tests assert on the returned ``statement`` field the tool echoes back.""" + monkeypatch.setattr( + "cb_mcp.tools.index_admin.run_cluster_query", + lambda ctx, statement, **kw: [{"ok": True}], + ) + yield + + +CTX = object() # tools never introspect ctx directly; they pass it through + + +# -------------------------------------------------------------------------- +# create_index - structured path +# -------------------------------------------------------------------------- + + +def test_create_secondary_index_builds_expected_statement(): + out = create_index( + CTX, + index_name="idx_name", + bucket_name="travel-sample", + scope_name="inventory", + collection_name="airline", + fields=["name", "country"], + ) + assert out["statement"] == ( + "CREATE INDEX `idx_name` ON " + "`travel-sample`.`inventory`.`airline` (`name`, `country`)" + ) + + +def test_create_primary_index_with_name(): + out = create_index(CTX, index_name="pk", bucket_name="b", is_primary=True) + assert out["statement"] == "CREATE PRIMARY INDEX `pk` ON `b`.`_default`.`_default`" + + +def test_create_primary_index_without_name(): + out = create_index(CTX, bucket_name="b", is_primary=True) + assert out["statement"] == "CREATE PRIMARY INDEX ON `b`.`_default`.`_default`" + + +def test_create_with_num_replica_and_defer(): + out = create_index( + CTX, + index_name="i", + bucket_name="b", + fields=["x"], + num_replica=2, + defer_build=True, + ) + assert out["statement"].endswith('WITH {"num_replica": 2, "defer_build": true}') + + +def test_create_missing_bucket_raises(): + with pytest.raises(ValueError, match="bucket_name is required"): + create_index(CTX, index_name="i", fields=["x"]) + + +def test_create_secondary_missing_fields_raises(): + with pytest.raises(ValueError, match="fields are required"): + create_index(CTX, index_name="i", bucket_name="b") + + +def test_create_secondary_missing_name_raises(): + with pytest.raises(ValueError, match="index_name is required"): + create_index(CTX, bucket_name="b", fields=["x"]) + + +# -------------------------------------------------------------------------- +# create_index - raw statement path +# -------------------------------------------------------------------------- + + +def test_create_raw_statement_allowed(): + out = create_index(CTX, statement="CREATE INDEX i ON b.s.c (x)") + assert out["statement"] == "CREATE INDEX i ON b.s.c (x)" + + +def test_create_raw_statement_rejected(): + with pytest.raises(ValueError, match="index-create DDL"): + create_index(CTX, statement="DELETE FROM b.s.c") + + +# -------------------------------------------------------------------------- +# drop_index +# -------------------------------------------------------------------------- + + +def test_drop_secondary_index_statement(): + out = drop_index( + CTX, index_name="i", bucket_name="b", scope_name="s", collection_name="c" + ) + assert out["statement"] == "DROP INDEX `i` ON `b`.`s`.`c`" + + +def test_drop_primary_index_statement(): + out = drop_index(CTX, bucket_name="b", is_primary=True) + assert out["statement"] == "DROP PRIMARY INDEX ON `b`.`_default`.`_default`" + + +def test_drop_raw_statement_rejected(): + with pytest.raises(ValueError, match="DROP INDEX"): + drop_index(CTX, statement="DROP SCOPE b.s") + + +def test_drop_missing_name_raises(): + with pytest.raises(ValueError, match="index_name is required"): + drop_index(CTX, bucket_name="b") + + +# -------------------------------------------------------------------------- +# build_deferred_indexes +# -------------------------------------------------------------------------- + + +def test_build_bucket_level(): + out = build_deferred_indexes(CTX, bucket_name="b", index_names=["i1", "i2"]) + assert out["statement"] == "BUILD INDEX ON `b` (`i1`, `i2`)" + + +def test_build_collection_level(): + out = build_deferred_indexes( + CTX, + bucket_name="b", + index_names=["i1"], + scope_name="s", + collection_name="c", + ) + assert out["statement"] == "BUILD INDEX ON `b`.`s`.`c` (`i1`)" + + +def test_build_partial_keyspace_raises(): + with pytest.raises(ValueError, match="must be provided together"): + build_deferred_indexes(CTX, bucket_name="b", index_names=["i"], scope_name="s") + + +def test_build_empty_names_raises(): + with pytest.raises(ValueError, match="at least one"): + build_deferred_indexes(CTX, bucket_name="b", index_names=[]) + + +# -------------------------------------------------------------------------- +# write-scope gating (mirrors run_sql_plus_plus_query semantics) +# -------------------------------------------------------------------------- + + +def test_no_token_allows_write(): + # stdio / OAuth disabled: no token in context -> no scope enforcement + with _mock_token(None): + out = create_index(CTX, bucket_name="b", is_primary=True) + assert out["statement"].startswith("CREATE PRIMARY INDEX") + + +def test_token_with_write_scope_allows(): + with _mock_token(["couchbase-mcp:write"]): + out = drop_index(CTX, bucket_name="b", is_primary=True) + assert out["statement"].startswith("DROP PRIMARY INDEX") + + +def test_token_without_write_scope_denied(): + with ( + _mock_token(["couchbase-mcp:read"]), + pytest.raises(PermissionError, match="requires the 'couchbase-mcp:write'"), + ): + create_index(CTX, bucket_name="b", is_primary=True) + + +def test_build_denied_without_write_scope(): + with _mock_token([]), pytest.raises(PermissionError): + build_deferred_indexes(CTX, bucket_name="b", index_names=["i"]) diff --git a/tests/unit/test_index_ddl_unit.py b/tests/unit/test_index_ddl_unit.py new file mode 100644 index 00000000..e18faa82 --- /dev/null +++ b/tests/unit/test_index_ddl_unit.py @@ -0,0 +1,151 @@ +""" +Unit tests for index DDL safety primitives and statement construction. + +These run without a Couchbase cluster: they exercise the pure logic +(identifier quoting, statement allow-listing, statement building) and the +write-scope check via a stubbed access token. The cluster execution path +(``run_cluster_query``) is monkeypatched. + +License: MIT - Copyright (c) 2026 Chris Ahrendt +""" + +import pytest + +from cb_mcp.utils.index_ddl import ( + assert_index_create_ddl, + assert_index_drop_ddl, + safe_ident, +) + +# -------------------------------------------------------------------------- +# safe_ident +# -------------------------------------------------------------------------- + + +def test_safe_ident_wraps_in_backticks(): + assert safe_ident("users") == "`users`" + + +def test_safe_ident_doubles_embedded_backtick(): + # A name containing a backtick must not be able to terminate its quoting. + assert safe_ident("we`ird") == "`we``ird`" + + +def test_safe_ident_handles_empty_and_none(): + assert safe_ident("") == "``" + assert safe_ident(None) == "``" + + +def test_safe_ident_injection_attempt_is_neutralized(): + # Attempt to break out and append a DROP; the backtick is doubled so the + # whole thing stays a single (absurd) identifier. + malicious = "x` ON b.s.c; DROP INDEX `y" + quoted = safe_ident(malicious) + assert quoted.startswith("`") and quoted.endswith("`") + # No unescaped backtick remains that could close the identifier early. + inner = quoted[1:-1] + assert "``" in inner + assert not any( + inner[i] == "`" and inner[i - 1] != "`" and inner[i + 1 : i + 2] != "`" + for i in range(1, len(inner) - 1) + ) + + +# -------------------------------------------------------------------------- +# assert_index_create_ddl +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "stmt", + [ + "CREATE INDEX idx ON b.s.c (name)", + "create index idx on b.s.c (name)", + "CREATE PRIMARY INDEX ON b.s.c", + " CREATE PRIMARY INDEX ON b.s.c", + "CREATE VECTOR INDEX v ON b.s.c (emb VECTOR)", + "CREATE HYPERSCALE VECTOR INDEX v ON b.s.c (emb VECTOR)", + "CREATE COMPOSITE VECTOR INDEX v ON b.s.c (emb VECTOR)", + ], +) +def test_create_ddl_allows_index_statements(stmt): + assert assert_index_create_ddl(stmt) is None + + +@pytest.mark.parametrize( + "stmt", + [ + "SELECT * FROM b.s.c", + "DELETE FROM b.s.c", + "UPDATE b.s.c SET x = 1", + "DROP INDEX idx ON b.s.c", + "INSERT INTO b.s.c VALUES (1, {})", + "CREATE SCOPE b.s", + "CREATE COLLECTION b.s.c", + "", + " ", + "; CREATE INDEX idx ON b.s.c (name)", # leading junk before keyword + "BUILD INDEX ON b.s.c (i)", # BUILD routes to build_deferred_indexes + # multi-statement injection: permitted head + smuggled tail + "CREATE INDEX i ON b.s.c (x); DELETE FROM b.s.c WHERE 1=1", + "CREATE PRIMARY INDEX ON b.s.c ; DROP SCOPE b.s", + "CREATE INDEX i ON b.s.c (x);SELECT 1", + ], +) +def test_create_ddl_rejects_non_index_statements(stmt): + msg = assert_index_create_ddl(stmt) + assert msg is not None + assert "index-create DDL" in msg + + +def test_create_ddl_handles_none(): + assert assert_index_create_ddl(None) is not None + + +@pytest.mark.parametrize( + "stmt", + [ + "CREATE INDEX i ON b.s.c (name);", # single trailing semicolon + "CREATE INDEX i ON b.s.c (name) ; ", # trailing semicolon + whitespace + "CREATE INDEX `weird;name` ON b.s.c (x)", # ';' inside a quoted ident + ], +) +def test_create_ddl_allows_trailing_semicolon_and_quoted_semicolon(stmt): + assert assert_index_create_ddl(stmt) is None + + +# -------------------------------------------------------------------------- +# assert_index_drop_ddl +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "stmt", + [ + "DROP INDEX idx ON b.s.c", + "drop index idx on b.s.c", + "DROP PRIMARY INDEX ON b.s.c", + "DROP VECTOR INDEX v ON b.s.c", + " DROP INDEX idx ON b.s.c", + ], +) +def test_drop_ddl_allows_drop_statements(stmt): + assert assert_index_drop_ddl(stmt) is None + + +@pytest.mark.parametrize( + "stmt", + [ + "CREATE INDEX idx ON b.s.c (name)", + "SELECT * FROM b.s.c", + "DELETE FROM b.s.c", + "DROP SCOPE b.s", + "DROP COLLECTION b.s.c", + "DROP PRIMARY VECTOR INDEX ON b.s.c", # not a real statement shape + "DROP INDEX i ON b.s.c; CREATE PRIMARY INDEX ON b.s.c", # injection + "", + None, + ], +) +def test_drop_ddl_rejects_non_drop_statements(stmt): + assert assert_index_drop_ddl(stmt) is not None diff --git a/tests/unit/test_index_settings_unit.py b/tests/unit/test_index_settings_unit.py new file mode 100644 index 00000000..c4e32a4b --- /dev/null +++ b/tests/unit/test_index_settings_unit.py @@ -0,0 +1,186 @@ +""" +Unit tests for the GSI settings tools and REST helper pure logic. + +The tools' network calls (get_gsi_settings / set_gsi_settings) are +monkeypatched so these tests exercise parameter mapping and gating without a +cluster. The REST helper's pure functions (_form_value, _settings_base) are +tested directly. + +License: MIT - Copyright (c) 2026 Chris Ahrendt +""" + +from contextlib import contextmanager +from unittest.mock import patch + +import pytest + +from cb_mcp.tools import index_admin +from cb_mcp.utils import index_settings + + +@contextmanager +def _mock_token(scopes): + class _Tok: + def __init__(self, scopes_): + self.scopes = scopes_ + + token = _Tok(scopes) if scopes is not None else None + with patch("cb_mcp.tools.index_admin.get_access_token", return_value=token): + yield + + +@pytest.fixture(autouse=True) +def _stub_get_settings(monkeypatch): + """Patch get_settings so settings tools don't require a live AppContext.""" + monkeypatch.setattr( + "cb_mcp.tools.index_admin.get_settings", + lambda ctx: { + "connection_string": "couchbase://localhost", + "username": "u", + "password": "p", + "ca_cert_path": None, + }, + ) + yield + + +CTX = object() + + +# -------------------------------------------------------------------------- +# REST helper pure logic +# -------------------------------------------------------------------------- + + +def test_form_value_bools_lowercased(): + assert index_settings._form_value(True) == "true" + assert index_settings._form_value(False) == "false" + + +def test_form_value_ints_and_strings(): + assert index_settings._form_value(4) == "4" + assert index_settings._form_value("plasma") == "plasma" + + +def test_settings_base_non_tls(): + assert index_settings._settings_base("couchbase://localhost") == ("http", 8091) + + +def test_settings_base_tls(): + assert index_settings._settings_base("couchbases://cb.example") == ( + "https", + 18091, + ) + + +def test_set_gsi_settings_rejects_empty(): + with pytest.raises(ValueError, match="at least one setting"): + index_settings.set_gsi_settings("couchbase://h", "u", "p", params={}) + + +def test_set_gsi_settings_rejects_all_none(): + with pytest.raises(ValueError, match="no non-null settings"): + index_settings.set_gsi_settings( + "couchbase://h", "u", "p", params={"numReplica": None} + ) + + +# -------------------------------------------------------------------------- +# admin_index_settings_get +# -------------------------------------------------------------------------- + + +def test_settings_get_returns_settings(monkeypatch): + captured = {} + + def fake_get(**kwargs): + captured.update(kwargs) + return {"indexerThreads": 4, "logLevel": "info"} + + monkeypatch.setattr(index_admin, "get_gsi_settings", fake_get) + out = index_admin.admin_index_settings_get(CTX) + assert out == {"indexerThreads": 4, "logLevel": "info"} + assert captured["connection_string"] == "couchbase://localhost" + + +# -------------------------------------------------------------------------- +# admin_index_settings_set - param mapping +# -------------------------------------------------------------------------- + + +def test_settings_set_maps_named_params(monkeypatch): + captured = {} + + def fake_set(**kwargs): + captured.update(kwargs) + return {"applied": True} + + monkeypatch.setattr(index_admin, "set_gsi_settings", fake_set) + index_admin.admin_index_settings_set( + CTX, + indexer_threads=8, + log_level="verbose", + redistribute_indexes=False, + ) + assert captured["params"] == { + "indexerThreads": 8, + "logLevel": "verbose", + "redistributeIndexes": False, + } + + +def test_settings_set_merges_extra(monkeypatch): + captured = {} + monkeypatch.setattr( + index_admin, + "set_gsi_settings", + lambda **kw: captured.update(kw) or {}, + ) + # extra keys must be documented camelCase GSI keys; enableShardAffinity is + # valid but has no named parameter in an older tool version, so it is a + # representative forward-compat use of the escape hatch. + index_admin.admin_index_settings_set( + CTX, num_replica=2, extra={"enableShardAffinity": True} + ) + assert captured["params"] == {"numReplica": 2, "enableShardAffinity": True} + + +def test_settings_set_rejects_unknown_extra_key(monkeypatch): + monkeypatch.setattr(index_admin, "set_gsi_settings", lambda **kw: {}) + with pytest.raises(ValueError, match="Unknown GSI setting key"): + index_admin.admin_index_settings_set(CTX, extra={"arbitraryKey": "x"}) + + +def test_settings_set_requires_at_least_one(monkeypatch): + monkeypatch.setattr(index_admin, "set_gsi_settings", lambda **kw: {}) + with pytest.raises(ValueError, match="at least one setting"): + index_admin.admin_index_settings_set(CTX) + + +def test_settings_set_omits_none_values(monkeypatch): + captured = {} + monkeypatch.setattr( + index_admin, + "set_gsi_settings", + lambda **kw: captured.update(kw) or {}, + ) + index_admin.admin_index_settings_set(CTX, storage_mode="plasma") + assert captured["params"] == {"storageMode": "plasma"} + + +# -------------------------------------------------------------------------- +# write-scope gating on set (get is read-only, not gated) +# -------------------------------------------------------------------------- + + +def test_settings_set_denied_without_write_scope(monkeypatch): + monkeypatch.setattr(index_admin, "set_gsi_settings", lambda **kw: {}) + with _mock_token(["couchbase-mcp:read"]), pytest.raises(PermissionError): + index_admin.admin_index_settings_set(CTX, indexer_threads=4) + + +def test_settings_set_allowed_with_write_scope(monkeypatch): + monkeypatch.setattr(index_admin, "set_gsi_settings", lambda **kw: {"ok": True}) + with _mock_token(["couchbase-mcp:write"]): + out = index_admin.admin_index_settings_set(CTX, indexer_threads=4) + assert out == {"ok": True} diff --git a/tests/unit/test_read_only_mode.py b/tests/unit/test_read_only_mode.py index 825f1b7d..0722ef85 100644 --- a/tests/unit/test_read_only_mode.py +++ b/tests/unit/test_read_only_mode.py @@ -24,7 +24,7 @@ "delete_document_by_id", } -# Read-only tool names that should always be available (20 tools) +# Read-only tool names that should always be available (21 tools) READ_ONLY_TOOL_NAMES = { # Server/Cluster management tools (7) "get_buckets_in_cluster", @@ -43,6 +43,8 @@ # Index tools (2) "get_index_advisor_recommendations", "list_indexes", + # Index settings read tool (1) + "admin_index_settings_get", # Query performance analysis tools (7) "get_queries_not_selective", "get_queries_not_using_covering_index", @@ -157,13 +159,13 @@ def test_read_only_mode_tool_count(self): """Verify correct number of tools in read-only mode.""" tools = get_tools(read_only_mode=True) assert len(tools) == len(READ_ONLY_TOOLS) - assert len(tools) == 20 # Expected count of read-only tools + assert len(tools) == 21 # Expected count of read-only tools def test_all_tools_mode_tool_count(self): """Verify correct number of tools when all write tools are enabled.""" tools = get_tools(read_only_mode=False) assert len(tools) == len(ALL_TOOLS) - assert len(tools) == 24 # Expected total count (20 read-only + 4 KV write) + assert len(tools) == 25 # Expected total count (21 read-only + 4 KV write) def test_kv_write_tools_count(self): """Verify exactly 4 KV write tools exist.""" From 8839f894e13739332d6016581b9db732ad025a3a Mon Sep 17 00:00:00 2001 From: Chris Ahrendt Date: Mon, 6 Jul 2026 22:43:43 -0400 Subject: [PATCH 2/2] Add scope/collection management tools (create/drop scope, create/drop/update/get_settings collection) --- src/cb_mcp/tools/__init__.py | 31 ++ src/cb_mcp/tools/collections_admin.py | 416 +++++++++++++++++++++++++ tests/unit/test_collections_admin.py | 432 ++++++++++++++++++++++++++ tests/unit/test_read_only_mode.py | 8 +- 4 files changed, 884 insertions(+), 3 deletions(-) create mode 100644 src/cb_mcp/tools/collections_admin.py create mode 100644 tests/unit/test_collections_admin.py diff --git a/src/cb_mcp/tools/__init__.py b/src/cb_mcp/tools/__init__.py index a9993a3b..1e1b88b8 100644 --- a/src/cb_mcp/tools/__init__.py +++ b/src/cb_mcp/tools/__init__.py @@ -12,6 +12,16 @@ from mcp.types import ToolAnnotations +# Scope and collection management tools (write) + get_collection_settings (read) +from .collections_admin import ( + create_collection, + create_scope, + drop_collection, + drop_scope, + get_collection_settings, + update_collection, +) + # Index tools from .index import get_index_advisor_recommendations, list_indexes @@ -79,6 +89,8 @@ list_indexes, # Index settings (read) admin_index_settings_get, + # Collection settings (read) + get_collection_settings, # Query performance analysis tools get_queries_not_selective, get_queries_not_using_covering_index, @@ -106,6 +118,11 @@ drop_index, build_deferred_indexes, admin_index_settings_set, + create_scope, + drop_scope, + create_collection, + drop_collection, + update_collection, ] # List of all tools for easy registration (kept for backward compatibility) @@ -152,6 +169,14 @@ "admin_index_settings_set": ToolAnnotations( destructiveHint=False, idempotentHint=True ), + # Scope/collection management (write) + "create_scope": ToolAnnotations(idempotentHint=False), + "drop_scope": ToolAnnotations(destructiveHint=True, idempotentHint=True), + "create_collection": ToolAnnotations(idempotentHint=False), + "drop_collection": ToolAnnotations(destructiveHint=True, idempotentHint=True), + "update_collection": ToolAnnotations(destructiveHint=False, idempotentHint=True), + # Collection settings (read) + "get_collection_settings": ToolAnnotations(readOnlyHint=True), } @@ -203,6 +228,12 @@ def get_tools( "build_deferred_indexes", "admin_index_settings_get", "admin_index_settings_set", + "create_scope", + "drop_scope", + "create_collection", + "drop_collection", + "update_collection", + "get_collection_settings", "get_cluster_health_and_services", "get_queries_not_selective", "get_queries_not_using_covering_index", diff --git a/src/cb_mcp/tools/collections_admin.py b/src/cb_mcp/tools/collections_admin.py new file mode 100644 index 00000000..43681a03 --- /dev/null +++ b/src/cb_mcp/tools/collections_admin.py @@ -0,0 +1,416 @@ +""" +Tools for scope and collection management. + +These tools create, update, drop, and read settings for scopes and +collections within a bucket. They complement the existing read-only +tools ``get_scopes_in_bucket``, ``get_collections_in_scope``, and +``get_scopes_and_collections_in_bucket``. + +Write gating +------------ +Scope and collection lifecycle is a per-bucket namespace change. These +tools are loaded only when the server is *not* in read-only mode AND +admin-write mode is enabled (see ``tools/__init__.py`` ``get_tools`` +and ``ADMIN_WRITE_TOOLS``). When an OAuth token is present, the caller +must additionally hold the write scope; this mirrors the scope +enforcement in ``run_sql_plus_plus_query``. + +Drop confirmation +----------------- +``drop_scope`` and ``drop_collection`` require both ``confirm=True`` AND +a ``confirm_name`` that matches the target scope/collection name. This +is a UX guard against fat-finger destruction; combined with +admin-write-mode gating and the write-scope check, three independent +gates precede a drop. + +Implementation +-------------- +This module uses the Couchbase SDK's ``CollectionManager`` for both +reads and writes so behavior stays consistent with the existing scope +and collection read tools in ``tools/server.py``. TTL and history- +retention settings are represented by the SDK's ``CollectionSpec`` +type, which correctly handles the ``max_expiry`` (Couchbase 7.6+) vs. +``max_ttl`` (7.0 - 7.5) transition per the SDK's compatibility layer. + +License: MIT - Copyright (c) 2026 Chris Ahrendt +""" + +import logging +from datetime import timedelta +from typing import Any + +from couchbase.management.collections import CollectionSpec +from fastmcp import Context +from fastmcp.server.dependencies import get_access_token + +from ..utils.connection import connect_to_bucket +from ..utils.constants import MCP_SERVER_NAME, SCOPE_WRITE +from ..utils.context import get_cluster_connection + +logger = logging.getLogger(f"{MCP_SERVER_NAME}.tools.collections_admin") + + +def _require_write_scope() -> None: + """Raise ``PermissionError`` if a token is present but lacks the write scope. + + No-op when no token is in context (stdio transport or OAuth disabled), + matching the behavior of ``run_sql_plus_plus_query`` so the same tool + body serves authenticated HTTP and unauthenticated stdio without + branching at registration time. + """ + token = get_access_token() + if token is not None and SCOPE_WRITE not in (token.scopes or []): + held = sorted(set(token.scopes or [])) + msg = ( + f"Scope/collection admin requires the '{SCOPE_WRITE}' scope; " + f"token scopes are {held}." + ) + logger.warning(msg) + raise PermissionError(msg) + + +def _timedelta_or_none(seconds: int | None) -> timedelta | None: + """Convert an optional integer of seconds to timedelta or return None. + + ``0`` maps to ``timedelta(seconds=0)`` explicitly because Couchbase uses + it to mean "no TTL" (whereas ``None`` means "leave unchanged" on update). + """ + if seconds is None: + return None + if seconds < 0: + raise ValueError(f"TTL/max_expiry cannot be negative, got {seconds}") + return timedelta(seconds=seconds) + + +# -------------------------------------------------------------------------- +# Scope management +# -------------------------------------------------------------------------- + + +def create_scope( + ctx: Context, + bucket_name: str, + scope_name: str, +) -> dict[str, Any]: + """Create a scope within a bucket. + + Both ``bucket_name`` and ``scope_name`` are required. Returns a dict + with the created bucket and scope names. + + Fails with ``ScopeAlreadyExistsException`` (from the SDK) if a scope + with that name already exists in the bucket. + """ + _require_write_scope() + + cluster = get_cluster_connection(ctx) + bucket = connect_to_bucket(cluster, bucket_name) + try: + logger.info(f"Creating scope {scope_name!r} in bucket {bucket_name!r}") + bucket.collections().create_scope(scope_name) + return {"bucket": bucket_name, "scope": scope_name, "created": True} + except Exception as e: + logger.error( + f"Error creating scope {scope_name!r} in bucket {bucket_name!r}: {e}", + exc_info=True, + ) + raise + + +def drop_scope( + ctx: Context, + bucket_name: str, + scope_name: str, + confirm: bool = False, + confirm_name: str | None = None, +) -> dict[str, Any]: + """Drop a scope from a bucket. Irreversible; deletes all collections + (and thus all documents) in the scope. + + Requires BOTH: + + - ``confirm=True`` + - ``confirm_name`` set to the exact scope name + + The name-match is a UX guard against fat-finger deletion; combined + with the admin-write-mode gate and the OAuth write-scope check, + three independent gates precede a drop. + + Cannot drop ``_default`` (Couchbase server rejects it). + """ + _require_write_scope() + + if not confirm: + raise ValueError( + "drop_scope requires confirm=True. This operation is " + "irreversible and deletes every collection and document in " + "the scope." + ) + if confirm_name != scope_name: + raise ValueError( + "drop_scope requires confirm_name to exactly match " + f"scope_name ({scope_name!r}). This guard against fat-finger " + "deletion matches the delete_bucket pattern." + ) + if scope_name == "_default": + raise ValueError( + "The _default scope cannot be dropped (Couchbase server " + "rejects this operation)." + ) + + cluster = get_cluster_connection(ctx) + bucket = connect_to_bucket(cluster, bucket_name) + try: + logger.warning(f"Dropping scope {scope_name!r} in bucket {bucket_name!r}") + bucket.collections().drop_scope(scope_name) + return {"bucket": bucket_name, "scope": scope_name, "dropped": True} + except Exception as e: + logger.error( + f"Error dropping scope {scope_name!r} in bucket {bucket_name!r}: {e}", + exc_info=True, + ) + raise + + +# -------------------------------------------------------------------------- +# Collection management +# -------------------------------------------------------------------------- + + +def create_collection( + ctx: Context, + bucket_name: str, + scope_name: str, + collection_name: str, + max_expiry_seconds: int | None = None, + history: bool | None = None, +) -> dict[str, Any]: + """Create a collection inside a scope. + + Required: ``bucket_name``, ``scope_name``, ``collection_name``. + + Optional per-collection settings: + + - ``max_expiry_seconds``: default document TTL in seconds. ``0`` = no + TTL (documents don't expire). ``None`` = inherit from the bucket + default. + - ``history``: enable per-document history retention (Couchbase 7.2+ + with Magma storage engine). ``None`` = server default. + + Returns a dict with the created keyspace and settings applied. + """ + _require_write_scope() + + spec = CollectionSpec( + collection_name=collection_name, + scope_name=scope_name, + max_expiry=_timedelta_or_none(max_expiry_seconds), + history=history, + ) + + cluster = get_cluster_connection(ctx) + bucket = connect_to_bucket(cluster, bucket_name) + try: + logger.info(f"Creating collection {bucket_name}.{scope_name}.{collection_name}") + bucket.collections().create_collection(spec) + return { + "bucket": bucket_name, + "scope": scope_name, + "collection": collection_name, + "max_expiry_seconds": max_expiry_seconds, + "history": history, + "created": True, + } + except Exception as e: + logger.error( + f"Error creating collection " + f"{bucket_name}.{scope_name}.{collection_name}: {e}", + exc_info=True, + ) + raise + + +def drop_collection( + ctx: Context, + bucket_name: str, + scope_name: str, + collection_name: str, + confirm: bool = False, + confirm_name: str | None = None, +) -> dict[str, Any]: + """Drop a collection. Irreversible; deletes all documents in the + collection. + + Requires BOTH: + + - ``confirm=True`` + - ``confirm_name`` set to the exact collection name + + The name-match is a UX guard against fat-finger deletion; combined + with the admin-write-mode gate and the OAuth write-scope check, + three independent gates precede a drop. + + Cannot drop ``_default`` collection from ``_default`` scope (Couchbase + server rejects it). + """ + _require_write_scope() + + if not confirm: + raise ValueError( + "drop_collection requires confirm=True. This operation is " + "irreversible and deletes every document in the collection." + ) + if confirm_name != collection_name: + raise ValueError( + "drop_collection requires confirm_name to exactly match " + f"collection_name ({collection_name!r}). This guard against " + "fat-finger deletion matches the delete_bucket pattern." + ) + if scope_name == "_default" and collection_name == "_default": + raise ValueError( + "The _default collection in the _default scope cannot be " + "dropped (Couchbase server rejects this operation)." + ) + + cluster = get_cluster_connection(ctx) + bucket = connect_to_bucket(cluster, bucket_name) + try: + logger.warning( + f"Dropping collection {bucket_name}.{scope_name}.{collection_name}" + ) + # The SDK's drop_collection takes a CollectionSpec but only reads + # scope_name and collection_name from it, so a minimal spec is fine. + spec = CollectionSpec(collection_name=collection_name, scope_name=scope_name) + bucket.collections().drop_collection(spec) + return { + "bucket": bucket_name, + "scope": scope_name, + "collection": collection_name, + "dropped": True, + } + except Exception as e: + logger.error( + f"Error dropping collection " + f"{bucket_name}.{scope_name}.{collection_name}: {e}", + exc_info=True, + ) + raise + + +def update_collection( + ctx: Context, + bucket_name: str, + scope_name: str, + collection_name: str, + max_expiry_seconds: int | None = None, + history: bool | None = None, +) -> dict[str, Any]: + """Update mutable settings on an existing collection. + + Both ``max_expiry_seconds`` and ``history`` are optional; only the + values explicitly provided are sent to the server. Passing both as + ``None`` (the default) is a no-op — raises an error since the caller + likely made a mistake. + + Returns a dict with the settings sent to the server. + """ + _require_write_scope() + + if max_expiry_seconds is None and history is None: + raise ValueError( + "update_collection requires at least one of " + "max_expiry_seconds or history to be provided." + ) + + spec = CollectionSpec( + collection_name=collection_name, + scope_name=scope_name, + max_expiry=_timedelta_or_none(max_expiry_seconds), + history=history, + ) + + cluster = get_cluster_connection(ctx) + bucket = connect_to_bucket(cluster, bucket_name) + try: + logger.info(f"Updating collection {bucket_name}.{scope_name}.{collection_name}") + bucket.collections().update_collection(spec) + return { + "bucket": bucket_name, + "scope": scope_name, + "collection": collection_name, + "max_expiry_seconds": max_expiry_seconds, + "history": history, + "updated": True, + } + except Exception as e: + logger.error( + f"Error updating collection " + f"{bucket_name}.{scope_name}.{collection_name}: {e}", + exc_info=True, + ) + raise + + +# -------------------------------------------------------------------------- +# Collection settings read +# -------------------------------------------------------------------------- + + +def get_collection_settings( + ctx: Context, + bucket_name: str, + scope_name: str, + collection_name: str, +) -> dict[str, Any]: + """Get the settings for a single collection. + + Returns a dict with: + + - ``bucket``, ``scope``, ``collection``: the keyspace identifiers + - ``max_expiry_seconds``: default TTL in seconds; ``0`` = no TTL, + ``-1`` = inherit bucket default (per Couchbase 7.6+ semantics), + ``None`` = server did not return a value + - ``history``: whether per-document history retention is enabled + + Raises ``ValueError`` if the requested scope or collection does not + exist in the bucket. + """ + cluster = get_cluster_connection(ctx) + bucket = connect_to_bucket(cluster, bucket_name) + try: + logger.debug( + f"Reading settings for {bucket_name}.{scope_name}.{collection_name}" + ) + scopes = bucket.collections().get_all_scopes() + for scope in scopes: + if scope.name != scope_name: + continue + for coll in scope.collections: + if coll.name != collection_name: + continue + # SDK exposes max_expiry as timedelta or None + exp = getattr(coll, "max_expiry", None) + if isinstance(exp, timedelta): + max_expiry_seconds = int(exp.total_seconds()) + else: + max_expiry_seconds = None + return { + "bucket": bucket_name, + "scope": scope_name, + "collection": collection_name, + "max_expiry_seconds": max_expiry_seconds, + "history": getattr(coll, "history", None), + } + raise ValueError( + f"Collection {collection_name!r} not found in scope " + f"{scope_name!r} of bucket {bucket_name!r}" + ) + raise ValueError(f"Scope {scope_name!r} not found in bucket {bucket_name!r}") + except ValueError: + raise + except Exception as e: + logger.error( + f"Error reading settings for " + f"{bucket_name}.{scope_name}.{collection_name}: {e}", + exc_info=True, + ) + raise diff --git a/tests/unit/test_collections_admin.py b/tests/unit/test_collections_admin.py new file mode 100644 index 00000000..f174a04e --- /dev/null +++ b/tests/unit/test_collections_admin.py @@ -0,0 +1,432 @@ +""" +Unit tests for the scope/collection management tool bodies (parameter +validation, argument construction, confirmation gates, and write-scope +gating), exercised against stubbed CollectionManager / cluster objects +so no live Couchbase cluster is required. + +License: MIT - Copyright (c) 2026 Chris Ahrendt +""" + +from contextlib import contextmanager +from datetime import timedelta +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from cb_mcp.tools.collections_admin import ( + _require_write_scope, + _timedelta_or_none, + create_collection, + create_scope, + drop_collection, + drop_scope, + get_collection_settings, + update_collection, +) +from cb_mcp.utils.constants import SCOPE_WRITE + +# -------------------------------------------------------------------------- +# Helpers +# -------------------------------------------------------------------------- + + +@contextmanager +def _mock_token(scopes): + """Patch get_access_token at the collections_admin call site.""" + + class _Tok: + def __init__(self, scopes_): + self.scopes = scopes_ + + token = _Tok(scopes) if scopes is not None else None + with patch("cb_mcp.tools.collections_admin.get_access_token", return_value=token): + yield + + +def _ctx(): + """Minimal Context stub — the tools pass it to helper functions we stub.""" + return SimpleNamespace() + + +def _stub_cluster_bucket(monkeypatch, coll_mgr=None): + """Stub get_cluster_connection and connect_to_bucket so we get a + controllable CollectionManager mock back.""" + coll_mgr = coll_mgr or MagicMock() + bucket = MagicMock() + bucket.collections.return_value = coll_mgr + cluster = MagicMock() + monkeypatch.setattr( + "cb_mcp.tools.collections_admin.get_cluster_connection", + lambda ctx: cluster, + ) + monkeypatch.setattr( + "cb_mcp.tools.collections_admin.connect_to_bucket", + lambda cluster, bucket_name: bucket, + ) + return coll_mgr + + +CTX = _ctx() + + +# -------------------------------------------------------------------------- +# _timedelta_or_none +# -------------------------------------------------------------------------- + + +def test_timedelta_or_none_none(): + assert _timedelta_or_none(None) is None + + +def test_timedelta_or_none_zero(): + """Zero is 'no TTL', explicit — must return timedelta(0), not None.""" + assert _timedelta_or_none(0) == timedelta(seconds=0) + + +def test_timedelta_or_none_positive(): + assert _timedelta_or_none(3600) == timedelta(seconds=3600) + + +def test_timedelta_or_none_negative_rejected(): + with pytest.raises(ValueError, match="cannot be negative"): + _timedelta_or_none(-1) + + +# -------------------------------------------------------------------------- +# _require_write_scope +# -------------------------------------------------------------------------- + + +def test_require_write_scope_noop_without_token(): + with _mock_token(None): + _require_write_scope() + + +def test_require_write_scope_raises_when_missing(): + with _mock_token(["read"]), pytest.raises(PermissionError, match="write"): + _require_write_scope() + + +def test_require_write_scope_passes_when_present(): + with _mock_token(["read", SCOPE_WRITE]): + _require_write_scope() + + +# -------------------------------------------------------------------------- +# create_scope +# -------------------------------------------------------------------------- + + +def test_create_scope_calls_sdk_create_scope(monkeypatch): + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + result = create_scope(CTX, bucket_name="b1", scope_name="s1") + coll_mgr.create_scope.assert_called_once_with("s1") + assert result == {"bucket": "b1", "scope": "s1", "created": True} + + +def test_create_scope_propagates_sdk_error(monkeypatch): + coll_mgr = _stub_cluster_bucket(monkeypatch) + coll_mgr.create_scope.side_effect = RuntimeError("boom") + with _mock_token(None), pytest.raises(RuntimeError, match="boom"): + create_scope(CTX, bucket_name="b1", scope_name="s1") + + +# -------------------------------------------------------------------------- +# drop_scope +# -------------------------------------------------------------------------- + + +def test_drop_scope_requires_confirm_true(): + with _mock_token(None), pytest.raises(ValueError, match="confirm=True"): + drop_scope(CTX, bucket_name="b1", scope_name="s1") + + +def test_drop_scope_requires_matching_confirm_name(): + with _mock_token(None), pytest.raises(ValueError, match="confirm_name"): + drop_scope( + CTX, + bucket_name="b1", + scope_name="s1", + confirm=True, + confirm_name="other", + ) + + +def test_drop_scope_requires_confirm_name_explicitly(): + with _mock_token(None), pytest.raises(ValueError, match="confirm_name"): + drop_scope(CTX, bucket_name="b1", scope_name="s1", confirm=True) + + +def test_drop_scope_rejects_default_scope(): + with _mock_token(None), pytest.raises(ValueError, match="_default scope"): + drop_scope( + CTX, + bucket_name="b1", + scope_name="_default", + confirm=True, + confirm_name="_default", + ) + + +def test_drop_scope_succeeds_with_matching_name(monkeypatch): + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + result = drop_scope( + CTX, + bucket_name="b1", + scope_name="s1", + confirm=True, + confirm_name="s1", + ) + coll_mgr.drop_scope.assert_called_once_with("s1") + assert result == {"bucket": "b1", "scope": "s1", "dropped": True} + + +# -------------------------------------------------------------------------- +# create_collection +# -------------------------------------------------------------------------- + + +def test_create_collection_minimal(monkeypatch): + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + result = create_collection( + CTX, + bucket_name="b1", + scope_name="s1", + collection_name="c1", + ) + call = coll_mgr.create_collection.call_args + spec = call.args[0] + assert spec.name == "c1" + assert spec.scope_name == "s1" + assert result["created"] is True + + +def test_create_collection_with_ttl_and_history(monkeypatch): + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + result = create_collection( + CTX, + bucket_name="b1", + scope_name="s1", + collection_name="c1", + max_expiry_seconds=3600, + history=True, + ) + spec = coll_mgr.create_collection.call_args.args[0] + assert spec.max_expiry == timedelta(seconds=3600) + assert spec.history is True + assert result["max_expiry_seconds"] == 3600 + assert result["history"] is True + + +def test_create_collection_zero_ttl_means_no_expiry(monkeypatch): + """max_expiry_seconds=0 means 'no TTL', distinct from None ('inherit').""" + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + create_collection( + CTX, + bucket_name="b1", + scope_name="s1", + collection_name="c1", + max_expiry_seconds=0, + ) + spec = coll_mgr.create_collection.call_args.args[0] + assert spec.max_expiry == timedelta(seconds=0) + + +def test_create_collection_negative_ttl_rejected(): + with _mock_token(None), pytest.raises(ValueError, match="cannot be negative"): + create_collection( + CTX, + bucket_name="b1", + scope_name="s1", + collection_name="c1", + max_expiry_seconds=-5, + ) + + +# -------------------------------------------------------------------------- +# drop_collection +# -------------------------------------------------------------------------- + + +def test_drop_collection_requires_confirm_true(): + with _mock_token(None), pytest.raises(ValueError, match="confirm=True"): + drop_collection(CTX, bucket_name="b1", scope_name="s1", collection_name="c1") + + +def test_drop_collection_requires_matching_confirm_name(): + with _mock_token(None), pytest.raises(ValueError, match="confirm_name"): + drop_collection( + CTX, + bucket_name="b1", + scope_name="s1", + collection_name="c1", + confirm=True, + confirm_name="c2", + ) + + +def test_drop_collection_rejects_default_default(): + with _mock_token(None), pytest.raises(ValueError, match="_default"): + drop_collection( + CTX, + bucket_name="b1", + scope_name="_default", + collection_name="_default", + confirm=True, + confirm_name="_default", + ) + + +def test_drop_collection_allows_default_scope_non_default_collection(monkeypatch): + """A user-created collection in the _default scope IS droppable.""" + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + result = drop_collection( + CTX, + bucket_name="b1", + scope_name="_default", + collection_name="my_coll", + confirm=True, + confirm_name="my_coll", + ) + coll_mgr.drop_collection.assert_called_once() + assert result["dropped"] is True + + +def test_drop_collection_succeeds(monkeypatch): + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + result = drop_collection( + CTX, + bucket_name="b1", + scope_name="s1", + collection_name="c1", + confirm=True, + confirm_name="c1", + ) + spec = coll_mgr.drop_collection.call_args.args[0] + assert spec.name == "c1" + assert spec.scope_name == "s1" + assert result["dropped"] is True + + +# -------------------------------------------------------------------------- +# update_collection +# -------------------------------------------------------------------------- + + +def test_update_collection_requires_at_least_one_change(): + with _mock_token(None), pytest.raises(ValueError, match="at least one"): + update_collection(CTX, bucket_name="b1", scope_name="s1", collection_name="c1") + + +def test_update_collection_ttl_only(monkeypatch): + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + result = update_collection( + CTX, + bucket_name="b1", + scope_name="s1", + collection_name="c1", + max_expiry_seconds=7200, + ) + spec = coll_mgr.update_collection.call_args.args[0] + assert spec.max_expiry == timedelta(seconds=7200) + assert result["updated"] is True + + +def test_update_collection_history_only(monkeypatch): + coll_mgr = _stub_cluster_bucket(monkeypatch) + with _mock_token(None): + update_collection( + CTX, + bucket_name="b1", + scope_name="s1", + collection_name="c1", + history=False, + ) + spec = coll_mgr.update_collection.call_args.args[0] + assert spec.history is False + + +# -------------------------------------------------------------------------- +# get_collection_settings +# -------------------------------------------------------------------------- + + +def _fake_scope_spec(name, collections): + """Build a minimal object shaped like the SDK's ScopeSpec.""" + scope = SimpleNamespace(name=name, collections=collections) + return scope + + +def _fake_collection_spec(name, max_expiry=None, history=None): + c = SimpleNamespace(name=name) + if max_expiry is not None: + c.max_expiry = max_expiry + if history is not None: + c.history = history + return c + + +def test_get_collection_settings_returns_ttl_and_history(monkeypatch): + coll = _fake_collection_spec("c1", max_expiry=timedelta(seconds=1800), history=True) + scope = _fake_scope_spec("s1", [coll]) + coll_mgr = MagicMock() + coll_mgr.get_all_scopes.return_value = [scope] + _stub_cluster_bucket(monkeypatch, coll_mgr=coll_mgr) + + result = get_collection_settings( + CTX, bucket_name="b1", scope_name="s1", collection_name="c1" + ) + assert result == { + "bucket": "b1", + "scope": "s1", + "collection": "c1", + "max_expiry_seconds": 1800, + "history": True, + } + + +def test_get_collection_settings_none_max_expiry(monkeypatch): + """When the SDK returns no max_expiry, we return None (not 0).""" + coll = _fake_collection_spec("c1", history=False) + scope = _fake_scope_spec("s1", [coll]) + coll_mgr = MagicMock() + coll_mgr.get_all_scopes.return_value = [scope] + _stub_cluster_bucket(monkeypatch, coll_mgr=coll_mgr) + + result = get_collection_settings( + CTX, bucket_name="b1", scope_name="s1", collection_name="c1" + ) + assert result["max_expiry_seconds"] is None + assert result["history"] is False + + +def test_get_collection_settings_scope_not_found(monkeypatch): + coll_mgr = MagicMock() + coll_mgr.get_all_scopes.return_value = [_fake_scope_spec("other", [])] + _stub_cluster_bucket(monkeypatch, coll_mgr=coll_mgr) + + with pytest.raises(ValueError, match="Scope 's1' not found"): + get_collection_settings( + CTX, bucket_name="b1", scope_name="s1", collection_name="c1" + ) + + +def test_get_collection_settings_collection_not_found(monkeypatch): + scope = _fake_scope_spec("s1", [_fake_collection_spec("other")]) + coll_mgr = MagicMock() + coll_mgr.get_all_scopes.return_value = [scope] + _stub_cluster_bucket(monkeypatch, coll_mgr=coll_mgr) + + with pytest.raises(ValueError, match="Collection 'c1' not found"): + get_collection_settings( + CTX, bucket_name="b1", scope_name="s1", collection_name="c1" + ) diff --git a/tests/unit/test_read_only_mode.py b/tests/unit/test_read_only_mode.py index 0722ef85..1ce988b9 100644 --- a/tests/unit/test_read_only_mode.py +++ b/tests/unit/test_read_only_mode.py @@ -24,7 +24,7 @@ "delete_document_by_id", } -# Read-only tool names that should always be available (21 tools) +# Read-only tool names that should always be available (22 tools) READ_ONLY_TOOL_NAMES = { # Server/Cluster management tools (7) "get_buckets_in_cluster", @@ -45,6 +45,8 @@ "list_indexes", # Index settings read tool (1) "admin_index_settings_get", + # Collection settings read tool (1) + "get_collection_settings", # Query performance analysis tools (7) "get_queries_not_selective", "get_queries_not_using_covering_index", @@ -159,13 +161,13 @@ def test_read_only_mode_tool_count(self): """Verify correct number of tools in read-only mode.""" tools = get_tools(read_only_mode=True) assert len(tools) == len(READ_ONLY_TOOLS) - assert len(tools) == 21 # Expected count of read-only tools + assert len(tools) == 22 # Expected count of read-only tools def test_all_tools_mode_tool_count(self): """Verify correct number of tools when all write tools are enabled.""" tools = get_tools(read_only_mode=False) assert len(tools) == len(ALL_TOOLS) - assert len(tools) == 25 # Expected total count (21 read-only + 4 KV write) + assert len(tools) == 26 # Expected total count (22 read-only + 4 KV write) def test_kv_write_tools_count(self): """Verify exactly 4 KV write tools exist."""