diff --git a/nextcloud_mcp_server/api/__init__.py b/nextcloud_mcp_server/api/__init__.py index 6b79a44e..05d7bb39 100644 --- a/nextcloud_mcp_server/api/__init__.py +++ b/nextcloud_mcp_server/api/__init__.py @@ -36,6 +36,9 @@ from nextcloud_mcp_server.api.passwords import ( get_app_password_status, provision_app_password, ) +from nextcloud_mcp_server.api.vector_sync import ( + purge_doc_types_route, +) from nextcloud_mcp_server.api.visualization import ( get_chunk_context, get_pdf_preview, @@ -78,6 +81,8 @@ __all__ = [ "list_webhooks", "create_webhook", "delete_webhook", + # Vector-sync admin endpoints (from vector_sync.py) + "purge_doc_types_route", # Visualization endpoints (from visualization.py) "unified_search", "vector_search", diff --git a/nextcloud_mcp_server/api/vector_sync.py b/nextcloud_mcp_server/api/vector_sync.py new file mode 100644 index 00000000..7668e2e6 --- /dev/null +++ b/nextcloud_mcp_server/api/vector_sync.py @@ -0,0 +1,125 @@ +"""Vector-sync admin API endpoints. + +Provides the purge endpoint Astrolabe calls when an admin disables a content +source for semantic search. Consent is binding on data-at-rest, so the +already-indexed content for the disabled source's doc type(s) is deleted +globally (every owner) — see :mod:`nextcloud_mcp_server.vector.purge`. + +Auth: the OAuth bearer identifies the caller (``validate_token_and_get_user``); +because the purge deletes every owner's content for a doc type, it is further +restricted to Nextcloud administrators (verified via the ``admin`` group using +the caller's app password). This is stricter than the per-user webhook routes +in :mod:`nextcloud_mcp_server.api.webhooks` precisely because the blast radius +is global. +""" + +import logging + +import httpx +from starlette.requests import Request +from starlette.responses import JSONResponse + +from nextcloud_mcp_server.api._auth import get_basic_auth_for_user +from nextcloud_mcp_server.api.management import ( + _sanitize_error_for_client, + validate_token_and_get_user, +) +from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError +from nextcloud_mcp_server.client.users import UsersClient +from nextcloud_mcp_server.vector.purge import purge_doc_types + +from ..http import nextcloud_httpx_client + +logger = logging.getLogger(__name__) + + +async def purge_doc_types_route(request: Request) -> JSONResponse: + """POST /api/v1/vector-sync/purge — delete indexed vectors by doc type. + + Request body:: + + {"doc_types": ["file", "note"]} + + Returns ``{"purged": {doc_type: deleted_count}}``. Admin-only. + + Requires OAuth bearer token for authentication. + """ + try: + user_id, _ = await validate_token_and_get_user(request) + except Exception as e: + logger.warning("Unauthorized access to /api/v1/vector-sync/purge: %s", e) + return JSONResponse( + { + "error": "Unauthorized", + "message": _sanitize_error_for_client(e, "purge_doc_types"), + }, + status_code=401, + ) + + try: + body = await request.json() + except Exception as e: + logger.warning("Purge payload was not valid JSON: %s", e) + return JSONResponse( + {"error": "Bad request", "message": "invalid JSON"}, + status_code=400, + ) + + raw = body.get("doc_types") + if not isinstance(raw, list) or not all(isinstance(d, str) for d in raw): + return JSONResponse( + { + "error": "Bad request", + "message": "doc_types must be a list of strings", + }, + status_code=400, + ) + doc_types = [d for d in raw if d] + if not doc_types: + return JSONResponse({"purged": {}}) + + try: + username, app_password = await get_basic_auth_for_user(user_id) + + oauth_ctx = request.app.state.oauth_context + nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "") + if not nextcloud_host: + raise ValueError("Nextcloud host not configured") + + # Verify admin via the caller's own app password before any deletion. + async with nextcloud_httpx_client( + base_url=nextcloud_host, + auth=httpx.BasicAuth(username, app_password), + timeout=30.0, + ) as client: + users_client = UsersClient(client, username) + user_groups = await users_client.get_user_groups(username) + if "admin" not in user_groups: + logger.warning("Non-admin user %s attempted vector-sync purge", user_id) + return JSONResponse( + { + "error": "Forbidden", + "message": "Administrator privileges required", + }, + status_code=403, + ) + + purged = await purge_doc_types(doc_types) + logger.info("Vector-sync purge by admin %s: %s", user_id, purged) + return JSONResponse({"purged": purged}) + + except ProvisioningRequiredError as e: + logger.info("Provisioning required for user %s: %s", user_id, e) + return JSONResponse( + {"error": "Provisioning required", "message": str(e)}, + status_code=428, + ) + except Exception as e: + logger.error("Error purging doc types for user %s: %s", user_id, e) + return JSONResponse( + { + "error": "Internal error", + "message": _sanitize_error_for_client(e, "purge_doc_types"), + }, + status_code=500, + ) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 6168fbc2..f5b2fa87 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -47,6 +47,7 @@ from nextcloud_mcp_server.api import ( list_supported_scopes, list_webhooks, provision_app_password, + purge_doc_types_route, revoke_user_access, unified_search, update_user_scopes, @@ -2422,6 +2423,15 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = routes.append( Route("/api/v1/webhooks/{webhook_id}", delete_webhook, methods=["DELETE"]) ) + # Vector-sync admin: purge indexed vectors by doc type (admin consent — + # called by Astrolabe when a source is disabled for semantic search) + routes.append( + Route( + "/api/v1/vector-sync/purge", + purge_doc_types_route, + methods=["POST"], + ) + ) # Access and scope management endpoints (ADR-022) routes.append( Route( diff --git a/nextcloud_mcp_server/capabilities.py b/nextcloud_mcp_server/capabilities.py new file mode 100644 index 00000000..90cc1850 --- /dev/null +++ b/nextcloud_mcp_server/capabilities.py @@ -0,0 +1,115 @@ +"""Reads admin-approved searchable sources from the Astrolabe capability. + +The Astrolabe Nextcloud app advertises, per user, which content sources an +admin has approved for semantic search, under +``capabilities.astrolabe.semantic_search.enabled_doc_types`` on the OCS +capabilities endpoint (``/ocs/v2.php/cloud/capabilities``). This is the single +source of truth for admin consent: the search layer filters results to these +doc types, and the indexing layer (scanner + webhook ingest) skips everything +else (the hard data-at-rest guarantee is the eager purge Astrolabe triggers on +disable; see ``vector.purge``). + +Fail-open for *availability*: if the capability block is absent (an older +Astrolabe that predates this feature) or the OCS call fails, ``allowed_doc_types`` +returns ``None`` meaning "no restriction", so search keeps working. ``None`` is +distinct from an empty set, which means "the admin disabled every source". +""" + +from __future__ import annotations + +import logging +import time +from collections import OrderedDict +from typing import Any, Protocol + +logger = logging.getLogger(__name__) + +# Short-lived per-user cache for the OCS capabilities lookup. Admin consent +# changes rarely, but search/scan paths consult it frequently, so trade a little +# staleness for keeping the OCS round-trip off the hot path. Mirrors the +# list_accessible_owners cache in search/access_filter.py. +_CACHE_TTL_SECONDS = 30.0 +_CACHE_MAXSIZE = 1024 +# user_id -> (monotonic_ts, frozenset[doc_type] | None). None = no restriction. +_cache: OrderedDict[str, tuple[float, frozenset[str] | None]] = OrderedDict() + + +class _CapabilitiesClientProtocol(Protocol): + async def capabilities(self) -> Any: ... + + +def _parse_enabled_doc_types(payload: Any) -> set[str] | None: + """Extract ``enabled_doc_types`` from an OCS capabilities payload. + + Returns ``None`` when the ``astrolabe.semantic_search`` block is absent or + malformed (treated as "no restriction"). Returns a set (possibly empty) when + the block is present and well-formed; an empty set means the admin disabled + every source. + """ + if not isinstance(payload, dict): + return None + try: + caps = payload["ocs"]["data"]["capabilities"] + except (KeyError, TypeError): + return None + if not isinstance(caps, dict): + return None + block = caps.get("astrolabe") + if not isinstance(block, dict): + return None + semantic = block.get("semantic_search") + if not isinstance(semantic, dict): + return None + raw = semantic.get("enabled_doc_types") + if not isinstance(raw, list): + return None + return {dt for dt in raw if isinstance(dt, str)} + + +async def allowed_doc_types( + client: _CapabilitiesClientProtocol, user_id: str +) -> frozenset[str] | None: + """Admin-approved doc types for ``user_id``, or ``None`` for "no restriction". + + Cached per user with a short TTL (+ LRU eviction). Failures are not cached so + a transient OCS hiccup retries on the next call. Fail-open: a missing + capability block or an error yields ``None`` so search remains available. + """ + now = time.monotonic() + cached = _cache.get(user_id) + if cached is not None and now - cached[0] < _CACHE_TTL_SECONDS: + _cache.move_to_end(user_id) # mark recently used (LRU) + return cached[1] + + try: + payload = await client.capabilities() + except Exception as exc: # noqa: BLE001 — degrade gracefully (fail-open) + logger.warning( + "Astrolabe capabilities unavailable for user %s (%s); " + "not restricting doc types this cycle", + user_id, + exc, + ) + return None # don't cache failures — retry next call + + parsed = _parse_enabled_doc_types(payload) + result = frozenset(parsed) if parsed is not None else None + _cache[user_id] = (now, result) + _cache.move_to_end(user_id) + while len(_cache) > _CACHE_MAXSIZE: + _cache.popitem(last=False) # evict least-recently-used + return result + + +def is_doc_type_allowed(doc_type: str, allowed: frozenset[str] | None) -> bool: + """Whether ``doc_type`` may be indexed/searched given an allow-set. + + ``allowed=None`` means "no restriction" (fail-open / older Astrolabe), so + everything is permitted. + """ + return allowed is None or doc_type in allowed + + +def clear_cache() -> None: + """Test hook: drop all cached entries.""" + _cache.clear() diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 0696fbcc..eec7bd41 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -20,6 +20,7 @@ from mcp.types import ( from pydantic import Field from nextcloud_mcp_server.auth import require_scopes +from nextcloud_mcp_server.capabilities import allowed_doc_types from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.context import get_client from nextcloud_mcp_server.models.semantic import ( @@ -300,6 +301,33 @@ def configure_semantic_tools(mcp: FastMCP): # files under their own user_id. accessible_owners = await list_accessible_owners(client.sharing, username) + # Admin consent gate: restrict to source types the Astrolabe admin has + # approved (and that are installed for this user). This mirrors + # Astrolabe's own server-side enforcement but is independent because + # this tool queries Qdrant directly. ``None`` = no restriction + # (fail-open / Astrolabe predating this feature). An empty allow-set + # means the admin disabled every source. + allowed = await allowed_doc_types(client, username) + if allowed is not None: + if doc_types is None: + doc_types = sorted(allowed) + else: + doc_types = [dt for dt in doc_types if dt in allowed] + if not doc_types: + logger.info( + "Semantic search short-circuited for user %s: no requested " + "doc_type is both installed and admin-approved", + username, + ) + return SemanticSearchResponse( + results=[], + query=query, + total_found=0, + search_method=f"bm25_hybrid_{fusion}", + verified_chunk_count=0, + dropped_document_count=0, + ) + try: # The nc_semantic_search tool deliberately uses BM25-hybrid (dense + # sparse with RRF/DBSF fusion) as the single tool-layer algorithm. diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 64b968e1..5d93b7d6 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -21,6 +21,7 @@ if TYPE_CHECKING: from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry from nextcloud_mcp_server.acl_hash import compute_acl_hash +from nextcloud_mcp_server.capabilities import allowed_doc_types, is_doc_type_allowed from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service @@ -574,6 +575,24 @@ async def process_document( ): await _reconcile_tag_event(doc_task, nc_client) + # Admin consent gate (Astrolabe): never index a source the admin has + # disabled for semantic search — this catches near-real-time webhook + # events that bypass the scanner's discovery gate. Deletes always + # proceed (removing data honours consent). ``None`` from the reader + # means no restriction (fail-open / older Astrolabe), so a transient + # capabilities failure never silently drops indexing. + if doc_task.operation == "index": + allowed = await allowed_doc_types(nc_client, doc_task.user_id) + if not is_doc_type_allowed(doc_task.doc_type, allowed): + logger.info( + "Skipping index of %s_%s for %s: doc_type disabled by admin", + doc_task.doc_type, + doc_task.doc_id, + doc_task.user_id, + ) + record_vector_sync_processing(time.time() - start_time, "skipped") + return + # Handle deletion if doc_task.operation == "delete": # Release this user rather than blind-delete: a file shared across diff --git a/nextcloud_mcp_server/vector/purge.py b/nextcloud_mcp_server/vector/purge.py new file mode 100644 index 00000000..88d8baca --- /dev/null +++ b/nextcloud_mcp_server/vector/purge.py @@ -0,0 +1,73 @@ +"""Global purge of indexed vectors by doc type (admin consent enforcement). + +When an admin disables a content source for semantic search in Astrolabe, +consent is binding on data-at-rest: the already-indexed content for that +source's doc type(s) must be deleted, not merely hidden. Astrolabe calls the +``/api/v1/vector-sync/purge`` route on disable, which delegates here. + +The purge is global (every owner) because the admin disable is a global +decision. It is safe to call for a doc type with no indexed points — Qdrant +deletes zero points and reports a count of 0. +""" + +from __future__ import annotations + +import logging + +from qdrant_client.models import FieldCondition, Filter, MatchValue + +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client + +logger = logging.getLogger(__name__) + + +def _doc_type_filter(doc_type: str) -> Filter: + return Filter( + must=[FieldCondition(key="doc_type", match=MatchValue(value=doc_type))] + ) + + +async def purge_doc_types(doc_types: list[str]) -> dict[str, int]: + """Delete every indexed point whose ``doc_type`` is in ``doc_types``. + + Returns a mapping of doc_type -> number of points deleted (counted before + deletion). Each doc type is purged independently so a failure on one does + not abort the rest; failures re-raise after the loop only if every doc type + failed, otherwise partial progress is returned. + """ + qdrant_client = await get_qdrant_client() + collection = get_settings().get_collection_name() + + purged: dict[str, int] = {} + last_error: Exception | None = None + for doc_type in dict.fromkeys(doc_types): # de-dupe, preserve order + flt = _doc_type_filter(doc_type) + try: + count_result = await qdrant_client.count( + collection_name=collection, + count_filter=flt, + exact=True, + ) + await qdrant_client.delete( + collection_name=collection, + points_selector=flt, + ) + purged[doc_type] = int(count_result.count) + logger.info( + "Purged %d indexed point(s) for disabled doc_type=%s", + purged[doc_type], + doc_type, + ) + except Exception as exc: # noqa: BLE001 — record and continue + last_error = exc + logger.error( + "Failed to purge indexed points for doc_type=%s: %s", + doc_type, + exc, + ) + + if not purged and last_error is not None: + # Nothing succeeded — surface the failure to the caller (HTTP 500). + raise last_error + return purged diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index b1085aa0..da802c48 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -16,6 +16,7 @@ from httpx import HTTPStatusError from qdrant_client import AsyncQdrantClient from qdrant_client.models import FieldCondition, Filter, MatchValue, Record +from nextcloud_mcp_server.capabilities import allowed_doc_types, is_doc_type_allowed from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.client.news import NewsItemType from nextcloud_mcp_server.config import get_settings @@ -365,6 +366,13 @@ async def scan_user_documents( # detection failed: fall back to scanning every app (prior behaviour). enabled_apps = await _get_enabled_apps_or_none(nc_client, user_id, scan_id) + # Admin consent gate (Astrolabe): only index sources the admin has + # approved for semantic search. ``None`` = no restriction (fail-open / + # older Astrolabe), so a transient capabilities failure never silently + # halts (or worse, mass-deletes) indexing. This is independent of + # ``enabled_apps``, which reflects only what the user has installed. + allowed = await allowed_doc_types(nc_client, user_id) + # Notes (isolated so an uninstalled or disabled Notes app — whose API # returns 404 — cannot abort scanning of the other apps; this mirrors the # per-app try/except guards already wrapping files/news/deck below). @@ -373,7 +381,7 @@ async def scan_user_documents( current_time = time.time() queued = 0 - if _app_enabled("notes", enabled_apps): + if _app_enabled("notes", enabled_apps) and is_doc_type_allowed("note", allowed): try: queued += await scan_notes( user_id=user_id, @@ -454,9 +462,21 @@ async def scan_user_documents( # folder applies to every PDF beneath it. settings = get_settings() tag_name = settings.vector_sync_pdf_tag - tagged_files = await nc_client.find_files_by_tag( - tag_name, mime_type_filter="application/pdf" - ) + if is_doc_type_allowed("file", allowed): + tagged_files = await nc_client.find_files_by_tag( + tag_name, mime_type_filter="application/pdf" + ) + else: + # Files disabled by admin: discover nothing so no new file is + # indexed. The deletion-reconcile below then sees every indexed + # file as "missing" and purges it after the grace period — the + # backstop for the eager purge Astrolabe runs on disable. + logger.debug( + "[SCAN-%s] Files disabled by admin for %s; skipping tagged-file discovery", + scan_id, + user_id, + ) + tagged_files = [] # Apply EXCLUDED_TAGS as defense-in-depth: a folder marked # off-limits via the exclusion tag must not be indexed even if @@ -710,7 +730,9 @@ async def scan_user_documents( # Scan News items (starred + unread) news_queued = 0 - if _app_enabled("news", enabled_apps): + if _app_enabled("news", enabled_apps) and is_doc_type_allowed( + "news_item", allowed + ): try: news_queued = await scan_news_items( user_id=user_id, @@ -731,7 +753,9 @@ async def scan_user_documents( # Scan Deck cards deck_queued = 0 - if _app_enabled("deck", enabled_apps): + if _app_enabled("deck", enabled_apps) and is_doc_type_allowed( + "deck_card", allowed + ): try: deck_queued = await scan_deck_cards( user_id=user_id, diff --git a/tests/unit/test_capabilities.py b/tests/unit/test_capabilities.py new file mode 100644 index 00000000..9feb280c --- /dev/null +++ b/tests/unit/test_capabilities.py @@ -0,0 +1,144 @@ +"""Unit tests for the Astrolabe searchable-sources capability reader.""" + +from __future__ import annotations + +import nextcloud_mcp_server.capabilities as cap +from nextcloud_mcp_server.capabilities import ( + _parse_enabled_doc_types, + allowed_doc_types, + clear_cache, + is_doc_type_allowed, +) + + +def _payload(enabled_doc_types) -> dict: + """Build an OCS capabilities envelope carrying the astrolabe block. + + ``enabled_doc_types=...`` (Ellipsis) omits the key entirely. + """ + semantic: dict = {} + if enabled_doc_types is not ...: + semantic["enabled_doc_types"] = enabled_doc_types + return { + "ocs": { + "meta": {"status": "ok"}, + "data": {"capabilities": {"astrolabe": {"semantic_search": semantic}}}, + } + } + + +# --------------------------------------------------------------------------- +# _parse_enabled_doc_types +# --------------------------------------------------------------------------- + + +def test_parse_present_list_returns_set(): + assert _parse_enabled_doc_types(_payload(["note", "file"])) == {"note", "file"} + + +def test_parse_empty_list_returns_empty_set(): + # Admin disabled every source — distinct from "no restriction". + assert _parse_enabled_doc_types(_payload([])) == set() + + +def test_parse_missing_astrolabe_block_returns_none(): + payload = {"ocs": {"data": {"capabilities": {}}}} + assert _parse_enabled_doc_types(payload) is None + + +def test_parse_missing_enabled_key_returns_none(): + assert _parse_enabled_doc_types(_payload(...)) is None + + +def test_parse_malformed_payload_returns_none(): + assert _parse_enabled_doc_types(None) is None + assert _parse_enabled_doc_types({"ocs": "nope"}) is None + assert _parse_enabled_doc_types(_payload("not-a-list")) is None + + +def test_parse_drops_non_string_entries(): + assert _parse_enabled_doc_types(_payload(["note", 5, None])) == {"note"} + + +# --------------------------------------------------------------------------- +# is_doc_type_allowed +# --------------------------------------------------------------------------- + + +def test_is_doc_type_allowed_none_means_no_restriction(): + assert is_doc_type_allowed("anything", None) is True + + +def test_is_doc_type_allowed_respects_set(): + allowed = frozenset({"note"}) + assert is_doc_type_allowed("note", allowed) is True + assert is_doc_type_allowed("file", allowed) is False + + +def test_is_doc_type_allowed_empty_set_blocks_all(): + assert is_doc_type_allowed("note", frozenset()) is False + + +# --------------------------------------------------------------------------- +# allowed_doc_types (cache + fail-open) +# --------------------------------------------------------------------------- + + +class _FakeClient: + def __init__(self, payload=None, raises: Exception | None = None): + self._payload = payload + self._raises = raises + self.calls = 0 + + async def capabilities(self): + self.calls += 1 + if self._raises is not None: + raise self._raises + return self._payload + + +async def test_allowed_doc_types_parses_and_caches(): + clear_cache() + client = _FakeClient(_payload(["note", "file"])) + + first = await allowed_doc_types(client, "alice") + second = await allowed_doc_types(client, "alice") + + assert first == frozenset({"note", "file"}) + assert second == frozenset({"note", "file"}) + # Second call served from the cache — only one OCS round-trip. + assert client.calls == 1 + + +async def test_allowed_doc_types_missing_block_returns_none(): + clear_cache() + client = _FakeClient({"ocs": {"data": {"capabilities": {}}}}) + assert await allowed_doc_types(client, "bob") is None + + +async def test_allowed_doc_types_fail_open_not_cached(): + clear_cache() + client = _FakeClient(raises=RuntimeError("ocs down")) + + assert await allowed_doc_types(client, "carol") is None + # Failures are not cached — the next call retries the OCS lookup. + assert await allowed_doc_types(client, "carol") is None + assert client.calls == 2 + + +async def test_allowed_doc_types_cache_is_per_user(): + clear_cache() + alice = _FakeClient(_payload(["note"])) + bob = _FakeClient(_payload(["file"])) + + assert await allowed_doc_types(alice, "alice") == frozenset({"note"}) + assert await allowed_doc_types(bob, "bob") == frozenset({"file"}) + + +async def test_clear_cache_forces_refetch(): + clear_cache() + client = _FakeClient(_payload(["note"])) + await allowed_doc_types(client, "dave") + cap.clear_cache() + await allowed_doc_types(client, "dave") + assert client.calls == 2 diff --git a/tests/unit/test_vector_sync_purge_route.py b/tests/unit/test_vector_sync_purge_route.py new file mode 100644 index 00000000..8b929cf9 --- /dev/null +++ b/tests/unit/test_vector_sync_purge_route.py @@ -0,0 +1,138 @@ +"""Unit tests for the /api/v1/vector-sync/purge admin route. + +The purge is global and destructive (deletes every owner's content for a doc +type), so the route must: authenticate the bearer, restrict to Nextcloud +admins, validate the body, and only then delegate to the global purge. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest +from starlette.applications import Starlette +from starlette.routing import Route +from starlette.testclient import TestClient + +from nextcloud_mcp_server.api.vector_sync import purge_doc_types_route + +pytestmark = pytest.mark.unit + + +def _build_app() -> Starlette: + app = Starlette( + routes=[ + Route( + "/api/v1/vector-sync/purge", + purge_doc_types_route, + methods=["POST"], + ) + ] + ) + app.state.oauth_context = {"config": {"nextcloud_host": "http://nc.test"}} + return app + + +def _patch_token(mocker, user_id="admin"): + mocker.patch( + "nextcloud_mcp_server.api.vector_sync.validate_token_and_get_user", + new=AsyncMock(return_value=(user_id, {"sub": user_id})), + ) + + +def _patch_basic_auth(mocker, username="admin"): + mocker.patch( + "nextcloud_mcp_server.api.vector_sync.get_basic_auth_for_user", + new=AsyncMock(return_value=(username, "app-pwd")), + ) + + +def _patch_outbound_client(mocker): + client = AsyncMock() + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=False) + mocker.patch( + "nextcloud_mcp_server.api.vector_sync.nextcloud_httpx_client", + MagicMock(return_value=client), + ) + return client + + +def _patch_groups(mocker, groups): + instance = MagicMock() + instance.get_user_groups = AsyncMock(return_value=groups) + mocker.patch( + "nextcloud_mcp_server.api.vector_sync.UsersClient", + MagicMock(return_value=instance), + ) + + +def _patch_purge(mocker, result=None): + return mocker.patch( + "nextcloud_mcp_server.api.vector_sync.purge_doc_types", + new=AsyncMock(return_value=result or {}), + ) + + +async def test_unauthorized_when_token_invalid(mocker): + mocker.patch( + "nextcloud_mcp_server.api.vector_sync.validate_token_and_get_user", + new=AsyncMock(side_effect=ValueError("bad token")), + ) + purge = _patch_purge(mocker) + + client = TestClient(_build_app()) + resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]}) + + assert resp.status_code == 401 + purge.assert_not_called() + + +async def test_bad_request_when_doc_types_not_list(mocker): + _patch_token(mocker) + purge = _patch_purge(mocker) + + client = TestClient(_build_app()) + resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": "file"}) + + assert resp.status_code == 400 + purge.assert_not_called() + + +async def test_forbidden_when_not_admin(mocker): + _patch_token(mocker, "bob") + _patch_basic_auth(mocker, "bob") + _patch_outbound_client(mocker) + _patch_groups(mocker, ["users"]) # not an admin + purge = _patch_purge(mocker) + + client = TestClient(_build_app()) + resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]}) + + assert resp.status_code == 403 + purge.assert_not_called() + + +async def test_empty_doc_types_is_noop(mocker): + _patch_token(mocker) + purge = _patch_purge(mocker) + + client = TestClient(_build_app()) + resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": []}) + + assert resp.status_code == 200 + assert resp.json() == {"purged": {}} + purge.assert_not_called() + + +async def test_admin_purge_happy_path(mocker): + _patch_token(mocker, "admin") + _patch_basic_auth(mocker, "admin") + _patch_outbound_client(mocker) + _patch_groups(mocker, ["admin"]) + purge = _patch_purge(mocker, {"file": 12}) + + client = TestClient(_build_app()) + resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]}) + + assert resp.status_code == 200 + assert resp.json() == {"purged": {"file": 12}} + purge.assert_awaited_once_with(["file"]) diff --git a/tests/unit/vector/test_purge.py b/tests/unit/vector/test_purge.py new file mode 100644 index 00000000..4529ccdb --- /dev/null +++ b/tests/unit/vector/test_purge.py @@ -0,0 +1,83 @@ +"""Unit tests for global purge-by-doc-type (admin consent enforcement).""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +import nextcloud_mcp_server.vector.purge as purge_module +from nextcloud_mcp_server.vector.purge import purge_doc_types + + +def _patch_qdrant(monkeypatch, *, counts: dict[str, int], delete_raises=None): + """Wire a fake Qdrant client whose ``count`` reflects ``counts`` per + doc_type (read off the filter's MatchValue) and whose ``delete`` optionally + raises for given doc_types.""" + client = AsyncMock() + + def _doc_type_of(flt): + return flt.must[0].match.value + + async def fake_count(*, collection_name, count_filter, exact): + return SimpleNamespace(count=counts.get(_doc_type_of(count_filter), 0)) + + async def fake_delete(*, collection_name, points_selector): + dt = _doc_type_of(points_selector) + if delete_raises and dt in delete_raises: + raise RuntimeError(f"delete failed for {dt}") + + client.count.side_effect = fake_count + client.delete.side_effect = fake_delete + + async def fake_get_qdrant_client(): + return client + + monkeypatch.setattr(purge_module, "get_qdrant_client", fake_get_qdrant_client) + monkeypatch.setattr( + purge_module, + "get_settings", + lambda: SimpleNamespace(get_collection_name=lambda: "test_collection"), + ) + return client + + +async def test_purges_each_doc_type_and_reports_counts(monkeypatch): + client = _patch_qdrant(monkeypatch, counts={"file": 7, "note": 3}) + + result = await purge_doc_types(["file", "note"]) + + assert result == {"file": 7, "note": 3} + assert client.delete.await_count == 2 + + +async def test_dedupes_doc_types(monkeypatch): + client = _patch_qdrant(monkeypatch, counts={"file": 2}) + + result = await purge_doc_types(["file", "file"]) + + assert result == {"file": 2} + assert client.delete.await_count == 1 + + +async def test_zero_points_is_safe(monkeypatch): + _patch_qdrant(monkeypatch, counts={}) + assert await purge_doc_types(["deck_card"]) == {"deck_card": 0} + + +async def test_partial_failure_returns_partial(monkeypatch): + _patch_qdrant( + monkeypatch, + counts={"file": 5, "note": 4}, + delete_raises={"note"}, + ) + # "note" delete fails, "file" succeeds — partial progress is returned. + result = await purge_doc_types(["file", "note"]) + assert result == {"file": 5} + + +async def test_total_failure_raises(monkeypatch): + _patch_qdrant(monkeypatch, counts={"file": 5}, delete_raises={"file"}) + with pytest.raises(RuntimeError): + await purge_doc_types(["file"])