Chris CoutinhoandClaude Opus 4.7
8a0d06107e
The /api/v1/chunk-context and /api/v1/pdf-preview handlers in api/visualization.py forwarded the incoming OAuth bearer directly to Nextcloud via NextcloudClient.from_token. In multi-user BasicAuth mode Nextcloud has no validator for those bearers on Notes/WebDAV, so it treats the request as anonymous and returns 401 — surfaced to the user as a 500 from /apps/astrolabe/api/chunk-context. Search worked because it only hits Qdrant. Architecturally, OAuth is only for Astrolabe→MCP server; MCP server→ Nextcloud always uses the per-user app password stored during provision (background sync already does this via vector.oauth_sync). - Resolve the Nextcloud client through get_user_client_basic_auth in both get_chunk_context and get_pdf_preview, surfacing NotProvisionedError as a clean 401 instead of opaque 500. - Apply the same fix to the session-cookie variant in auth/viz_routes.chunk_context_endpoint for the internal viz UI. Tests: - New unit file test_management_chunk_context_endpoint.py, including a regression guard that asserts get_user_client_basic_auth is awaited (so reverting to from_token fails without needing a live Nextcloud). - Updated test_management_pdf_preview_endpoint.py to mock the new auth path (drops extract_bearer_token / NextcloudClient.from_token patches). - New integration test test_astrolabe_chunk_context.py drives the full chain (browser → Astrolabe → MCP → Nextcloud) in multi-user BasicAuth mode, plus bare-bones 401 checks on the MCP endpoint. Full unit suite: 546 passed. Companion PR on astrolabe (cbcoutinho/astrolabe#66) sends the Nextcloud UID as loginName in the app-password POST body so the stored record is complete. Submodule bump to that branch will follow once CI reproduces the failure on the old submodule. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
283 lines
11 KiB
Python
283 lines
11 KiB
Python
"""
|
|
Unit tests for Management API chunk-context endpoint.
|
|
|
|
Tests the /api/v1/chunk-context endpoint focusing on:
|
|
- Parameter validation (doc_type, doc_id, start, end, context)
|
|
- OAuth token validation
|
|
- Nextcloud credential path (must use get_user_client_basic_auth, not
|
|
NextcloudClient.from_token — see regression in api/visualization.py)
|
|
- Error handling (missing params, invalid ranges, missing credentials)
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from starlette.applications import Starlette
|
|
from starlette.routing import Route
|
|
from starlette.testclient import TestClient
|
|
|
|
from nextcloud_mcp_server.api.visualization import get_chunk_context
|
|
from nextcloud_mcp_server.vector.oauth_sync import NotProvisionedError
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
def create_test_app():
|
|
"""Create a test Starlette app with the chunk-context endpoint."""
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/api/v1/chunk-context", get_chunk_context, methods=["GET"]),
|
|
]
|
|
)
|
|
app.state.oauth_context = {"config": {"nextcloud_host": "http://localhost:8080"}}
|
|
return app
|
|
|
|
|
|
def _make_mock_chunk_context(chunk_text="chunk", before="before", after="after"):
|
|
"""Mock the ChunkContext dataclass with enough fields for the handler."""
|
|
ctx = MagicMock()
|
|
ctx.chunk_text = chunk_text
|
|
ctx.before_context = before
|
|
ctx.after_context = after
|
|
ctx.has_before_truncation = False
|
|
ctx.has_after_truncation = False
|
|
ctx.page_number = None
|
|
ctx.chunk_index = 0
|
|
ctx.total_chunks = 1
|
|
return ctx
|
|
|
|
|
|
def _make_mock_nc_client():
|
|
"""Mock NextcloudClient that supports `async with`."""
|
|
mock_client = MagicMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
return mock_client
|
|
|
|
|
|
class TestChunkContextParameterValidation:
|
|
"""Tests for parameter validation in the chunk-context endpoint."""
|
|
|
|
def test_missing_params_returns_400(self):
|
|
with patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
# Missing end
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=1&start=0",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "required parameters" in data["error"].lower()
|
|
|
|
def test_end_less_than_or_equal_start_returns_400(self):
|
|
with patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=1&start=100&end=100",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "end must be greater than start" in data["error"].lower()
|
|
|
|
def test_non_numeric_start_returns_400(self):
|
|
with patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=1&start=abc&end=10",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
class TestChunkContextTokenValidation:
|
|
"""Tests for OAuth token validation."""
|
|
|
|
def test_missing_token_returns_401(self):
|
|
with patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
side_effect=ValueError("Missing Authorization header"),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=1&start=0&end=10"
|
|
)
|
|
assert response.status_code == 401
|
|
data = response.json()
|
|
assert data["error"] == "Unauthorized"
|
|
|
|
def test_invalid_token_returns_401(self):
|
|
with patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
side_effect=Exception("Token expired"),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=1&start=0&end=10",
|
|
headers={"Authorization": "Bearer invalid-token"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestChunkContextCredentialPath:
|
|
"""Regression tests: the handler MUST use get_user_client_basic_auth.
|
|
|
|
The earlier bug (reproduced in homelab 2026-04-22) forwarded the OAuth
|
|
bearer directly to Nextcloud via NextcloudClient.from_token. That works in
|
|
single-user env-BasicAuth mode but fails with 401 in multi-user BasicAuth
|
|
mode because Nextcloud won't validate that bearer on the Notes API.
|
|
"""
|
|
|
|
def test_successful_fetch_uses_app_password_client(self):
|
|
"""Handler must build its NC client via get_user_client_basic_auth and
|
|
return the chunk text from get_chunk_with_context."""
|
|
mock_nc_client = _make_mock_nc_client()
|
|
mock_ctx = _make_mock_chunk_context(
|
|
chunk_text="hello world",
|
|
before="before the chunk",
|
|
after="after the chunk",
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.get_user_client_basic_auth",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_nc_client,
|
|
) as mock_basic_auth,
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.get_chunk_with_context",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_ctx,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=42&start=0&end=11",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["chunk_text"] == "hello world"
|
|
assert data["before_context"] == "before the chunk"
|
|
assert data["after_context"] == "after the chunk"
|
|
|
|
# Regression guard: credential resolution must go through the
|
|
# app-password helper, not NextcloudClient.from_token.
|
|
mock_basic_auth.assert_awaited_once()
|
|
args, kwargs = mock_basic_auth.call_args
|
|
positional = list(args)
|
|
if "user_id" in kwargs:
|
|
positional.insert(0, kwargs["user_id"])
|
|
assert positional[0] == "testuser"
|
|
|
|
def test_not_provisioned_returns_401(self):
|
|
"""If the user has no stored app password, the handler must surface
|
|
NotProvisionedError as a clean 401 (not 500)."""
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.get_user_client_basic_auth",
|
|
new_callable=AsyncMock,
|
|
side_effect=NotProvisionedError(
|
|
"User testuser has not provisioned an app password."
|
|
),
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=1&start=0&end=10",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 401
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "app password" in data["error"].lower()
|
|
|
|
def test_chunk_fetch_returns_none_yields_404(self):
|
|
"""When get_chunk_with_context returns None (doc missing / offsets
|
|
out of range), the handler reports 404 with a structured error body."""
|
|
mock_nc_client = _make_mock_nc_client()
|
|
|
|
with (
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.get_user_client_basic_auth",
|
|
new_callable=AsyncMock,
|
|
return_value=mock_nc_client,
|
|
),
|
|
patch(
|
|
"nextcloud_mcp_server.api.visualization.get_chunk_with_context",
|
|
new_callable=AsyncMock,
|
|
return_value=None,
|
|
),
|
|
):
|
|
app = create_test_app()
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=999&start=0&end=10",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert data["success"] is False
|
|
assert "failed to fetch chunk context" in data["error"].lower()
|
|
|
|
|
|
class TestChunkContextConfigErrors:
|
|
"""Tests for configuration failure paths."""
|
|
|
|
def test_missing_nextcloud_host_config(self):
|
|
with patch(
|
|
"nextcloud_mcp_server.api.visualization.validate_token_and_get_user",
|
|
new_callable=AsyncMock,
|
|
return_value=("testuser", True),
|
|
):
|
|
app = create_test_app()
|
|
app.state.oauth_context = {"config": {"nextcloud_host": ""}}
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/v1/chunk-context?doc_type=note&doc_id=1&start=0&end=10",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
)
|
|
# Handler wraps ValueError via _sanitize_error_for_client → 500
|
|
assert response.status_code == 500
|