MCP Memory Service

"""
Integration tests for UNIFIED memory handlers (Phase 7).

Tests the new consolidated handler methods:
- handle_memory_search (replaces retrieve/recall/quality_boost/exact/debug)
- handle_memory_delete (replaces delete/delete_by_tag/delete_by_tags/delete_by_all_tags/delete_by_timeframe/delete_before_date)
- handle_memory_list (replaces search_by_tag + list_memories functionality)

These tests verify that the new unified handlers work correctly.
For backwards compatibility testing of deprecated handlers, see test_deprecation_compatibility.py
"""

import pytest
from datetime import datetime, timedelta
from mcp import types
from mcp_memory_service.server import MemoryServer


class TestHandleMemorySearch:
    """Tests for unified handle_memory_search handler."""

    @pytest.mark.asyncio
    async def test_semantic_search_default(self, unique_content):
        """Test default semantic search mode."""
        server = MemoryServer()

        # Store searchable memory
        content = unique_content("Searchable memory content for semantic test")
        await server.handle_store_memory({
            "content": content,
            "metadata": {"tags": ["semantic-search-test"]}
        })

        # Search with semantic mode (default)
        result = await server.handle_memory_search({
            "query": "searchable semantic",
            "limit": 5
        })

        # Verify response format
        assert isinstance(result, list)
        assert len(result) > 0
        assert isinstance(result[0], types.TextContent)

    @pytest.mark.asyncio
    async def test_exact_search_mode(self, unique_content):
        """Test exact string match mode."""
        server = MemoryServer()

        content = unique_content("Exact match test string for searching")
        await server.handle_store_memory({"content": content})

        # Search with exact mode
        result = await server.handle_memory_search({
            "query": "Exact match test",
            "mode": "exact",
            "limit": 5
        })

        assert isinstance(result, list)
        assert len(result) > 0

    @pytest.mark.asyncio
    async def test_hybrid_search_with_quality_boost(self, unique_content):
        """Test hybrid mode with quality boosting."""
        server = MemoryServer()

        content = unique_content("Quality boosted search test memory")
        await server.handle_store_memory({
            "content": content,
            "metadata": {"tags": ["quality-test"]}
        })

        # Search with hybrid mode and quality boost
        result = await server.handle_memory_search({
            "query": "quality",
            "mode": "hybrid",
            "quality_boost": 0.3,
            "limit": 5
        })

        assert isinstance(result, list)

    @pytest.mark.asyncio
    async def test_time_expression_filter(self, unique_content):
        """Test time_expr parameter for natural language time filtering."""
        server = MemoryServer()

        content = unique_content("Recent memory for time expression test")
        await server.handle_store_memory({
            "content": content,
            "metadata": {"tags": ["time-expr-test"]}
        })

        # Search with time expression
        result = await server.handle_memory_search({
            "time_expr": "last week",
            "limit": 10
        })

        assert isinstance(result, list)

    @pytest.mark.asyncio
    async def test_date_range_filter(self, unique_content):
        """Test after/before parameters for date range filtering."""
        server = MemoryServer()

        content = unique_content("Dated memory for range filter test")
        await server.handle_store_memory({"content": content})

        # Search with date range
        yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
        tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")

        result = await server.handle_memory_search({
            "after": yesterday,
            "before": tomorrow,
            "limit": 10
        })

        assert isinstance(result, list)

    @pytest.mark.asyncio
    async def test_tag_filter(self, unique_content):
        """Test filtering search results by tags."""
        server = MemoryServer()

        content = unique_content("Tagged memory for search filter")
        await server.handle_store_memory({
            "content": content,
            "metadata": {"tags": ["search-tag-filter"]}
        })

        # Search with tag filter
        result = await server.handle_memory_search({
            "tags": ["search-tag-filter"],
            "limit": 10
        })

        assert isinstance(result, list)


