mcp-nextcloud/tests/unit/test_app_password_storage.py

"""
Unit tests for App Password Storage functionality.

Tests the app password methods in RefreshTokenStorage for multi-user
BasicAuth mode background sync.

These tests are parametrized over both supported backends so the storage
layer is exercised against SQLite (default, always runs) and Postgres (gated
on ``TEST_DATABASE_URL``; bring up ``docker compose --profile postgres up
-d postgres-test`` and export ``TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp``
to opt in).
"""

import tempfile
from pathlib import Path

import pytest
from cryptography.fernet import Fernet

from nextcloud_mcp_server.auth.storage import RefreshTokenStorage

pytestmark = pytest.mark.unit


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


@pytest.fixture
async def temp_storage(encryption_key, storage_backend):
    """Create a storage instance backed by either SQLite or Postgres.

    The ``storage_backend`` fixture is parametrized by pytest, so every test
    that uses ``temp_storage`` runs once per backend that is available in
    the current environment.
    """
    if storage_backend["kind"] == "sqlite":
        with tempfile.TemporaryDirectory() as tmpdir:
            db_path = Path(tmpdir) / "test_app_passwords.db"
            storage = RefreshTokenStorage(
                db_path=str(db_path), encryption_key=encryption_key
            )
            await storage.initialize()
            yield storage
    else:
        storage = RefreshTokenStorage(
            database_url=storage_backend["url"], encryption_key=encryption_key
        )
        await storage.initialize()
        try:
            yield storage
        finally:
            # Each test gets an isolated schema; tear it down so the next
            # parametrized run starts clean.
            await storage_backend["reset"]()


async def test_store_app_password(temp_storage):
    """Test storing an app password."""
    await temp_storage.store_app_password(
        user_id="testuser",
        app_password="JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB",
    )

    # Verify it can be retrieved
    retrieved = await temp_storage.get_app_password("testuser")
    assert retrieved == "JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB"


async def test_store_app_password_replaces_existing(temp_storage):
    """Test that storing a new app password replaces the existing one."""
    await temp_storage.store_app_password(
        user_id="testuser",
        app_password="aaaaa-bbbbb-ccccc-ddddd-eeeee",
    )
    await temp_storage.store_app_password(
        user_id="testuser",
        app_password="fffff-ggggg-hhhhh-iiiii-jjjjj",
    )

    retrieved = await temp_storage.get_app_password("testuser")
    assert retrieved == "fffff-ggggg-hhhhh-iiiii-jjjjj"


async def test_get_app_password_nonexistent(temp_storage):
    """Test retrieving app password for non-existent user."""
    retrieved = await temp_storage.get_app_password("nonexistent")
    assert retrieved is None


async def test_delete_app_password(temp_storage):
    """Test deleting an app password."""
    await temp_storage.store_app_password(
        user_id="testuser",
        app_password="JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB",
    )

    deleted = await temp_storage.delete_app_password("testuser")
    assert deleted is True

    # Verify it's gone
    retrieved = await temp_storage.get_app_password("testuser")
    assert retrieved is None


async def test_delete_app_password_nonexistent(temp_storage):
    """Test deleting non-existent app password."""
    deleted = await temp_storage.delete_app_password("nonexistent")
    assert deleted is False


async def test_get_all_app_password_user_ids(temp_storage):
    """Test listing all users with app passwords."""
    await temp_storage.store_app_password("alice", "aaaaa-aaaaa-aaaaa-aaaaa-aaaaa")
    await temp_storage.store_app_password("bob", "bbbbb-bbbbb-bbbbb-bbbbb-bbbbb")
    await temp_storage.store_app_password("charlie", "ccccc-ccccc-ccccc-ccccc-ccccc")

    user_ids = await temp_storage.get_all_app_password_user_ids()
    assert len(user_ids) == 3
    assert "alice" in user_ids
    assert "bob" in user_ids
    assert "charlie" in user_ids


async def test_get_all_app_password_user_ids_empty(temp_storage):
    """Test listing users when none have app passwords."""
    user_ids = await temp_storage.get_all_app_password_user_ids()
    assert len(user_ids) == 0


