mempalace--mempalace/tests/conftest.py

"""
conftest.py — Shared fixtures for MemPalace tests.

Provides isolated palace and knowledge graph instances so tests never
touch the user's real data or leak temp files on failure.

HOME is redirected to a temp directory at module load time — before any
mempalace imports — so that module-level initialisations (e.g.
``_kg = KnowledgeGraph()`` in mcp_server) write to a throwaway location
instead of the real user profile.
"""

import os
import shutil
import tempfile

# ── Isolate HOME before any mempalace imports ──────────────────────────
_original_env = {}
_session_tmp = tempfile.mkdtemp(prefix="mempalace_session_")

for _var in ("HOME", "USERPROFILE", "HOMEDRIVE", "HOMEPATH"):
    _original_env[_var] = os.environ.get(_var)

os.environ["HOME"] = _session_tmp
os.environ["USERPROFILE"] = _session_tmp
os.environ["HOMEDRIVE"] = os.path.splitdrive(_session_tmp)[0] or "C:"
os.environ["HOMEPATH"] = os.path.splitdrive(_session_tmp)[1] or _session_tmp

# Now it is safe to import mempalace modules that trigger initialisation.
import chromadb  # noqa: E402
import pytest  # noqa: E402

from mempalace.config import MempalaceConfig  # noqa: E402
from mempalace.knowledge_graph import KnowledgeGraph  # noqa: E402

# Redirect ChromaDB's ONNX model cache back to the real user's cache so tests
# don't re-download the 79 MB model on every run. The HOME redirect above
# would otherwise point ONNXMiniLM_L6_V2.DOWNLOAD_PATH at the empty temp dir.
try:
    from pathlib import Path  # noqa: E402
    from chromadb.utils.embedding_functions.onnx_mini_lm_l6_v2 import (  # noqa: E402
        ONNXMiniLM_L6_V2,
    )

    _real_home = _original_env.get("USERPROFILE") or _original_env.get("HOME")
    if _real_home:
        _real_cache = Path(_real_home) / ".cache" / "chroma" / "onnx_models" / "all-MiniLM-L6-v2"
        if _real_cache.exists():
            ONNXMiniLM_L6_V2.DOWNLOAD_PATH = _real_cache
except ImportError:
    pass


@pytest.fixture(autouse=True)
def _reset_mcp_cache():
    """Reset cached MCP state between tests without importing mcp_server.

    If mempalace.mcp_server is already imported, close/clear its KG cache and
    Chroma client cache. If it has not been imported, leave it unloaded so
    fork/spawn-based tests do not inherit extra Chroma/SQLite state.
    """

    def _clear_cache():
        try:
            import sys

            mcp_server = sys.modules.get("mempalace.mcp_server")
            if mcp_server is not None:
                for kg in list(getattr(mcp_server, "_kg_by_path", {}).values()):
                    close = getattr(kg, "close", None)
                    if close is not None:
                        try:
                            close()
                        except Exception:
                            pass

                if hasattr(mcp_server, "_kg_by_path"):
                    mcp_server._kg_by_path.clear()

                # Close (not just dereference) the cached chromadb client so its
                # rust-side file handles are released; on Windows a bare deref
                # leaves them locked and leaks across the session (#1128).
                cached_client = getattr(mcp_server, "_client_cache", None)
                if cached_client is not None:
                    close = getattr(cached_client, "close", None)
                    if callable(close):
                        try:
                            close()
                        except Exception:
                            pass
                mcp_server._client_cache = None
                mcp_server._collection_cache = None
                if hasattr(mcp_server, "_collection_cache_backend"):
                    mcp_server._collection_cache_backend = None
                if hasattr(mcp_server, "_collection_cache_palace"):
                    mcp_server._collection_cache_palace = None
                if hasattr(mcp_server, "_collection_open_error"):
                    mcp_server._collection_open_error = None
        except AttributeError:
            pass

        try:
            # Reset the per-process quarantine gate so tests don't leak
            # state through ChromaBackend._quarantined_paths.
            from mempalace.backends.chroma import ChromaBackend

            ChromaBackend._quarantined_paths.clear()
        except (ImportError, AttributeError):
            pass

        # Release chromadb clients opened through the backend layer. Many tests
        # reach the store via palace.get_collection() (sweep, repair, CLI, ...),
        # which caches one PersistentClient per palace_path on the long-lived
        # backend singleton and never closes it. chromadb frees the rust-side
        # SQLite/HNSW file handles only on client.close(); on POSIX the open
        # handles are harmless, but on Windows they stay locked and accumulate
        # across the session until a later test's HNSW segment write fails
        # (#1128 Windows CI). close_palace() closes the client and drops the
        # handle without marking the backend closed, so it stays reusable.
        try:
            from mempalace import palace as _palace

            backend = getattr(_palace, "_DEFAULT_BACKEND", None)
            clients = getattr(backend, "_clients", None)
            if clients:
                for path in list(clients):
                    try:
                        backend.close_palace(path)
                    except Exception:
                        pass
        except (ImportError, AttributeError):
            pass

    _clear_cache()
    yield
    _clear_cache()


