"""Tests for Mem.ai API client."""
import os
from uuid import uuid4
import httpx
import pytest
from pytest_httpx import HTTPXMock
from mcp_mem.client import (
MemAPIError,
MemAuthenticationError,
MemClient,
MemNotFoundError,
MemValidationError,
)
from mcp_mem.models import Collection, Note, RequestResponse
@pytest.fixture
def mock_api_key(monkeypatch):
"""Set a mock API key for testing."""
monkeypatch.setenv("MEM_API_KEY", "test_api_key_12345")
@pytest.fixture
async def client(mock_api_key):
"""Create a test client."""
async with MemClient(api_key="test_api_key") as client:
yield client
class TestMemClient:
"""Test suite for MemClient."""
def test_client_init_without_api_key(self, monkeypatch):
"""Test that client raises error without API key."""
monkeypatch.delenv("MEM_API_KEY", raising=False)
with pytest.raises(MemAuthenticationError, match="MEM_API_KEY"):
MemClient()
def test_client_init_with_api_key(self, mock_api_key):
"""Test client initialization with API key."""
client = MemClient(api_key="test_key")
assert client.api_key == "test_key"
assert client.base_url == "https://api.mem.ai/v2"
def test_client_init_with_custom_base_url(self, mock_api_key):
"""Test client initialization with custom base URL."""
custom_url = "https://custom.api.com/v1"
client = MemClient(api_key="test_key", base_url=custom_url)
assert client.base_url == custom_url
@pytest.mark.asyncio
async def test_mem_it_success(self, client, httpx_mock: HTTPXMock):
"""Test successful Mem It request."""
request_id = "api-request-123"
httpx_mock.add_response(
method="POST",
url="https://api.mem.ai/v2/mem-it",
json={"request_id": request_id},
status_code=200,
)
response = await client.mem_it(
input="Test content",
instructions="Save this",
context="Testing",
)
assert isinstance(response, RequestResponse)
assert response.request_id == request_id
@pytest.mark.asyncio
async def test_mem_it_authentication_error(self, client, httpx_mock: HTTPXMock):
"""Test Mem It with authentication error."""
httpx_mock.add_response(
method="POST",
url="https://api.mem.ai/v2/mem-it",
status_code=401,
text="Unauthorized",
)
with pytest.raises(MemAuthenticationError):
await client.mem_it(input="Test content")
@pytest.mark.asyncio
async def test_create_note_success(self, client, httpx_mock: HTTPXMock):
"""Test successful note creation."""
note_id = uuid4()
httpx_mock.add_response(
method="POST",
url="https://api.mem.ai/v2/notes",
json={
"id": str(note_id),
"title": "Test Note",
"content": "# Test\n\nContent here",
"collection_ids": [],
"created_at": "2024-01-15T10:00:00Z",
"updated_at": "2024-01-15T10:00:00Z",
"request_id": "req-123",
},
status_code=200,
)
note = await client.create_note(content="# Test\n\nContent here")
assert isinstance(note, Note)
assert note.id == note_id
assert note.title == "Test Note"
assert note.content == "# Test\n\nContent here"
@pytest.mark.asyncio
async def test_create_note_validation_error(self, client, httpx_mock: HTTPXMock):
"""Test note creation with validation error."""
httpx_mock.add_response(
method="POST",
url="https://api.mem.ai/v2/notes",
status_code=422,
text="Validation failed",
)
with pytest.raises(MemValidationError):
await client.create_note(content="Test")
@pytest.mark.asyncio
async def test_read_note_success(self, client, httpx_mock: HTTPXMock):
"""Test successful note reading."""
note_id = uuid4()
httpx_mock.add_response(
method="GET",
url=f"https://api.mem.ai/v2/notes/{note_id}",
json={
"id": str(note_id),
"title": "Test Note",
"content": "Content",
"collection_ids": [],
"created_at": "2024-01-15T10:00:00Z",
"updated_at": "2024-01-15T10:00:00Z",
},
status_code=200,
)
note = await client.read_note(str(note_id))
assert isinstance(note, Note)
assert note.id == note_id
assert note.title == "Test Note"
@pytest.mark.asyncio
async def test_read_note_not_found(self, client, httpx_mock: HTTPXMock):
"""Test reading non-existent note."""
note_id = uuid4()
httpx_mock.add_response(
method="GET",
url=f"https://api.mem.ai/v2/notes/{note_id}",
status_code=404,
text="Not found",
)
with pytest.raises(MemNotFoundError):
await client.read_note(str(note_id))
@pytest.mark.asyncio
async def test_delete_note_success(self, client, httpx_mock: HTTPXMock):
"""Test successful note deletion."""
note_id = uuid4()
request_id = "req-delete-123"
httpx_mock.add_response(
method="DELETE",
url=f"https://api.mem.ai/v2/notes/{note_id}",
json={"request_id": request_id},
status_code=200,
)
response = await client.delete_note(str(note_id))
assert isinstance(response, RequestResponse)
assert response.request_id == request_id
@pytest.mark.asyncio
async def test_create_collection_success(self, client, httpx_mock: HTTPXMock):
"""Test successful collection creation."""
collection_id = uuid4()
httpx_mock.add_response(
method="POST",
url="https://api.mem.ai/v2/collections",
json={
"id": str(collection_id),
"title": "Test Collection",
"description": "A test collection",
"created_at": "2024-01-15T10:00:00Z",
"updated_at": "2024-01-15T10:00:00Z",
},
status_code=200,
)
collection = await client.create_collection(
title="Test Collection",
description="A test collection",
)
assert isinstance(collection, Collection)
assert collection.id == collection_id
assert collection.title == "Test Collection"
@pytest.mark.asyncio
async def test_delete_collection_success(self, client, httpx_mock: HTTPXMock):
"""Test successful collection deletion."""
collection_id = uuid4()
request_id = "req-delete-coll-123"
httpx_mock.add_response(
method="DELETE",
url=f"https://api.mem.ai/v2/collections/{collection_id}",
json={"request_id": request_id},
status_code=200,
)
response = await client.delete_collection(str(collection_id))
assert isinstance(response, RequestResponse)
assert response.request_id == request_id
@pytest.mark.asyncio
async def test_context_manager(self, mock_api_key):
"""Test client as async context manager."""
async with MemClient(api_key="test_key") as client:
assert client.api_key == "test_key"
assert client.client is not None
# Client should be closed after context
assert client.client.is_closed
