basicmachines-co-basic-memory/test-int/conftest.py

"""
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.
"""

import os
from typing import AsyncGenerator, Literal

import pytest
import pytest_asyncio
from pathlib import Path
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool
from testcontainers.postgres import PostgresContainer

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.models.base import Base
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


# =============================================================================
# Database Backend Selection (env var approach)
# =============================================================================
# By default, integration tests run against SQLite.
# Set BASIC_MEMORY_TEST_POSTGRES=1 to run against Postgres (uses testcontainers).


@pytest.fixture(scope="session")
def db_backend() -> Literal["sqlite", "postgres"]:
    """Determine database backend from environment variable.

    Default: sqlite
    Set BASIC_MEMORY_TEST_POSTGRES=1 to use postgres
    """
    if os.environ.get("BASIC_MEMORY_TEST_POSTGRES", "").lower() in ("1", "true", "yes"):
        return "postgres"
    return "sqlite"


@pytest.fixture(scope="session")
def postgres_container(db_backend):
    """Session-scoped Postgres container for integration tests.

    Uses testcontainers to spin up a real Postgres instance.
    Only starts if db_backend is "postgres".
    """
    if db_backend != "postgres":
        yield None
        return

    with PostgresContainer("postgres:16-alpine") as postgres:
        yield postgres


@pytest_asyncio.fixture
async def engine_factory(
    app_config,
    config_manager,
    db_backend: Literal["sqlite", "postgres"],
    postgres_container,
    tmp_path,
) -> AsyncGenerator[tuple, None]:
    """Create engine and session factory for the configured database backend."""
    from basic_memory.models.search import (
        CREATE_SEARCH_INDEX,
        CREATE_POSTGRES_SEARCH_INDEX_TABLE,
        CREATE_POSTGRES_SEARCH_INDEX_FTS,
        CREATE_POSTGRES_SEARCH_INDEX_METADATA,
        CREATE_POSTGRES_SEARCH_INDEX_PERMALINK,
    )
    from basic_memory import db

    if db_backend == "postgres":
        # Postgres mode using testcontainers
        sync_url = postgres_container.get_connection_url()
        async_url = sync_url.replace("postgresql+psycopg2", "postgresql+asyncpg")

        engine = create_async_engine(
            async_url,
            echo=False,
            poolclass=NullPool,
        )

        session_maker = async_sessionmaker(
            bind=engine,
            class_=AsyncSession,
            expire_on_commit=False,
            autoflush=False,
        )

        # Set module-level state to prevent MCP lifespan from re-initializing
        # This ensures get_or_create_db() sees an existing engine and skips initialization
        db._engine = engine
        db._session_maker = session_maker

        # Drop and recreate all tables for test isolation
        async with engine.begin() as conn:
            await conn.execute(text("DROP TABLE IF EXISTS search_index CASCADE"))
            await conn.run_sync(Base.metadata.drop_all)
            await conn.run_sync(Base.metadata.create_all)
            # asyncpg requires separate execute calls for each statement
            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_TABLE)
            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_FTS)
            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_METADATA)
            await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_PERMALINK)

        yield engine, session_maker

        # Clean up module-level state
        await engine.dispose()
        db._engine = None
        db._session_maker = None

    else:
        # SQLite: Create fresh database (fast with tmp files)
        db_path = tmp_path / "test.db"
        db_type = DatabaseType.FILESYSTEM

        async with engine_session_factory(db_path, db_type) as (engine, session_maker):
            # Create all tables via ORM
            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"],
    postgres_container,
    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 env var
    if db_backend == "postgres":
        database_backend = DatabaseBackend.POSTGRES
        # Get URL from testcontainer and convert to asyncpg driver
        sync_url = postgres_container.get_connection_url()
        database_url = sync_url.replace("postgresql+psycopg2", "postgresql+asyncpg")
    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
        sync_changes=False,  # Disable file sync in tests - prevents lifespan from starting blocking task
        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