basicmachines-co-basic-memory/tests/services/test_project_removal_bug.py

"""Test for project removal bug #254."""

import os
import tempfile
from datetime import timezone, datetime
from pathlib import Path

import pytest
from sqlalchemy import text

from basic_memory import db
from basic_memory.services.project_service import ProjectService


@pytest.mark.asyncio
async def test_remove_project_with_related_entities(project_service: ProjectService):
    """Test removing a project that has related entities (reproduces issue #254).

    This test verifies that projects with related entities (entities, observations, relations)
    can be properly deleted without foreign key constraint violations.

    The bug was caused by missing foreign key constraints with CASCADE DELETE after
    the project table was recreated in migration 647e7a75e2cd.
    """
    test_project_name = f"test-remove-with-entities-{os.urandom(4).hex()}"
    with tempfile.TemporaryDirectory() as temp_dir:
        test_root = Path(temp_dir)
        test_project_path = str(test_root / "test-remove-with-entities")

        # Make sure the test directory exists
        os.makedirs(test_project_path, exist_ok=True)

        try:
            # Step 1: Add the test project
            await project_service.add_project(test_project_name, test_project_path)

            # Verify project exists
            project = await project_service.get_project(test_project_name)
            assert project is not None

            # Step 2: Create related entities for this project
            from basic_memory.repository.entity_repository import EntityRepository

            entity_repo = EntityRepository(
                project_service.repository.session_maker, project_id=project.id
            )

            entity_data = {
                "title": "Test Entity for Deletion",
                "note_type": "note",
                "content_type": "text/markdown",
                "project_id": project.id,
                "permalink": "test-deletion-entity",
                "file_path": "test-deletion-entity.md",
                "checksum": "test123",
                "created_at": datetime.now(timezone.utc),
                "updated_at": datetime.now(timezone.utc),
            }
            entity = await entity_repo.create(entity_data)
            assert entity is not None

            # Step 3: Create observations for the entity
            from basic_memory.repository.observation_repository import ObservationRepository

            obs_repo = ObservationRepository(
                project_service.repository.session_maker, project_id=project.id
            )

            observation_data = {
                "entity_id": entity.id,
                "content": "This is a test observation",
                "category": "note",
            }
            observation = await obs_repo.create(observation_data)
            assert observation is not None

            # Step 4: Create relations involving the entity
            from basic_memory.repository.relation_repository import RelationRepository

            rel_repo = RelationRepository(
                project_service.repository.session_maker, project_id=project.id
            )

            relation_data = {
                "from_id": entity.id,
                "to_name": "some-target-entity",
                "relation_type": "relates-to",
            }
            relation = await rel_repo.create(relation_data)
            assert relation is not None

            # Step 5: Attempt to remove the project
            # This should work with proper cascade delete, or fail with foreign key constraint
            await project_service.remove_project(test_project_name)

            # Step 6: Verify everything was properly deleted

            # Project should be gone
            removed_project = await project_service.get_project(test_project_name)
            assert removed_project is None, "Project should have been removed"

            # Related entities should be cascade deleted
            remaining_entity = await entity_repo.find_by_id(entity.id)
            assert remaining_entity is None, "Entity should have been cascade deleted"

            # Observations should be cascade deleted
            remaining_obs = await obs_repo.find_by_id(observation.id)
            assert remaining_obs is None, "Observation should have been cascade deleted"

            # Relations should be cascade deleted
            remaining_rel = await rel_repo.find_by_id(relation.id)
            assert remaining_rel is None, "Relation should have been cascade deleted"

        except Exception as e:
            # Check if this is the specific foreign key constraint error from the bug report
            if "FOREIGN KEY constraint failed" in str(e):
                pytest.fail(
                    f"Bug #254 reproduced: {e}. "
                    "This indicates missing foreign key constraints with CASCADE DELETE. "
                    "Run migration a1b2c3d4e5f6_fix_project_foreign_keys.py to fix this."
                )
            else:
                # Re-raise other unexpected errors
                raise e

        finally:
            # Clean up - remove project if it still exists
            if test_project_name in project_service.projects:
                try:
                    await project_service.remove_project(test_project_name)
                except Exception:
                    # Manual cleanup if remove_project fails
                    try:
                        project_service.config_manager.remove_project(test_project_name)
                    except Exception:
                        pass

                    project = await project_service.get_project(test_project_name)
                    if project:
                        await project_service.repository.delete(project.id)


