nextcloud-mcp-server/tests/integration/test_app_password_provisioning.py

"""Integration tests for app password provisioning via management API.

Tests the complete flow for multi-user BasicAuth mode:
1. User stores app password via management API endpoint
2. MCP server stores it locally (encrypted)
3. Background sync uses locally stored password to access Nextcloud

These tests verify that BasicAuth and OAuth are completely separate concerns
with no fallback between them.
"""

import tempfile
from pathlib import Path

import pytest
from cryptography.fernet import Fernet

from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
from nextcloud_mcp_server.vector.oauth_sync import (
    NotProvisionedError,
    get_user_client,
    get_user_client_basic_auth,
    get_user_client_oauth,
)


@pytest.fixture
def encryption_key():
    """Generate a test encryption key."""
    return Fernet.generate_key().decode()


@pytest.fixture
async def temp_storage(encryption_key):
    """Create temporary storage instance with encryption for testing."""
    with tempfile.TemporaryDirectory() as tmpdir:
        db_path = Path(tmpdir) / "test_provisioning.db"
        storage = RefreshTokenStorage(
            db_path=str(db_path), encryption_key=encryption_key
        )
        await storage.initialize()
        yield storage


@pytest.mark.integration
async def test_basic_auth_mode_uses_local_storage(temp_storage, mocker):
    """Test that BasicAuth mode uses locally stored app passwords.

    In multi-user BasicAuth mode, app passwords are stored locally
    in the MCP server's database after being provisioned via the API.
    """
    # Store an app password in local storage
    await temp_storage.store_app_password("test_user", "JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB")

    # Call get_user_client_basic_auth with local storage
    client = await get_user_client_basic_auth(
        user_id="test_user",
        nextcloud_host="http://localhost:8080",
        storage=temp_storage,
    )

    # Verify client was created with correct credentials
    assert client is not None
    assert client.username == "test_user"


@pytest.mark.integration
async def test_basic_auth_mode_raises_error_without_app_password(temp_storage):
    """Test that BasicAuth mode raises NotProvisionedError if no app password.

    There is NO fallback to OAuth - if no app password, user must provision one.
    """
    # Don't store any app password

    # Call get_user_client_basic_auth - should raise NotProvisionedError
    with pytest.raises(NotProvisionedError) as exc_info:
        await get_user_client_basic_auth(
            user_id="test_user",
            nextcloud_host="http://localhost:8080",
            storage=temp_storage,
        )

    # Verify error message mentions app password provisioning
    assert "app password" in str(exc_info.value).lower()
    assert "test_user" in str(exc_info.value)


@pytest.mark.integration
async def test_get_user_client_dispatches_to_basic_auth(temp_storage, mocker):
    """Test that get_user_client dispatches to BasicAuth mode correctly."""
    # Store an app password
    await temp_storage.store_app_password("alice", "aaaaa-bbbbb-ccccc-ddddd-eeeee")

    # Mock RefreshTokenStorage.from_env at the source module
    mocker.patch(
        "nextcloud_mcp_server.auth.storage.RefreshTokenStorage.from_env",
        return_value=temp_storage,
    )
    # Also mock initialize since from_env returns an uninitialized instance
    mocker.patch.object(temp_storage, "initialize", return_value=None)

    # Call get_user_client in BasicAuth mode
    client = await get_user_client(
        user_id="alice",
        token_broker=None,  # No token broker needed for BasicAuth mode
        nextcloud_host="http://localhost:8080",
        use_basic_auth=True,
    )

    # Verify client was created successfully
    assert client is not None
    assert client.username == "alice"


@pytest.mark.integration
async def test_oauth_mode_uses_refresh_token_only(mocker):
    """Test that OAuth mode uses ONLY refresh tokens, NOT app passwords.

    In OAuth mode, app passwords are NOT used.
    This is a complete separation of concerns.
    """
    from nextcloud_mcp_server.auth.token_broker import TokenBrokerService

    # Mock TokenBrokerService to return an access token
    mock_token_broker = mocker.AsyncMock(spec=TokenBrokerService)
    mock_token_broker.get_background_token.return_value = "test-access-token"

    # Call get_user_client in OAuth mode
    _client = await get_user_client(
        user_id="test_user",
        token_broker=mock_token_broker,
        nextcloud_host="http://localhost:8080",
        use_basic_auth=False,  # OAuth mode
    )

    # Verify token broker was called
    mock_token_broker.get_background_token.assert_called_once()


