"""
Shared fixtures for integration tests.
Integration tests verify the complete flow: MCP Client → MCP Server → FastAPI → Database.
Unlike unit tests which use in-memory databases and mocks, integration tests use real SQLite
files and test the full application stack to ensure all components work together correctly.
## Architecture
The integration test setup creates this flow:
```
Test → MCP Client → MCP Server → HTTP Request (ASGITransport) → FastAPI App → Database
↑
Dependency overrides
point to test database
```
## Key Components
1. **Real SQLite Database**: Uses `DatabaseType.FILESYSTEM` with actual SQLite files
in temporary directories instead of in-memory databases.
2. **Shared Database Connection**: Both MCP server and FastAPI app use the same
database via dependency injection overrides.
3. **Project Session Management**: Initializes the MCP project session with test
project configuration so tools know which project to operate on.
4. **Search Index Initialization**: Creates the FTS5 search index tables that
the application requires for search functionality.
5. **Global Configuration Override**: Modifies the global `basic_memory_app_config`
so MCP tools use test project settings instead of user configuration.
## Usage
Integration tests should include both `mcp_server` and `app` fixtures to ensure
the complete stack is wired correctly:
```python
@pytest.mark.asyncio
async def test_my_mcp_tool(mcp_server, app):
async with Client(mcp_server) as client:
result = await client.call_tool("tool_name", {"param": "value"})
# Assert on results...
```
The `app` fixture ensures FastAPI dependency overrides are active, and
`mcp_server` provides the MCP server with proper project session initialization.
"""
from typing import AsyncGenerator, Literal
import pytest
import pytest_asyncio
from pathlib import Path
from sqlalchemy import text
from httpx import AsyncClient, ASGITransport
from basic_memory.config import BasicMemoryConfig, ProjectConfig, ConfigManager, DatabaseBackend
from basic_memory.db import engine_session_factory, DatabaseType
from basic_memory.models import Project
from basic_memory.repository.project_repository import ProjectRepository
from fastapi import FastAPI
from basic_memory.deps import get_project_config, get_engine_factory, get_app_config
# Import MCP tools so they're available for testing
from basic_memory.mcp import tools # noqa: F401
@pytest.fixture(
params=[
pytest.param("sqlite", id="sqlite"),
pytest.param("postgres", id="postgres", marks=pytest.mark.postgres),
]
)
def db_backend(request) -> Literal["sqlite", "postgres"]:
"""Parametrize tests to run against both SQLite and Postgres.
Usage:
pytest # Runs tests against SQLite only (default)
pytest -m postgres # Runs tests against Postgres only
pytest -m "not postgres" # Runs tests against SQLite only
pytest --run-all-backends # Runs tests against both backends
Note: Only tests that use database fixtures (engine_factory, session_maker, etc.)
will be parametrized. Tests that don't use the database won't be affected.
"""
return request.param
@pytest_asyncio.fixture
async def engine_factory(
app_config,
config_manager,
db_backend: Literal["sqlite", "postgres"],
tmp_path,
) -> AsyncGenerator[tuple, None]:
"""Create engine and session factory for the configured database backend."""
from basic_memory.models.search import CREATE_SEARCH_INDEX
from basic_memory import db
# Determine database type based on backend
if db_backend == "postgres":
db_type = DatabaseType.FILESYSTEM
else:
db_type = DatabaseType.FILESYSTEM # Integration tests use file-based SQLite
# Use tmp_path for SQLite, use config database_path for Postgres
if db_backend == "sqlite":
db_path = tmp_path / "test.db"
else:
db_path = app_config.database_path
if db_backend == "postgres":
# Postgres: Create fresh engine for each test with full schema reset
config_manager._config = app_config
# Use context manager to handle engine disposal properly
async with engine_session_factory(db_path, db_type) as (engine, session_maker):
# Drop and recreate schema for complete isolation
async with engine.begin() as conn:
await conn.execute(text("DROP SCHEMA IF EXISTS public CASCADE"))
await conn.execute(text("CREATE SCHEMA public"))
await conn.execute(text("GRANT ALL ON SCHEMA public TO basic_memory_user"))
await conn.execute(text("GRANT ALL ON SCHEMA public TO public"))
# Run migrations to create production tables
from basic_memory.db import run_migrations
await run_migrations(app_config, db_type)
yield engine, session_maker
else:
# SQLite: Create fresh database (fast with tmp files)
async with engine_session_factory(db_path, db_type) as (engine, session_maker):
# Create all tables via ORM
from basic_memory.models.base import Base
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Drop any SearchIndex ORM table, then create FTS5 virtual table
async with db.scoped_session(session_maker) as session:
await session.execute(text("DROP TABLE IF EXISTS search_index"))
await session.execute(CREATE_SEARCH_INDEX)
await session.commit()
yield engine, session_maker
@pytest_asyncio.fixture
async def test_project(config_home, engine_factory) -> Project:
"""Create a test project."""
project_data = {
"name": "test-project",
"description": "Project used for integration tests",
"path": str(config_home),
"is_active": True,
"is_default": True,
}
engine, session_maker = engine_factory
project_repository = ProjectRepository(session_maker)
project = await project_repository.create(project_data)
return project
@pytest.fixture
def config_home(tmp_path, monkeypatch) -> Path:
monkeypatch.setenv("HOME", str(tmp_path))
# Set BASIC_MEMORY_HOME to the test directory
monkeypatch.setenv("BASIC_MEMORY_HOME", str(tmp_path / "basic-memory"))
return tmp_path
@pytest.fixture
def app_config(
config_home, db_backend: Literal["sqlite", "postgres"], tmp_path, monkeypatch
) -> BasicMemoryConfig:
"""Create test app configuration."""
# Disable cloud mode for CLI tests
monkeypatch.setenv("BASIC_MEMORY_CLOUD_MODE", "false")
# Create a basic config with test-project like unit tests do
projects = {"test-project": str(config_home)}
# Configure database backend based on test parameter
if db_backend == "postgres":
database_backend = DatabaseBackend.POSTGRES
database_url = (
"postgresql+asyncpg://basic_memory_user:dev_password@localhost:5433/basic_memory_test"
)
else:
database_backend = DatabaseBackend.SQLITE
database_url = None
app_config = BasicMemoryConfig(
env="test",
projects=projects,
default_project="test-project",
default_project_mode=False, # Match real-world usage - tools must pass explicit project
update_permalinks_on_move=True,
cloud_mode=False, # Explicitly disable cloud mode
database_backend=database_backend,
database_url=database_url,
)
return app_config
@pytest.fixture
def config_manager(app_config: BasicMemoryConfig, config_home) -> ConfigManager:
# Invalidate config cache to ensure clean state for each test
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_manager = ConfigManager()
# Update its paths to use the test directory
config_manager.config_dir = config_home / ".basic-memory"
config_manager.config_file = config_manager.config_dir / "config.json"
config_manager.config_dir.mkdir(parents=True, exist_ok=True)
# Ensure the config file is written to disk
config_manager.save_config(app_config)
return config_manager
@pytest.fixture
def project_config(test_project):
"""Create test project configuration."""
project_config = ProjectConfig(
name=test_project.name,
home=Path(test_project.path),
)
return project_config
@pytest.fixture
def app(app_config, project_config, engine_factory, test_project, config_manager) -> FastAPI:
"""Create test FastAPI application with single project."""
# Import the FastAPI app AFTER the config_manager has written the test config to disk
# This ensures that when the app's lifespan manager runs, it reads the correct test config
from basic_memory.api.app import app as fastapi_app
app = fastapi_app
app.dependency_overrides[get_project_config] = lambda: project_config
app.dependency_overrides[get_engine_factory] = lambda: engine_factory
app.dependency_overrides[get_app_config] = lambda: app_config
return app
@pytest_asyncio.fixture
async def search_service(engine_factory, test_project, app_config):
"""Create and initialize search service for integration tests.
Uses app_config fixture to determine database backend - no patching needed.
"""
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.services.file_service import FileService
from basic_memory.services.search_service import SearchService
from basic_memory.markdown.markdown_processor import MarkdownProcessor
from basic_memory.markdown import EntityParser
from basic_memory.repository.search_repository import create_search_repository
engine, session_maker = engine_factory
# Use factory function to create appropriate search repository
search_repository = create_search_repository(session_maker, project_id=test_project.id)
entity_repository = EntityRepository(session_maker, project_id=test_project.id)
# Create file service
entity_parser = EntityParser(Path(test_project.path))
markdown_processor = MarkdownProcessor(entity_parser)
file_service = FileService(Path(test_project.path), markdown_processor)
# Create and initialize search service
service = SearchService(search_repository, entity_repository, file_service)
await service.init_search_index()
return service
@pytest.fixture
def mcp_server(config_manager, search_service):
# Import mcp instance
from basic_memory.mcp.server import mcp as server
# Import mcp tools to register them
import basic_memory.mcp.tools # noqa: F401
# Import prompts to register them
import basic_memory.mcp.prompts # noqa: F401
return server
@pytest_asyncio.fixture
async def client(app: FastAPI) -> AsyncGenerator[AsyncClient, None]:
"""Create test client that both MCP and tests will use."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
yield client