FabledGrimoire/tests/test_api_keys.py

"""Unit tests for api_keys service — pure logic, DB calls mocked."""
import hashlib
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from grimoire.services.api_keys import (
    hash_token,
    _key_prefix,
    generate_key,
    create_api_key,
    list_api_keys,
    revoke_api_key,
    lookup_key,
)
from tests.helpers import make_mock_session


def test_generate_key_format():
    key = generate_key()
    assert key.startswith("fmcp_")
    assert len(key) > 12


def test_generate_key_uniqueness():
    keys = {generate_key() for _ in range(100)}
    assert len(keys) == 100


def test_hash_key_is_sha256():
    key = "fmcp_testkey"
    h = hash_token(key)
    expected = hashlib.sha256(key.encode()).hexdigest()
    assert h == expected


def test_key_prefix_returns_first_12_chars():
    key = "fmcp_abcdefghijklmnop"
    assert _key_prefix(key) == "fmcp_abcdefg"


@pytest.mark.asyncio
async def test_create_api_key_returns_full_key():
    mock_key_obj = MagicMock()
    mock_key_obj.id = 1
    mock_key_obj.to_dict.return_value = {"id": 1, "name": "test", "scope": "read", "key_prefix": "fmcp_xxx"}

    with patch("grimoire.services.api_keys.async_session") as mock_session_ctx:
        mock_session = make_mock_session()
        mock_session.add = MagicMock()
        mock_session.commit = AsyncMock()
        mock_session.refresh = AsyncMock(side_effect=lambda obj: None)
        # Return our fake object from refresh
        mock_session.refresh = AsyncMock()
        mock_key_obj.to_dict.return_value = {"id": 1, "name": "test", "scope": "read", "key_prefix": "fmcp_xxx"}
        mock_session_ctx.return_value = mock_session

        # Patch the select result
        with patch("grimoire.services.api_keys.ApiKey") as mock_model:
            mock_model.return_value = mock_key_obj
            full_key, key_dict = await create_api_key(user_id=1, name="test", scope="read")

    assert full_key.startswith("fmcp_")


@pytest.mark.asyncio
async def test_lookup_key_returns_none_for_unknown():
    with patch("grimoire.services.api_keys.async_session") as mock_session_ctx:
        mock_session = make_mock_session()
        mock_result = MagicMock()
        mock_result.scalars.return_value.first.return_value = None
        mock_session.execute = AsyncMock(return_value=mock_result)
        mock_session_ctx.return_value = mock_session

        result = await lookup_key("fmcp_doesnotexist")

    assert result is None


def test_hash_key_deterministic():
    key = "fmcp_some_test_key_value"
    assert hash_token(key) == hash_token(key)


@pytest.mark.asyncio
async def test_scope_validation():
    with pytest.raises(ValueError, match="scope must be"):
        await create_api_key(1, "bad", "superadmin")


# --- Auth middleware tests ---

@pytest.mark.asyncio
async def test_read_only_key_blocked_on_post():
    """Read-only API key returns 403 on non-GET requests."""
    from unittest.mock import AsyncMock, MagicMock, patch
    from quart import Quart

    fake_key = MagicMock()
    fake_key.user_id = 7
    fake_key.scope = "read"

    fake_user = MagicMock()
    fake_user.role = "user"

    app = Quart(__name__)

    async def dummy():
        from quart import jsonify
        return jsonify({"ok": True})

    from grimoire.auth import _check_auth
    wrapped = _check_auth(dummy)

    async with app.test_request_context(
        "/test", method="POST",
        headers={"Authorization": "Bearer fmcp_readonly"},
    ):
        with patch("grimoire.auth.lookup_key", AsyncMock(return_value=fake_key)), \
             patch("grimoire.auth.get_user_by_id", AsyncMock(return_value=fake_user)):
            resp = await wrapped()
    assert resp[1] == 403


@pytest.mark.asyncio
async def test_bearer_token_valid_write_key_passes():
    """Valid write-scoped bearer token passes auth on POST."""
    from unittest.mock import AsyncMock, MagicMock, patch
    from quart import Quart, g

    fake_key = MagicMock()
    fake_key.user_id = 7
    fake_key.scope = "write"

    fake_user = MagicMock()
    fake_user.role = "user"
    fake_user.id = 7

    app = Quart(__name__)

    result_holder = {}

    async def dummy():
        result_holder["user"] = g.user
        result_holder["api_key"] = g.api_key
        from quart import jsonify
        return jsonify({"ok": True})

    from grimoire.auth import _check_auth
    wrapped = _check_auth(dummy)

    async with app.test_request_context(
        "/test", method="POST",
        headers={"Authorization": "Bearer fmcp_writekey"},
    ):
        with patch("grimoire.auth.lookup_key", AsyncMock(return_value=fake_key)), \
             patch("grimoire.auth.get_user_by_id", AsyncMock(return_value=fake_user)):
            resp = await wrapped()

    # Should not be a tuple (no error response)
    assert not isinstance(resp, tuple) or resp[1] == 200
    assert result_holder.get("user") is fake_user
    assert result_holder.get("api_key") is fake_key