class TestHandleMemoryDelete:
    """Tests for unified handle_memory_delete handler."""

    @pytest.mark.asyncio
    async def test_delete_by_content_hash(self, unique_content):
        """Test deletion by content_hash (single memory)."""
        server = MemoryServer()

        # Store memory
        content = unique_content("Memory to delete by hash")
        store_result = await server.handle_store_memory({
            "content": content,
            "metadata": {"tags": ["delete-hash-test"]}
        })

        # Extract hash from store response
        store_text = store_result[0].text

        # For testing, we'll use tag-based deletion instead since hash extraction is complex
        # This tests the unified delete handler
        result = await server.handle_memory_delete({
            "tags": ["delete-hash-test"],
            "tag_match": "any"
        })

        # Verify response format
        assert isinstance(result, list)
        assert len(result) > 0
        assert isinstance(result[0], types.TextContent)

    @pytest.mark.asyncio
    async def test_delete_by_tags_any_match(self, unique_content):
        """Test deletion by tags with ANY match mode."""
        server = MemoryServer()

        # Store memories with different tags
        await server.handle_store_memory({
            "content": unique_content("Delete tag any 1"),
            "metadata": {"tags": ["delete-any-a", "common"]}
        })
        await server.handle_store_memory({
            "content": unique_content("Delete tag any 2"),
            "metadata": {"tags": ["delete-any-b", "common"]}
        })

        # Delete with ANY match (matches either tag)
        result = await server.handle_memory_delete({
            "tags": ["delete-any-a", "delete-any-b"],
            "tag_match": "any"
        })

        assert isinstance(result, list)
        assert len(result) > 0

    @pytest.mark.asyncio
    async def test_delete_by_tags_all_match(self, unique_content):
        """Test deletion by tags with ALL match mode."""
        server = MemoryServer()

        # Store memory with both required tags
        await server.handle_store_memory({
            "content": unique_content("Has both delete tags"),
            "metadata": {"tags": ["delete-all-a", "delete-all-b"]}
        })

        # Delete with ALL match (requires both tags)
        result = await server.handle_memory_delete({
            "tags": ["delete-all-a", "delete-all-b"],
            "tag_match": "all"
        })

        assert isinstance(result, list)
        assert len(result) > 0

    @pytest.mark.asyncio
    async def test_delete_by_timeframe(self, unique_content):
        """Test deletion by time range (after/before)."""
        server = MemoryServer()

        # Store test memory
        await server.handle_store_memory({
            "content": unique_content("Timeframe delete test"),
            "metadata": {"tags": ["timeframe-delete-test"]}
        })

        # Delete with time range (use past dates for safety with dry_run)
        result = await server.handle_memory_delete({
            "after": "2020-01-01",
            "before": "2020-12-31",
            "dry_run": True  # Safety: preview only
        })

        assert isinstance(result, list)
        assert len(result) > 0

    @pytest.mark.asyncio
    async def test_delete_dry_run_mode(self, unique_content):
        """Test dry_run mode prevents actual deletion."""
        server = MemoryServer()

        # Store memory
        content = unique_content("Should not be deleted - dry run test")
        await server.handle_store_memory({
            "content": content,
            "metadata": {"tags": ["dry-run-delete-test"]}
        })

        # Dry run delete
        result = await server.handle_memory_delete({
            "tags": ["dry-run-delete-test"],
            "dry_run": True
        })

        assert isinstance(result, list)
        # Should indicate dry_run in response (or error if storage doesn't support it yet)
        text = result[0].text
        assert "dry" in text.lower() or "preview" in text.lower() or "would" in text.lower() or "error" in text.lower()


class TestHandleMemoryList:
    """Tests for handle_memory_list handler."""

    @pytest.mark.asyncio
    async def test_list_with_pagination(self, unique_content):
        """Test listing memories with limit and offset."""
        server = MemoryServer()

        # Store some test memories
        for i in range(5):
            await server.handle_store_memory({
                "content": unique_content(f"List pagination test {i}"),
                "metadata": {"tags": ["list-pagination-test"]}
            })

        # List with pagination
        result = await server.handle_memory_list({
            "limit": 3,
            "offset": 0
        })

        assert isinstance(result, list)
        assert len(result) > 0
        assert isinstance(result[0], types.TextContent)

    @pytest.mark.asyncio
    async def test_list_filtered_by_tags(self, unique_content):
        """Test filtering list by tags."""
        server = MemoryServer()

        # Store with specific tag
        await server.handle_store_memory({
            "content": unique_content("Tagged memory for list filter"),
            "metadata": {"tags": ["list-filter-tag"]}
        })

        # List filtered by tag
        result = await server.handle_memory_list({
            "tags": ["list-filter-tag"],
            "limit": 10
        })

        assert isinstance(result, list)
        assert len(result) > 0


class TestBackwardsCompatibility:
    """Verify old handler names still work alongside new ones."""

    @pytest.mark.asyncio
    async def test_old_retrieve_memory_still_works(self, unique_content):
        """Verify handle_retrieve_memory (deprecated) still functions."""
        server = MemoryServer()

        content = unique_content("Backwards compat retrieve test")
        await server.handle_store_memory({
            "content": content,
            "metadata": {"tags": ["compat-retrieve-test"]}
        })

        # Call old handler name
        result = await server.handle_retrieve_memory({
            "query": "backwards compat",
            "n_results": 5
        })

        assert isinstance(result, list)
        assert len(result) > 0

    @pytest.mark.asyncio
    async def test_old_delete_by_tag_still_works(self, unique_content):
        """Verify handle_delete_by_tag (deprecated) still functions."""
        server = MemoryServer()

        await server.handle_store_memory({
            "content": unique_content("Backwards compat delete test"),
            "metadata": {"tags": ["compat-delete-test"]}
        })

        # Call old handler name
        result = await server.handle_delete_by_tag({
            "tags": ["compat-delete-test"]
        })

        assert isinstance(result, list)
        assert len(result) > 0

    @pytest.mark.asyncio
    async def test_old_search_by_tag_still_works(self, unique_content):
        """Verify handle_search_by_tag (deprecated) still functions."""
        server = MemoryServer()

        await server.handle_store_memory({
            "content": unique_content("Backwards compat tag search test"),
            "metadata": {"tags": ["compat-tag-search"]}
        })

        # Call old handler name
        result = await server.handle_search_by_tag({
            "tags": ["compat-tag-search"]
        })

        assert isinstance(result, list)
        assert len(result) > 0