admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(astrolabe): migrate suite to session-JWT auth model

Browse Source 
Astrolabe was refactored to mint session-derived JWTs (TokenGenerationRequest
Event) and a one-click background-indexing opt-in, dropping the OAuth
authorize/callback/refresh surface. Bump the submodule and bring the test
suite in line:

- New test_astrolabe_session_jwt_search.py: a logged-in user searches via the
  minted JWT with no provisioning (replaces the obsolete login_flow_provisioning
  OAuth-authorize test; token_refresh test deleted — refresh flow is gone).
- settings_buttons: assert the new revoke endpoint + that oauth/disconnect is
  gone (404).
- multi_user_background_sync / plotly / chunk_context: drop the OAuth authorize
  step; provision via the one-click "Enable background indexing" button
  (#mcp-enable-background-button -> #mcp-revoke-background-button) instead of
  generating + pasting an app password.
- docker-compose.yml: mount the astrolabe submodule into the app container.
- third_party/astrolabe: bump to the one-click opt-in commit.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-05-28 23:06:27 +02:00
co-authored by Claude Opus 4.8
parent ae54956f27
commit 4ed228613e
7 changed files with 229 additions and 776 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker-compose.yml
+1 -1
View File
@@ -36,7 +36,7 @@ services:
# Mount OIDC development directory outside /var/www/html to avoid rsync conflicts
# The post-installation hook will register /opt/apps as an additional app directory
#- ./third_party:/opt/apps:ro
#- ./third_party/astrolabe:/opt/apps/astrolabe:ro
- ./third_party/astrolabe:/opt/apps/astrolabe:ro
#- ./third_party/oidc:/opt/apps/oidc:ro
environment:
- NEXTCLOUD_TRUSTED_DOMAINS=app
tests/integration/test_astrolabe_chunk_context.py
-1
View File
@@ -133,7 +133,6 @@ async def test_chunk_context_endpoint_uses_app_password(
try:
await login_to_nextcloud(page, username, password)
auth_result = await complete_astrolabe_authorization(page, username, password)
assert auth_result["step1"], "OAuth authorization did not complete"
assert auth_result["step2"], "App password provisioning did not complete"
auth_header = _build_basic_auth_header(username, password)
tests/integration/test_astrolabe_multi_user_background_sync.py
+70 -692
View File
@@ -7,18 +7,17 @@ lives in a separate repository (https://github.com/cbcoutinho/astrolabe).
This test verifies that multiple users can independently:
1. Log in to Nextcloud
2. Generate an app password in Security settings
3. Enter the app password in Astrolabe personal settings
4. Enable background sync for the mcp-multi-user-basic service
5. Verify app password is stored in the database
2. Click the one-click "Enable background indexing" opt-in in Astrolabe settings
3. Have a dedicated app password minted from their session and handed to the
MCP server (core/getapppassword — no Security-settings step, no copy-paste)
4. Verify the app password is stored in the database
Tests the complete app password provisioning flow:
user login → Security settings → app password generation → Astrolabe settings →
app password entry → background sync activation → database verification.
Tests the one-click background-indexing provisioning flow:
user login → Astrolabe settings → Enable background indexing → session app
password minted + forwarded to MCP → background sync active → DB verification.
"""
import logging
import re
import subprocess
import anyio
@@ -91,621 +90,62 @@ async def navigate_to_astrolabe_settings(page: Page):
logger.info("✓ Successfully loaded Astrolabe settings page")
async def authorize_search_access(page: Page, username: str) -> bool:
"""Complete Step 1: OAuth Authorization for Astrolabe.
async def enable_background_sync(page: Page, username: str) -> bool:
"""Provision background indexing via the one-click opt-in button.
Handles the OAuth flow:
1. Check if already authorized (Step 1 shows "Complete")
2. Click "Authorize" link
3. Handle Nextcloud OIDC consent screen
4. Wait for redirect back to Astrolabe settings
5. Verify "Complete" badge appears on Step 1
The refactored settings page mints a dedicated app password from the
current Nextcloud session (core/getapppassword) and hands it to the MCP
server — there is no app-password generation in Security settings and no
copy-paste. Idempotent: if already enabled (the revoke form is shown
instead of the enable button), returns True without acting.
Args:
page: Playwright page instance (must be on Astrolabe settings page)
username: Username for logging
Returns:
True if authorization completed successfully
"""
nextcloud_url = "http://localhost:8080"
logger.info("Authorizing search access (Step 1) for %s...", username)
# Check if already on Astrolabe settings page, if not navigate there
if "/settings/user/astrolabe" not in page.url:
await navigate_to_astrolabe_settings(page)
# Wait for page to fully render
await anyio.sleep(1)
# Check if already authorized (either "Active" badge or Step 1 "Complete" badge)
try:
# Check for "Active" badge (fully configured state)
active_badge = page.get_by_text("Active", exact=True)
if await active_badge.count() > 0 and await active_badge.is_visible():
logger.info("✓ Already fully authorized for %s (Active badge)", username)
return True
except Exception:
pass
try:
step1_section = page.locator('h4:has-text("Step 1")')
if await step1_section.count() > 0:
# Look for "Complete" text in the Step 1 section's parent
step1_parent = step1_section.locator("..")
complete_badge = step1_parent.get_by_text("Complete", exact=True)
if await complete_badge.count() > 0 and await complete_badge.is_visible():
logger.info("✓ Step 1 already complete for %s", username)
return True
except Exception:
pass
# Find and click the "Authorize" button
authorize_button = page.locator('a.button.primary:has-text("Authorize")')
try:
await authorize_button.wait_for(timeout=5000, state="visible")
logger.info("Found Authorize button for %s", username)
except Exception:
# Take screenshot for debugging
screenshot_path = f"/tmp/astrolabe_no_authorize_button_{username}.png"
await page.screenshot(path=screenshot_path)
logger.error(
"Could not find Authorize button for %s. Screenshot: %s",
username,
screenshot_path,
)
raise ValueError(f"Authorize button not found for {username}")
# Click the Authorize button - this will redirect to OAuth provider
# Use force=True to bypass stability check which can timeout due to CSS transitions
await authorize_button.click(force=True)
logger.info("Clicked Authorize button for %s", username)
# Wait for OAuth redirect to complete
await page.wait_for_load_state("networkidle", timeout=30000)
logger.info("After networkidle, current URL: %s", page.url)
# Take screenshot to see current state
await page.screenshot(path=f"/tmp/astrolabe_after_authorize_{username}.png")
logger.info("Screenshot saved: /tmp/astrolabe_after_authorize_%s.png", username)
# Handle OIDC consent screen if present
consent_handled = await _handle_oauth_consent_screen(page, username)
if consent_handled:
logger.info("✓ OAuth consent granted for %s", username)
else:
logger.info(
"No consent screen required for %s (may be previously authorized)", username
)
# Wait for redirect back to Astrolabe settings
# The OAuth callback will redirect back to /settings/user/astrolabe
try:
await page.wait_for_url(
f"**{nextcloud_url}/settings/user/astrolabe**", timeout=30000
)
logger.info("Redirected back to Astrolabe settings for %s", username)
except Exception:
# Check if we're already on settings page
if "/settings/user/astrolabe" not in page.url:
logger.warning(
"Not redirected to Astrolabe settings, current URL: %s", page.url
)
# Navigate manually
await page.goto(
f"{nextcloud_url}/settings/user/astrolabe", wait_until="networkidle"
)
# Wait for page to reload and render
await anyio.sleep(2)
# Verify authorization completed - check for various success indicators
# When fully configured, shows "Active" badge; when only Step 1 done, shows "Complete"
try:
# First check if "Active" badge is shown (fully configured state)
active_badge = page.get_by_text("Active", exact=True)
if await active_badge.count() > 0 and await active_badge.is_visible():
logger.info(
"✓ OAuth authorization complete for %s (Active badge)", username
)
return True
except Exception:
pass
try:
# Check for Step 1 "Complete" badge (partial configuration)
step1_section = page.locator('h4:has-text("Step 1")')
if await step1_section.count() > 0:
step1_parent = step1_section.locator("..")
complete_badge = step1_parent.get_by_text("Complete", exact=True)
await complete_badge.wait_for(timeout=5000, state="visible")
logger.info("✓ Step 1 OAuth authorization complete for %s", username)
return True
except Exception:
pass
# Neither badge found - authorization failed
screenshot_path = f"/tmp/astrolabe_step1_not_complete_{username}.png"
await page.screenshot(path=screenshot_path)
logger.error(
"Authorization badge not visible for %s. Screenshot: %s",
username,
screenshot_path,
)
raise ValueError(f"OAuth authorization did not complete for {username}")
async def _handle_oauth_consent_screen(page: Page, username: str) -> bool:
"""Handle the OIDC consent screen during OAuth flow.
Reuses the proven pattern from tests/conftest.py.
Args:
page: Playwright page instance
username: Username for logging
Returns:
True if consent was handled, False if no consent screen was found
"""
try:
logger.info("Checking for consent screen at URL: %s", page.url)
# Check if consent screen is present - try multiple selectors
# The consent screen may be #oidc-consent or use a different format
consent_div = await page.query_selector("#oidc-consent")
if consent_div:
logger.info("Consent screen detected via #oidc-consent for %s", username)
# Get consent screen data attributes for logging
client_name = await consent_div.get_attribute("data-client-name")
scopes_attr = await consent_div.get_attribute("data-scopes")
logger.info(" Client: %s", client_name)
logger.info(" Requested scopes: %s", scopes_attr)
else:
# Check for Allow button directly (different consent screen format)
allow_button = page.locator('button:has-text("Allow")')
if await allow_button.count() > 0:
logger.info("Consent screen detected via Allow button for %s", username)
else:
logger.info("No consent screen found for %s at %s", username, page.url)
await page.screenshot(path=f"/tmp/no_consent_screen_{username}.png")
logger.info("Screenshot: /tmp/no_consent_screen_%s.png", username)
return False
# Wait for Vue.js to render the Allow button
try:
await page.wait_for_selector('button:has-text("Allow")', timeout=10000)
logger.info(" Allow button rendered by Vue.js")
except Exception as e:
screenshot_path = f"/tmp/consent_no_allow_button_{username}.png"
await page.screenshot(path=screenshot_path)
logger.error(" Timeout waiting for Allow button: %s", e)
raise
# Check all scope checkboxes
scope_checkboxes = await page.query_selector_all('input[type="checkbox"]')
if scope_checkboxes:
logger.info(" Found %s scope checkboxes", len(scope_checkboxes))
for i, checkbox in enumerate(scope_checkboxes):
is_checked = await checkbox.is_checked()
is_disabled = await checkbox.is_disabled()
if not is_checked and not is_disabled:
await checkbox.check()
logger.info(" ✓ Checked scope checkbox %s", i + 1)
# Click the Allow button using JavaScript (handles viewport issues)
allow_button_locator = page.locator('button:has-text("Allow")')
# Debug: take screenshot before clicking Allow
await page.screenshot(path=f"/tmp/consent_before_allow_{username}.png")
logger.info(
" Screenshot before Allow: /tmp/consent_before_allow_%s.png", username
)
button_count = await allow_button_locator.count()
logger.info(" Found %s Allow button(s)", button_count)
if button_count > 0:
current_url = page.url
logger.info(" Current URL: %s", current_url)
logger.info(" Clicking Allow button for %s...", username)
# Use JavaScript click to handle consent buttons (proven pattern from conftest.py)
# This is more reliable than Playwright's click for Vue.js rendered buttons
await page.evaluate(
"""
const buttons = document.querySelectorAll('button');
for (const btn of buttons) {
if (btn.textContent.trim() === 'Allow') {
btn.click();
break;
}
}
"""
)
# Wait for URL to change (Vue.js uses window.location.href after fetch)
# networkidle doesn't detect fetch-based redirects
try:
await page.wait_for_url(
lambda url: url != current_url,
timeout=30000,
)
logger.info(" URL changed to: %s", page.url)
except Exception as wait_error:
# If URL didn't change, check console for errors
logger.warning(" URL didn't change after click: %s", wait_error)
await page.screenshot(path=f"/tmp/consent_after_allow_{username}.png")
# Try alternative: manually POST consent and navigate
logger.info(" Trying manual consent submission...")
try:
redirect_url = await page.evaluate(
"""
async () => {
const selectedScopes = Array.from(document.querySelectorAll('input[type="checkbox"]:checked'))
.map(cb => cb.value).join(' ');
const response = await fetch('/index.php/apps/oidc/consent/grant', {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
'requesttoken': OC.requestToken,
},
body: 'scopes=' + encodeURIComponent(selectedScopes),
redirect: 'follow',
});
return response.url || '/index.php/apps/oidc/authorize';
}
"""
)
logger.info(" Manual consent returned URL: %s", redirect_url)
await page.goto(redirect_url, wait_until="networkidle")
except Exception as manual_error:
logger.error(" Manual consent also failed: %s", manual_error)
raise
await page.screenshot(path=f"/tmp/consent_after_allow_{username}.png")
logger.info(" Consent granted for %s", username)
return True
else:
logger.error(" Allow button not found for %s", username)
return False
except Exception as e:
logger.error("Error handling consent screen for %s: %s", username, e)
raise
async def generate_app_password(
page: Page, username: str, app_name: str = "Astrolabe Background Sync"
) -> str:
"""Generate an app password in Nextcloud Security settings.
Args:
page: Playwright page instance (must be authenticated)
page: Playwright page instance (must be logged in)
username: Username (for logging)
app_name: Name for the app password
Returns:
The generated app password string
True once background indexing is enabled.
"""
logger.info("Generating app password for %s...", username)
nextcloud_url = "http://localhost:8080"
# Navigate to Security settings
await page.goto(f"{nextcloud_url}/settings/user/security", wait_until="networkidle")
logger.info("Navigated to Security settings")
# Fill the app password input field (selector confirmed via Playwright MCP)
app_password_input = page.locator('input[placeholder="App name"]')
await app_password_input.fill(app_name)
logger.info("Entered app name: %s", app_name)
# Wait for Vue.js to react and enable the button (needs 1 second, not 0.5)
await anyio.sleep(1.0)
logger.info("Waited for Vue.js to process input and enable button")
# Click the create button - use force=True to bypass stability check (CSS transitions)
create_button = page.locator(
'button[type="submit"]:has-text("Create new app password")'
)
try:
await create_button.click(force=True, timeout=10000)
except Exception:
# Fallback: JavaScript click
logger.info("Using JavaScript click for create button...")
await page.evaluate(
"""
const btn = document.querySelector('button[type="submit"]');
if (btn) btn.click();
"""
)
logger.info("Clicked create app password button")
# Wait for app password to be generated and displayed in the dialog
await anyio.sleep(3) # Give it more time to generate and display
# Debug screenshot after clicking create
await page.screenshot(path=f"/tmp/app_password_after_create_{username}.png")
logger.info(
"Screenshot after create: /tmp/app_password_after_create_%s.png", username
)
# Find the Login input field which should have the username value
# Then find the Password input field which is in the same form
app_password = None
try:
# Wait for heading "New app password" to appear
await page.wait_for_selector('text="New app password"', timeout=10000)
logger.info("App password dialog appeared with heading")
# Get all visible input elements
all_inputs = await page.locator('input[type="text"]').all()
logger.info("Found %s text input elements", len(all_inputs))
# Check each input to find the one with the app password
for idx, input_elem in enumerate(all_inputs):
try:
value = await input_elem.input_value()
if value and "-" in value and len(value) > 20:
app_password = value.strip()
logger.info(
"Found app password in input %s: '%s' (length: %s)",
idx,
app_password,
len(app_password),
)
break
except Exception as e:
logger.debug("Could not get value from input %s: %s", idx, e)
continue
except Exception as e:
logger.error("Failed to find app password dialog or extract password: %s", e)
if not app_password:
# Take screenshot for debugging
screenshot_path = f"/tmp/app_password_generation_{username}.png"
await page.screenshot(path=screenshot_path)
raise ValueError(
f"Could not find generated app password. Screenshot: {screenshot_path}"
)
# Validate password format before returning
if not re.match(
r"^[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}$",
app_password,
):
logger.error(
"Extracted password does not match expected format: '%s'", app_password
)
logger.error("Password repr: %s", repr(app_password))
screenshot_path = f"/tmp/app_password_invalid_format_{username}.png"
await page.screenshot(path=screenshot_path)
raise ValueError(
f"App password format validation failed. Screenshot: {screenshot_path}"
)
logger.info(
"✓ Generated app password for %s: %s... (validated)",
username,
app_password[:10],
)
# Close dialog with Escape key (bypasses CSS layout issues with h2 intercepting clicks)
logger.info("Closing app password dialog with Escape key...")
await page.keyboard.press("Escape")
await anyio.sleep(0.5) # Wait for dialog close animation
logger.info("Closed app password dialog")
return app_password
async def enable_background_sync_via_app_password(
page: Page, username: str, app_password: str
):
"""Enable background sync by entering app password in Astrolabe settings.
Args:
page: Playwright page instance
username: Username (for logging)
app_password: App password to enter
Returns:
True if background sync was enabled successfully
"""
logger.info("Enabling background sync via app password for %s...", username)
nextcloud_url = "http://localhost:8080"
# Set up network request and console listeners BEFORE navigation
network_requests = []
network_responses = []
console_messages = []
def log_request(req):
network_requests.append(f"{req.method} {req.url}")
def log_response(resp):
response_info = f"{resp.status} {resp.url}"
network_responses.append(response_info)
logger.info("Response: %s", response_info)
def log_console(msg):
console_messages.append(f"[{msg.type}] {msg.text}")
page.on("request", log_request)
page.on("response", log_response)
page.on("console", log_console)
# Navigate to Astrolabe settings
logger.info("Enabling background indexing for %s...", username)
await page.goto(
f"{nextcloud_url}/settings/user/astrolabe", wait_until="networkidle"
"http://localhost:8080/settings/user/astrolabe", wait_until="networkidle"
)
# Wait for page to load
await anyio.sleep(1)
# Check if already complete (look for Step 2 "Complete" badge or overall "Active" state)
try:
# First check for overall "Active" badge (both steps complete)
active_text = page.get_by_text("Active", exact=True)
if await active_text.is_visible(timeout=2000):
logger.info("✓ Background sync already active for %s", username)
return True
except Exception:
pass
if await page.locator("#mcp-revoke-background-button").count() > 0:
logger.info("✓ Background indexing already enabled for %s", username)
return True
try:
# Check for Step 2 "Complete" badge (app password already set)
step2_section = page.locator('h4:has-text("Step 2")')
if await step2_section.count() > 0:
step2_parent = step2_section.locator("..")
complete_badge = step2_parent.get_by_text("Complete", exact=True)
if await complete_badge.count() > 0 and await complete_badge.is_visible():
logger.info("✓ Step 2 (app password) already complete for %s", username)
return True
except Exception:
pass
# Find the app password input field using the placeholder text
# Based on manual testing: textbox with placeholder "xxxxx-xxxxx-xxxxx-xxxxx-xxxxx"
app_password_input = page.get_by_placeholder("xxxxx-xxxxx-xxxxx-xxxxx-xxxxx")
enable_button = page.locator("#mcp-enable-background-button")
await enable_button.wait_for(timeout=5000, state="visible")
await enable_button.click()
logger.info("Clicked 'Enable background indexing' for %s", username)
# On success the page JS reloads to the enabled state (revoke form shown).
try:
await app_password_input.wait_for(timeout=5000, state="visible")
logger.info("Found app password input field")
await page.locator("#mcp-revoke-background-button").wait_for(
timeout=15000, state="visible"
)
logger.info("✓ Background indexing enabled for %s", username)
return True
except Exception:
# Take screenshot for debugging
screenshot_path = f"/tmp/astrolabe_no_password_field_{username}.png"
screenshot_path = f"/tmp/astrolabe_enable_failed_{username}.png"
await page.screenshot(path=screenshot_path)
raise ValueError(
f"Could not find app password input field for {username}. Screenshot: {screenshot_path}"
f"Background indexing did not enable for {username}. "
f"Screenshot: {screenshot_path}"
)
# Enter the app password
await app_password_input.fill(app_password)
logger.info("Entered app password for %s", username)
# Wait a moment for any validation to complete
await anyio.sleep(0.5)
# Take screenshot before clicking Save to check for warnings
screenshot_path = f"/tmp/before_save_{username}.png"
await page.screenshot(path=screenshot_path)
logger.info("Screenshot taken before Save: %s", screenshot_path)
# Find and click the Save button
save_button = page.get_by_role("button", name="Save")
# Check if Save button is enabled
is_disabled = await save_button.is_disabled()
logger.info("Save button disabled state: %s", is_disabled)
await save_button.click()
logger.info("Clicked Save button")
# Give the request time to complete before checking logs
await anyio.sleep(0.5)
# Log network requests after clicking Save
logger.info("Network requests after Save for %s:", username)
for req in network_requests[-10:]: # Last 10 requests
logger.info(" %s", req)
# Log network responses after clicking Save
logger.info("Network responses after Save for %s:", username)
for resp in network_responses[-10:]: # Last 10 responses
logger.info(" %s", resp)
# Check specifically for the credentials POST response
credentials_responses = [
r for r in network_responses if "background-sync/credentials" in r
]
if credentials_responses:
logger.info("Credentials endpoint response: %s", credentials_responses[-1])
if "200" not in credentials_responses[-1]:
logger.error(
"Credentials POST did not return 200 OK: %s", credentials_responses[-1]
)
else:
logger.warning("No response found for credentials endpoint!")
# Wait for the page to reload after successful save
# The JavaScript in personalSettings.js does: setTimeout(() => window.location.reload(), 1000)
await page.wait_for_load_state("networkidle", timeout=15000)
await anyio.sleep(2)
# Log any console messages
if console_messages:
logger.info("Console messages for %s:", username)
for msg in console_messages:
logger.info(" %s", msg)
# Check for error notifications (toast messages)
try:
error_toast = page.locator(".toastify.toast-error, .toast-error")
if await error_toast.count() > 0:
error_text = await error_toast.first.text_content()
logger.error("Error notification for %s: %s", username, error_text)
except Exception:
pass
# Verify Step 2 "Complete" badge or overall "Active" badge appears after reload
try:
# First try to find "Active" badge (both steps complete)
active_text = page.get_by_text("Active", exact=True)
if await active_text.count() > 0:
await active_text.wait_for(timeout=5000, state="visible")
logger.info(
"✓ Background sync enabled for %s - Active badge visible", username
)
return True
except Exception:
pass
try:
# Check for Step 2 "Complete" badge
step2_section = page.locator('h4:has-text("Step 2")')
if await step2_section.count() > 0:
step2_parent = step2_section.locator("..")
complete_badge = step2_parent.get_by_text("Complete", exact=True)
await complete_badge.wait_for(timeout=5000, state="visible")
logger.info(
"✓ Step 2 (app password) enabled for %s - Complete badge visible",
username,
)
return True
except Exception:
pass
# If neither badge found, raise error
screenshot_path = f"/tmp/astrolabe_after_password_{username}.png"
await page.screenshot(path=screenshot_path)
logger.error(
"Neither Active nor Complete badge appeared for %s. Screenshot: %s",
username,
screenshot_path,
)
raise ValueError(f"Background sync setup did not complete for {username}")
async def complete_astrolabe_authorization(
page: Page, username: str, password: str
) -> dict:
"""Complete full Astrolabe two-step authorization.
"""Provision background indexing for a user (one-click app-password opt-in).
Performs the complete authorization flow:
1. Navigate to Astrolabe settings
2. OAuth authorization (Step 1) if needed
3. Generate app password in Security settings
4. App password entry (Step 2) if needed
The auth refactor dropped the per-user OAuth step entirely — search now
uses a session-minted JWT, so the only remaining "authorization" is the
one-click background-indexing opt-in (a dedicated app password minted from
the session and handed to the MCP server).
Args:
page: Playwright page instance (must be logged in)
@@ -713,64 +153,15 @@ async def complete_astrolabe_authorization(
password: Nextcloud password (for reference, not used directly)
Returns:
Dict with {"step1": bool, "step2": bool, "app_password": str | None}
Dict with {"step1": True (no-op, kept for caller compat),
"step2": bool, "app_password": None}
"""
logger.info("Starting full Astrolabe authorization for %s...", username)
logger.info("Provisioning Astrolabe background indexing for %s...", username)
result = {"step1": False, "step2": False, "app_password": None}
# Navigate to Astrolabe settings
await navigate_to_astrolabe_settings(page)
# Step 1: OAuth authorization
try:
result["step1"] = await authorize_search_access(page, username)
logger.info("✓ Step 1 complete for %s", username)
except Exception as e:
logger.error("Step 1 failed for %s: %s", username, e)
raise
# Navigate back to settings if needed (OAuth might have redirected elsewhere)
if "/settings/user/astrolabe" not in page.url:
await navigate_to_astrolabe_settings(page)
# Check if Step 2 is already complete
try:
step2_section = page.locator('h4:has-text("Step 2")')
if await step2_section.count() > 0:
step2_parent = step2_section.locator("..")
complete_badge = step2_parent.get_by_text("Complete", exact=True)
if await complete_badge.count() > 0 and await complete_badge.is_visible():
logger.info("✓ Step 2 already complete for %s", username)
result["step2"] = True
return result
except Exception:
pass
# Also check for overall "Active" badge
try:
active_text = page.get_by_text("Active", exact=True)
if await active_text.count() > 0 and await active_text.is_visible():
logger.info("✓ Authorization already fully active for %s", username)
result["step2"] = True
return result
except Exception:
pass
# Step 2: Generate app password and enter it
app_password = await generate_app_password(page, username)
result["app_password"] = app_password
try:
result["step2"] = await enable_background_sync_via_app_password(
page, username, app_password
)
logger.info("✓ Step 2 complete for %s", username)
except Exception as e:
logger.error("Step 2 failed for %s: %s", username, e)
raise
logger.info("✓ Full Astrolabe authorization complete for %s", username)
# step1 is retained as always-True for backward compat with callers — there
# is no longer an OAuth authorize step to perform.
result = {"step1": True, "step2": False, "app_password": None}
result["step2"] = await enable_background_sync(page, username)
return result
@@ -993,15 +384,11 @@ async def test_multi_user_astrolabe_background_sync_enablement(
# Step 1: Login to Nextcloud
await login_to_nextcloud(page, username, password)
# Step 2: Generate app password in Security settings
app_password = await generate_app_password(page, username)
# Step 2: One-click "Enable background indexing" (mints a dedicated
# app password from the session and hands it to the MCP server).
sync_enabled = await enable_background_sync(page, username)
# Step 3: Enable background sync by entering app password in Astrolabe
sync_enabled = await enable_background_sync_via_app_password(
page, username, app_password
)
# Step 4: Verify app password was stored in database
# Step 3: Verify app password was stored in database
app_password_stored = await verify_app_password_created(username)
# Give it time to complete
@@ -1009,7 +396,6 @@ async def test_multi_user_astrolabe_background_sync_enablement(
results[username] = {
"settings_accessed": True,
"app_password_generated": bool(app_password),
"sync_enabled": sync_enabled,
"app_password_stored": app_password_stored,
"background_sync_active": sync_enabled and app_password_stored,
@@ -1017,7 +403,6 @@ async def test_multi_user_astrolabe_background_sync_enablement(
logger.info("\\n%s results:", username)
logger.info(" Settings accessed: ✓")
logger.info(" App password generated: %s", "✓" if app_password else "✗")
logger.info(" Sync enabled: %s", "✓" if sync_enabled else "✗")
logger.info(
" App password stored: %s", "✓" if app_password_stored else "✗"
@@ -1081,7 +466,7 @@ async def test_multi_user_astrolabe_background_sync_enablement(
async def revoke_background_sync_access(page: Page, username: str) -> bool:
"""Revoke background sync access by clicking the Revoke Access button.
"""Revoke background sync access by clicking the "Disable background indexing" button.
Args:
page: Playwright page instance (must be authenticated)
@@ -1122,37 +507,35 @@ async def revoke_background_sync_access(page: Page, username: str) -> bool:
# Wait for page to load
await anyio.sleep(1)
# Check if "Active" badge is visible (indicating background sync is enabled)
# The revoke form (#mcp-revoke-background-form) is only rendered while
# background indexing is enabled.
revoke_button = page.locator("#mcp-revoke-background-button")
try:
active_text = page.get_by_text("Active", exact=True)
if not await active_text.is_visible(timeout=2000):
if await revoke_button.count() == 0:
logger.warning(
"Background sync not active for %s, nothing to revoke", username
"Background indexing not enabled for %s, nothing to revoke", username
)
return False
except Exception:
logger.warning("Could not find Active badge for %s", username)
logger.warning("Could not find revoke button for %s", username)
return False
# Find the "Revoke Access" button
revoke_button = page.get_by_role("button", name="Revoke Access")
try:
await revoke_button.wait_for(timeout=5000, state="visible")
logger.info("Found Revoke Access button")
logger.info("Found 'Disable background indexing' button")
except Exception:
screenshot_path = f"/tmp/astrolabe_no_revoke_button_{username}.png"
await page.screenshot(path=screenshot_path)
raise ValueError(
f"Could not find Revoke Access button for {username}. Screenshot: {screenshot_path}"
f"Could not find revoke button for {username}. Screenshot: {screenshot_path}"
)
# Set up dialog handler for confirmation dialog
page.once("dialog", lambda dialog: dialog.accept())
# Click the Revoke Access button
# Click the "Disable background indexing" button
await revoke_button.click()
logger.info("Clicked Revoke Access button")
logger.info("Clicked the revoke button")
# Wait for the request to complete and page to reload
await page.wait_for_load_state("networkidle", timeout=15000)
@@ -1198,12 +581,12 @@ async def revoke_background_sync_access(page: Page, username: str) -> bool:
except Exception:
pass
# Verify "Active" badge is no longer visible
# After revoke + reload the settings page returns to the un-provisioned
# state: the revoke button is gone and the app-password input is shown again.
try:
active_text = page.get_by_text("Active", exact=True)
if await active_text.is_visible(timeout=2000):
logger.error("Active badge still visible for %s after revoke!", username)
screenshot_path = f"/tmp/astrolabe_revoke_still_active_{username}.png"
if await page.locator("#mcp-revoke-background-button").is_visible(timeout=2000):
logger.error("Revoke button still visible for %s after revoke!", username)
screenshot_path = f"/tmp/astrolabe_revoke_still_enabled_{username}.png"
await page.screenshot(path=screenshot_path)
return False
except Exception:
@@ -1280,11 +663,11 @@ async def test_revoke_background_sync_access(
test_users_setup,
configure_astrolabe_for_mcp_server,
):
"""Test that users can revoke background sync access via the Revoke Access button.
"""Test that users can revoke background sync access via the "Disable background indexing" button.
This test verifies:
1. User enables background sync via app password
2. User clicks "Revoke Access" button
2. User clicks "Disable background indexing" button
3. Confirmation dialog is handled
4. POST request is sent to /api/v1/background-sync/credentials/revoke
5. "Active" badge disappears from settings page
@@ -1319,14 +702,9 @@ async def test_revoke_background_sync_access(
# Step 1: Login to Nextcloud
await login_to_nextcloud(page, username, password)
# Step 2: Complete full authorization (OAuth Step 1 + App Password Step 2)
# Provision background indexing (app-password opt-in; no OAuth step).
auth_result = await complete_astrolabe_authorization(page, username, password)
assert auth_result["step1"], (
f"OAuth authorization (Step 1) failed for {username}"
)
assert auth_result["step2"], (
f"App password setup (Step 2) failed for {username}"
)
assert auth_result["step2"], f"App password provisioning failed for {username}"
# Step 3: Verify background sync is enabled
assert await verify_app_password_created(username), (
tests/integration/test_astrolabe_plotly_visualization.py
+2 -2
View File
@@ -108,7 +108,7 @@ async def navigate_to_astrolabe_main(page: Page):
@pytest.mark.multi_user_basic
@pytest.mark.timeout(
300
) # 5 minutes - this test involves OAuth, app password, and vector sync
) # 5 minutes - this test involves app-password provisioning + vector sync
async def test_astrolabe_plotly_visualization_with_basic_auth(
browser,
test_users_setup,
@@ -143,7 +143,7 @@ async def test_astrolabe_plotly_visualization_with_basic_auth(
page = await context.new_page()
try:
# Phase 2: Complete full Astrolabe authorization (OAuth + app password)
# 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)
tests/integration/test_astrolabe_session_jwt_search.py
+109
View File
@@ -0,0 +1,109 @@
"""Astrolabe session-derived JWT search path (card 120 auth refactor).
Cross-system interface test. The astrolabe app was refactored to mint a
short-lived JWT for the current Nextcloud session user on demand (via the
`oidc` app's ``TokenGenerationRequestEvent``), replacing the old OAuth
authorize + offline_access + stored-refresh-token flow. This test proves the
new path end-to-end:
logged-in NC user → GET /apps/astrolabe/api/search → astrolabe mints a JWT
(McpTokenMinter) → calls the MCP server with ``Authorization: Bearer`` →
MCP validates the JWT (unified_verifier, aud=astrolabe_client_id) → results.
The headline behavioural change is that a user needs **no provisioning** to
search: there is no authorize redirect and ``has_background_access`` stays
False (app-password provisioning is now only for *background indexing*, a
separate opt-in covered by the background-sync tests).
Astrolabe is installed + configured (astrolabe_client_id, mcp_server_url) by
the container app-hooks; the test skips if that wiring is absent. Driven over
HTTP with BasicAuth (which establishes a Nextcloud session for the request) —
no browser needed.
"""
import os
import httpx
import pytest
pytestmark = [pytest.mark.integration, pytest.mark.login_flow]
NEXTCLOUD_URL = "http://localhost:8080"
ASTROLABE_API = f"{NEXTCLOUD_URL}/apps/astrolabe/api"
_HEADERS = {"OCS-APIRequest": "true"}
async def _astrolabe_configured(client: httpx.AsyncClient, auth) -> bool:
"""Readiness probe: astrolabe must be able to reach its MCP server."""
try:
resp = await client.get(
f"{ASTROLABE_API}/vector-status", auth=auth, headers=_HEADERS
)
except httpx.HTTPError:
return False
if resp.status_code != 200:
return False
return bool(resp.json().get("success"))
async def test_session_user_searches_without_provisioning(test_users_setup):
"""A non-admin session user searches with no OAuth/provisioning step.
success=True proves astrolabe minted a JWT from the session and the MCP
server accepted it. Results may be empty (nothing indexed) — the auth
chain, not recall, is under test here.
"""
auth = httpx.BasicAuth("bob", test_users_setup["bob"]["password"])
async with httpx.AsyncClient(timeout=30) as client:
if not await _astrolabe_configured(client, auth):
pytest.skip("Astrolabe not wired to an MCP server in this stack")
# No provisioning: search must work purely from the session JWT.
status = await client.get(
f"{ASTROLABE_API}/v1/background-sync/status", auth=auth, headers=_HEADERS
)
assert status.json()["has_background_access"] is False, (
"precondition: bob has not opted into background indexing"
)
resp = await client.get(
f"{ASTROLABE_API}/search",
params={"query": "quarterly planning", "limit": 3},
auth=auth,
headers=_HEADERS,
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["success"] is True, (
f"session-JWT search must succeed without provisioning; got {body}"
)
assert "results" in body and "algorithm_used" in body
async def test_admin_session_search_succeeds():
"""The same JWT-mint path works for the admin session user."""
admin_pw = os.environ["NEXTCLOUD_PASSWORD"]
auth = httpx.BasicAuth(os.environ["NEXTCLOUD_USERNAME"], admin_pw)
async with httpx.AsyncClient(timeout=30) as client:
if not await _astrolabe_configured(client, auth):
pytest.skip("Astrolabe not wired to an MCP server in this stack")
resp = await client.get(
f"{ASTROLABE_API}/search",
params={"query": "infrastructure", "limit": 3},
auth=auth,
headers=_HEADERS,
)
assert resp.status_code == 200, resp.text
assert resp.json()["success"] is True
async def test_search_requires_authentication():
"""Unauthenticated search is rejected (no anonymous JWT minting)."""
async with httpx.AsyncClient(follow_redirects=False, timeout=30) as client:
resp = await client.get(
f"{ASTROLABE_API}/search",
params={"query": "x"},
headers=_HEADERS,
)
assert resp.status_code in (401, 302, 303, 307, 308), resp.status_code
tests/integration/test_astrolabe_settings_buttons.py
+46 -79
View File
@@ -1,96 +1,63 @@
"""Integration tests for Astrolabe personal settings page buttons.
"""Integration tests for Astrolabe personal-settings background-sync endpoints.
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).
Cross-system interface test. The astrolabe app (installed by
app-hooks/post-installation/20-install-astrolabe-app.sh; source in
./third_party/astrolabe) was refactored to session-minted JWTs — the old
per-user OAuth flow and its ``/apps/astrolabe/oauth/disconnect`` route are
gone. Background indexing is now an app-password opt-in with a single revoke
endpoint.
Tests the button functionality on /settings/user/astrolabe:
1. Disable Indexing button (POST to /apps/astrolabe/api/revoke)
2. Disconnect button (POST to /apps/astrolabe/oauth/disconnect)
These tests verify that:
- The endpoints respond correctly to POST requests
- CSRF token validation works
- User actions are properly handled
- Appropriate redirects occur
These tests assert the *current* HTTP surface of the settings page:
- the revoke endpoint exists and is auth-gated
(POST /apps/astrolabe/api/v1/background-sync/credentials/revoke)
- the obsolete OAuth disconnect route is gone (404)
- the personal settings page route resolves
"""
import httpx
import pytest
pytestmark = pytest.mark.integration
@pytest.mark.integration
async def test_disable_indexing_button_endpoint_exists():
"""Test that the Disable Indexing endpoint is accessible."""
async with httpx.AsyncClient() as client:
# Try without authentication - should return 401 or redirect
response = await client.post(
"http://localhost:8080/apps/astrolabe/api/revoke",
follow_redirects=False,
)
NEXTCLOUD_URL = "http://localhost:8080"
ASTROLABE = f"{NEXTCLOUD_URL}/apps/astrolabe"
# Should get 401 Unauthorized or 30x redirect
assert response.status_code in [401, 301, 302, 303, 307, 308], (
f"Expected 401 or redirect without auth, got {response.status_code}"
# Auth failures (no session) surface as 401 or a login redirect.
_UNAUTH = {401, 302, 303, 307, 308}
async def test_revoke_endpoint_requires_auth():
"""The background-sync revoke endpoint exists and rejects anonymous calls."""
async with httpx.AsyncClient(follow_redirects=False) as client:
resp = await client.post(
f"{ASTROLABE}/api/v1/background-sync/credentials/revoke",
headers={"OCS-APIRequest": "true"},
)
# Must NOT be 404 — the route must exist — and must be auth-gated.
assert resp.status_code != 404, "revoke route missing"
assert resp.status_code in _UNAUTH, (
f"expected auth rejection, got {resp.status_code}"
)
@pytest.mark.integration
async def test_disconnect_button_endpoint_exists():
"""Test that the Disconnect endpoint is accessible."""
async with httpx.AsyncClient() as client:
# Try without authentication - should return 401 or redirect
response = await client.post(
"http://localhost:8080/apps/astrolabe/oauth/disconnect",
follow_redirects=False,
)
async def test_obsolete_oauth_disconnect_route_removed():
"""The pre-refactor OAuth disconnect route must no longer exist.
# Should get 401 Unauthorized or 30x redirect
assert response.status_code in [401, 301, 302, 303, 307, 308], (
f"Expected 401 or redirect without auth, got {response.status_code}"
)
@pytest.mark.integration
async def test_settings_page_renders_buttons():
"""Test that the settings page template includes button forms.
This test verifies that the PHP template renders the form elements.
It doesn't require authentication since we're just checking the route exists.
Regression guard for the auth refactor: ``/apps/astrolabe/oauth/disconnect``
(and the rest of the OAuth authorize/callback/disconnect surface) was
removed in favour of session-minted JWTs.
"""
async with httpx.AsyncClient(follow_redirects=False) as client:
# Try to access settings page
response = await client.get("http://localhost:8080/settings/user/astrolabe")
# Should get 401/redirect if not authenticated (expected)
# or 200 if user session exists from browser testing
assert response.status_code in [200, 401, 302, 303, 307, 308], (
f"Unexpected status code: {response.status_code}"
)
resp = await client.post(f"{ASTROLABE}/oauth/disconnect")
assert resp.status_code == 404, (
f"obsolete oauth/disconnect route still resolves ({resp.status_code})"
)
@pytest.mark.integration
@pytest.mark.skip(
reason="Requires manual authentication - test with Playwright instead"
)
async def test_disconnect_button_functionality():
"""Test that clicking Disconnect button clears user OAuth tokens.
NOTE: This test is skipped because programmatic login to Nextcloud is complex.
Use Playwright-based tests or manual testing instead.
"""
pass
@pytest.mark.integration
@pytest.mark.skip(
reason="Requires manual authentication - test with Playwright instead"
)
async def test_disable_indexing_button_functionality():
"""Test that clicking Disable Indexing button revokes background access.
NOTE: This test is skipped because programmatic login to Nextcloud is complex.
Use Playwright-based tests or manual testing instead.
"""
pass
async def test_settings_page_route_resolves():
"""The personal settings page route exists (auth-gated when no session)."""
async with httpx.AsyncClient(follow_redirects=False) as client:
resp = await client.get(f"{NEXTCLOUD_URL}/settings/user/astrolabe")
assert resp.status_code in ({200} | _UNAUTH), (
f"unexpected status for settings page: {resp.status_code}"
)
third_party/astrolabe
Vendored
+1 -1
Submodule third_party/astrolabe updated: 64153f5a3f...71a18287d4