@pytest.fixture(scope="session", autouse=True)
def _isolate_home():
    """Ensure HOME points to a temp dir for the entire test session.

    The env vars were already set at module level (above) so that
    module-level initialisations are captured.  This fixture simply
    restores the originals on teardown and cleans up the temp dir.
    """
    yield
    for var, orig in _original_env.items():
        if orig is None:
            os.environ.pop(var, None)
        else:
            os.environ[var] = orig
    shutil.rmtree(_session_tmp, ignore_errors=True)


@pytest.fixture
def tmp_dir():
    """Create and auto-cleanup a temporary directory."""
    d = tempfile.mkdtemp(prefix="mempalace_test_")
    yield d
    shutil.rmtree(d, ignore_errors=True)


@pytest.fixture
def palace_path(tmp_dir):
    """Path to an empty palace directory inside tmp_dir."""
    p = os.path.join(tmp_dir, "palace")
    os.makedirs(p)
    return p


@pytest.fixture
def config(tmp_dir, palace_path):
    """A MempalaceConfig pointing at the temp palace."""
    cfg_dir = os.path.join(tmp_dir, "config")
    os.makedirs(cfg_dir)
    import json

    with open(os.path.join(cfg_dir, "config.json"), "w") as f:
        json.dump({"palace_path": palace_path}, f)
    return MempalaceConfig(config_dir=cfg_dir)


@pytest.fixture
def collection(palace_path):
    """A ChromaDB collection pre-seeded in the temp palace."""
    client = chromadb.PersistentClient(path=palace_path)
    col = client.get_or_create_collection("mempalace_drawers", metadata={"hnsw:space": "cosine"})
    yield col
    client.delete_collection("mempalace_drawers")
    # close() (not a bare dereference) releases chromadb's rust-side SQLite/HNSW
    # file handles. On Windows a mere `del` leaves them locked, so the temp
    # palace cannot be removed and handles leak across the whole test session
    # until a later test's HNSW write fails (#1128 Windows CI).
    client.close()


@pytest.fixture
def seeded_collection(collection):
    """Collection with a handful of representative drawers."""
    collection.add(
        ids=[
            "drawer_proj_backend_aaa",
            "drawer_proj_backend_bbb",
            "drawer_proj_frontend_ccc",
            "drawer_notes_planning_ddd",
        ],
        documents=[
            "The authentication module uses JWT tokens for session management. "
            "Tokens expire after 24 hours. Refresh tokens are stored in HttpOnly cookies.",
            "Database migrations are handled by Alembic. We use PostgreSQL 15 "
            "with connection pooling via pgbouncer.",
            "The React frontend uses TanStack Query for server state management. "
            "All API calls go through a centralized fetch wrapper.",
            "Sprint planning: migrate auth to passkeys by Q3. "
            "Evaluate ChromaDB alternatives for vector search.",
        ],
        metadatas=[
            {
                "wing": "project",
                "room": "backend",
                "source_file": "auth.py",
                "chunk_index": 0,
                "added_by": "miner",
                "filed_at": "2026-01-01T00:00:00",
            },
            {
                "wing": "project",
                "room": "backend",
                "source_file": "db.py",
                "chunk_index": 0,
                "added_by": "miner",
                "filed_at": "2026-01-02T00:00:00",
            },
            {
                "wing": "project",
                "room": "frontend",
                "source_file": "App.tsx",
                "chunk_index": 0,
                "added_by": "miner",
                "filed_at": "2026-01-03T00:00:00",
            },
            {
                "wing": "notes",
                "room": "planning",
                "source_file": "sprint.md",
                "chunk_index": 0,
                "added_by": "miner",
                "filed_at": "2026-01-04T00:00:00",
            },
        ],
    )
    return collection


@pytest.fixture
def kg(tmp_dir):
    """An isolated KnowledgeGraph using a temp SQLite file."""
    db_path = os.path.join(tmp_dir, "test_kg.sqlite3")
    graph = KnowledgeGraph(db_path=db_path)
    yield graph
    graph.close()


@pytest.fixture
def seeded_kg(kg):
    """KnowledgeGraph pre-loaded with sample triples."""
    kg.add_entity("Alice", entity_type="person")
    kg.add_entity("Max", entity_type="person")
    kg.add_entity("swimming", entity_type="activity")
    kg.add_entity("chess", entity_type="activity")

    kg.add_triple("Alice", "parent_of", "Max", valid_from="2015-04-01")
    kg.add_triple("Max", "does", "swimming", valid_from="2025-01-01")
    kg.add_triple("Max", "does", "chess", valid_from="2024-06-01")
    kg.add_triple("Alice", "works_at", "Acme Corp", valid_from="2020-01-01", valid_to="2024-12-31")
    kg.add_triple("Alice", "works_at", "NewCo", valid_from="2025-01-01")

    return kg