@pytest.mark.integration
async def test_oauth_mode_raises_error_without_token(mocker):
    """Test that OAuth mode raises NotProvisionedError if no refresh token.

    There is NO fallback to app passwords - if no token, user must provision.
    """
    from nextcloud_mcp_server.auth.token_broker import TokenBrokerService

    # Mock TokenBrokerService to return None (no token)
    mock_token_broker = mocker.AsyncMock(spec=TokenBrokerService)
    mock_token_broker.get_background_token.return_value = None

    # Call get_user_client in OAuth mode - should raise NotProvisionedError
    with pytest.raises(NotProvisionedError) as exc_info:
        await get_user_client(
            user_id="test_user",
            token_broker=mock_token_broker,
            nextcloud_host="http://localhost:8080",
            use_basic_auth=False,
        )

    # Verify error message mentions OAuth provisioning
    assert "oauth" in str(exc_info.value).lower()
    assert "test_user" in str(exc_info.value)


@pytest.mark.integration
async def test_get_user_client_oauth_function(mocker):
    """Test the dedicated get_user_client_oauth function."""
    from nextcloud_mcp_server.auth.token_broker import TokenBrokerService

    # Mock TokenBrokerService
    mock_token_broker = mocker.AsyncMock(spec=TokenBrokerService)
    mock_token_broker.get_background_token.return_value = "test-bearer-token"

    # Call dedicated function
    client = await get_user_client_oauth(
        user_id="alice",
        token_broker=mock_token_broker,
        nextcloud_host="http://localhost:8080",
    )

    assert client is not None
    assert client.username == "alice"
    mock_token_broker.get_background_token.assert_called_once()


@pytest.mark.integration
async def test_oauth_mode_requires_token_broker():
    """Test that OAuth mode requires a token broker."""
    with pytest.raises(ValueError, match="token_broker required"):
        await get_user_client(
            user_id="test_user",
            token_broker=None,  # Missing token broker
            nextcloud_host="http://localhost:8080",
            use_basic_auth=False,  # OAuth mode
        )


@pytest.mark.integration
async def test_multiple_users_basic_auth_mode(temp_storage, mocker):
    """Test that multiple users can be provisioned independently."""
    # Store app passwords for multiple users
    users = {
        "alice": "aaaaa-aaaaa-aaaaa-aaaaa-aaaaa",
        "bob": "bbbbb-bbbbb-bbbbb-bbbbb-bbbbb",
        "charlie": "ccccc-ccccc-ccccc-ccccc-ccccc",
    }

    for user_id, password in users.items():
        await temp_storage.store_app_password(user_id, password)

    # Verify each user can get a client
    for user_id in users.keys():
        client = await get_user_client_basic_auth(
            user_id=user_id,
            nextcloud_host="http://localhost:8080",
            storage=temp_storage,
        )
        assert client is not None
        assert client.username == user_id


@pytest.mark.integration
async def test_get_all_provisioned_users(temp_storage):
    """Test that we can list all provisioned users for BasicAuth mode."""
    # Store app passwords for multiple users
    await temp_storage.store_app_password("alice", "aaaaa-aaaaa-aaaaa-aaaaa-aaaaa")
    await temp_storage.store_app_password("bob", "bbbbb-bbbbb-bbbbb-bbbbb-bbbbb")

    # Get all provisioned users
    user_ids = await temp_storage.get_all_app_password_user_ids()

    assert len(user_ids) == 2
    assert "alice" in user_ids
    assert "bob" in user_ids


@pytest.mark.integration
async def test_revoke_app_password(temp_storage):
    """Test that deleting app password revokes background access."""
    # Provision user
    await temp_storage.store_app_password("alice", "aaaaa-aaaaa-aaaaa-aaaaa-aaaaa")

    # Verify user is provisioned
    user_ids = await temp_storage.get_all_app_password_user_ids()
    assert "alice" in user_ids

    # Revoke access
    deleted = await temp_storage.delete_app_password("alice")
    assert deleted is True

    # Verify user is no longer provisioned
    user_ids = await temp_storage.get_all_app_password_user_ids()
    assert "alice" not in user_ids

    # Verify get_user_client now raises NotProvisionedError
    with pytest.raises(NotProvisionedError):
        await get_user_client_basic_auth(
            user_id="alice",
            nextcloud_host="http://localhost:8080",
            storage=temp_storage,
        )