basicmachines-co-basic-memory/test-int/mcp/test_pagination_integration.py

"""
Integration tests for pagination across search and recent_activity MCP tools.

Verifies that page 1 and page 2 return disjoint result sets all the way
through to the database, and that has_more correctly signals when more
results are available.
"""

import json

import pytest
from fastmcp import Client


def _json_content(tool_result) -> dict | list:
    """Parse a FastMCP tool result content block into JSON."""
    assert len(tool_result.content) == 1
    assert tool_result.content[0].type == "text"
    return json.loads(tool_result.content[0].text)


# ---------------------------------------------------------------------------
# search_notes pagination
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_search_notes_pages_are_disjoint(mcp_server, app, test_project):
    """Page 1 and page 2 of search_notes must return different results."""

    async with Client(mcp_server) as client:
        # Create 6 notes so page_size=3 gives us at least 2 full pages
        for i in range(6):
            await client.call_tool(
                "write_note",
                {
                    "project": test_project.name,
                    "title": f"Pagination Note {i:02d}",
                    "directory": "pagination",
                    "content": f"# Pagination Note {i:02d}\n\nContent for pagination testing.",
                },
            )

        # Fetch page 1
        page1_result = await client.call_tool(
            "search_notes",
            {
                "project": test_project.name,
                "query": "Pagination Note",
                "page": 1,
                "page_size": 3,
                "output_format": "json",
            },
        )
        page1 = _json_content(page1_result)
        page1_titles = {r["title"] for r in page1["results"]}

        assert len(page1["results"]) == 3
        assert page1["has_more"] is True

        # Fetch page 2
        page2_result = await client.call_tool(
            "search_notes",
            {
                "project": test_project.name,
                "query": "Pagination Note",
                "page": 2,
                "page_size": 3,
                "output_format": "json",
            },
        )
        page2 = _json_content(page2_result)
        page2_titles = {r["title"] for r in page2["results"]}

        assert len(page2["results"]) == 3

        # The two pages must not overlap
        assert page1_titles.isdisjoint(page2_titles), (
            f"Pages overlap: {page1_titles & page2_titles}"
        )

        # Together they should cover all 6 notes
        assert len(page1_titles | page2_titles) == 6


@pytest.mark.asyncio
async def test_search_notes_has_more_becomes_false(mcp_server, app, test_project):
    """has_more should be False when the last page is reached."""

    async with Client(mcp_server) as client:
        # Create 4 notes, page_size=3 → page 1 has_more=True, page 2 has_more=False
        for i in range(4):
            await client.call_tool(
                "write_note",
                {
                    "project": test_project.name,
                    "title": f"LastPage Note {i:02d}",
                    "directory": "lastpage",
                    "content": f"# LastPage Note {i:02d}\n\nContent for last-page testing.",
                },
            )

        page1 = _json_content(
            await client.call_tool(
                "search_notes",
                {
                    "project": test_project.name,
                    "query": "LastPage Note",
                    "page": 1,
                    "page_size": 3,
                    "output_format": "json",
                },
            )
        )
        assert page1["has_more"] is True
        assert len(page1["results"]) == 3

        page2 = _json_content(
            await client.call_tool(
                "search_notes",
                {
                    "project": test_project.name,
                    "query": "LastPage Note",
                    "page": 2,
                    "page_size": 3,
                    "output_format": "json",
                },
            )
        )
        assert page2["has_more"] is False
        assert len(page2["results"]) == 1


