Merge pull request #707 from cbcoutinho/fix/uvx-pypi-deployment
fix: enable uvx/PyPI deployments without Docker assumptions
This commit is contained in:
@@ -7,6 +7,7 @@ __pycache__/
|
||||
|
||||
# Dynaconf (ADR-024)
|
||||
.secrets.toml
|
||||
settings.toml
|
||||
settings.local.toml
|
||||
|
||||
# Git
|
||||
|
||||
+10
-5
@@ -40,11 +40,16 @@ path_separator = os
|
||||
# Default: utf-8
|
||||
# output_encoding = utf-8
|
||||
|
||||
# Database URL - can be overridden by:
|
||||
# 1. Passing -x database_url=... to alembic commands
|
||||
# 2. Setting in environment via get_database_url() in env.py
|
||||
# Default: sqlite:///app/data/tokens.db
|
||||
sqlalchemy.url = sqlite+aiosqlite:////app/data/tokens.db
|
||||
# Database URL - placeholder only.
|
||||
# Runtime code (migrations.get_alembic_config / alembic/env.get_database_url)
|
||||
# overrides this with config.get_token_db_path() which defaults to an
|
||||
# ephemeral per-process tempfile unless TOKEN_STORAGE_DB is set.
|
||||
# This value only matters for manual `alembic` invocations (e.g.
|
||||
# `uv run alembic current`). The `./tokens.db` path is intentionally
|
||||
# resolved relative to the shell's current working directory at the time
|
||||
# of invocation — run alembic from the repo root for predictable behaviour,
|
||||
# or pass `-x database_url=sqlite+aiosqlite:///<abs-path>`.
|
||||
sqlalchemy.url = sqlite+aiosqlite:///./tokens.db
|
||||
|
||||
[post_write_hooks]
|
||||
# Post-write hooks allow you to run scripts after generating migration files
|
||||
|
||||
+8
-4
@@ -86,13 +86,17 @@ services:
|
||||
- 127.0.0.1:9090:9090
|
||||
volumes:
|
||||
- mcp-data:/app/data
|
||||
- ./settings.toml:/app/settings.toml:ro
|
||||
- ./settings.toml.example:/app/settings.toml:ro
|
||||
environment:
|
||||
- NEXTCLOUD_HOST=http://app:80
|
||||
- NEXTCLOUD_USERNAME=admin
|
||||
- NEXTCLOUD_PASSWORD=admin
|
||||
- NEXTCLOUD_PUBLIC_ISSUER_URL=http://localhost:8080
|
||||
|
||||
# Pin the token DB inside the mcp-data volume so the runtime default
|
||||
# (ephemeral tempfile) doesn't silently apply inside containers.
|
||||
- TOKEN_STORAGE_DB=/app/data/tokens.db
|
||||
|
||||
# Semantic search configuration (ADR-007, ADR-021)
|
||||
- ENABLE_SEMANTIC_SEARCH=true
|
||||
- VECTOR_SYNC_SCAN_INTERVAL=5
|
||||
@@ -167,7 +171,7 @@ services:
|
||||
# NO admin credentials - credentials come from client Authorization header
|
||||
volumes:
|
||||
- multi-user-basic-data:/app/data
|
||||
- ./settings.toml:/app/settings.toml:ro
|
||||
- ./settings.toml.example:/app/settings.toml:ro
|
||||
profiles:
|
||||
- multi-user-basic
|
||||
|
||||
@@ -243,7 +247,7 @@ services:
|
||||
volumes:
|
||||
- keycloak-tokens:/app/data
|
||||
- keycloak-oauth-storage:/app/.oauth
|
||||
- ./settings.toml:/app/settings.toml:ro
|
||||
- ./settings.toml.example:/app/settings.toml:ro
|
||||
profiles:
|
||||
- keycloak
|
||||
|
||||
@@ -282,7 +286,7 @@ services:
|
||||
volumes:
|
||||
- login-flow-data:/app/data
|
||||
- login-flow-oauth-storage:/app/.oauth
|
||||
- ./settings.toml:/app/settings.toml:ro
|
||||
- ./settings.toml.example:/app/settings.toml:ro
|
||||
profiles:
|
||||
- login-flow
|
||||
|
||||
|
||||
@@ -51,8 +51,14 @@ def get_database_url() -> str:
|
||||
url = config.get_main_option("sqlalchemy.url")
|
||||
|
||||
if not url:
|
||||
# Default to /app/data/tokens.db for Docker deployments
|
||||
db_path = Path("/app/data/tokens.db")
|
||||
# Fall back to the same resolver the runtime uses (ephemeral tempfile
|
||||
# unless TOKEN_STORAGE_DB is set). Imported lazily to avoid pulling
|
||||
# the full config module into offline alembic invocations.
|
||||
from nextcloud_mcp_server.config import ( # noqa: PLC0415
|
||||
get_token_db_path,
|
||||
)
|
||||
|
||||
db_path = Path(get_token_db_path())
|
||||
url = f"sqlite+aiosqlite:///{db_path}"
|
||||
logger.warning(
|
||||
f"No database URL configured, using default: {url}. "
|
||||
|
||||
@@ -39,6 +39,7 @@ import httpx
|
||||
from anyio import to_thread
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
from nextcloud_mcp_server.config import get_token_db_path, is_ephemeral_token_db
|
||||
from nextcloud_mcp_server.migrations import stamp_database, upgrade_database
|
||||
from nextcloud_mcp_server.observability.metrics import record_db_operation
|
||||
|
||||
@@ -82,7 +83,10 @@ class RefreshTokenStorage:
|
||||
Create storage instance from environment variables.
|
||||
|
||||
Environment variables:
|
||||
TOKEN_STORAGE_DB: Path to database file (default: /app/data/tokens.db)
|
||||
TOKEN_STORAGE_DB: Path to database file. If unset, a per-process
|
||||
tempfile is allocated and deleted at interpreter exit —
|
||||
tokens are ephemeral and wiped on restart. Set this to a
|
||||
filesystem path to persist tokens across restarts.
|
||||
TOKEN_ENCRYPTION_KEY: Optional base64-encoded Fernet key (required for token storage)
|
||||
|
||||
Returns:
|
||||
@@ -92,7 +96,13 @@ class RefreshTokenStorage:
|
||||
If TOKEN_ENCRYPTION_KEY is not set, token storage operations will fail,
|
||||
but webhook tracking will still work.
|
||||
"""
|
||||
db_path = os.getenv("TOKEN_STORAGE_DB", "/app/data/tokens.db")
|
||||
db_path = get_token_db_path()
|
||||
if is_ephemeral_token_db(db_path):
|
||||
logger.info(
|
||||
"Using ephemeral token storage at %s "
|
||||
"(set TOKEN_STORAGE_DB to persist tokens across restarts)",
|
||||
db_path,
|
||||
)
|
||||
encryption_key_b64 = os.getenv("TOKEN_ENCRYPTION_KEY")
|
||||
|
||||
encryption_key = None
|
||||
|
||||
+31
-12
@@ -6,6 +6,8 @@ import uvicorn
|
||||
|
||||
from nextcloud_mcp_server.config import (
|
||||
get_settings,
|
||||
get_token_db_path,
|
||||
is_ephemeral_token_db,
|
||||
)
|
||||
from nextcloud_mcp_server.migrations import (
|
||||
create_migration,
|
||||
@@ -285,13 +287,25 @@ def db():
|
||||
pass
|
||||
|
||||
|
||||
def _warn_if_ephemeral(database_path: str) -> None:
|
||||
if is_ephemeral_token_db(database_path):
|
||||
click.echo(
|
||||
click.style(
|
||||
f"⚠ Using ephemeral tempfile {database_path}; changes "
|
||||
"will be lost on exit. Pass --database-path or set "
|
||||
"TOKEN_STORAGE_DB to operate on a persistent database.",
|
||||
fg="yellow",
|
||||
),
|
||||
err=True,
|
||||
)
|
||||
|
||||
|
||||
@db.command()
|
||||
@click.option(
|
||||
"--database-path",
|
||||
"-d",
|
||||
envvar="TOKEN_STORAGE_DB",
|
||||
default="/app/data/tokens.db",
|
||||
show_default=True,
|
||||
default=None,
|
||||
help="Path to token storage database (can also use TOKEN_STORAGE_DB env var)",
|
||||
)
|
||||
@click.option(
|
||||
@@ -301,7 +315,7 @@ def db():
|
||||
show_default=True,
|
||||
help="Target revision (default: head for latest)",
|
||||
)
|
||||
def upgrade(database_path: str, revision: str):
|
||||
def upgrade(database_path: str | None, revision: str):
|
||||
"""Upgrade database to a specific revision.
|
||||
|
||||
\b
|
||||
@@ -315,6 +329,8 @@ def upgrade(database_path: str, revision: str):
|
||||
# Use custom database path
|
||||
$ nextcloud-mcp-server db upgrade -d /path/to/tokens.db
|
||||
"""
|
||||
database_path = database_path or get_token_db_path()
|
||||
_warn_if_ephemeral(database_path)
|
||||
try:
|
||||
click.echo(f"Upgrading database to revision: {revision}")
|
||||
upgrade_database(database_path, revision)
|
||||
@@ -329,8 +345,7 @@ def upgrade(database_path: str, revision: str):
|
||||
"--database-path",
|
||||
"-d",
|
||||
envvar="TOKEN_STORAGE_DB",
|
||||
default="/app/data/tokens.db",
|
||||
show_default=True,
|
||||
default=None,
|
||||
help="Path to token storage database",
|
||||
)
|
||||
@click.option(
|
||||
@@ -343,7 +358,7 @@ def upgrade(database_path: str, revision: str):
|
||||
@click.confirmation_option(
|
||||
prompt="Are you sure you want to downgrade the database? This may result in data loss."
|
||||
)
|
||||
def downgrade(database_path: str, revision: str):
|
||||
def downgrade(database_path: str | None, revision: str):
|
||||
"""Downgrade database to a specific revision.
|
||||
|
||||
WARNING: This may result in data loss! Use with caution.
|
||||
@@ -359,6 +374,8 @@ def downgrade(database_path: str, revision: str):
|
||||
# Downgrade to base (empty database)
|
||||
$ nextcloud-mcp-server db downgrade --revision base
|
||||
"""
|
||||
database_path = database_path or get_token_db_path()
|
||||
_warn_if_ephemeral(database_path)
|
||||
try:
|
||||
click.echo(f"Downgrading database to revision: {revision}")
|
||||
downgrade_database(database_path, revision)
|
||||
@@ -373,17 +390,18 @@ def downgrade(database_path: str, revision: str):
|
||||
"--database-path",
|
||||
"-d",
|
||||
envvar="TOKEN_STORAGE_DB",
|
||||
default="/app/data/tokens.db",
|
||||
show_default=True,
|
||||
default=None,
|
||||
help="Path to token storage database",
|
||||
)
|
||||
def current(database_path: str):
|
||||
def current(database_path: str | None):
|
||||
"""Show current database revision.
|
||||
|
||||
\b
|
||||
Example:
|
||||
$ nextcloud-mcp-server db current
|
||||
"""
|
||||
database_path = database_path or get_token_db_path()
|
||||
_warn_if_ephemeral(database_path)
|
||||
try:
|
||||
revision = get_current_revision(database_path)
|
||||
if revision:
|
||||
@@ -406,17 +424,18 @@ def current(database_path: str):
|
||||
"--database-path",
|
||||
"-d",
|
||||
envvar="TOKEN_STORAGE_DB",
|
||||
default="/app/data/tokens.db",
|
||||
show_default=True,
|
||||
default=None,
|
||||
help="Path to token storage database",
|
||||
)
|
||||
def history(database_path: str):
|
||||
def history(database_path: str | None):
|
||||
"""Show migration history.
|
||||
|
||||
\b
|
||||
Example:
|
||||
$ nextcloud-mcp-server db history
|
||||
"""
|
||||
database_path = database_path or get_token_db_path()
|
||||
_warn_if_ephemeral(database_path)
|
||||
try:
|
||||
click.echo("Migration history:")
|
||||
show_migration_history(database_path)
|
||||
|
||||
+186
-11
@@ -1,34 +1,160 @@
|
||||
import atexit
|
||||
import logging
|
||||
import logging.config
|
||||
import os
|
||||
import socket
|
||||
import ssl
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from dynaconf import Dynaconf, Validator
|
||||
|
||||
# Resolve root_path for dynaconf settings files.
|
||||
# Editable installs: parent.parent is the project root (has settings.toml).
|
||||
# Non-editable installs (Docker): settings.toml is mounted at WORKDIR (/app/).
|
||||
_config_root = Path(__file__).parent.parent
|
||||
if not (_config_root / "settings.toml").exists():
|
||||
_config_root = Path.cwd()
|
||||
|
||||
# Sentinel for "key not in dynaconf at all" vs "explicitly set to None".
|
||||
_UNSET = object()
|
||||
|
||||
# Dynaconf instance — loads settings.toml + .secrets.toml + env vars.
|
||||
# Env vars always win (12-factor). See ADR-024 for architecture.
|
||||
# Built-in defaults — declared in Python so env vars work without any settings
|
||||
# file being present (e.g., `uvx` / `pip install` deployments). Mirrors the
|
||||
# [default] section that used to live in settings.toml. Keys set here are
|
||||
# "known" to dynaconf, which is required because we run with
|
||||
# ignore_unknown_envvars=True. See ADR-024/025.
|
||||
_DEFAULTS: dict[str, Any] = {
|
||||
# Deployment mode (ADR-021)
|
||||
"mcp_deployment_mode": None,
|
||||
# Nextcloud core
|
||||
"nextcloud_host": None,
|
||||
"nextcloud_username": None,
|
||||
"nextcloud_password": None,
|
||||
"nextcloud_app_password": None,
|
||||
"nextcloud_verify_ssl": True,
|
||||
"nextcloud_ca_bundle": None,
|
||||
"nextcloud_mcp_server_url": None,
|
||||
"nextcloud_resource_uri": None,
|
||||
# OAuth/OIDC
|
||||
"oidc_discovery_url": None,
|
||||
"nextcloud_oidc_client_id": None,
|
||||
"nextcloud_oidc_client_secret": None,
|
||||
"oidc_issuer": None,
|
||||
"jwks_uri": None,
|
||||
"introspection_uri": None,
|
||||
"userinfo_uri": None,
|
||||
"oidc_resource_server_id": None,
|
||||
# Mode flags
|
||||
"enable_multi_user_basic_auth": False,
|
||||
"enable_login_flow": False,
|
||||
"enable_semantic_search": False,
|
||||
"enable_background_operations": False,
|
||||
"vector_sync_enabled": False,
|
||||
"enable_offline_access": False,
|
||||
"enable_token_exchange": False,
|
||||
# Token storage
|
||||
"token_encryption_key": None,
|
||||
# None = ephemeral per-process tempfile (see get_token_db_path()).
|
||||
# Set TOKEN_STORAGE_DB to persist tokens across restarts.
|
||||
"token_storage_db": None,
|
||||
# Vector sync
|
||||
"vector_sync_scan_interval": 300,
|
||||
"vector_sync_processor_workers": 3,
|
||||
"vector_sync_queue_max_size": 10000,
|
||||
"vector_sync_user_poll_interval": 60,
|
||||
# Qdrant
|
||||
"qdrant_url": None,
|
||||
"qdrant_location": None,
|
||||
"qdrant_api_key": None,
|
||||
"qdrant_collection": "nextcloud_content",
|
||||
# Ollama
|
||||
"ollama_base_url": None,
|
||||
"ollama_embedding_model": "nomic-embed-text",
|
||||
"ollama_verify_ssl": True,
|
||||
# OpenAI
|
||||
"openai_api_key": None,
|
||||
"openai_base_url": None,
|
||||
"openai_embedding_model": "text-embedding-3-small",
|
||||
# Document chunking
|
||||
"document_chunk_size": 2048,
|
||||
"document_chunk_overlap": 200,
|
||||
# Observability
|
||||
"metrics_enabled": True,
|
||||
"metrics_port": 9090,
|
||||
"otel_exporter_otlp_endpoint": None,
|
||||
"otel_exporter_verify_ssl": False,
|
||||
"otel_service_name": "nextcloud-mcp-server",
|
||||
"otel_traces_sampler": "always_on",
|
||||
"otel_traces_sampler_arg": 1.0,
|
||||
"log_format": "text",
|
||||
"log_level": "INFO",
|
||||
"log_include_trace_context": True,
|
||||
# Document processing
|
||||
"enable_document_processing": False,
|
||||
"document_processor": "unstructured",
|
||||
"enable_unstructured": False,
|
||||
"unstructured_api_url": "http://unstructured:8000",
|
||||
"unstructured_timeout": 120,
|
||||
"unstructured_strategy": "auto",
|
||||
"unstructured_languages": "eng,deu",
|
||||
"progress_interval": 10,
|
||||
"enable_tesseract": False,
|
||||
"tesseract_cmd": None,
|
||||
"tesseract_lang": "eng",
|
||||
"enable_pymupdf": True,
|
||||
"pymupdf_extract_images": True,
|
||||
"pymupdf_image_dir": None,
|
||||
"enable_custom_processor": False,
|
||||
"custom_processor_url": None,
|
||||
"custom_processor_types": "application/pdf",
|
||||
"custom_processor_name": "custom",
|
||||
"custom_processor_api_key": None,
|
||||
"custom_processor_timeout": 60,
|
||||
}
|
||||
|
||||
|
||||
def _resolve_settings_files() -> list[str]:
|
||||
"""Find optional external settings files.
|
||||
|
||||
Priority:
|
||||
1. NEXTCLOUD_MCP_SETTINGS_FILE env var (absolute or relative path).
|
||||
If set but the file does not exist, raise FileNotFoundError —
|
||||
silently falling back to defaults on a typo would be a footgun.
|
||||
.secrets.toml is looked for alongside the explicit file.
|
||||
2. Otherwise ./settings.toml in cwd (for docker / dev workflows),
|
||||
with .secrets.toml also looked for in cwd.
|
||||
|
||||
Returns an empty list if nothing is configured — that's fine, defaults
|
||||
and env vars still apply.
|
||||
"""
|
||||
files: list[str] = []
|
||||
explicit = os.environ.get("NEXTCLOUD_MCP_SETTINGS_FILE")
|
||||
if explicit:
|
||||
p = Path(explicit)
|
||||
if not p.exists():
|
||||
raise FileNotFoundError(
|
||||
f"NEXTCLOUD_MCP_SETTINGS_FILE points to a file that does "
|
||||
f"not exist: {explicit}"
|
||||
)
|
||||
files.append(str(p))
|
||||
secrets = p.parent / ".secrets.toml"
|
||||
else:
|
||||
cwd_settings = Path.cwd() / "settings.toml"
|
||||
if cwd_settings.exists():
|
||||
files.append(str(cwd_settings))
|
||||
secrets = Path.cwd() / ".secrets.toml"
|
||||
if secrets.exists():
|
||||
files.append(str(secrets))
|
||||
return files
|
||||
|
||||
|
||||
# Dynaconf instance — env vars always win (12-factor). Settings files are
|
||||
# optional; when absent the defaults above provide the full key schema so
|
||||
# env vars still override correctly. See ADR-024/025 for architecture.
|
||||
_dynaconf = Dynaconf(
|
||||
settings_files=["settings.toml", ".secrets.toml"],
|
||||
settings_files=_resolve_settings_files(),
|
||||
environments=True,
|
||||
envvar_prefix=False,
|
||||
env_switcher="MCP_DEPLOYMENT_MODE",
|
||||
ignore_unknown_envvars=True,
|
||||
root_path=str(_config_root),
|
||||
load_dotenv=False,
|
||||
**_DEFAULTS,
|
||||
validators=[
|
||||
# Port ranges
|
||||
Validator("METRICS_PORT", gte=1, lte=65535),
|
||||
@@ -73,6 +199,55 @@ def _reload_config():
|
||||
_dynaconf.validators.validate_all()
|
||||
|
||||
|
||||
_ephemeral_db_path: str | None = None
|
||||
|
||||
|
||||
def get_token_db_path() -> str:
|
||||
"""Resolve the token SQLite database path.
|
||||
|
||||
Priority:
|
||||
1. TOKEN_STORAGE_DB if explicitly set — docker-compose pins
|
||||
/app/data/tokens.db this way. Read via dynaconf, which picks up
|
||||
the env var because TOKEN_STORAGE_DB is declared in _DEFAULTS.
|
||||
2. Otherwise a per-process tempfile under tempfile.gettempdir(),
|
||||
allocated lazily and deleted at interpreter exit via atexit.
|
||||
Ephemeral: tokens are wiped on restart, matching the Qdrant
|
||||
":memory:" default pattern used elsewhere in this project.
|
||||
"""
|
||||
explicit = _dynaconf.get("TOKEN_STORAGE_DB")
|
||||
if explicit:
|
||||
return str(explicit)
|
||||
global _ephemeral_db_path
|
||||
if _ephemeral_db_path is None:
|
||||
fd, path = tempfile.mkstemp(
|
||||
prefix=f"nextcloud-mcp-tokens-{os.getpid()}-", suffix=".db"
|
||||
)
|
||||
os.close(fd)
|
||||
_ephemeral_db_path = path
|
||||
|
||||
def _cleanup(p: str = path) -> None:
|
||||
try:
|
||||
if os.path.exists(p):
|
||||
os.unlink(p)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
atexit.register(_cleanup)
|
||||
return _ephemeral_db_path
|
||||
|
||||
|
||||
def is_ephemeral_token_db(path: str) -> bool:
|
||||
"""Return True if the given path is the process-local ephemeral tempfile.
|
||||
|
||||
Precondition: `get_token_db_path()` must have been called at least once
|
||||
in this process to allocate the tempfile. If called before allocation,
|
||||
this returns False for any input (including the eventual tempfile path),
|
||||
because there is nothing to compare against yet. In practice every call
|
||||
site in this repo resolves the path via `get_token_db_path()` first.
|
||||
"""
|
||||
return path == _ephemeral_db_path
|
||||
|
||||
|
||||
LOGGING_CONFIG = {
|
||||
"version": 1,
|
||||
"disable_existing_loggers": False,
|
||||
|
||||
@@ -13,6 +13,7 @@ from alembic.config import Config
|
||||
|
||||
import nextcloud_mcp_server.alembic as alembic_package
|
||||
from alembic import command
|
||||
from nextcloud_mcp_server.config import get_token_db_path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -25,8 +26,9 @@ def get_alembic_config(database_path: str | Path | None = None) -> Config:
|
||||
package location instead of alembic.ini file.
|
||||
|
||||
Args:
|
||||
database_path: Path to SQLite database file. If None, uses default
|
||||
(/app/data/tokens.db for Docker)
|
||||
database_path: Path to SQLite database file. If None, resolves via
|
||||
config.get_token_db_path() (ephemeral tempfile unless
|
||||
TOKEN_STORAGE_DB is set).
|
||||
|
||||
Returns:
|
||||
Alembic Config object configured for the specified database
|
||||
@@ -45,7 +47,7 @@ def get_alembic_config(database_path: str | Path | None = None) -> Config:
|
||||
if database_path:
|
||||
db_path = Path(database_path).resolve()
|
||||
else:
|
||||
db_path = Path("/app/data/tokens.db") # Default for Docker
|
||||
db_path = Path(get_token_db_path()).resolve()
|
||||
|
||||
url = f"sqlite+aiosqlite:///{db_path}"
|
||||
config.set_main_option("sqlalchemy.url", url)
|
||||
@@ -100,7 +102,7 @@ def get_current_revision(database_path: str | Path | None = None) -> str | None:
|
||||
"""
|
||||
|
||||
if database_path is None:
|
||||
database_path = "/app/data/tokens.db"
|
||||
database_path = get_token_db_path()
|
||||
|
||||
db_path = Path(database_path).resolve()
|
||||
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
"""Tests for config.py path resolution helpers added in PR #707.
|
||||
|
||||
Covers:
|
||||
- get_token_db_path() / is_ephemeral_token_db() — ephemeral tempfile default
|
||||
with TOKEN_STORAGE_DB override.
|
||||
- _resolve_settings_files() — optional external settings file discovery,
|
||||
including the NEXTCLOUD_MCP_SETTINGS_FILE env var and its colocation
|
||||
semantics for .secrets.toml.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import nextcloud_mcp_server.config as cfg
|
||||
from nextcloud_mcp_server.config import (
|
||||
_reload_config,
|
||||
_resolve_settings_files,
|
||||
get_token_db_path,
|
||||
is_ephemeral_token_db,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_ephemeral_state(monkeypatch):
|
||||
"""Reset module-global ephemeral tempfile state between tests.
|
||||
|
||||
get_token_db_path() memoizes its result in a module-level global and
|
||||
registers an atexit hook for cleanup. Tests need an isolated slate so
|
||||
assertions about "already allocated" vs "not yet" are meaningful.
|
||||
"""
|
||||
old = cfg._ephemeral_db_path
|
||||
cfg._ephemeral_db_path = None
|
||||
monkeypatch.delenv("TOKEN_STORAGE_DB", raising=False)
|
||||
monkeypatch.delenv("NEXTCLOUD_MCP_SETTINGS_FILE", raising=False)
|
||||
_reload_config()
|
||||
yield
|
||||
if cfg._ephemeral_db_path and os.path.exists(cfg._ephemeral_db_path):
|
||||
os.unlink(cfg._ephemeral_db_path)
|
||||
cfg._ephemeral_db_path = old
|
||||
_reload_config()
|
||||
|
||||
|
||||
class TestGetTokenDbPath:
|
||||
def test_explicit_env_var_returned(self, monkeypatch, tmp_path):
|
||||
target = tmp_path / "explicit.db"
|
||||
monkeypatch.setenv("TOKEN_STORAGE_DB", str(target))
|
||||
_reload_config()
|
||||
|
||||
assert get_token_db_path() == str(target)
|
||||
# No tempfile should have been allocated since we took the explicit
|
||||
# branch.
|
||||
assert cfg._ephemeral_db_path is None
|
||||
|
||||
def test_ephemeral_tempfile_when_unset(self):
|
||||
path = get_token_db_path()
|
||||
|
||||
assert path.startswith(tempfile.gettempdir())
|
||||
assert f"nextcloud-mcp-tokens-{os.getpid()}-" in os.path.basename(path)
|
||||
assert path.endswith(".db")
|
||||
assert os.path.exists(path)
|
||||
|
||||
def test_ephemeral_tempfile_is_memoized(self):
|
||||
first = get_token_db_path()
|
||||
second = get_token_db_path()
|
||||
|
||||
assert first == second
|
||||
# Only the memoized file should exist — no stray siblings.
|
||||
parent = Path(first).parent
|
||||
matches = list(parent.glob(f"nextcloud-mcp-tokens-{os.getpid()}-*.db"))
|
||||
assert matches == [Path(first)]
|
||||
|
||||
def test_is_ephemeral_token_db_detects_allocated_path(self):
|
||||
path = get_token_db_path()
|
||||
|
||||
assert is_ephemeral_token_db(path) is True
|
||||
assert is_ephemeral_token_db("/some/other/path") is False
|
||||
|
||||
def test_is_ephemeral_token_db_before_allocation(self):
|
||||
# The autouse fixture reset _ephemeral_db_path to None; do not call
|
||||
# get_token_db_path() first. Nothing is allocated yet.
|
||||
assert cfg._ephemeral_db_path is None
|
||||
assert is_ephemeral_token_db("/any/path") is False
|
||||
assert is_ephemeral_token_db("") is False
|
||||
|
||||
def test_explicit_path_does_not_trigger_tempfile(self, monkeypatch, tmp_path):
|
||||
"""Regression guard: the explicit branch must short-circuit cleanly."""
|
||||
monkeypatch.setenv("TOKEN_STORAGE_DB", str(tmp_path / "pinned.db"))
|
||||
_reload_config()
|
||||
|
||||
get_token_db_path()
|
||||
# Nothing under the tempfile prefix should have been created.
|
||||
matches = list(
|
||||
Path(tempfile.gettempdir()).glob(f"nextcloud-mcp-tokens-{os.getpid()}-*.db")
|
||||
)
|
||||
assert matches == []
|
||||
|
||||
|
||||
class TestResolveSettingsFiles:
|
||||
def test_empty_list_when_nothing_present(self, monkeypatch, tmp_path):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
assert _resolve_settings_files() == []
|
||||
|
||||
def test_picks_up_cwd_settings(self, monkeypatch, tmp_path):
|
||||
(tmp_path / "settings.toml").write_text("[default]\n")
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
result = _resolve_settings_files()
|
||||
assert str(tmp_path / "settings.toml") in result
|
||||
|
||||
def test_picks_up_cwd_secrets(self, monkeypatch, tmp_path):
|
||||
(tmp_path / ".secrets.toml").write_text("[default]\n")
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
result = _resolve_settings_files()
|
||||
assert str(tmp_path / ".secrets.toml") in result
|
||||
|
||||
def test_explicit_settings_file_included(self, monkeypatch, tmp_path):
|
||||
explicit = tmp_path / "nested" / "my-settings.toml"
|
||||
explicit.parent.mkdir()
|
||||
explicit.write_text("[default]\n")
|
||||
monkeypatch.setenv("NEXTCLOUD_MCP_SETTINGS_FILE", str(explicit))
|
||||
# cwd has no settings.toml / .secrets.toml
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
result = _resolve_settings_files()
|
||||
assert result == [str(explicit)]
|
||||
|
||||
def test_explicit_settings_file_secrets_colocated(self, monkeypatch, tmp_path):
|
||||
"""PR #707 reviewer feedback: .secrets.toml should live beside
|
||||
the explicit settings file, not always in cwd."""
|
||||
config_dir = tmp_path / "etc"
|
||||
config_dir.mkdir()
|
||||
explicit = config_dir / "settings.toml"
|
||||
explicit.write_text("[default]\n")
|
||||
secrets = config_dir / ".secrets.toml"
|
||||
secrets.write_text("[default]\n")
|
||||
|
||||
monkeypatch.setenv("NEXTCLOUD_MCP_SETTINGS_FILE", str(explicit))
|
||||
# cwd is elsewhere and deliberately contains *no* secrets file
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
result = _resolve_settings_files()
|
||||
assert str(explicit) in result
|
||||
assert str(secrets) in result
|
||||
|
||||
def test_explicit_missing_raises(self, monkeypatch, tmp_path):
|
||||
"""PR #707 reviewer feedback: missing explicit path must not be
|
||||
silently ignored — users will think their config is applied when
|
||||
it isn't."""
|
||||
missing = tmp_path / "does-not-exist.toml"
|
||||
monkeypatch.setenv("NEXTCLOUD_MCP_SETTINGS_FILE", str(missing))
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
with pytest.raises(FileNotFoundError, match="does-not-exist.toml"):
|
||||
_resolve_settings_files()
|
||||
Reference in New Issue
Block a user