async def _table_exists(session_maker, table: str) -> bool:
    """Return True if the named table is present on the current connection."""
    from sqlalchemy import inspect as sa_inspect

    async with db.scoped_session(session_maker) as session:
        return await session.run_sync(
            lambda sync_session: table in sa_inspect(sync_session.connection()).get_table_names()
        )


@pytest.mark.asyncio
async def test_remove_project_purges_search_rows(project_service: ProjectService):
    """Project deletion must sweep the derived search tables.

    SQLite stores search_index as an FTS5 virtual table, which cannot carry a
    foreign key, so without an explicit purge the FTS rows survive the project
    and leak into the next project that reuses the same auto-increment id.
    Postgres has the cascade FK, but we expect the same end-state on either
    backend. This test fails on the pre-fix code: search_index still holds the
    project's rows after remove_project completes.
    """
    test_project_name = f"test-search-cleanup-{os.urandom(4).hex()}"
    with tempfile.TemporaryDirectory() as temp_dir:
        test_project_path = str(Path(temp_dir) / "test-search-cleanup")
        os.makedirs(test_project_path, exist_ok=True)

        await project_service.add_project(test_project_name, test_project_path)
        project = await project_service.get_project(test_project_name)
        assert project is not None
        project_id = project.id

        # Seed both derived tables directly. The bug is in the cleanup path,
        # not the indexer, so a synthetic row is enough to prove the sweep.
        async with db.scoped_session(project_service.repository.session_maker) as session:
            await session.execute(
                text(
                    "INSERT INTO search_index "
                    "(id, title, content_stems, content_snippet, permalink, "
                    " file_path, type, project_id) "
                    "VALUES (:id, :title, :stems, :snippet, :permalink, "
                    " :file_path, :type, :project_id)"
                ),
                {
                    "id": 999_001,
                    "title": "leak canary",
                    "stems": "leak canary",
                    "snippet": "leak canary",
                    "permalink": f"leak-canary-{project_id}",
                    "file_path": "leak-canary.md",
                    "type": "entity",
                    "project_id": project_id,
                },
            )
            await session.execute(
                text(
                    "INSERT INTO search_vector_chunks "
                    "(entity_id, project_id, chunk_key, chunk_text, source_hash, "
                    " entity_fingerprint, embedding_model) "
                    "VALUES (:entity_id, :project_id, :chunk_key, :chunk_text, "
                    " :source_hash, :entity_fingerprint, :embedding_model)"
                ),
                {
                    "entity_id": 999_001,
                    "project_id": project_id,
                    "chunk_key": "canary",
                    "chunk_text": "leak canary",
                    "source_hash": "abc",
                    "entity_fingerprint": "",
                    "embedding_model": "",
                },
            )

        async with db.scoped_session(project_service.repository.session_maker) as session:
            pre_index = (
                await session.execute(
                    text("SELECT COUNT(*) FROM search_index WHERE project_id = :pid"),
                    {"pid": project_id},
                )
            ).scalar_one()
            pre_chunks = (
                await session.execute(
                    text("SELECT COUNT(*) FROM search_vector_chunks WHERE project_id = :pid"),
                    {"pid": project_id},
                )
            ).scalar_one()
        assert pre_index >= 1, "seed row should exist before removal"
        assert pre_chunks >= 1, "seed chunk should exist before removal"

        await project_service.remove_project(test_project_name)

        async with db.scoped_session(project_service.repository.session_maker) as session:
            post_index = (
                await session.execute(
                    text("SELECT COUNT(*) FROM search_index WHERE project_id = :pid"),
                    {"pid": project_id},
                )
            ).scalar_one()
            post_chunks = (
                await session.execute(
                    text("SELECT COUNT(*) FROM search_vector_chunks WHERE project_id = :pid"),
                    {"pid": project_id},
                )
            ).scalar_one()

        assert post_index == 0, (
            f"search_index still has {post_index} rows for deleted project_id={project_id} "
            "— project deletion did not sweep the FTS table."
        )
        assert post_chunks == 0, (
            f"search_vector_chunks still has {post_chunks} rows for deleted "
            f"project_id={project_id}."
        )