async def test_app_password_encryption(encryption_key):
    """Test that app passwords are encrypted at rest."""
    with tempfile.TemporaryDirectory() as tmpdir:
        db_path = Path(tmpdir) / "test_encryption.db"
        storage = RefreshTokenStorage(
            db_path=str(db_path), encryption_key=encryption_key
        )
        await storage.initialize()

        # Store a password
        test_password = "JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB"
        await storage.store_app_password("testuser", test_password)

        # Read directly from database to verify encryption
        import aiosqlite

        async with aiosqlite.connect(str(db_path)) as db:
            async with db.execute(
                "SELECT encrypted_password FROM app_passwords WHERE user_id = ?",
                ("testuser",),
            ) as cursor:
                row = await cursor.fetchone()

        # The stored value should be encrypted (not plain text)
        encrypted_bytes = row[0]
        assert encrypted_bytes != test_password.encode()
        # Encrypted data should be longer due to Fernet overhead
        assert len(encrypted_bytes) > len(test_password)


async def test_app_password_requires_encryption_key():
    """Test that app password operations require encryption key."""
    with tempfile.TemporaryDirectory() as tmpdir:
        db_path = Path(tmpdir) / "test_no_key.db"
        storage = RefreshTokenStorage(db_path=str(db_path), encryption_key=None)
        await storage.initialize()

        # Storing should fail without encryption key
        with pytest.raises(RuntimeError, match="Encryption key not configured"):
            await storage.store_app_password(
                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
            )

        # Getting should also fail without encryption key
        with pytest.raises(RuntimeError, match="Encryption key not configured"):
            await storage.get_app_password("testuser")


async def test_multiple_users_independence(temp_storage):
    """Test that different users maintain independent app passwords."""
    users = ["alice", "bob", "charlie", "diana"]

    # Store unique passwords for each user
    for i, user in enumerate(users):
        password = (
            f"{user[0]}{user[0]}{user[0]}{user[0]}{user[0]}-" * 4
            + f"{user[0]}{user[0]}{user[0]}{user[0]}{user[0]}"
        )
        await temp_storage.store_app_password(user, password)

    # Verify each user has their correct password
    for user in users:
        expected = (
            f"{user[0]}{user[0]}{user[0]}{user[0]}{user[0]}-" * 4
            + f"{user[0]}{user[0]}{user[0]}{user[0]}{user[0]}"
        )
        retrieved = await temp_storage.get_app_password(user)
        assert retrieved == expected

    # Delete one user's password
    await temp_storage.delete_app_password("bob")

    # Verify other users unchanged
    for user in ["alice", "charlie", "diana"]:
        retrieved = await temp_storage.get_app_password(user)
        assert retrieved is not None

    # Verify bob's password is gone
    assert await temp_storage.get_app_password("bob") is None


async def test_app_password_with_special_characters(temp_storage):
    """Test storing passwords with various alphanumeric patterns."""
    # Nextcloud app passwords use alphanumeric characters
    passwords = [
        "AAAAA-BBBBB-CCCCC-DDDDD-EEEEE",  # uppercase
        "aaaaa-bbbbb-ccccc-ddddd-eeeee",  # lowercase
        "12345-67890-12345-67890-12345",  # numbers
        "aB1cD-eF2gH-iJ3kL-mN4oP-qR5sT",  # mixed
    ]

    for i, password in enumerate(passwords):
        user = f"user{i}"
        await temp_storage.store_app_password(user, password)
        retrieved = await temp_storage.get_app_password(user)
        assert retrieved == password


async def test_decryption_with_wrong_key(encryption_key):
    """Test that decryption fails with wrong key."""
    with tempfile.TemporaryDirectory() as tmpdir:
        db_path = Path(tmpdir) / "test_wrong_key.db"

        # Store with original key
        storage1 = RefreshTokenStorage(
            db_path=str(db_path), encryption_key=encryption_key
        )
        await storage1.initialize()
        await storage1.store_app_password("testuser", "JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB")

        # Try to read with different key
        wrong_key = Fernet.generate_key().decode()
        storage2 = RefreshTokenStorage(db_path=str(db_path), encryption_key=wrong_key)
        await storage2.initialize()

        # Decryption should fail and return None (graceful handling)
        retrieved = await storage2.get_app_password("testuser")
        assert retrieved is None