MCP Memory Service

import pytest
import os
import sys
import tempfile
import shutil
import uuid
from typing import Callable, Optional, List

# Add src directory to Python path
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(__file__)), 'src'))

# Disable semantic deduplication during tests to avoid interference with test expectations
# Tests often use similar content patterns (e.g., "Test memory 1", "Test memory 2")
# which would be caught by semantic dedup and cause unexpected failures
os.environ['MCP_SEMANTIC_DEDUP_ENABLED'] = 'false'

# CRITICAL: Force sqlite_vec backend for tests to prevent accidental Cloudflare operations
# This prevents tests from soft-deleting production memories in Cloudflare D1
#
# To explicitly test Cloudflare/Hybrid backends, set:
#   MCP_TEST_ALLOW_CLOUD_BACKEND=true pytest tests/integration/test_cloudflare_*.py
#
_original_backend = os.environ.get('MCP_MEMORY_STORAGE_BACKEND')
_allow_cloud = os.environ.get('MCP_TEST_ALLOW_CLOUD_BACKEND', 'false').lower() == 'true'

if not _allow_cloud:
    os.environ['MCP_MEMORY_STORAGE_BACKEND'] = 'sqlite_vec'
    if _original_backend and _original_backend != 'sqlite_vec':
        print(f"\n[Test Safety] ⚠️  Overriding MCP_MEMORY_STORAGE_BACKEND from '{_original_backend}' to 'sqlite_vec'")
        print(f"[Test Safety] This prevents tests from modifying production Cloudflare data.")
        print(f"[Test Safety] To allow cloud backends: MCP_TEST_ALLOW_CLOUD_BACKEND=true pytest ...")

# CRITICAL: Force test database path to prevent production database access
# This ensures tests NEVER touch production databases even if they use the wrong fixtures
_test_db_dir = tempfile.mkdtemp(prefix='mcp-test-db-')
os.environ['MCP_MEMORY_SQLITE_PATH'] = os.path.join(_test_db_dir, 'test.db')
print(f"\n[Test Safety] 🔒 Test database path: {os.environ['MCP_MEMORY_SQLITE_PATH']}")
print(f"[Test Safety] Production databases are isolated and protected.")

# Reserved tag for test memories - enables automatic cleanup
TEST_MEMORY_TAG = "__test__"

@pytest.fixture
def temp_db_path():
    '''Create a temporary directory for database testing.'''
    temp_dir = tempfile.mkdtemp()
    yield temp_dir
    # Clean up after test
    shutil.rmtree(temp_dir)

@pytest.fixture
def unique_content() -> Callable[[str], str]:
    """
    Generate unique test content to avoid duplicate content errors.

    Usage:
        def test_example(unique_content):
            content = unique_content("Test memory about authentication")
            hash1 = store(content, tags=["test"])

    Returns:
        A function that takes a base string and returns a unique version.
    """
    def _generator(base: str = "test") -> str:
        return f"{base} [{uuid.uuid4()}]"
    return _generator


@pytest.fixture
def test_store():
    """
    Store function that auto-tags memories with TEST_MEMORY_TAG for cleanup.

    All memories created with this fixture will be automatically deleted
    at the end of the test session via pytest_sessionfinish hook.

    Usage:
        def test_example(test_store, unique_content):
            hash1 = test_store(unique_content("Test memory"), tags=["auth"])
            # Memory will have tags: ["__test__", "auth"]
    """
    from mcp_memory_service.api import store

    def _store(content: str, tags: Optional[List[str]] = None, **kwargs):
        all_tags = [TEST_MEMORY_TAG] + (tags or [])
        return store(content, tags=all_tags, **kwargs)

    return _store