# ---------------------------------------------------------------------------
# recent_activity pagination
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_recent_activity_pages_are_disjoint(mcp_server, app, test_project):
    """Page 1 and page 2 of recent_activity must return different results."""

    async with Client(mcp_server) as client:
        # Create 6 notes so page_size=3 gives us 2 full pages of entities
        for i in range(6):
            await client.call_tool(
                "write_note",
                {
                    "project": test_project.name,
                    "title": f"Recent Page Note {i:02d}",
                    "directory": "recent-page",
                    "content": f"# Recent Page Note {i:02d}\n\nContent for recent-activity pagination.",
                },
            )

        # Fetch page 1 (entity-only default)
        page1_result = await client.call_tool(
            "recent_activity",
            {
                "project": test_project.name,
                "page": 1,
                "page_size": 3,
                "timeframe": "7d",
                "output_format": "json",
            },
        )
        page1 = _json_content(page1_result)
        page1_titles = {item["title"] for item in page1}

        assert len(page1) == 3

        # Fetch page 2
        page2_result = await client.call_tool(
            "recent_activity",
            {
                "project": test_project.name,
                "page": 2,
                "page_size": 3,
                "timeframe": "7d",
                "output_format": "json",
            },
        )
        page2 = _json_content(page2_result)
        page2_titles = {item["title"] for item in page2}

        assert len(page2) == 3

        # The two pages must not overlap
        assert page1_titles.isdisjoint(page2_titles), (
            f"Pages overlap: {page1_titles & page2_titles}"
        )

        # Together they should cover all 6 notes
        assert len(page1_titles | page2_titles) == 6


@pytest.mark.asyncio
async def test_recent_activity_has_more_becomes_false(mcp_server, app, test_project):
    """has_more should be False on the last page of recent_activity."""

    async with Client(mcp_server) as client:
        # Create 4 notes, page_size=3 → page 1 has more, page 2 does not
        for i in range(4):
            await client.call_tool(
                "write_note",
                {
                    "project": test_project.name,
                    "title": f"Recent Last Note {i:02d}",
                    "directory": "recent-last",
                    "content": f"# Recent Last Note {i:02d}\n\nContent for has_more testing.",
                },
            )

        # Page 1: text output should mention "page=2 to see more"
        page1_result = await client.call_tool(
            "recent_activity",
            {
                "project": test_project.name,
                "page": 1,
                "page_size": 3,
                "timeframe": "7d",
            },
        )
        page1_text = page1_result.content[0].text
        assert "Use page=2 to see more" in page1_text

        # Page 2: should NOT mention further pagination
        page2_result = await client.call_tool(
            "recent_activity",
            {
                "project": test_project.name,
                "page": 2,
                "page_size": 3,
                "timeframe": "7d",
            },
        )
        page2_text = page2_result.content[0].text
        assert "Use page=" not in page2_text
        assert "items found" in page2_text


@pytest.mark.asyncio
async def test_recent_activity_entity_only_default_with_relations(mcp_server, app, test_project):
    """Without explicit type, recent_activity returns only entities even when
    observations and relations exist in the database."""

    async with Client(mcp_server) as client:
        # Create a note with observations and relations
        await client.call_tool(
            "write_note",
            {
                "project": test_project.name,
                "title": "Entity With Relations",
                "directory": "dedup",
                "content": (
                    "# Entity With Relations\n\n"
                    "## Observations\n"
                    "- [status] This entity has observations\n"
                    "- [note] Another observation here\n\n"
                    "## Relations\n"
                    "- links_to [[Target Note]]\n"
                ),
            },
        )
        await client.call_tool(
            "write_note",
            {
                "project": test_project.name,
                "title": "Target Note",
                "directory": "dedup",
                "content": "# Target Note\n\nTarget for relation.",
            },
        )

        # Default call (no type specified) should return only entities
        result = await client.call_tool(
            "recent_activity",
            {
                "project": test_project.name,
                "timeframe": "7d",
                "output_format": "json",
            },
        )
        payload = _json_content(result)
        assert isinstance(payload, list)
        assert len(payload) >= 2
        assert all(item["type"] == "entity" for item in payload)

        # Explicit type=["entity", "observation", "relation"] returns mixed types
        all_types_result = await client.call_tool(
            "recent_activity",
            {
                "project": test_project.name,
                "timeframe": "7d",
                "type": ["entity", "observation", "relation"],
                "output_format": "json",
            },
        )
        all_payload = _json_content(all_types_result)
        types_returned = {item["type"] for item in all_payload}
        # Observations and relations should now appear alongside entities
        assert "entity" in types_returned
        assert len(all_payload) > len(payload), (
            "Requesting all types should return more items than entity-only"
        )