test-int/conftest.py
Rama main · réplica de lectura del repositorio original.
"""
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, Generator, Literal
import pytest
import pytest_asyncio
from fake_embeddings import use_fake_embeddings
from postgres_test_server import configured_postgres_sync_url
from pathlib import Path
from alembic import command
from alembic.config import Config
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.pool import NullPool
from testcontainers.postgres import PostgresContainer
from httpx import AsyncClient, ASGITransport
import basic_memory.api.container as api_container_module
from basic_memory import db
from basic_memory.config import (
BasicMemoryConfig,
ProjectConfig,
ProjectEntry,
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.postgres_search_repository import PostgresSearchRepository
from basic_memory.repository.project_repository import ProjectRepository
from basic_memory.repository.search_repository import create_search_repository
from fastapi import FastAPI
from basic_memory.deps import get_engine_factory, get_app_config
# Import MCP tools so they're available for testing
from basic_memory.mcp import tools # noqa: F401
from basic_memory.mcp.server import mcp as mcp_instance, set_project_delete_visibility
# =============================================================================
# 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(autouse=True)
def _fake_embeddings(request, monkeypatch):
"""Use the deterministic test embedder; see tests/fake_embeddings.py."""
use_fake_embeddings(request, monkeypatch)
@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" or _configured_postgres_sync_url():
yield None
return
# Use pgvector image so CREATE EXTENSION vector succeeds in search repository
with PostgresContainer("pgvector/pgvector:pg16") as postgres:
yield postgres
@pytest_asyncio.fixture(autouse=True)
async def cleanup_global_db_after_test() -> AsyncGenerator[None, None]:
"""Close any module-level DB engine created outside fixture ownership."""
yield
# Trigger: integration tests invoke CLI/MCP routes through the production
# client fallback, bypassing this file's engine_factory fixture.
# Why: those fallback engines live in basic_memory.db module state and can
# otherwise leave a non-daemon aiosqlite worker alive after pytest finishes.
# Outcome: every test boundary becomes a cleanup point for fallback engines.
from basic_memory import db
await db.shutdown_db()
@pytest.fixture(autouse=True)
def clean_routing_env(monkeypatch) -> None:
"""Keep CLI routing env mutations from leaking between integration tests."""
# Trigger: CLI integration tests exercise long-running MCP entrypoints that set routing env.
# Why: those commands normally own the process lifetime, but pytest keeps reusing it.
# Outcome: every integration test starts from neutral routing unless it opts in explicitly.
monkeypatch.delenv("BASIC_MEMORY_FORCE_LOCAL", raising=False)
monkeypatch.delenv("BASIC_MEMORY_FORCE_CLOUD", raising=False)
monkeypatch.delenv("BASIC_MEMORY_EXPLICIT_ROUTING", raising=False)
@pytest.fixture(autouse=True)
def clean_api_container(monkeypatch) -> None:
"""Keep an API lifespan's container from leaking into later tests.
The API lifespan installs its container in a module global and never removes
it, which a real process can rely on because the lifespan owns it. In one
pytest process the leaked container would make every later local ASGI client
look like it runs inside an API host, so it would not open the standalone
read cache a CLI process opens.
"""
monkeypatch.setattr(api_container_module, "_container", None)
@pytest.fixture(autouse=True)
def isolate_data_dir_env(monkeypatch) -> None:
"""Keep host data-dir env vars from leaking into integration tests.
Why: GitHub Actions Ubuntu runners set ``XDG_CONFIG_HOME=/home/runner/.config``,
and ``resolve_data_dir()`` honors it ahead of ``Path.home() / ".basic-memory"``.
Without clearing it, the MCP tool process reads config.json from the host XDG
path instead of the tmp dir the ``config_manager`` fixture wrote to — so
``test-project`` is missing from ``config.projects``, ``get_project_mode``
falls through to its CLOUD default (#837), and every tool call fails with
"Cloud routing requested but no credentials found."
"""
monkeypatch.delenv("BASIC_MEMORY_CONFIG_DIR", raising=False)
monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
POSTGRES_EPHEMERAL_TABLES = [
"search_vector_embeddings",
"search_vector_chunks",
"search_vector_index",
]
def _configured_postgres_sync_url() -> str | None:
"""Prefer an externally managed Postgres server when CI provides one."""
return configured_postgres_sync_url()
def _postgres_reset_tables() -> list[str]:
"""Resolve the current ORM table set at reset time."""
return [table.name for table in Base.metadata.sorted_tables] + ["search_index"]
def _resolve_postgres_sync_url(postgres_container) -> str:
"""Use CI's shared service when configured, otherwise fall back to testcontainers."""
configured_url = _configured_postgres_sync_url()
if configured_url:
return configured_url
assert postgres_container is not None
return postgres_container.get_connection_url()
def _postgres_alembic_config(async_url: str) -> Config:
"""Build Alembic config for stamping the shared Postgres integration schema."""
alembic_dir = Path(db.__file__).parent / "alembic"
cfg = Config()
cfg.set_main_option("script_location", str(alembic_dir))
cfg.set_main_option(
"file_template",
"%%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s",
)
cfg.set_main_option("timezone", "UTC")
cfg.set_main_option("revision_environment", "false")
cfg.set_main_option("sqlalchemy.url", async_url)
return cfg
async def _reset_postgres_integration_schema(engine: AsyncEngine, async_url: str) -> None:
"""Restore the shared Postgres integration schema to a clean baseline."""
from basic_memory.models.search import (
CREATE_POSTGRES_SEARCH_INDEX_FTS,
CREATE_POSTGRES_SEARCH_INDEX_FTS_CHUNKS_INDEX,
CREATE_POSTGRES_SEARCH_INDEX_FTS_CHUNKS_TABLE,
CREATE_POSTGRES_SEARCH_INDEX_METADATA,
CREATE_POSTGRES_SEARCH_INDEX_PERMALINK,
CREATE_POSTGRES_SEARCH_INDEX_TABLE,
)
async with engine.begin() as conn:
# Trigger: integration tests may leave behind temporary search/vector tables while
# exercising full-stack recovery paths.
# Why: recreating only the missing schema is much cheaper than dropping every table.
# Outcome: each integration test gets the same baseline without paying repeated full DDL cost.
await conn.run_sync(Base.metadata.create_all)
await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_TABLE)
await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_FTS)
await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_FTS_CHUNKS_TABLE)
await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_FTS_CHUNKS_INDEX)
await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_METADATA)
await conn.execute(CREATE_POSTGRES_SEARCH_INDEX_PERMALINK)
for table_name in POSTGRES_EPHEMERAL_TABLES:
await conn.execute(text(f"DROP TABLE IF EXISTS {table_name} CASCADE"))
await conn.execute(
text(f"TRUNCATE TABLE {', '.join(_postgres_reset_tables())} RESTART IDENTITY CASCADE")
)
alembic_version_exists = (
await conn.execute(text("SELECT to_regclass('public.alembic_version')"))
).scalar() is not None
if not alembic_version_exists:
command.stamp(_postgres_alembic_config(async_url), "head")
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def postgres_engine(
db_backend: Literal["sqlite", "postgres"], postgres_container
) -> AsyncGenerator[AsyncEngine | None, None]:
"""Create the shared Postgres engine once per integration test session."""
if db_backend != "postgres":
yield None
return
sync_url = _resolve_postgres_sync_url(postgres_container)
async_url = sync_url.replace("postgresql+psycopg2", "postgresql+asyncpg")
engine = create_async_engine(
async_url,
echo=False,
poolclass=NullPool,
)
try:
yield engine
finally:
await engine.dispose()
@pytest_asyncio.fixture
async def engine_factory(
app_config,
config_manager,
db_backend: Literal["sqlite", "postgres"],
postgres_container,
postgres_engine,
tmp_path,
) -> AsyncGenerator[
tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
None,
]:
"""Create engine and session factory for the configured database backend."""
if db_backend == "postgres":
assert postgres_engine is not None
# Trigger: full-stack MCP/CLI tests exercise sync/indexing code that can
# recover from DB errors by rolling back and opening later scoped sessions.
# Why: one savepoint-backed connection is too brittle for that flow.
# Outcome: reuse the engine, but reset rows/schema before each test and
# let app code use normal transaction boundaries.
async_url = postgres_engine.url.render_as_string(hide_password=False)
await _reset_postgres_integration_schema(postgres_engine, async_url)
session_maker = async_sessionmaker(
bind=postgres_engine,
class_=AsyncSession,
expire_on_commit=False,
autoflush=False,
)
# Semantic search is always on, and production creates vector storage (the
# chunk manifest and pgvector tables) at database initialization. The reset
# above drops it, so run that same initialization here. Vector storage is
# shared by every project, so any valid project id builds it.
await PostgresSearchRepository(
session_maker, project_id=1, app_config=app_config
).init_search_index()
# 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 = postgres_engine
db._session_maker = session_maker
try:
yield postgres_engine, session_maker
finally:
# Clean up module-level state
if db._engine is postgres_engine:
db._engine = None
if db._session_maker is session_maker:
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)
# create_all made search_index a plain table from the ORM model; drop it so
# the search repository creates the FTS5 virtual table instead.
async with db.scoped_session(session_maker) as session:
await session.execute(text("DROP TABLE IF EXISTS search_index"))
await session.commit()
# Trigger: production initializes search storage (the FTS5 table and the
# vector storage) once, when the database is opened, before any request.
# Why: left to the first embedding, the vector-table DDL runs while other
# connections search; each schema change makes those connections reload
# the schema and reconnect search_index, and a reconnect that overlaps
# the DDL fails with "vtable constructor failed: search_index" (#1955).
# Outcome: tests open the database the way production does.
await create_search_repository(
session_maker, project_id=1, app_config=app_config
).init_search_index()
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,
}
_, session_maker = engine_factory
project_repository = ProjectRepository()
from basic_memory import db
async with db.scoped_session(session_maker) as session:
project = await project_repository.create(session, project_data)
return project
@pytest.fixture
def config_home(tmp_path, monkeypatch) -> Path:
# Patch both HOME and USERPROFILE so Path.home() returns the test dir on
# every platform — Path.home() reads HOME on POSIX and USERPROFILE on
# Windows, and ConfigManager.data_dir_path now goes through Path.home()
# via resolve_data_dir(). Must mirror tests/conftest.py:config_home.
monkeypatch.setenv("HOME", str(tmp_path))
if os.name == "nt":
monkeypatch.setenv("USERPROFILE", 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": ProjectEntry(path=str(config_home))}
# Configure database backend based on env var
if db_backend == "postgres":
database_backend = DatabaseBackend.POSTGRES
# Trigger: CI jobs can provide a shared Postgres service instead of per-session containers.
# Why: reusing one pgvector-enabled server avoids Docker startup churn on every job.
# Outcome: local runs keep using testcontainers, while CI injects a stable service URL.
sync_url = _resolve_postgres_sync_url(postgres_container)
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",
update_permalinks_on_move=True,
index_changes=False, # Disable file indexing 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_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = 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
) -> Generator[FastAPI, None, None]:
"""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
previous_overrides = dict(app.dependency_overrides)
app.dependency_overrides[get_engine_factory] = lambda: engine_factory
app.dependency_overrides[get_app_config] = lambda: app_config
try:
yield app
finally:
# Restore overrides so one test's injected dependencies don't leak into
# subsequent tests that use the same global FastAPI app instance.
app.dependency_overrides = previous_overrides
@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
_, session_maker = engine_factory
# Use factory function to create appropriate search repository
search_repository = create_search_repository(
session_maker, project_id=test_project.id, app_config=app_config
)
entity_repository = EntityRepository(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,
session_maker=session_maker,
)
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 resources to register them
import basic_memory.mcp.resources # noqa: F401
import basic_memory.mcp.resources.skills # noqa: F401
# Import prompts to register them
import basic_memory.mcp.prompts # noqa: F401
return server
@pytest.fixture
def project_delete_enabled(config_manager: ConfigManager) -> Generator[None, None, None]:
"""Turn on enable_project_delete, so delete_project and bm_project_delete are listed.
The server lifespan applies the flag when a client connects. Visibility marks
outlive the test on the shared server, so the teardown hides the tools again.
"""
config = config_manager.load_config()
config.enable_project_delete = True
config_manager.save_config(config)
yield
set_project_delete_visibility(mcp_instance, False)
@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
@pytest_asyncio.fixture
async def indexed_project(client: AsyncClient, test_project: Project) -> Project:
"""Complete an initial pass for tests asserting honest misses after reads/deletes."""
from basic_memory.mcp.clients.project import ProjectClient
await ProjectClient(client).index(test_project.external_id, run_in_background=False)
return test_project