def pytest_sessionfinish(session, exitstatus):
    """
    Cleanup all test memories at end of test session.

    SAFETY: Uses TRIPLE verification to prevent production database deletion.
    Only cleans test databases, never production.
    """
    try:
        from mcp_memory_service.config import SQLITE_VEC_PATH, STORAGE_BACKEND
        import tempfile

        # Get database path based on backend
        if STORAGE_BACKEND == 'hybrid':
            # Hybrid uses SQLite-vec as primary
            db_path = str(SQLITE_VEC_PATH) if SQLITE_VEC_PATH else None
        elif STORAGE_BACKEND == 'sqlite_vec':
            db_path = str(SQLITE_VEC_PATH) if SQLITE_VEC_PATH else None
        else:
            # Cloudflare or other backends - skip cleanup (no local DB)
            print(f"\n[Test Cleanup] ℹ️  Backend '{STORAGE_BACKEND}' has no local database to clean")
            return

        if not db_path:
            print(f"\n[Test Cleanup] ⚠️  No database path configured, skipping cleanup")
            return

        # === TRIPLE SAFETY CHECK SYSTEM ===

        # SAFETY CHECK 1: Must be in temp directory
        temp_prefix = tempfile.gettempdir()
        if not db_path.startswith(temp_prefix):
            print(f"\n[Test Cleanup] 🛑 SAFETY ABORT: Database not in temp directory!")
            print(f"[Test Cleanup] Database path: {db_path}")
            print(f"[Test Cleanup] Expected prefix: {temp_prefix}")
            print(f"[Test Cleanup] This indicates a configuration error - tests must use isolated databases.")
            return

        # SAFETY CHECK 2: Must NOT contain production indicators
        production_indicators = [
            "Library/Application Support/mcp-memory",
            ".mcp-memory",
            "mcp-memory-service/data",
            "/Users/",  # Any user home directory
            "/home/",   # Linux home directories
            "C:\\Users\\",  # Windows user directories
        ]

        for indicator in production_indicators:
            if indicator in db_path:
                print(f"\n[Test Cleanup] 🛑 SAFETY ABORT: Production path indicator detected!")
                print(f"[Test Cleanup] Database path: {db_path}")
                print(f"[Test Cleanup] Detected indicator: '{indicator}'")
                print(f"[Test Cleanup] Tests MUST use temp_db_path fixture for isolation.")
                return

        # SAFETY CHECK 3: Must contain test-specific markers
        test_markers = [
            'mcp-test-',  # Our custom test prefix
            'pytest-',    # pytest temp directories
            '/tmp/',      # Unix temp
            '/var/tmp/',  # Unix alternative temp
            'Temp\\',     # Windows temp
        ]

        has_test_marker = any(marker in db_path for marker in test_markers)
        if not has_test_marker:
            print(f"\n[Test Cleanup] 🛑 SAFETY ABORT: No test marker found in database path!")
            print(f"[Test Cleanup] Database path: {db_path}")
            print(f"[Test Cleanup] Expected one of: {test_markers}")
            return

        # === ALL SAFETY CHECKS PASSED ===
        print(f"\n[Test Cleanup] ✅ Safety checks passed")
        print(f"[Test Cleanup] Database path: {db_path}")
        print(f"[Test Cleanup] Cleaning test database...")

        # Import delete function
        from mcp_memory_service.api import delete_by_tag

        # Perform cleanup - DO NOT use allow_production flag
        # The delete_by_tag function has its own safety checks
        # If our checks above passed, the delete_by_tag checks should also pass
        result = delete_by_tag([TEST_MEMORY_TAG], allow_production=False)
        deleted = result.get('deleted', 0) if isinstance(result, dict) else 0

        if deleted > 0:
            print(f"[Test Cleanup] ✅ Deleted {deleted} test memories tagged with '{TEST_MEMORY_TAG}'")
        else:
            print(f"[Test Cleanup] ✅ No test memories to clean")

    except ValueError as e:
        # Safety check from delete_by_tag - this is GOOD, it means safety worked!
        if "production database" in str(e).lower():
            print(f"\n[Test Cleanup] ✅ Safety check prevented production deletion!")
            print(f"[Test Cleanup] Error: {e}")
            print(f"[Test Cleanup] This is the safety system working correctly.")
        else:
            print(f"\n[Test Cleanup] ❌ Unexpected ValueError: {e}")
    except Exception as e:
        # Make errors VISIBLE instead of silent
        print(f"\n[Test Cleanup] ❌ ERROR during cleanup: {e}")
        print(f"[Test Cleanup] This may indicate a configuration problem.")
        # Still don't fail the test session


def pytest_sessionstart(session):
    """
    Verify test isolation before running any tests.

    This hook runs BEFORE any tests execute to ensure proper configuration.
    """
    print("\n" + "="*80)
    print("🔒 TEST ISOLATION VERIFICATION")
    print("="*80)

    try:
        from mcp_memory_service.config import SQLITE_VEC_PATH, STORAGE_BACKEND
        import tempfile

        # Check backend
        if STORAGE_BACKEND not in ['sqlite_vec', 'cloudflare']:
            if not _allow_cloud:
                print(f"⚠️  WARNING: Backend '{STORAGE_BACKEND}' may access production!")
                print(f"⚠️  Tests should use 'sqlite_vec' backend for isolation.")
                print(f"⚠️  Set MCP_TEST_ALLOW_CLOUD_BACKEND=true to explicitly allow this.")

        # Check database path
        if SQLITE_VEC_PATH:
            db_path = str(SQLITE_VEC_PATH)
            temp_prefix = tempfile.gettempdir()

            # Verify it's in temp directory
            if db_path.startswith(temp_prefix):
                print(f"✅ Database in temp directory: {db_path}")
            else:
                print(f"❌ CRITICAL: Database NOT in temp directory!")
                print(f"   Path: {db_path}")
                print(f"   Expected prefix: {temp_prefix}")
                print(f"   Tests may affect production data!")

                # Check for production indicators
                production_indicators = [
                    "Library/Application Support/mcp-memory",
                    ".mcp-memory",
                    "mcp-memory-service/data"
                ]

                for indicator in production_indicators:
                    if indicator in db_path:
                        print(f"   🛑 PRODUCTION PATH DETECTED: '{indicator}'")
                        print(f"   🛑 ABORTING TEST RUN FOR SAFETY")
                        pytest.exit("Test configuration would affect production database", returncode=1)

        print(f"✅ Storage backend: {STORAGE_BACKEND}")
        print(f"✅ Cloud backends: {'ALLOWED' if _allow_cloud else 'BLOCKED'}")
        print("="*80 + "\n")

    except Exception as e:
        print(f"❌ Error during isolation verification: {e}")
        print("="*80 + "\n")


# Cleanup temp test database directory at session end
def pytest_unconfigure(config):
    """Clean up global test database directory."""
    try:
        if os.path.exists(_test_db_dir):
            shutil.rmtree(_test_db_dir, ignore_errors=True)
            print(f"\n[Test Cleanup] 🗑️  Removed test database directory: {_test_db_dir}")
    except Exception as e:
        print(f"\n[Test Cleanup] ⚠️  Failed to remove test database directory: {e}")