From b11d1b17a30eaf4f7bd12e3583e53dad9429e588 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 7 Jun 2026 20:04:00 +0200 Subject: [PATCH] test(vector): address PR #873 round-1 review - Extract `_app_enabled` to a module-level helper so the gate predicate is unit-tested directly instead of via an inline copy that could drift. - Move `import logging` to module scope in test_scanner_app_gating.py. - Harden `get_enabled_apps` OCS-envelope parsing (`X or {}` / `or []`) so a present-but-null `ocs`/`data` coerces to empty instead of raising on `None.get`; add parametrized malformed-envelope tests. - Use https:// in the test request URL (SonarCloud S5332 hotspot). Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/client/__init__.py | 6 +++- nextcloud_mcp_server/vector/scanner.py | 19 +++++++---- tests/unit/client/test_nextcloud_client.py | 14 ++++++++ tests/unit/vector/test_scanner_app_gating.py | 35 +++++++++++--------- 4 files changed, 51 insertions(+), 23 deletions(-) diff --git a/nextcloud_mcp_server/client/__init__.py b/nextcloud_mcp_server/client/__init__.py index db92b853..41f4c974 100644 --- a/nextcloud_mcp_server/client/__init__.py +++ b/nextcloud_mcp_server/client/__init__.py @@ -214,7 +214,11 @@ class NextcloudClient: ) response.raise_for_status() data = response.json() - entries = data.get("ocs", {}).get("data", []) or [] + # ``X or {}``/``or []`` (not ``.get(k, default)``) so a present-but-null + # ``ocs``/``data`` (``{"ocs": null}``) coerces to empty instead of + # raising AttributeError on ``None.get``. + ocs = data.get("ocs") or {} + entries = ocs.get("data") or [] enabled: set[str] = set() for entry in entries: # ``app`` is the canonical app id; ``id`` matches it for the apps we diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index 8fc7c707..b1085aa0 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -269,6 +269,16 @@ async def _get_enabled_apps_or_none( return None +def _app_enabled(app_id: str, enabled_apps: set[str] | None) -> bool: + """Whether ``app_id`` should be scanned for the current user. + + ``enabled_apps is None`` means detection failed — every app is treated as + enabled (the scan-all fallback) so a transient navigation-endpoint failure + never silently halts indexing. + """ + return enabled_apps is None or app_id in enabled_apps + + async def scan_user_documents( user_id: str, send_stream: TaskProducer, @@ -355,9 +365,6 @@ 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) - def _app_enabled(app_id: str) -> bool: - return enabled_apps is None or app_id in enabled_apps - # 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). @@ -366,7 +373,7 @@ async def scan_user_documents( current_time = time.time() queued = 0 - if _app_enabled("notes"): + if _app_enabled("notes", enabled_apps): try: queued += await scan_notes( user_id=user_id, @@ -703,7 +710,7 @@ async def scan_user_documents( # Scan News items (starred + unread) news_queued = 0 - if _app_enabled("news"): + if _app_enabled("news", enabled_apps): try: news_queued = await scan_news_items( user_id=user_id, @@ -724,7 +731,7 @@ async def scan_user_documents( # Scan Deck cards deck_queued = 0 - if _app_enabled("deck"): + if _app_enabled("deck", enabled_apps): try: deck_queued = await scan_deck_cards( user_id=user_id, diff --git a/tests/unit/client/test_nextcloud_client.py b/tests/unit/client/test_nextcloud_client.py index 14ffae61..58f9eaf4 100644 --- a/tests/unit/client/test_nextcloud_client.py +++ b/tests/unit/client/test_nextcloud_client.py @@ -94,6 +94,20 @@ class TestGetEnabledApps: assert await client.get_enabled_apps() == {"notes"} + @pytest.mark.parametrize("body", [{}, {"ocs": None}, {"ocs": {"data": None}}]) + async def test_malformed_envelope_returns_empty_set(self, body): + """A missing/null ``ocs``/``data`` envelope yields an empty set rather + than raising — the scanner then gates every app off, and its own + fallback (``_get_enabled_apps_or_none``) keeps indexing safe.""" + client = _make_client() + response = MagicMock() + response.raise_for_status = MagicMock() + response.json.return_value = body + client._client = AsyncMock() + client._client.get = AsyncMock(return_value=response) + + assert await client.get_enabled_apps() == set() + class TestNormaliseSearchResult: def test_adds_leading_slash_to_path(self): diff --git a/tests/unit/vector/test_scanner_app_gating.py b/tests/unit/vector/test_scanner_app_gating.py index dbaf08c3..ddca0e11 100644 --- a/tests/unit/vector/test_scanner_app_gating.py +++ b/tests/unit/vector/test_scanner_app_gating.py @@ -1,17 +1,22 @@ -"""Unit tests for the vector scanner's enabled-app gating helper. +"""Unit tests for the vector scanner's enabled-app gating helpers. ``scan_user_documents`` skips polling apps the user doesn't have enabled (those polls 404 and flood tenant logs). ``_get_enabled_apps_or_none`` resolves the enabled-app set, returning ``None`` on any failure so the caller falls back to -scanning every app (the prior behaviour) rather than silently halting indexing. +scanning every app (the prior behaviour) rather than silently halting indexing; +``_app_enabled`` is the gate predicate applied per app. """ +import logging from unittest.mock import AsyncMock import pytest from httpx import HTTPStatusError, Request, Response -from nextcloud_mcp_server.vector.scanner import _get_enabled_apps_or_none +from nextcloud_mcp_server.vector.scanner import ( + _app_enabled, + _get_enabled_apps_or_none, +) pytestmark = pytest.mark.unit @@ -27,33 +32,31 @@ async def test_returns_enabled_set_on_success(): async def test_returns_none_when_detection_raises(caplog): nc_client = AsyncMock() - request = Request("GET", "http://nc.test/ocs/v2.php/core/navigation/apps") + request = Request("GET", "https://nc.test/ocs/v2.php/core/navigation/apps") nc_client.get_enabled_apps = AsyncMock( side_effect=HTTPStatusError( "boom", request=request, response=Response(503, request=request) ) ) - import logging - caplog.set_level(logging.WARNING, logger="nextcloud_mcp_server.vector.scanner") result = await _get_enabled_apps_or_none(nc_client, "alice", scan_id=1234) - # None signals scan-all fallback; the inline gate treats `None` as + # None signals scan-all fallback; _app_enabled treats `None` as # "every app enabled" so indexing never silently stops. assert result is None assert "scanning all apps" in caplog.text def test_none_set_enables_every_app(): - """The gate predicate used in scan_user_documents: a None set means - detection failed, so every app must be scanned (back-compat).""" + """A None set means detection failed, so every app must be scanned.""" + assert _app_enabled("news", None) is True + assert _app_enabled("deck", None) is True - def app_enabled(app_id: str, enabled: set[str] | None) -> bool: - return enabled is None or app_id in enabled - assert app_enabled("news", None) is True - assert app_enabled("deck", None) is True - # And a concrete set gates precisely. - assert app_enabled("news", {"notes"}) is False - assert app_enabled("notes", {"notes"}) is True +def test_concrete_set_gates_precisely(): + """A resolved set scans only the apps it contains.""" + enabled = {"notes", "files"} + assert _app_enabled("notes", enabled) is True + assert _app_enabled("news", enabled) is False + assert _app_enabled("deck", enabled) is False