@pytest.mark.asyncio
async def test_delete_returns_false_for_missing_project_id(project_service: ProjectService):
    """ProjectRepository.delete must return False when the project id is gone.

    The override loses the base Repository.delete contract if the NoResultFound
    branch isn't covered — a silent True would mislead callers into thinking
    a non-existent project was removed.
    """
    result = await project_service.repository.delete(9_999_999)
    assert result is False


@pytest.mark.asyncio
async def test_remove_project_purges_vector_embeddings(project_service: ProjectService):
    """Project deletion must also drop sqlite-vec embeddings keyed by chunk rowid.

    sqlite-vec stores vectors in a vec0 virtual table that has no cascade
    behavior. If embeddings linger after the chunks they reference are gone,
    `_run_vector_query` pulls them as top-k candidates and crowds out live
    results. The test only runs when the embeddings table is present, which
    matches the install path that exercises semantic search.
    """
    test_project_name = f"test-vec-cleanup-{os.urandom(4).hex()}"
    session_maker = project_service.repository.session_maker

    # The embeddings table only exists once semantic search has initialized.
    # Skipping when it's absent keeps this test honest on minimal CI DBs.
    if not await _table_exists(session_maker, "search_vector_embeddings"):
        pytest.skip("search_vector_embeddings is not present on this connection")

    with tempfile.TemporaryDirectory() as temp_dir:
        test_project_path = str(Path(temp_dir) / "test-vec-cleanup")
        os.makedirs(test_project_path, exist_ok=True)

        await project_service.add_project(test_project_name, test_project_path)
        project = await project_service.get_project(test_project_name)
        assert project is not None
        project_id = project.id

        async with db.scoped_session(session_maker) as session:
            await session.execute(
                text(
                    "INSERT INTO search_vector_chunks "
                    "(id, entity_id, project_id, chunk_key, chunk_text, source_hash, "
                    " entity_fingerprint, embedding_model) "
                    "VALUES (:id, :entity_id, :project_id, :chunk_key, :chunk_text, "
                    " :source_hash, :entity_fingerprint, :embedding_model)"
                ),
                {
                    "id": 999_201,
                    "entity_id": 999_201,
                    "project_id": project_id,
                    "chunk_key": "vec-canary",
                    "chunk_text": "vec canary",
                    "source_hash": "abc",
                    "entity_fingerprint": "",
                    "embedding_model": "",
                },
            )
            # vec0 requires a vector matching the configured dimensions, but the
            # delete path filters by rowid; a non-existing dimension would block
            # this seed step. Skip the insert if the embeddings DDL hasn't run.
            try:
                await session.execute(
                    text(
                        "INSERT INTO search_vector_embeddings (rowid, embedding) "
                        "VALUES (:rowid, :embedding)"
                    ),
                    {"rowid": 999_201, "embedding": "[" + ",".join(["0.0"] * 384) + "]"},
                )
            except Exception:
                pytest.skip("search_vector_embeddings rejected the synthetic seed row")

        async with db.scoped_session(session_maker) as session:
            pre = (
                await session.execute(
                    text("SELECT COUNT(*) FROM search_vector_embeddings WHERE rowid = :rowid"),
                    {"rowid": 999_201},
                )
            ).scalar_one()
        assert pre >= 1, "seed embedding should exist before removal"

        await project_service.remove_project(test_project_name)

        async with db.scoped_session(session_maker) as session:
            post = (
                await session.execute(
                    text("SELECT COUNT(*) FROM search_vector_embeddings WHERE rowid = :rowid"),
                    {"rowid": 999_201},
                )
            ).scalar_one()

        assert post == 0, (
            f"search_vector_embeddings still has {post} rows for rowid 999_201 "
            "— project deletion did not sweep the sqlite-vec embeddings table."
        )