MCP Memory Service

"""Tests for database integrity health monitoring."""

import asyncio
import json
import os
import sqlite3
import tempfile

import pytest

# Skip entire module if health subpackage can't be imported
# (happens in uvx/coverage environments where package namespace is shadowed)
pytest.importorskip("mcp_memory_service.health")


@pytest.fixture
def temp_db():
    """Create a temporary SQLite database with a memories table."""
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
        db_path = f.name

    conn = sqlite3.connect(db_path)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute(
        """CREATE TABLE memories (
            content_hash TEXT PRIMARY KEY,
            content TEXT,
            created_at REAL,
            updated_at REAL,
            metadata TEXT,
            tags TEXT,
            type TEXT
        )"""
    )
    conn.execute(
        "INSERT INTO memories VALUES (?, ?, ?, ?, ?, ?, ?)",
        ("hash1", "test memory", 1.0, 1.0, "{}", "test", "note"),
    )
    conn.commit()
    conn.close()

    yield db_path

    os.unlink(db_path)
    for suffix in ("-wal", "-shm"):
        path = db_path + suffix
        if os.path.exists(path):
            os.unlink(path)


@pytest.fixture
def monitor(temp_db, monkeypatch):
    """Create an IntegrityMonitor with test config."""
    from mcp_memory_service import config
    from mcp_memory_service.health import integrity as integrity_mod

    # Patch config values without importlib.reload (which breaks in uvx)
    monkeypatch.setattr(config, "INTEGRITY_CHECK_ENABLED", True)
    monkeypatch.setattr(config, "INTEGRITY_CHECK_INTERVAL", 60)
    monkeypatch.setattr(integrity_mod, "INTEGRITY_CHECK_ENABLED", True)
    monkeypatch.setattr(integrity_mod, "INTEGRITY_CHECK_INTERVAL", 60)

    from mcp_memory_service.health.integrity import IntegrityMonitor
    return IntegrityMonitor(temp_db)


class TestIntegrityCheck:
    """Tests for PRAGMA integrity_check."""

    @pytest.mark.asyncio
    async def test_healthy_database(self, monitor):
        is_healthy, detail = await monitor.check_integrity()
        assert is_healthy is True
        assert detail == "ok"

    @pytest.mark.asyncio
    async def test_missing_database(self, tmp_path, monkeypatch):
        from mcp_memory_service import config
        from mcp_memory_service.health import integrity as integrity_mod

        monkeypatch.setattr(config, "INTEGRITY_CHECK_ENABLED", True)
        monkeypatch.setattr(config, "INTEGRITY_CHECK_INTERVAL", 60)
        monkeypatch.setattr(integrity_mod, "INTEGRITY_CHECK_ENABLED", True)
        monkeypatch.setattr(integrity_mod, "INTEGRITY_CHECK_INTERVAL", 60)

        from mcp_memory_service.health.integrity import IntegrityMonitor
        m = IntegrityMonitor(str(tmp_path / "nonexistent.db"))
        # SQLite creates the file on connect, so it will be "healthy" but empty
        is_healthy, detail = await m.check_integrity()
        assert is_healthy is True

    @pytest.mark.asyncio
    async def test_check_increments_counters(self, monitor):
        assert monitor.total_checks == 0
        await monitor.run_check()
        assert monitor.total_checks == 1
        assert monitor.last_check_healthy is True


class TestWalRepair:
    """Tests for WAL checkpoint repair."""

    @pytest.mark.asyncio
    async def test_repair_on_healthy_db(self, monitor):
        success, detail = await monitor.attempt_wal_repair()
        assert success is True
        assert "successful" in detail


class TestMemoryExport:
    """Tests for emergency memory export."""

    @pytest.mark.asyncio
    async def test_export_memories(self, monitor, tmp_path):
        export_path = str(tmp_path / "export.json")
        success, count = await monitor.export_memories(export_path)
        assert success is True
        assert count == 1

        with open(export_path) as f:
            data = json.load(f)
        assert len(data) == 1
        assert data[0]["hash"] == "hash1"
        assert data[0]["content"] == "test memory"

    @pytest.mark.asyncio
    async def test_export_empty_database(self, tmp_path, monkeypatch):
        from mcp_memory_service import config
        from mcp_memory_service.health import integrity as integrity_mod

        monkeypatch.setattr(config, "INTEGRITY_CHECK_ENABLED", True)
        monkeypatch.setattr(config, "INTEGRITY_CHECK_INTERVAL", 60)
        monkeypatch.setattr(integrity_mod, "INTEGRITY_CHECK_ENABLED", True)
        monkeypatch.setattr(integrity_mod, "INTEGRITY_CHECK_INTERVAL", 60)

        from mcp_memory_service.health.integrity import IntegrityMonitor

        db_path = str(tmp_path / "empty.db")
        conn = sqlite3.connect(db_path)
        conn.execute(
            """CREATE TABLE memories (
                content_hash TEXT PRIMARY KEY, content TEXT,
                created_at REAL, updated_at REAL,
                metadata TEXT, tags TEXT, type TEXT
            )"""
        )
        conn.close()

        m = IntegrityMonitor(db_path)
        export_path = str(tmp_path / "export.json")
        success, count = await m.export_memories(export_path)
        assert success is True
        assert count == 0


class TestRunCheck:
    """Tests for the full check + repair flow."""

    @pytest.mark.asyncio
    async def test_healthy_check_result(self, monitor):
        result = await monitor.run_check()
        assert result["healthy"] is True
        assert result["repaired"] is False
        assert result["exported"] is False
        assert result["check_ms"] < 1000  # Should be fast


class TestGetStatus:
    """Tests for status reporting."""

    def test_initial_status(self, monitor):
        status = monitor.get_status()
        assert status["running"] is False
        assert status["total_checks"] == 0
        assert status["total_auto_repairs"] == 0
        assert status["total_unrecoverable"] == 0

    @pytest.mark.asyncio
    async def test_status_after_check(self, monitor):
        await monitor.run_check()
        status = monitor.get_status()
        assert status["total_checks"] == 1
        assert status["last_check_healthy"] is True