Chris CoutinhoandClaude Opus 4.7
79ea4e9e21
`_get_background_operations_enabled()` was emitting three advisory log lines (1 INFO + 2 deprecation WARNINGs) on every call. Because `get_settings()` is intentionally non-cached and runs on every MCP tool invocation via `get_client()`, the "Automatically enabled background operations for semantic search in multi-user mode" INFO line was firing per-request — 569 entries/hour in one production tenant. Gate the three log emissions behind a module-level `_bg_ops_advisories_logged` flag, mirroring the existing `_warn_missing_secret_once` precedent in `vector/webhook_receiver.py`. The boolean-derivation path stays unchanged, so the `Settings` value remains fresh per call. Extends the autouse `_reload_dynaconf_after_test` fixture to reset the new flag between tests, and adds two regression tests that call `get_settings()` five times and assert each advisory fires exactly once. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1064 lines
39 KiB
Python
1064 lines
39 KiB
Python
"""Unit tests for configuration validation and mode detection.
|
|
|
|
Tests cover:
|
|
- Mode detection logic
|
|
- Configuration validation for each mode
|
|
- Error message generation
|
|
- Edge cases and boundary conditions
|
|
"""
|
|
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from nextcloud_mcp_server.config import Settings, _reload_config
|
|
from nextcloud_mcp_server.config_validators import (
|
|
AuthMode,
|
|
detect_auth_mode,
|
|
get_mode_summary,
|
|
validate_configuration,
|
|
)
|
|
|
|
|
|
class TestModeDetection:
|
|
"""Test auth mode detection from configuration."""
|
|
|
|
def test_multi_user_basic_mode_detection(self):
|
|
"""Test multi-user BasicAuth mode is selected via explicit deployment_mode.
|
|
|
|
ADR-022 follow-up: the ENABLE_MULTI_USER_BASIC_AUTH auto-detection branch
|
|
was removed; the only way to opt in is `MCP_DEPLOYMENT_MODE=multi_user_basic`.
|
|
Coverage for the explicit-mode path also lives in
|
|
TestExplicitModeSelection::test_explicit_multi_user_basic_mode.
|
|
"""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
deployment_mode="multi_user_basic",
|
|
)
|
|
|
|
mode = detect_auth_mode(settings)
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
assert settings.enable_multi_user_basic_auth is True
|
|
|
|
def test_single_user_basic_mode_detection(self):
|
|
"""Test single-user BasicAuth mode is detected."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
)
|
|
|
|
mode = detect_auth_mode(settings)
|
|
assert mode == AuthMode.SINGLE_USER_BASIC
|
|
|
|
def test_login_flow_default(self):
|
|
"""Test Login Flow v2 is the default multi-user mode."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
)
|
|
|
|
mode = detect_auth_mode(settings)
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
|
|
|
|
class TestSingleUserBasicValidation:
|
|
"""Test validation for single-user BasicAuth mode."""
|
|
|
|
def test_valid_minimal_config(self):
|
|
"""Test valid minimal single-user BasicAuth config."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.SINGLE_USER_BASIC
|
|
assert len(errors) == 0
|
|
|
|
def test_valid_with_vector_sync(self):
|
|
"""Test valid config with vector sync enabled."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
vector_sync_enabled=True,
|
|
qdrant_location=":memory:",
|
|
ollama_base_url="http://ollama:11434",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.SINGLE_USER_BASIC
|
|
assert len(errors) == 0
|
|
|
|
def test_missing_required_host(self):
|
|
"""Test error when NEXTCLOUD_HOST is missing."""
|
|
settings = Settings(
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.SINGLE_USER_BASIC
|
|
assert any("nextcloud_host" in err.lower() for err in errors)
|
|
|
|
def test_missing_required_username(self):
|
|
"""Test that partial credentials fall back to OAuth mode."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_password="password", # Password without username
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
# Mode detection requires BOTH username AND password for single-user BasicAuth
|
|
# If only one is present, it defaults to OAuth single-audience
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
# In OAuth mode, having a password set is forbidden
|
|
assert any("nextcloud_password" in err.lower() for err in errors)
|
|
|
|
def test_missing_required_password(self):
|
|
"""Test that partial credentials fall back to OAuth mode."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="admin", # Username without password
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
# Mode detection requires BOTH username AND password for single-user BasicAuth
|
|
# If only one is present, it defaults to OAuth single-audience
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
# In OAuth mode, having a username set is forbidden
|
|
assert any("nextcloud_username" in err.lower() for err in errors)
|
|
|
|
def test_forbidden_multi_user_basic_when_credentials_present(self):
|
|
"""Test multi-user mode rejects single-user credentials.
|
|
|
|
When MCP_DEPLOYMENT_MODE=multi_user_basic is set explicitly but
|
|
NEXTCLOUD_USERNAME/PASSWORD are also set (a misconfiguration),
|
|
the explicit mode wins and validation reports the credentials as
|
|
forbidden.
|
|
"""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
deployment_mode="multi_user_basic",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
# Should report errors for forbidden username/password
|
|
assert len(errors) > 0
|
|
|
|
def test_vector_sync_without_embedding_provider_uses_fallback(self):
|
|
"""Test that vector sync works with Simple provider fallback (no config needed)."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
vector_sync_enabled=True,
|
|
qdrant_location=":memory:",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.SINGLE_USER_BASIC
|
|
# Should pass - Simple provider is always available as fallback
|
|
assert len(errors) == 0
|
|
|
|
|
|
class TestMultiUserBasicValidation:
|
|
"""Test validation for multi-user BasicAuth mode."""
|
|
|
|
def test_valid_minimal_config(self):
|
|
"""Test valid minimal multi-user BasicAuth config."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
deployment_mode="multi_user_basic",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
assert len(errors) == 0
|
|
|
|
def test_valid_with_offline_access(self):
|
|
"""Test valid config with offline access enabled."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
deployment_mode="multi_user_basic",
|
|
enable_offline_access=True,
|
|
oidc_client_id="test-client",
|
|
oidc_client_secret="test-secret",
|
|
token_encryption_key="test-key-" + "a" * 32,
|
|
token_storage_db="/tmp/tokens.db",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
assert len(errors) == 0
|
|
|
|
def test_missing_required_host(self):
|
|
"""Test error when NEXTCLOUD_HOST is missing."""
|
|
settings = Settings(
|
|
deployment_mode="multi_user_basic",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
assert any("nextcloud_host" in err.lower() for err in errors)
|
|
|
|
def test_forbidden_username_password(self):
|
|
"""Test error when NEXTCLOUD_USERNAME/PASSWORD are set."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
deployment_mode="multi_user_basic",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
# Explicit MCP_DEPLOYMENT_MODE wins over auto-detection from credentials
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
# Should report errors for forbidden username/password
|
|
assert any("nextcloud_username" in err.lower() for err in errors)
|
|
assert any("nextcloud_password" in err.lower() for err in errors)
|
|
|
|
def test_offline_access_missing_oauth_credentials(self):
|
|
"""Test that offline access works without OAuth credentials (will use DCR)."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
deployment_mode="multi_user_basic",
|
|
enable_offline_access=True,
|
|
token_encryption_key="test-key-" + "a" * 32,
|
|
token_storage_db="/tmp/tokens.db",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
# No errors - DCR will be used as fallback (consistent with OAuth modes)
|
|
assert len(errors) == 0
|
|
|
|
def test_offline_access_missing_encryption_key(self):
|
|
"""Test error when offline access enabled but encryption key missing."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
deployment_mode="multi_user_basic",
|
|
enable_offline_access=True,
|
|
oidc_client_id="test-client",
|
|
oidc_client_secret="test-secret",
|
|
token_storage_db="/tmp/tokens.db",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
assert any("token_encryption_key" in err.lower() for err in errors)
|
|
|
|
def test_vector_sync_auto_enables_background_ops_in_multi_user_mode(self):
|
|
"""Test vector sync automatically enables background operations in multi-user mode (ADR-021)."""
|
|
# Before ADR-021: This would have failed validation (required explicit ENABLE_OFFLINE_ACCESS)
|
|
# After ADR-021: vector_sync_enabled auto-enables background operations
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "multi_user_basic",
|
|
"VECTOR_SYNC_ENABLED": "true", # Using old name for backward compat test
|
|
"QDRANT_LOCATION": ":memory:",
|
|
"OLLAMA_BASE_URL": "http://ollama:11434",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
"NEXTCLOUD_OIDC_CLIENT_ID": "test-client-id",
|
|
"NEXTCLOUD_OIDC_CLIENT_SECRET": "test-client-secret",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
# Should have no errors - background operations auto-enabled
|
|
assert len(errors) == 0
|
|
# Verify background operations were auto-enabled
|
|
assert settings.enable_offline_access is True
|
|
|
|
|
|
class TestLoginFlowValidation:
|
|
"""Test validation for Login Flow v2 mode (formerly OAUTH_SINGLE_AUDIENCE)."""
|
|
|
|
def test_valid_minimal_config(self):
|
|
"""Test valid minimal Login Flow v2 config — enable_login_flow is now derived."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
assert len(errors) == 0
|
|
# ADR-022 follow-up: enable_login_flow is derived from the resolved mode.
|
|
assert settings.enable_login_flow is True
|
|
|
|
def test_valid_with_static_credentials(self):
|
|
"""Test valid config with static OAuth credentials."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
oidc_client_id="test-client",
|
|
oidc_client_secret="test-secret",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
assert len(errors) == 0
|
|
|
|
def test_valid_with_offline_access(self):
|
|
"""Test valid config with offline access."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
oidc_client_id="test-client",
|
|
oidc_client_secret="test-secret",
|
|
enable_offline_access=True,
|
|
token_encryption_key="test-key-" + "a" * 32,
|
|
token_storage_db="/tmp/tokens.db",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
assert len(errors) == 0
|
|
|
|
def test_forbidden_username_password(self):
|
|
"""Test that username/password trigger single-user mode instead."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
# This should detect as SINGLE_USER_BASIC
|
|
assert mode == AuthMode.SINGLE_USER_BASIC
|
|
|
|
def test_offline_access_missing_encryption_key(self):
|
|
"""Test error when offline access enabled but encryption key missing."""
|
|
settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
enable_offline_access=True,
|
|
token_storage_db="/tmp/tokens.db",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
assert any("token_encryption_key" in err.lower() for err in errors)
|
|
|
|
def test_login_flow_mode_auto_derives_enable_login_flow_flag(self):
|
|
"""ADR-022 follow-up: deployment_mode=login_flow auto-derives the flag.
|
|
|
|
Users no longer need to set ENABLE_LOGIN_FLOW=true (env var was
|
|
removed); Settings.__post_init__ populates settings.enable_login_flow
|
|
from the resolved mode at construction time, so every Settings
|
|
instance carries correct flags regardless of how it was built.
|
|
"""
|
|
# Default-fallback case: no auth env vars → LOGIN_FLOW.
|
|
settings = Settings(nextcloud_host="http://localhost")
|
|
assert settings.enable_login_flow is True
|
|
assert settings.enable_multi_user_basic_auth is False
|
|
assert detect_auth_mode(settings) == AuthMode.LOGIN_FLOW
|
|
|
|
# Single-user BasicAuth (credentials set) → neither derived flag.
|
|
basic_settings = Settings(
|
|
nextcloud_host="http://localhost",
|
|
nextcloud_username="alice",
|
|
nextcloud_password="password",
|
|
)
|
|
assert basic_settings.enable_login_flow is False
|
|
assert basic_settings.enable_multi_user_basic_auth is False
|
|
assert detect_auth_mode(basic_settings) == AuthMode.SINGLE_USER_BASIC
|
|
|
|
def test_vector_sync_auto_enables_background_ops_in_login_flow_mode(self):
|
|
"""Test vector sync automatically enables background operations in Login Flow v2 mode (ADR-021)."""
|
|
# Before ADR-021: This would have failed validation (required explicit ENABLE_OFFLINE_ACCESS)
|
|
# After ADR-021: vector_sync_enabled auto-enables background operations in multi-user modes
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"VECTOR_SYNC_ENABLED": "true",
|
|
"QDRANT_LOCATION": ":memory:",
|
|
"OLLAMA_BASE_URL": "http://ollama:11434",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
# Note: No username/password = Login Flow v2 multi-user OAuth mode
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
# Should have no errors - background operations auto-enabled
|
|
assert len(errors) == 0
|
|
# Verify background operations were auto-enabled
|
|
assert settings.enable_offline_access is True
|
|
|
|
|
|
class TestModeSummary:
|
|
"""Test mode summary generation."""
|
|
|
|
def test_single_user_basic_summary(self):
|
|
"""Test summary for single-user BasicAuth mode."""
|
|
summary = get_mode_summary(AuthMode.SINGLE_USER_BASIC)
|
|
|
|
assert "single_user_basic" in summary
|
|
assert "NEXTCLOUD_HOST" in summary
|
|
assert "NEXTCLOUD_USERNAME" in summary
|
|
assert "NEXTCLOUD_PASSWORD" in summary
|
|
assert "VECTOR_SYNC_ENABLED" in summary
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Test edge cases and boundary conditions."""
|
|
|
|
def test_empty_string_treated_as_missing(self):
|
|
"""Test that empty strings are treated as missing values."""
|
|
settings = Settings(
|
|
nextcloud_host="", # Empty string
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
# Should fail because nextcloud_host is effectively missing
|
|
assert any("nextcloud_host" in err.lower() for err in errors)
|
|
|
|
def test_whitespace_treated_as_missing(self):
|
|
"""Test that whitespace-only strings are treated as missing."""
|
|
settings = Settings(
|
|
nextcloud_host=" ", # Whitespace only
|
|
nextcloud_username="admin",
|
|
nextcloud_password="password",
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
# Should fail because nextcloud_host is effectively missing
|
|
assert any("nextcloud_host" in err.lower() for err in errors)
|
|
|
|
def test_multiple_errors_reported(self):
|
|
"""Test that multiple errors are all reported."""
|
|
settings = Settings(
|
|
# Missing all required fields for single-user BasicAuth
|
|
)
|
|
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
# Should have errors for missing host (OAuth mode is default)
|
|
assert len(errors) > 0
|
|
|
|
|
|
class TestConfigurationConsolidation:
|
|
"""Test ADR-021 configuration consolidation and backward compatibility.
|
|
|
|
Tests verify:
|
|
- New variable names work (ENABLE_SEMANTIC_SEARCH, ENABLE_BACKGROUND_OPERATIONS)
|
|
- Old variable names still work (VECTOR_SYNC_ENABLED, ENABLE_OFFLINE_ACCESS)
|
|
- Deprecation warnings are logged
|
|
- Auto-enablement of background operations in multi-user modes
|
|
"""
|
|
|
|
def test_new_semantic_search_variable_name(self):
|
|
"""Test ENABLE_SEMANTIC_SEARCH (new name) works correctly."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLE_SEMANTIC_SEARCH": "true",
|
|
"QDRANT_LOCATION": ":memory:",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
assert settings.vector_sync_enabled is True
|
|
|
|
def test_old_vector_sync_variable_name_backward_compat(self):
|
|
"""Test VECTOR_SYNC_ENABLED (old name) still works for backward compatibility."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"VECTOR_SYNC_ENABLED": "true",
|
|
"QDRANT_LOCATION": ":memory:",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
assert settings.vector_sync_enabled is True
|
|
|
|
def test_new_background_operations_variable_name(self):
|
|
"""Test ENABLE_BACKGROUND_OPERATIONS (new name) works correctly."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLE_BACKGROUND_OPERATIONS": "true",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
assert settings.enable_offline_access is True
|
|
|
|
def test_old_offline_access_variable_name_backward_compat(self):
|
|
"""Test ENABLE_OFFLINE_ACCESS (old name) still works for backward compatibility."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLE_OFFLINE_ACCESS": "true",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
assert settings.enable_offline_access is True
|
|
|
|
def test_semantic_search_auto_enables_background_ops_in_login_flow_mode(self):
|
|
"""Test ENABLE_SEMANTIC_SEARCH automatically enables background operations in Login Flow v2 mode."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"ENABLE_SEMANTIC_SEARCH": "true",
|
|
"QDRANT_LOCATION": ":memory:",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
# Note: No NEXTCLOUD_USERNAME/PASSWORD = OAuth mode
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
|
|
# Semantic search enabled
|
|
assert settings.vector_sync_enabled is True
|
|
|
|
# Background operations auto-enabled (even though not explicitly set)
|
|
assert settings.enable_offline_access is True
|
|
|
|
def test_semantic_search_does_not_auto_enable_in_single_user_mode(self):
|
|
"""Test ENABLE_SEMANTIC_SEARCH does NOT auto-enable background ops in single-user mode."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"NEXTCLOUD_USERNAME": "admin",
|
|
"NEXTCLOUD_PASSWORD": "password",
|
|
"ENABLE_SEMANTIC_SEARCH": "true",
|
|
"QDRANT_LOCATION": ":memory:",
|
|
# Note: Username/password set = single-user BasicAuth mode
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
|
|
# Semantic search enabled
|
|
assert settings.vector_sync_enabled is True
|
|
|
|
# Background operations NOT auto-enabled (not needed in single-user mode)
|
|
assert settings.enable_offline_access is False
|
|
|
|
def test_explicit_background_ops_still_works(self):
|
|
"""Test explicitly setting ENABLE_BACKGROUND_OPERATIONS works even without semantic search."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"ENABLE_BACKGROUND_OPERATIONS": "true",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
# Note: No semantic search enabled
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
|
|
# Semantic search NOT enabled
|
|
assert settings.vector_sync_enabled is False
|
|
|
|
# Background operations explicitly enabled
|
|
assert settings.enable_offline_access is True
|
|
|
|
def test_both_old_and_new_semantic_search_names_prefers_new(self):
|
|
"""Test setting both ENABLE_SEMANTIC_SEARCH and VECTOR_SYNC_ENABLED uses new name."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLE_SEMANTIC_SEARCH": "true",
|
|
"VECTOR_SYNC_ENABLED": "false", # Old name says false
|
|
"QDRANT_LOCATION": ":memory:",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
|
|
# Should use new name value (true)
|
|
assert settings.vector_sync_enabled is True
|
|
|
|
def test_both_old_and_new_background_ops_names_prefers_new(self):
|
|
"""Test setting both ENABLE_BACKGROUND_OPERATIONS and ENABLE_OFFLINE_ACCESS uses new name."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLE_BACKGROUND_OPERATIONS": "true",
|
|
"ENABLE_OFFLINE_ACCESS": "false", # Old name says false
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
|
|
# Should use new name value (true)
|
|
assert settings.enable_offline_access is True
|
|
|
|
def test_validation_no_longer_requires_both_variables(self):
|
|
"""Test validation no longer requires explicit ENABLE_OFFLINE_ACCESS when semantic search enabled."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "multi_user_basic",
|
|
"ENABLE_SEMANTIC_SEARCH": "true",
|
|
"QDRANT_LOCATION": ":memory:",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
# OAuth credentials required for app password retrieval (when background ops enabled)
|
|
"NEXTCLOUD_OIDC_CLIENT_ID": "test-client-id",
|
|
"NEXTCLOUD_OIDC_CLIENT_SECRET": "test-client-secret",
|
|
# Note: ENABLE_OFFLINE_ACCESS not set - should auto-enable
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode, errors = validate_configuration(settings)
|
|
|
|
# Should have no validation errors
|
|
# (Previously would have required explicit ENABLE_OFFLINE_ACCESS)
|
|
assert len(errors) == 0
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
# Verify background operations were auto-enabled
|
|
assert settings.enable_offline_access is True
|
|
|
|
def test_auto_enable_info_log_emitted_at_most_once(self, caplog):
|
|
"""Auto-enable INFO advisory must fire once per process, not per get_settings() call.
|
|
|
|
Regression: `get_settings()` is non-cached and called per-request from
|
|
`get_client()`, so unguarded `logger.info` calls in
|
|
`_get_background_operations_enabled()` spammed every MCP tool invocation
|
|
(observed: 569 entries/hour in tenant `tenant-e2e-disc-0033`).
|
|
"""
|
|
import logging
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"ENABLE_SEMANTIC_SEARCH": "true",
|
|
"QDRANT_LOCATION": ":memory:",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
# No NEXTCLOUD_USERNAME/PASSWORD → multi-user mode → auto-enable triggers
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
caplog.set_level(logging.INFO, logger="nextcloud_mcp_server.config")
|
|
|
|
for _ in range(5):
|
|
settings = get_settings()
|
|
assert settings.enable_offline_access is True
|
|
|
|
auto_enable_records = [
|
|
r
|
|
for r in caplog.records
|
|
if r.name == "nextcloud_mcp_server.config"
|
|
and "Automatically enabled background operations" in r.message
|
|
]
|
|
assert len(auto_enable_records) == 1, (
|
|
f"Expected exactly one auto-enable advisory log, "
|
|
f"got {len(auto_enable_records)}: "
|
|
f"{[r.message for r in auto_enable_records]}"
|
|
)
|
|
|
|
def test_legacy_offline_access_deprecation_warning_emitted_at_most_once(
|
|
self, caplog
|
|
):
|
|
"""Legacy `ENABLE_OFFLINE_ACCESS` deprecation WARNING is also one-shot."""
|
|
import logging
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"ENABLE_OFFLINE_ACCESS": "true",
|
|
"TOKEN_ENCRYPTION_KEY": "test-key",
|
|
"TOKEN_STORAGE_DB": "/tmp/test.db",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
caplog.set_level(logging.WARNING, logger="nextcloud_mcp_server.config")
|
|
|
|
for _ in range(5):
|
|
get_settings()
|
|
|
|
deprecation_records = [
|
|
r
|
|
for r in caplog.records
|
|
if r.name == "nextcloud_mcp_server.config"
|
|
and "ENABLE_OFFLINE_ACCESS is deprecated" in r.message
|
|
]
|
|
assert len(deprecation_records) == 1, (
|
|
f"Expected exactly one deprecation warning, "
|
|
f"got {len(deprecation_records)}: "
|
|
f"{[r.message for r in deprecation_records]}"
|
|
)
|
|
|
|
|
|
class TestExplicitModeSelection:
|
|
"""Test ADR-021 explicit mode selection via MCP_DEPLOYMENT_MODE.
|
|
|
|
Tests verify:
|
|
- Explicit mode selection works for all modes
|
|
- Invalid mode names raise ValueError
|
|
- Explicit mode takes precedence over auto-detection
|
|
"""
|
|
|
|
def test_explicit_single_user_basic_mode(self):
|
|
"""Test explicit single_user_basic mode selection."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "single_user_basic",
|
|
"NEXTCLOUD_USERNAME": "admin",
|
|
"NEXTCLOUD_PASSWORD": "password",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode = detect_auth_mode(settings)
|
|
|
|
assert mode == AuthMode.SINGLE_USER_BASIC
|
|
|
|
def test_explicit_multi_user_basic_mode(self):
|
|
"""Test explicit multi_user_basic mode selection."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "multi_user_basic",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode = detect_auth_mode(settings)
|
|
|
|
assert mode == AuthMode.MULTI_USER_BASIC
|
|
|
|
def test_explicit_login_flow_mode(self):
|
|
"""Test explicit login_flow mode selection."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "login_flow",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode = detect_auth_mode(settings)
|
|
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
|
|
def test_invalid_deployment_mode_raises_error(self):
|
|
"""Test invalid MCP_DEPLOYMENT_MODE raises ValueError."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "invalid_mode",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
|
|
# Should raise ValueError with clear message
|
|
try:
|
|
detect_auth_mode(settings)
|
|
assert False, "Should have raised ValueError"
|
|
except ValueError as e:
|
|
assert "Invalid MCP_DEPLOYMENT_MODE" in str(e)
|
|
assert "invalid_mode" in str(e)
|
|
assert "Valid values:" in str(e)
|
|
|
|
def test_oauth_single_audience_migration_hint(self):
|
|
"""ADR-022: rejecting `oauth_single_audience` surfaces a rename hint.
|
|
|
|
Pins the special-case branch in detect_auth_mode that helps users
|
|
upgrading from ADR-021 configurations spot the rename without
|
|
having to grep the changelog.
|
|
"""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "oauth_single_audience",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
|
|
with pytest.raises(ValueError) as exc:
|
|
detect_auth_mode(settings)
|
|
|
|
msg = str(exc.value)
|
|
assert "oauth_single_audience" in msg
|
|
assert "login_flow" in msg
|
|
assert "ADR-022" in msg
|
|
|
|
def test_explicit_mode_overrides_auto_detection(self):
|
|
"""Test explicit mode takes precedence over auto-detection."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"NEXTCLOUD_USERNAME": "admin", # Would auto-detect as single_user_basic
|
|
"NEXTCLOUD_PASSWORD": "password",
|
|
"MCP_DEPLOYMENT_MODE": "login_flow", # Explicit override
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode = detect_auth_mode(settings)
|
|
|
|
# Should use explicit mode, not auto-detected mode
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
|
|
def test_case_insensitive_mode_names(self):
|
|
"""Test MCP_DEPLOYMENT_MODE is case-insensitive."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "LOGIN_FLOW", # Uppercase
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode = detect_auth_mode(settings)
|
|
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
|
|
def test_whitespace_in_mode_name_stripped(self):
|
|
"""Test whitespace in MCP_DEPLOYMENT_MODE is stripped."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": " login_flow ", # Whitespace
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings()
|
|
mode = detect_auth_mode(settings)
|
|
|
|
assert mode == AuthMode.LOGIN_FLOW
|
|
|
|
def test_legacy_enable_multi_user_basic_auth_env_var_errors(self):
|
|
"""ADR-022 follow-up: ENABLE_MULTI_USER_BASIC_AUTH=true must fail loudly.
|
|
|
|
The env-var alias was removed; users must migrate to
|
|
`MCP_DEPLOYMENT_MODE=multi_user_basic`. Silent removal would have
|
|
switched users to LOGIN_FLOW (the default) — wrong runtime mode.
|
|
The check lives in Settings.__post_init__ so it fires at config
|
|
load time, not only when detect_auth_mode is reached.
|
|
"""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"ENABLE_MULTI_USER_BASIC_AUTH": "true",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
|
|
with pytest.raises(ValueError) as exc:
|
|
get_settings()
|
|
|
|
assert "ENABLE_MULTI_USER_BASIC_AUTH" in str(exc.value)
|
|
assert "multi_user_basic" in str(exc.value)
|
|
|
|
def test_legacy_enable_login_flow_env_var_errors(self):
|
|
"""ADR-022 follow-up: ENABLE_LOGIN_FLOW=true must fail loudly.
|
|
|
|
Mirrors the ENABLE_MULTI_USER_BASIC_AUTH check — both legacy aliases
|
|
now error with a one-line migration message at Settings construction.
|
|
"""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"ENABLE_LOGIN_FLOW": "true",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
|
|
with pytest.raises(ValueError) as exc:
|
|
get_settings()
|
|
|
|
assert "ENABLE_LOGIN_FLOW" in str(exc.value)
|
|
assert "login_flow" in str(exc.value)
|
|
|
|
def test_legacy_env_var_check_ignores_falsy_strings(self):
|
|
"""ADR-022 follow-up: a leftover ENABLE_LOGIN_FLOW=false must NOT error.
|
|
|
|
Reviewer round 2 found that the legacy check used `os.getenv(legacy)`
|
|
which is truthy for the literal string 'false'. A user who had
|
|
explicitly set the flag to false (meaning 'I don't want this') would
|
|
get a startup ValueError after upgrading, which is wrong. The fix
|
|
only fires for explicitly-truthy values.
|
|
"""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"ENABLE_LOGIN_FLOW": "false",
|
|
"ENABLE_MULTI_USER_BASIC_AUTH": "0",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
settings = get_settings() # Must not raise.
|
|
|
|
# Falls through to LOGIN_FLOW (the auto-detect default) since
|
|
# no credentials are set.
|
|
assert detect_auth_mode(settings) == AuthMode.LOGIN_FLOW
|
|
|
|
def test_derived_flags_stable_across_get_settings_calls(self):
|
|
"""Regression: derived flags must persist across get_settings() calls.
|
|
|
|
`get_settings()` builds a fresh Settings on each invocation. Earlier
|
|
commits set the derived flags as a side effect of detect_auth_mode,
|
|
which meant the *next* get_settings() call started with default
|
|
False — breaking per-request handlers in the integration tests
|
|
for `mcp-multi-user-basic` and `mcp-login-flow`. The fix moves the
|
|
derivation into Settings.__post_init__ so every instance carries
|
|
correct flags.
|
|
"""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NEXTCLOUD_HOST": "http://localhost:8080",
|
|
"MCP_DEPLOYMENT_MODE": "multi_user_basic",
|
|
},
|
|
clear=True,
|
|
):
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
_reload_config()
|
|
|
|
s1 = get_settings()
|
|
s2 = get_settings()
|
|
|
|
assert s1 is not s2 # fresh instance each call
|
|
assert s1.enable_multi_user_basic_auth is True
|
|
assert s2.enable_multi_user_basic_auth is True
|
|
assert s1.enable_login_flow is False
|
|
assert s2.enable_login_flow is False
|