diff --git a/nextcloud_mcp_server/api/vector_sync.py b/nextcloud_mcp_server/api/vector_sync.py index 7668e2e6..58a1144e 100644 --- a/nextcloud_mcp_server/api/vector_sync.py +++ b/nextcloud_mcp_server/api/vector_sync.py @@ -65,6 +65,12 @@ async def purge_doc_types_route(request: Request) -> JSONResponse: status_code=400, ) + if not isinstance(body, dict): + return JSONResponse( + {"error": "Bad request", "message": "body must be a JSON object"}, + 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( @@ -115,7 +121,7 @@ async def purge_doc_types_route(request: Request) -> JSONResponse: status_code=428, ) except Exception as e: - logger.error("Error purging doc types for user %s: %s", user_id, e) + logger.exception("Error purging doc types for user %s", user_id) return JSONResponse( { "error": "Internal error", diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index f5b2fa87..34e05754 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -2454,7 +2454,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = "/api/v1/users/{user_id}/app-password, /api/v1/users/{user_id}/access, " "/api/v1/users/{user_id}/scopes, /api/v1/scopes, " "/api/v1/vector-viz/search, /api/v1/search, /api/v1/apps, " - "/api/v1/webhooks, /api/v1/pdf-preview" + "/api/v1/webhooks, /api/v1/vector-sync/purge, /api/v1/pdf-preview" ) # Note: Metrics endpoint is NOT exposed on main HTTP port for security reasons. diff --git a/nextcloud_mcp_server/capabilities.py b/nextcloud_mcp_server/capabilities.py index 90cc1850..cf181bc8 100644 --- a/nextcloud_mcp_server/capabilities.py +++ b/nextcloud_mcp_server/capabilities.py @@ -38,13 +38,13 @@ class _CapabilitiesClientProtocol(Protocol): async def capabilities(self) -> Any: ... -def _parse_enabled_doc_types(payload: Any) -> set[str] | None: +def _parse_enabled_doc_types(payload: Any) -> frozenset[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. + malformed (treated as "no restriction"). Returns a frozenset (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 @@ -63,7 +63,7 @@ def _parse_enabled_doc_types(payload: Any) -> set[str] | None: raw = semantic.get("enabled_doc_types") if not isinstance(raw, list): return None - return {dt for dt in raw if isinstance(dt, str)} + return frozenset(dt for dt in raw if isinstance(dt, str) and dt) async def allowed_doc_types( @@ -92,8 +92,7 @@ async def allowed_doc_types( ) 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 + result = _parse_enabled_doc_types(payload) _cache[user_id] = (now, result) _cache.move_to_end(user_id) while len(_cache) > _CACHE_MAXSIZE: diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 5d93b7d6..126e0d0e 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -590,6 +590,9 @@ async def process_document( doc_task.doc_id, doc_task.user_id, ) + # Alertable counter so a flood of webhook events for a + # disabled source is observable (not silently swallowed). + record_ingest_dropped("admin_disabled") record_vector_sync_processing(time.time() - start_time, "skipped") return diff --git a/nextcloud_mcp_server/vector/purge.py b/nextcloud_mcp_server/vector/purge.py index 88d8baca..7e363d3a 100644 --- a/nextcloud_mcp_server/vector/purge.py +++ b/nextcloud_mcp_server/vector/purge.py @@ -35,6 +35,11 @@ async def purge_doc_types(doc_types: list[str]) -> dict[str, int]: 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. + + The count is taken just before the delete (two separate Qdrant calls), so + it is approximate — a point indexed in the gap is deleted but not counted. + This is acceptable: indexing of a disabled source is already gated upstream, + so the window is effectively empty in practice. """ qdrant_client = await get_qdrant_client() collection = get_settings().get_collection_name() @@ -61,10 +66,9 @@ async def purge_doc_types(doc_types: list[str]) -> dict[str, int]: ) except Exception as exc: # noqa: BLE001 — record and continue last_error = exc - logger.error( - "Failed to purge indexed points for doc_type=%s: %s", + logger.exception( + "Failed to purge indexed points for doc_type=%s", doc_type, - exc, ) if not purged and last_error is not None: diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index da802c48..537ab94e 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -280,6 +280,75 @@ def _app_enabled(app_id: str, enabled_apps: set[str] | None) -> bool: return enabled_apps is None or app_id in enabled_apps +# Text doc types whose deletion-tracking lives *inside* their scan_* function, +# so skipping that function (when admin-disabled) leaves indexed points with no +# grace-period backstop. ``file`` is intentionally excluded: its scan path +# empties discovery and lets the existing reconcile loop purge on disable. +_TEXT_BACKSTOP_DOC_TYPES: tuple[str, ...] = ("note", "news_item", "deck_card") + + +async def _enqueue_deletes_for_disabled_types( + user_id: str, + send_stream: TaskProducer, + allowed: frozenset[str] | None, + scan_id: int, +) -> int: + """Enqueue delete tasks for indexed text-source points the admin disabled. + + Backstop for a failed eager purge: scrolls this user's indexed points for + each admin-disallowed text doc_type and queues a delete. No-op when + ``allowed`` is ``None`` (fail-open — never delete on a transient capability + read failure). Returns the number of delete tasks enqueued. + """ + if allowed is None: + return 0 + + disabled = [dt for dt in _TEXT_BACKSTOP_DOC_TYPES if dt not in allowed] + if not disabled: + return 0 + + qdrant_client = await get_qdrant_client() + collection = get_settings().get_collection_name() + queued = 0 + for doc_type in disabled: + points = await _scroll_all_points( + qdrant_client, + collection_name=collection, + scroll_filter=Filter( + must=[ + FieldCondition(key="user_id", match=MatchValue(value=user_id)), + FieldCondition(key="doc_type", match=MatchValue(value=doc_type)), + ] + ), + payload_fields=["doc_id"], + ) + doc_ids = { + str(p.payload["doc_id"]) + for p in points + if p.payload is not None and "doc_id" in p.payload + } + if doc_ids: + logger.info( + "[SCAN-%s] %s disabled by admin for %s; enqueueing %d delete(s) (backstop)", + scan_id, + doc_type, + user_id, + len(doc_ids), + ) + for doc_id in doc_ids: + await send_stream.send( + DocumentTask( + user_id=user_id, + doc_id=doc_id, + doc_type=doc_type, + operation="delete", + modified_at=0, + ) + ) + queued += 1 + return queued + + async def scan_user_documents( user_id: str, send_stream: TaskProducer, @@ -381,6 +450,16 @@ async def scan_user_documents( current_time = time.time() queued = 0 + # Backstop purge for admin-disabled text sources. Their deletion- + # tracking lives inside the scan_* function we skip below, so (unlike + # files, whose discovery-empties-then-reconcile path purges on disable) + # they'd linger if Astrolabe's eager purge failed. Enqueue deletes for + # any indexed points of a now-disallowed type. Gated on a concrete + # allow-set, so a fail-open None never triggers deletion. + queued += await _enqueue_deletes_for_disabled_types( + user_id, send_stream, allowed, scan_id + ) + if _app_enabled("notes", enabled_apps) and is_doc_type_allowed("note", allowed): try: queued += await scan_notes( diff --git a/tests/unit/test_capabilities.py b/tests/unit/test_capabilities.py index 9feb280c..162d3374 100644 --- a/tests/unit/test_capabilities.py +++ b/tests/unit/test_capabilities.py @@ -2,6 +2,8 @@ from __future__ import annotations +from unittest.mock import AsyncMock + import nextcloud_mcp_server.capabilities as cap from nextcloud_mcp_server.capabilities import ( _parse_enabled_doc_types, @@ -84,22 +86,19 @@ def test_is_doc_type_allowed_empty_set_blocks_all(): # --------------------------------------------------------------------------- -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 +def _client(payload=None, raises: Exception | None = None) -> AsyncMock: + """An object with an async ``capabilities()`` method (AsyncMock-backed).""" + m = AsyncMock() + if raises is not None: + m.capabilities.side_effect = raises + else: + m.capabilities.return_value = payload + return m async def test_allowed_doc_types_parses_and_caches(): clear_cache() - client = _FakeClient(_payload(["note", "file"])) + client = _client(_payload(["note", "file"])) first = await allowed_doc_types(client, "alice") second = await allowed_doc_types(client, "alice") @@ -107,38 +106,53 @@ async def test_allowed_doc_types_parses_and_caches(): 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 + assert client.capabilities.await_count == 1 async def test_allowed_doc_types_missing_block_returns_none(): clear_cache() - client = _FakeClient({"ocs": {"data": {"capabilities": {}}}}) + client = _client({"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")) + client = _client(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 + assert client.capabilities.await_count == 2 async def test_allowed_doc_types_cache_is_per_user(): clear_cache() - alice = _FakeClient(_payload(["note"])) - bob = _FakeClient(_payload(["file"])) + alice = _client(_payload(["note"])) + bob = _client(_payload(["file"])) assert await allowed_doc_types(alice, "alice") == frozenset({"note"}) assert await allowed_doc_types(bob, "bob") == frozenset({"file"}) +async def test_allowed_doc_types_refetches_after_ttl(monkeypatch): + clear_cache() + client = _client(_payload(["note"])) + + # Drive the module clock so the second call lands past the TTL window. + clock = {"now": 1000.0} + monkeypatch.setattr(cap.time, "monotonic", lambda: clock["now"]) + + await allowed_doc_types(client, "erin") + clock["now"] += cap._CACHE_TTL_SECONDS + 1 + await allowed_doc_types(client, "erin") + + assert client.capabilities.await_count == 2 + + async def test_clear_cache_forces_refetch(): clear_cache() - client = _FakeClient(_payload(["note"])) + client = _client(_payload(["note"])) await allowed_doc_types(client, "dave") cap.clear_cache() await allowed_doc_types(client, "dave") - assert client.calls == 2 + assert client.capabilities.await_count == 2 diff --git a/tests/unit/test_vector_sync_purge_route.py b/tests/unit/test_vector_sync_purge_route.py index 8b929cf9..df4e0bb4 100644 --- a/tests/unit/test_vector_sync_purge_route.py +++ b/tests/unit/test_vector_sync_purge_route.py @@ -13,6 +13,7 @@ from starlette.routing import Route from starlette.testclient import TestClient from nextcloud_mcp_server.api.vector_sync import purge_doc_types_route +from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError pytestmark = pytest.mark.unit @@ -72,7 +73,7 @@ def _patch_purge(mocker, result=None): ) -async def test_unauthorized_when_token_invalid(mocker): +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")), @@ -86,7 +87,7 @@ async def test_unauthorized_when_token_invalid(mocker): purge.assert_not_called() -async def test_bad_request_when_doc_types_not_list(mocker): +def test_bad_request_when_doc_types_not_list(mocker): _patch_token(mocker) purge = _patch_purge(mocker) @@ -97,7 +98,7 @@ async def test_bad_request_when_doc_types_not_list(mocker): purge.assert_not_called() -async def test_forbidden_when_not_admin(mocker): +def test_forbidden_when_not_admin(mocker): _patch_token(mocker, "bob") _patch_basic_auth(mocker, "bob") _patch_outbound_client(mocker) @@ -111,7 +112,7 @@ async def test_forbidden_when_not_admin(mocker): purge.assert_not_called() -async def test_empty_doc_types_is_noop(mocker): +def test_empty_doc_types_is_noop(mocker): _patch_token(mocker) purge = _patch_purge(mocker) @@ -123,7 +124,7 @@ async def test_empty_doc_types_is_noop(mocker): purge.assert_not_called() -async def test_admin_purge_happy_path(mocker): +def test_admin_purge_happy_path(mocker): _patch_token(mocker, "admin") _patch_basic_auth(mocker, "admin") _patch_outbound_client(mocker) @@ -136,3 +137,30 @@ async def test_admin_purge_happy_path(mocker): assert resp.status_code == 200 assert resp.json() == {"purged": {"file": 12}} purge.assert_awaited_once_with(["file"]) + + +def test_bad_request_when_body_not_object(mocker): + # A valid JSON non-object (e.g. a list) must 400, not 500. + _patch_token(mocker) + purge = _patch_purge(mocker) + + client = TestClient(_build_app()) + resp = client.post("/api/v1/vector-sync/purge", json=[1, 2, 3]) + + assert resp.status_code == 400 + purge.assert_not_called() + + +def test_provisioning_required_returns_428(mocker): + _patch_token(mocker, "admin") + mocker.patch( + "nextcloud_mcp_server.api.vector_sync.get_basic_auth_for_user", + new=AsyncMock(side_effect=ProvisioningRequiredError("not provisioned")), + ) + purge = _patch_purge(mocker) + + client = TestClient(_build_app()) + resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]}) + + assert resp.status_code == 428 + purge.assert_not_called() diff --git a/tests/unit/vector/test_purge.py b/tests/unit/vector/test_purge.py index 4529ccdb..5aa3e8dd 100644 --- a/tests/unit/vector/test_purge.py +++ b/tests/unit/vector/test_purge.py @@ -20,10 +20,12 @@ def _patch_qdrant(monkeypatch, *, counts: dict[str, int], delete_raises=None): def _doc_type_of(flt): return flt.must[0].match.value - async def fake_count(*, collection_name, count_filter, exact): + # Sync side_effects: AsyncMock awaits the call and returns the value, so the + # helpers don't need to be coroutines themselves. + 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): + 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}") @@ -31,10 +33,9 @@ def _patch_qdrant(monkeypatch, *, counts: dict[str, int], delete_raises=None): 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_qdrant_client", AsyncMock(return_value=client) + ) monkeypatch.setattr( purge_module, "get_settings", diff --git a/tests/unit/vector/test_scanner_consent_backstop.py b/tests/unit/vector/test_scanner_consent_backstop.py new file mode 100644 index 00000000..2a71be30 --- /dev/null +++ b/tests/unit/vector/test_scanner_consent_backstop.py @@ -0,0 +1,81 @@ +"""Unit tests for the scanner's admin-consent backstop deletion. + +When an admin disables a text source (note/news_item/deck_card), the scanner +skips its scan_* function, so the in-function deletion-tracking never runs. The +backstop enqueues deletes for any indexed points of the disabled type, mirroring +the files path — but only on a concrete allow-set (never on fail-open None). +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import cast +from unittest.mock import AsyncMock + +from nextcloud_mcp_server.vector import scanner as scanner_module +from nextcloud_mcp_server.vector.queue.ports import TaskProducer +from nextcloud_mcp_server.vector.scanner import _enqueue_deletes_for_disabled_types + + +def _producer(send: AsyncMock) -> TaskProducer: + """A minimal stand-in for the TaskProducer protocol (only ``send`` is used).""" + return cast(TaskProducer, SimpleNamespace(send=send)) + + +def _patch_qdrant(monkeypatch, points_by_type: dict[str, list[str]]): + client = AsyncMock() + + def fake_scroll( + *, collection_name, scroll_filter, with_payload, with_vectors, limit, offset + ): + # must=[user_id, doc_type] — doc_type is the second condition. + doc_type = scroll_filter.must[1].match.value + points = [ + SimpleNamespace(payload={"doc_id": doc_id}) + for doc_id in points_by_type.get(doc_type, []) + ] + return (points, None) + + client.scroll.side_effect = fake_scroll + monkeypatch.setattr( + scanner_module, "get_qdrant_client", AsyncMock(return_value=client) + ) + monkeypatch.setattr( + scanner_module, + "get_settings", + lambda: SimpleNamespace(get_collection_name=lambda: "c"), + ) + + +async def test_enqueues_deletes_for_disabled_text_type(monkeypatch): + _patch_qdrant(monkeypatch, {"note": ["n1", "n2"], "deck_card": ["d1"]}) + sent: list = [] + stream = _producer(AsyncMock(side_effect=lambda t: sent.append(t))) + + # note disabled; news_item + deck_card still allowed. + allowed = frozenset({"file", "news_item", "deck_card"}) + queued = await _enqueue_deletes_for_disabled_types("alice", stream, allowed, 1) + + assert queued == 2 + assert {t.doc_id for t in sent} == {"n1", "n2"} + assert all(t.operation == "delete" and t.doc_type == "note" for t in sent) + + +async def test_noop_when_allowed_is_none(monkeypatch): + # Fail-open: a transient capability read must never trigger deletion. + send = AsyncMock() + queued = await _enqueue_deletes_for_disabled_types( + "alice", _producer(send), None, 1 + ) + assert queued == 0 + send.assert_not_called() + + +async def test_noop_when_all_text_types_allowed(monkeypatch): + send = AsyncMock() + allowed = frozenset({"note", "news_item", "deck_card", "file"}) + queued = await _enqueue_deletes_for_disabled_types( + "alice", _producer(send), allowed, 1 + ) + assert queued == 0 + send.assert_not_called()