Chris CoutinhoandClaude Opus 4.8
b8a9400ee0
Root cause of the multi-user-basic/nc32 failure: the appstore installs DIFFERENT astrolabe versions per NC major (min-version jumped 31->32 at astrolabe 0.25.0). NC31 pulls astrolabe 0.24.0 (search box = NcTextField -> <input>, submits on Enter); NC32 pulls 0.29.0 (search box = NcTextArea -> <textarea>, submits on Ctrl/Cmd+Enter). The test's `.mcp-search-input input` selector + Enter never matched the textarea on nc32, so it timed out after the SPA mounted fine. This was latent all along but masked on nc32 by the vector-sync gauge flake, which failed the test earlier; fixing that flake unmasked it. Fix: match either `.mcp-search-input textarea, .mcp-search-input input` and submit based on the element tag (Ctrl+Enter for textarea, Enter for input). Verified against a live NC32 + astrolabe 0.29.0 stack: the textarea is found and Ctrl+Enter fires GET /apps/astrolabe/api/search. All other selectors the test uses (.mcp-loading/.mcp-error/.mcp-results/scatter3d) still exist in 0.29.0. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
432 lines
18 KiB
Python
432 lines
18 KiB
Python
"""Integration test for Astrolabe Plotly 3D visualization with multi-user BasicAuth mode.
|
|
|
|
Cross-system interface test: Tests the MCP server's integration with the
|
|
Astrolabe Nextcloud app, which is installed from the Nextcloud app store via
|
|
app-hooks/post-installation/20-install-astrolabe-app.sh. Astrolabe source
|
|
lives in a separate repository (https://github.com/cbcoutinho/astrolabe).
|
|
|
|
This test verifies that:
|
|
1. User can provision background sync access via app password
|
|
2. Content created via MCP tools is indexed by vector sync
|
|
3. Semantic search via Astrolabe UI returns results
|
|
4. Plotly 3D visualization container renders correctly
|
|
|
|
Requires:
|
|
- docker-compose up -d app db mcp-multi-user-basic
|
|
- ENABLE_SEMANTIC_SEARCH=true on the mcp-multi-user-basic container
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import re
|
|
import uuid
|
|
|
|
import anyio
|
|
import pytest
|
|
from playwright.async_api import Page
|
|
|
|
# Import helper functions from existing test
|
|
from tests.conftest import create_mcp_client_session
|
|
from tests.integration._search_helpers import document_is_searchable
|
|
from tests.integration.test_astrolabe_multi_user_background_sync import (
|
|
complete_astrolabe_authorization,
|
|
login_to_nextcloud,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.multi_user_basic]
|
|
|
|
|
|
async def wait_for_vector_sync(
|
|
mcp_client,
|
|
initial_indexed_count: int,
|
|
timeout_seconds: int = 60,
|
|
*,
|
|
search_term: str | None = None,
|
|
note_id: int | None = None,
|
|
) -> tuple[bool, dict | None]:
|
|
"""Wait for vector sync to index newly-created content.
|
|
|
|
Completion signal:
|
|
|
|
- When ``search_term`` is provided (preferred), poll ``nc_semantic_search``
|
|
until the new document is actually retrievable. This is robust against
|
|
full-corpus re-scan churn and doubles as a real end-to-end check — it is
|
|
exactly what callers assert downstream.
|
|
- Otherwise, fall back to the legacy gauge-delta predicate.
|
|
|
|
Why the gauge delta is unreliable: under ``VECTOR_SYNC_SCAN_INTERVAL`` the
|
|
background sync re-queues the whole corpus every scan, so the corpus-wide
|
|
``indexed_count`` is *non-monotonic* — it can be re-counted downward
|
|
mid-scan. ``indexed_count > initial_indexed_count`` can therefore never hold
|
|
even though the new document is indexed and the status has settled to
|
|
``idle`` / ``pending_count == 0``. That false failure was the dominant
|
|
multi-user-basic CI flake (``test_astrolabe_plotly_visualization`` /
|
|
``test_astrolabe_chunk_context``).
|
|
|
|
Args:
|
|
mcp_client: MCP client session
|
|
initial_indexed_count: Indexed document count before creating content
|
|
(only used by the legacy gauge-delta fallback)
|
|
timeout_seconds: Maximum time to wait for sync
|
|
search_term: Unique term contained in the new document; enables the
|
|
robust searchability-based completion signal
|
|
note_id: ID of the new document, used to match search results exactly
|
|
|
|
Returns:
|
|
Tuple of (success, status_data)
|
|
"""
|
|
wait_interval = 2
|
|
waited = 0
|
|
status_data = None
|
|
|
|
while waited < timeout_seconds:
|
|
sync_status = await mcp_client.call_tool("nc_get_vector_sync_status", {})
|
|
if sync_status.isError:
|
|
logger.warning("Vector sync status error: %s", sync_status)
|
|
return False, None
|
|
|
|
status_data = json.loads(sync_status.content[0].text)
|
|
indexed_count = status_data.get("indexed_count", 0)
|
|
pending_count = status_data.get("pending_count", 1)
|
|
|
|
logger.info(
|
|
"Sync status at %ss: indexed=%s, pending=%s, status=%s",
|
|
waited,
|
|
indexed_count,
|
|
pending_count,
|
|
status_data.get("status"),
|
|
)
|
|
|
|
if search_term is not None:
|
|
if await document_is_searchable(mcp_client, search_term, note_id):
|
|
logger.info(
|
|
"✓ Sync complete: document %s retrievable via semantic search",
|
|
note_id,
|
|
)
|
|
return True, status_data
|
|
elif indexed_count > initial_indexed_count and pending_count == 0:
|
|
logger.info(
|
|
"✓ Sync complete: %s documents indexed (was %s)",
|
|
indexed_count,
|
|
initial_indexed_count,
|
|
)
|
|
return True, status_data
|
|
|
|
await anyio.sleep(wait_interval)
|
|
waited += wait_interval
|
|
|
|
return False, status_data
|
|
|
|
|
|
async def navigate_to_astrolabe_main(page: Page):
|
|
"""Navigate to Astrolabe main app page (Semantic Search section).
|
|
|
|
Args:
|
|
page: Playwright page instance (must be authenticated)
|
|
"""
|
|
nextcloud_url = "http://localhost:8080"
|
|
|
|
logger.info("Navigating to Astrolabe main app...")
|
|
await page.goto(f"{nextcloud_url}/apps/astrolabe", wait_until="networkidle")
|
|
|
|
# Wait for the app to load
|
|
await anyio.sleep(1)
|
|
|
|
logger.info("✓ Successfully loaded Astrolabe main app")
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.multi_user_basic
|
|
@pytest.mark.timeout(
|
|
300
|
|
) # 5 minutes - this test involves app-password provisioning + vector sync
|
|
async def test_astrolabe_plotly_visualization_with_basic_auth(
|
|
browser,
|
|
test_users_setup,
|
|
configure_astrolabe_for_mcp_server,
|
|
):
|
|
"""Test Plotly 3D visualization in Astrolabe with multi-user BasicAuth mode.
|
|
|
|
This test:
|
|
1. Configures Astrolabe for the mcp-multi-user-basic service
|
|
2. Provisions background sync access for alice via app password
|
|
3. Creates a note with unique searchable content (as alice)
|
|
4. Waits for vector sync to index the note
|
|
5. Performs semantic search in Astrolabe UI
|
|
6. Verifies the Plotly visualization renders and results are displayed
|
|
"""
|
|
# Phase 1: Configure Astrolabe for mcp-multi-user-basic
|
|
await configure_astrolabe_for_mcp_server(
|
|
mcp_server_internal_url="http://mcp-multi-user-basic:8000",
|
|
mcp_server_public_url="http://localhost:8003",
|
|
)
|
|
|
|
username = "alice"
|
|
password = test_users_setup[username]["password"]
|
|
note_id = None
|
|
unique_term = None
|
|
|
|
# Create MCP client with alice's credentials for the multi-user BasicAuth server
|
|
credentials = base64.b64encode(f"{username}:{password}".encode()).decode("utf-8")
|
|
auth_header = f"Basic {credentials}"
|
|
|
|
context = await browser.new_context(ignore_https_errors=True)
|
|
page = await context.new_page()
|
|
|
|
try:
|
|
# Phase 2: Provision background indexing (app-password opt-in; no OAuth)
|
|
await login_to_nextcloud(page, username, password)
|
|
auth_result = await complete_astrolabe_authorization(page, username, password)
|
|
logger.info("Authorization result: %s", auth_result)
|
|
|
|
# Create MCP client session as alice - all MCP operations inside this block
|
|
async with create_mcp_client_session(
|
|
url="http://localhost:8003/mcp",
|
|
headers={"Authorization": auth_header},
|
|
client_name="Alice BasicAuth MCP",
|
|
) as alice_mcp_client:
|
|
# Phase 3: Get initial indexed count
|
|
initial_sync = await alice_mcp_client.call_tool(
|
|
"nc_get_vector_sync_status", {}
|
|
)
|
|
|
|
if initial_sync.isError:
|
|
pytest.skip("Vector sync not enabled on mcp-multi-user-basic")
|
|
|
|
initial_data = json.loads(initial_sync.content[0].text)
|
|
initial_count = initial_data.get("indexed_count", 0)
|
|
logger.info("Initial indexed count: %s", initial_count)
|
|
|
|
# Create note with unique searchable term
|
|
unique_term = f"plotly_viz_test_{uuid.uuid4().hex[:8]}"
|
|
note_response = await alice_mcp_client.call_tool(
|
|
"nc_notes_create_note",
|
|
{
|
|
"title": f"Visualization Test Note {unique_term}",
|
|
"content": f"""# Testing Plotly Visualization
|
|
|
|
This note contains the unique term: {unique_term}
|
|
|
|
It is used to test the 3D vector space visualization in the Astrolabe app.
|
|
The visualization should show this document as a point in PCA-reduced space.
|
|
|
|
## Key Features
|
|
- Semantic search with embeddings
|
|
- PCA dimension reduction to 3D
|
|
- Interactive Plotly scatter3d plot
|
|
""",
|
|
"category": "Test",
|
|
},
|
|
)
|
|
|
|
if note_response.isError:
|
|
pytest.fail(f"Failed to create test note: {note_response}")
|
|
|
|
note_data = json.loads(note_response.content[0].text)
|
|
note_id = note_data.get("id")
|
|
logger.info("Created test note ID: %s", note_id)
|
|
|
|
# Phase 4: Wait for vector indexing
|
|
sync_complete, status = await wait_for_vector_sync(
|
|
alice_mcp_client,
|
|
initial_count,
|
|
timeout_seconds=90,
|
|
search_term=unique_term,
|
|
note_id=note_id,
|
|
)
|
|
assert sync_complete, (
|
|
f"Note {note_id} ({unique_term}) never became searchable "
|
|
f"within timeout. Last sync status: {status}"
|
|
)
|
|
|
|
# Phase 5: Navigate to Astrolabe and perform search
|
|
await navigate_to_astrolabe_main(page)
|
|
|
|
# Find the Astrolabe search field. The published app differs across
|
|
# the NC matrix: NC31 pulls astrolabe <=0.24 (NcTextField -> <input>,
|
|
# submits on Enter); NC32 pulls astrolabe >=0.25 (NcTextArea ->
|
|
# <textarea>, submits on Ctrl/Cmd+Enter). Match either element so the
|
|
# test isn't pinned to one frontend revision.
|
|
# 30s (not 10s): the SPA can be slow to mount on a loaded CI runner.
|
|
search_input = page.locator(
|
|
".mcp-search-input textarea, .mcp-search-input input"
|
|
).first
|
|
await search_input.wait_for(timeout=30000, state="visible")
|
|
await search_input.fill(unique_term)
|
|
logger.info("Entered search query: %s", unique_term)
|
|
|
|
# Trigger search. NcTextField submits on Enter; NcTextArea inserts a
|
|
# newline on Enter and submits on Ctrl/Cmd+Enter — so key off the tag.
|
|
field_tag = await search_input.evaluate("el => el.tagName.toLowerCase()")
|
|
if field_tag == "textarea":
|
|
await search_input.press("Control+Enter")
|
|
else:
|
|
await search_input.press("Enter")
|
|
logger.info("Triggered search via %s submit", field_tag)
|
|
|
|
# Wait for loading to complete - watch for loading indicator to disappear
|
|
loading_indicator = page.locator(".mcp-loading")
|
|
try:
|
|
# If loading indicator appears, wait for it to disappear
|
|
if await loading_indicator.count() > 0:
|
|
await loading_indicator.wait_for(state="hidden", timeout=30000)
|
|
logger.info("Loading completed")
|
|
except Exception:
|
|
# Loading might be too fast to catch
|
|
pass
|
|
|
|
# Brief wait for UI to settle
|
|
await anyio.sleep(1)
|
|
|
|
# Take diagnostic screenshot
|
|
await page.screenshot(path="/tmp/astrolabe_search_after_click.png")
|
|
logger.info(
|
|
"Took diagnostic screenshot: /tmp/astrolabe_search_after_click.png"
|
|
)
|
|
|
|
# Wait for search results using text-based detection
|
|
# This is more reliable than class-based selectors
|
|
# The UI shows "N results" when search completes successfully
|
|
results_text_pattern = page.get_by_text(re.compile(r"\d+ results?"))
|
|
no_results_text = page.get_by_text("No results found")
|
|
error_note = page.locator(".mcp-error")
|
|
|
|
# Wait for one of: results count, no results message, or error
|
|
try:
|
|
# Poll for results or error states (don't rely on Nextcloud core CSS classes)
|
|
found_state = False
|
|
for attempt in range(60): # 60 attempts, 500ms each = 30s total
|
|
if await error_note.count() > 0:
|
|
error_text = await error_note.text_content()
|
|
logger.error("Search error: %s", error_text)
|
|
pytest.fail(f"Search failed with error: {error_text}")
|
|
|
|
if await no_results_text.count() > 0:
|
|
logger.warning(
|
|
"No results found - vector sync may not have completed"
|
|
)
|
|
await page.screenshot(path="/tmp/astrolabe_no_results.png")
|
|
pytest.fail(
|
|
f"Search returned no results for '{unique_term}'. "
|
|
"Check if vector sync completed for alice's content."
|
|
)
|
|
|
|
if await results_text_pattern.count() > 0:
|
|
results_text = await results_text_pattern.first.text_content()
|
|
logger.info("Found results: %s", results_text)
|
|
found_state = True
|
|
break
|
|
|
|
if attempt % 10 == 0:
|
|
logger.info(
|
|
"Waiting for results... (attempt %s/60)", attempt + 1
|
|
)
|
|
|
|
await anyio.sleep(0.5)
|
|
|
|
if not found_state:
|
|
await page.screenshot(path="/tmp/astrolabe_search_timeout.png")
|
|
page_content = await page.content()
|
|
logger.error("Search state not resolved. Page URL: %s", page.url)
|
|
logger.error("Page content snippet: %s", page_content[:2000])
|
|
raise AssertionError("Search did not complete within timeout")
|
|
|
|
except AssertionError:
|
|
raise # Re-raise AssertionError as-is
|
|
except Exception as e:
|
|
# Take another screenshot and get page content for debugging
|
|
await page.screenshot(path="/tmp/astrolabe_search_timeout.png")
|
|
page_content = await page.content()
|
|
logger.error("Search state not resolved. Page URL: %s", page.url)
|
|
logger.error("Page content snippet: %s", page_content[:2000])
|
|
raise AssertionError(f"Search did not complete: {e}")
|
|
|
|
logger.info("Results loaded")
|
|
|
|
# Phase 6: Verify visualization
|
|
# Check Plotly container is visible
|
|
viz_plot = page.locator("#viz-plot")
|
|
await viz_plot.wait_for(timeout=15000, state="visible")
|
|
logger.info("Plotly container is visible")
|
|
|
|
# Verify Plotly has rendered content (SVG/canvas elements inside)
|
|
has_viz_content = await page.evaluate(
|
|
"""
|
|
() => {
|
|
const plot = document.getElementById('viz-plot');
|
|
if (!plot) return false;
|
|
// Plotly creates .plotly class, canvas, or svg elements
|
|
return plot.children.length > 0 ||
|
|
plot.querySelector('.plotly, canvas, svg, .main-svg') !== null;
|
|
}
|
|
"""
|
|
)
|
|
assert has_viz_content, "Plotly visualization did not render any content"
|
|
logger.info("✓ Plotly visualization rendered content")
|
|
|
|
# Verify results are displayed
|
|
result_items = page.locator(".mcp-result-item")
|
|
result_count = await result_items.count()
|
|
assert result_count > 0, "No search results displayed"
|
|
logger.info("✓ Found %s search result(s)", result_count)
|
|
|
|
# Verify our note appears in results
|
|
found_note = False
|
|
for i in range(result_count):
|
|
item = result_items.nth(i)
|
|
title_elem = item.locator(".mcp-result-title")
|
|
title_text = await title_elem.text_content()
|
|
if title_text and unique_term in title_text:
|
|
found_note = True
|
|
logger.info("✓ Found test note in results: %s", title_text)
|
|
break
|
|
|
|
assert found_note, f"Created note with '{unique_term}' not found in results"
|
|
|
|
# Optional: Take screenshot for verification
|
|
await page.screenshot(path="/tmp/astrolabe_plotly_test_success.png")
|
|
logger.info("✓ All Plotly visualization assertions passed")
|
|
|
|
# Cleanup: delete the created note (inside the MCP client context)
|
|
if note_id:
|
|
try:
|
|
delete_response = await alice_mcp_client.call_tool(
|
|
"nc_notes_delete_note", {"note_id": note_id}
|
|
)
|
|
if not delete_response.isError:
|
|
logger.info("✓ Cleaned up test note %s", note_id)
|
|
note_id = None # Mark as cleaned
|
|
else:
|
|
logger.warning(
|
|
"Failed to delete note %s: %s", note_id, delete_response
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Cleanup failed for note %s: %s", note_id, e)
|
|
|
|
finally:
|
|
# Cleanup note if not already cleaned (create new client for cleanup)
|
|
if note_id:
|
|
try:
|
|
async with create_mcp_client_session(
|
|
url="http://localhost:8003/mcp",
|
|
headers={"Authorization": auth_header},
|
|
client_name="Cleanup MCP",
|
|
) as cleanup_client:
|
|
delete_response = await cleanup_client.call_tool(
|
|
"nc_notes_delete_note", {"note_id": note_id}
|
|
)
|
|
if not delete_response.isError:
|
|
logger.info("✓ Cleaned up test note %s (finally)", note_id)
|
|
else:
|
|
logger.warning(
|
|
"Failed to delete note %s: %s", note_id, delete_response
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Cleanup failed for note %s: %s", note_id, e)
|
|
|
|
# Close browser context
|
|
await context.close()
|