basicmachines-co-basic-memory/tests/mcp/test_tool_read_note.py

"""Tests for note tools that exercise the full stack with SQLite."""

from contextlib import asynccontextmanager
from types import SimpleNamespace
from textwrap import dedent

import pytest

from basic_memory.mcp.tools import write_note, read_note
from basic_memory.mcp.tools.read_note import _parse_opening_frontmatter
from basic_memory.utils import normalize_newlines
from tests.mcp.conftest import ContextState, ctx


def test_parse_opening_frontmatter_handles_crlf():
    """JSON read_note output should strip frontmatter from Windows-written markdown."""
    body, frontmatter = _parse_opening_frontmatter(
        "---\r\ntitle: Windows Note\r\ntype: note\r\n---\r\n\r\nBody text\r\n"
    )

    assert frontmatter == {"title": "Windows Note", "type": "note"}
    assert body.strip() == "Body text"


@pytest.mark.asyncio
async def test_read_note_by_title(app, test_project):
    """Test reading a note by its title."""
    # First create a note
    await write_note(
        project=test_project.name,
        title="Special Note",
        directory="test",
        content="Note content here",
    )

    # Should be able to read it by title
    content = await read_note("Special Note", project=test_project.name)
    assert "Note content here" in content


@pytest.mark.asyncio
async def test_read_note_title_search_fallback_fetches_by_permalink(monkeypatch, app, test_project):
    """Force direct resolve to fail so we exercise the title-search + fetch fallback path."""
    await write_note(
        project=test_project.name,
        title="Fallback Title Note",
        directory="test",
        content="fallback content",
    )

    import importlib
    from basic_memory.schemas.memory import memory_url_path

    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient
    direct_identifier = memory_url_path("Fallback Title Note")

    class SelectiveKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            # Fail on the direct identifier to force fallback to title search
            if identifier == direct_identifier:
                raise RuntimeError("force direct lookup failure")
            return await super().resolve_entity(identifier, strict=strict)

    monkeypatch.setattr(clients_mod, "KnowledgeClient", SelectiveKnowledgeClient)

    content = await read_note("Fallback Title Note", project=test_project.name)
    assert "fallback content" in content


@pytest.mark.asyncio
async def test_read_note_returns_related_results_when_text_search_finds_matches(
    monkeypatch, app, test_project
):
    """Exercise the related-results message when no exact note match exists."""
    import importlib

    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient

    async def fake_search_notes_fn(*, query, search_type, **kwargs):
        if search_type == "title":
            return {"results": [], "current_page": 1, "page_size": 10}

        return {
            "results": [
                {
                    "title": "Related One",
                    "permalink": "docs/related-one",
                    "content": "",
                    "type": "entity",
                    "score": 1.0,
                    "file_path": "docs/related-one.md",
                },
                {
                    "title": "Related Two",
                    "permalink": "docs/related-two",
                    "content": "",
                    "type": "entity",
                    "score": 0.9,
                    "file_path": "docs/related-two.md",
                },
            ],
            "current_page": 1,
            "page_size": 10,
        }

    # Ensure direct resolution doesn't short-circuit the fallback logic.
    class FailingKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            raise RuntimeError("force fallback")

    monkeypatch.setattr(clients_mod, "KnowledgeClient", FailingKnowledgeClient)
    monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)

    result = await read_note("missing-note", project=test_project.name)
    assert "I couldn't find an exact match" in result
    assert "## 1. Related One" in result
    assert "## 2. Related Two" in result


@pytest.mark.asyncio
async def test_read_note_direct_match_returns_full_content_regardless_of_paging(app, test_project):
    """page/page_size never chunk note content — a direct match returns the whole note."""
    content = "Line one of the note\nLine two of the note\nLine three of the note"
    await write_note(
        project=test_project.name,
        title="Paging Direct Note",
        directory="test",
        content=content,
    )

    result = await read_note(
        "test/paging-direct-note",
        project=test_project.name,
        page=3,
        page_size=1,
    )

    assert "Line one of the note" in result
    assert "Line three of the note" in result


@pytest.mark.asyncio
async def test_read_note_rejects_non_positive_pagination(app, test_project):
    """Fail fast on invalid pagination, matching search_notes/build_context."""
    with pytest.raises(ValueError, match="page must be >= 1"):
        await read_note("any-note", project=test_project.name, page=0)

    with pytest.raises(ValueError, match="page_size must be >= 1"):
        await read_note("any-note", project=test_project.name, page_size=0)


@pytest.mark.asyncio
async def test_read_note_forwards_pagination_to_fallback_search(monkeypatch, app, test_project):
    """page/page_size must reach the server-side fallback search, not be swallowed."""
    import importlib

    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient

    captured_pages: list[tuple[str, int, int]] = []

    async def fake_search_notes_fn(*, query, search_type, page, page_size, **kwargs):
        captured_pages.append((search_type, page, page_size))
        return {"results": [], "current_page": page, "page_size": page_size}

    class FailingKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            raise RuntimeError("force fallback")

    monkeypatch.setattr(clients_mod, "KnowledgeClient", FailingKnowledgeClient)
    monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)

    result = await read_note("missing-note", project=test_project.name, page=2, page_size=3)

    # Title lookup is pinned to page 1 with a fixed lookup size (it exists to
    # find THE note by exact title); caller page/page_size apply only to the
    # text-search suggestions.
    assert captured_pages == [("title", 1, 10), ("text", 2, 3)]
    assert "Note Not Found" in result


@pytest.mark.asyncio
async def test_read_note_title_fallback_finds_exact_match_on_later_page(
    monkeypatch, app, test_project
):
    """An exact title match is returned even when the caller asks for page > 1.

    The title-match lookup is pinned to page 1 of title results; without the pin,
    read_note("Exact Title", page=2) would page past the match and return
    unrelated suggestions instead of the note.
    """
    await write_note(
        project=test_project.name,
        title="Paged Title Note",
        directory="test",
        content="paged title content",
    )

    import importlib
    from basic_memory.schemas.memory import memory_url_path

    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient
    direct_identifier = memory_url_path("Paged Title Note")

    class SelectiveKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            # Fail on the direct identifier to force fallback to title search
            if identifier == direct_identifier:
                raise RuntimeError("force direct lookup failure")
            return await super().resolve_entity(identifier, strict=strict)

    monkeypatch.setattr(clients_mod, "KnowledgeClient", SelectiveKnowledgeClient)

    content = await read_note("Paged Title Note", project=test_project.name, page=2)
    assert "paged title content" in content


@pytest.mark.asyncio
async def test_read_note_title_fallback_finds_exact_match_with_small_page_size(
    monkeypatch, app, test_project
):
    """An exact title match is returned even when the caller asks for a tiny page_size.

    The title-match lookup uses a fixed lookup size; without it, a higher-ranked
    fuzzy title ("Foo Bar Foo Bar") would displace the exact title ("Foo Bar")
    out of a page_size=1 window and read_note("Foo Bar", page_size=1) would
    return suggestions instead of the note.
    """
    await write_note(
        project=test_project.name,
        title="Foo Bar Foo Bar",
        directory="test",
        content="fuzzy decoy content",
    )
    await write_note(
        project=test_project.name,
        title="Foo Bar",
        directory="test",
        content="exact title content",
    )

    import importlib
    from basic_memory.schemas.memory import memory_url_path

    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient
    direct_identifier = memory_url_path("Foo Bar")

    class SelectiveKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            # Fail on the direct identifier to force fallback to title search
            if identifier == direct_identifier:
                raise RuntimeError("force direct lookup failure")
            return await super().resolve_entity(identifier, strict=strict)

    monkeypatch.setattr(clients_mod, "KnowledgeClient", SelectiveKnowledgeClient)

    content = await read_note("Foo Bar", project=test_project.name, page_size=1)
    assert "exact title content" in content


@pytest.mark.asyncio
async def test_read_note_title_fallback_pages_past_higher_ranked_fuzzy_titles(
    monkeypatch, app, test_project
):
    """An exact title match is found even when it ranks beyond the first lookup page.

    bm25 ranks titles that repeat the queried phrase above the exact title, so with
    more than _TITLE_LOOKUP_PAGE_SIZE such decoys the exact match lands on page 2
    of title results. A single-page lookup would fall through to suggestions even
    though the note exists; the lookup must page until the exact title is found.
    """
    from basic_memory.mcp.tools.read_note import _TITLE_LOOKUP_PAGE_SIZE
    from basic_memory.mcp.tools.search import search_notes

    for index in range(1, _TITLE_LOOKUP_PAGE_SIZE + 2):
        await write_note(
            project=test_project.name,
            title=f"Deep Page Note Deep Page Note Deep Page Note {index:02d}",
            directory="test",
            content=f"fuzzy decoy content {index}",
        )
    await write_note(
        project=test_project.name,
        title="Deep Page Note",
        directory="test",
        content="deep page exact content",
    )

    # Precondition: the exact title must rank beyond the first lookup page,
    # otherwise this test would pass even with a single-page lookup.
    first_page = await search_notes(
        project=test_project.name,
        query="Deep Page Note",
        search_type="title",
        page=1,
        page_size=_TITLE_LOOKUP_PAGE_SIZE,
        output_format="json",
    )
    assert isinstance(first_page, dict)
    first_page_titles = [result["title"] for result in first_page["results"]]
    assert "Deep Page Note" not in first_page_titles
    assert first_page["has_more"] is True

    import importlib
    from basic_memory.schemas.memory import memory_url_path

    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient
    direct_identifier = memory_url_path("Deep Page Note")

    class SelectiveKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            # Fail on the direct identifier to force fallback to title search
            if identifier == direct_identifier:
                raise RuntimeError("force direct lookup failure")
            return await super().resolve_entity(identifier, strict=strict)

    monkeypatch.setattr(clients_mod, "KnowledgeClient", SelectiveKnowledgeClient)

    content = await read_note("Deep Page Note", project=test_project.name)
    assert "deep page exact content" in content


@pytest.mark.asyncio
async def test_read_note_title_lookup_stops_at_page_cap(monkeypatch, app, test_project):
    """The title lookup is bounded: after the page cap it falls through to suggestions."""
    import importlib

    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient

    captured_pages: list[tuple[str, int]] = []

    async def fake_search_notes_fn(*, query, search_type, page, page_size, **kwargs):
        captured_pages.append((search_type, page))
        if search_type == "title":
            # Endless fuzzy titles: every page is full and reports more available,
            # simulating a pathological knowledge base that never yields the note.
            return {
                "results": [
                    {
                        "title": f"Fuzzy {page}-{index}",
                        "permalink": f"docs/fuzzy-{page}-{index}",
                        "content": "",
                        "type": "entity",
                        "score": 1.0,
                        "file_path": f"docs/fuzzy-{page}-{index}.md",
                    }
                    for index in range(page_size)
                ],
                "current_page": page,
                "page_size": page_size,
                "has_more": True,
            }
        return {"results": [], "current_page": page, "page_size": page_size}

    class FailingKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            raise RuntimeError("force fallback")

    monkeypatch.setattr(clients_mod, "KnowledgeClient", FailingKnowledgeClient)
    monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)

    result = await read_note("Pathological Note", project=test_project.name)

    title_pages = [page for search_type, page in captured_pages if search_type == "title"]
    assert title_pages == list(range(1, read_note_module._TITLE_LOOKUP_MAX_PAGES + 1))
    assert "Note Not Found" in result


@pytest.mark.asyncio
async def test_read_note_related_results_list_full_search_page(monkeypatch, app, test_project):
    """Suggestions list the whole returned search page instead of a hardcoded cap of 5."""
    import importlib

    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient

    candidates = [
        {
            "title": f"Related {index}",
            "permalink": f"docs/related-{index}",
            "content": "",
            "type": "entity",
            "score": 1.0,
            "file_path": f"docs/related-{index}.md",
        }
        for index in range(1, 7)
    ]

    async def fake_search_notes_fn(*, query, search_type, **kwargs):
        if search_type == "title":
            return {"results": [], "current_page": 1, "page_size": 10}
        return {"results": candidates, "current_page": 1, "page_size": 10}

    class FailingKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            raise RuntimeError("force fallback")

    monkeypatch.setattr(clients_mod, "KnowledgeClient", FailingKnowledgeClient)
    monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)

    text_result = await read_note("missing-note", project=test_project.name)
    assert "## 6. Related 6" in text_result

    json_result = await read_note(
        "missing-note",
        project=test_project.name,
        output_format="json",
    )
    assert isinstance(json_result, dict)
    assert len(json_result["related_results"]) == 6


@pytest.mark.asyncio
async def test_read_note_title_fallback_requires_exact_title_match(monkeypatch, app, test_project):
    """Do not fetch note content when title-search returns only fuzzy matches."""
    await write_note(
        project=test_project.name,
        title="Existing Note",
        directory="test",
        content="existing note content",
    )

    import importlib

    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient

    class StrictFailingKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            if strict:
                raise RuntimeError("force strict direct lookup failure")
            return await super().resolve_entity(identifier, strict=strict)

    async def fake_search_notes_fn(*, query, search_type, **kwargs):
        if search_type == "title":
            return {
                "results": [
                    {
                        "title": "Existing Note",
                        "permalink": "test/existing-note",
                        "content": "",
                        "type": "entity",
                        "score": 1.0,
                        "file_path": "test/Existing Note.md",
                    }
                ],
                "current_page": 1,
                "page_size": 10,
            }
        return {"results": [], "current_page": 1, "page_size": 10}

    monkeypatch.setattr(clients_mod, "KnowledgeClient", StrictFailingKnowledgeClient)
    monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)

    result = await read_note("Missing Exact Title", project=test_project.name)
    assert "Note Not Found" in result
    assert "Missing Exact Title" in result
    assert "existing note content" not in result


@pytest.mark.asyncio
async def test_read_note_explicit_workspace_project_ignores_stale_cached_project(
    monkeypatch,
    config_manager,
):
    """Explicit workspace routing should use the resolved cloud UUID, not stale cache."""
    import importlib

    import basic_memory.mcp.project_context as project_context
    from basic_memory.mcp.project_context import WorkspaceProjectEntry
    from basic_memory.schemas.project_info import ProjectItem

    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
    clients_mod = importlib.import_module("basic_memory.mcp.clients")

    config = config_manager.load_config()
    config.cloud_api_key = "bmc_test123"
    config_manager.save_config(config)

    personal = project_context.WorkspaceInfo(
        tenant_id="personal-tenant",
        workspace_type="personal",
        slug="personal",
        name="Personal",
        role="owner",
        is_default=True,
    )
    expected_uuid = "22222222-2222-2222-2222-222222222222"
    expected_project = ProjectItem(
        id=2,
        external_id=expected_uuid,
        name="main",
        path="/tmp/main",
        is_default=False,
    )
    stale_project = ProjectItem(
        id=99,
        external_id="33333333-3333-3333-3333-333333333333",
        name="main",
        path="/tmp/stale-main",
        is_default=False,
    )
    context = ContextState()
    await context.set_state("active_workspace", personal.model_dump())
    await context.set_state("active_project", stale_project.model_dump())

    async def fake_resolve_workspace_project_identifier(project_name, context=None):
        assert project_name == "personal/main"
        return WorkspaceProjectEntry(workspace=personal, project=expected_project)

    @asynccontextmanager
    async def fake_get_client(project_name=None, workspace=None):
        assert project_name == "main"
        assert workspace == "personal-tenant"
        yield object()

    class FakeProjectResolveResponse:
        def json(self):
            return {
                "external_id": expected_uuid,
                "project_id": expected_project.id,
                "name": expected_project.name,
                "permalink": expected_project.permalink,
                "path": expected_project.path,
                "is_active": True,
                "is_default": False,
                "resolution_method": "permalink",
            }

    async def fake_call_post(*args, **kwargs):
        return FakeProjectResolveResponse()

    class FakeKnowledgeClient:
        def __init__(self, client, project_id):
            assert project_id == expected_uuid

        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            assert identifier == "personal/main/todo"
            return "entity-1"

        async def get_entity(self, entity_id: str):
            assert entity_id == "entity-1"
            return SimpleNamespace(
                title="TODO",
                permalink="personal/main/todo",
                file_path="TODO.md",
            )

    class FakeResourceClient:
        def __init__(self, client, project_id):
            assert project_id == expected_uuid

        async def read(self, entity_id: str):
            assert entity_id == "entity-1"
            return SimpleNamespace(
                status_code=200,
                text="---\ntitle: TODO\n---\n\n# TODO - Priorities & Tasks\n",
            )

    monkeypatch.setattr(
        project_context,
        "resolve_workspace_project_identifier",
        fake_resolve_workspace_project_identifier,
    )
    monkeypatch.setattr("basic_memory.mcp.async_client.get_client", fake_get_client)
    monkeypatch.setattr("basic_memory.mcp.async_client.is_factory_mode", lambda: False)
    monkeypatch.setattr("basic_memory.mcp.async_client._explicit_routing", lambda: False)
    monkeypatch.setattr("basic_memory.mcp.async_client._force_local_mode", lambda: False)
    monkeypatch.setattr("basic_memory.mcp.tools.utils.call_post", fake_call_post)
    monkeypatch.setattr(clients_mod, "KnowledgeClient", FakeKnowledgeClient)
    monkeypatch.setattr(clients_mod, "ResourceClient", FakeResourceClient)

    result = await read_note_module.read_note(
        "memory://todo",
        project="personal/main",
        output_format="json",
        context=ctx(context),
    )

    assert isinstance(result, dict)
    assert result["permalink"] == "personal/main/todo"
    assert result["content"].strip() == "# TODO - Priorities & Tasks"


@pytest.mark.asyncio
async def test_note_unicode_content(app, test_project):
    """Test handling of unicode content in"""
    content = "# Test 🚀\nThis note has emoji 🎉 and unicode ♠♣♥♦"
    result = await write_note(
        project=test_project.name, title="Unicode Test", directory="test", content=content
    )

    # Check that note was created (checksum is now "unknown" in v2)
    assert "# Created note" in result
    assert f"project: {test_project.name}" in result
    assert "file_path: test/Unicode Test.md" in result
    assert f"permalink: {test_project.name}/test/unicode-test" in result
    assert "checksum:" in result  # Checksum exists but may be "unknown"

    # Read back should preserve unicode
    result = await read_note("test/unicode-test", project=test_project.name)
    assert normalize_newlines(content) in result


@pytest.mark.asyncio
async def test_multiple_notes(app, test_project):
    """Test creating and managing multiple notes"""
    # Create several notes
    notes_data = [
        ("test/note-1", "Note 1", "test", "Content 1", ["tag1"]),
        ("test/note-2", "Note 2", "test", "Content 2", ["tag1", "tag2"]),
        ("test/note-3", "Note 3", "test", "Content 3", []),
    ]

    for _, title, folder, content, tags in notes_data:
        await write_note(
            project=test_project.name, title=title, directory=folder, content=content, tags=tags
        )

    # Should be able to read each one individually
    for permalink, title, folder, content, _ in notes_data:
        note = await read_note(permalink, project=test_project.name)
        assert content in note

    # Note: v2 API does not support glob patterns in read_note
    # Glob patterns should be used with build_context or list_directory instead
    # For reading multiple notes, use build_context with memory:// URLs


@pytest.mark.asyncio
async def test_read_multiple_notes_individually(app, test_project):
    """Test reading several notes back individually after writing them."""
    notes_data = [
        ("test/note-1", "Note 1", "test", "Content 1", ["tag1"]),
        ("test/note-2", "Note 2", "test", "Content 2", ["tag1", "tag2"]),
        ("test/note-3", "Note 3", "test", "Content 3", []),
    ]

    for _, title, folder, content, tags in notes_data:
        await write_note(
            project=test_project.name, title=title, directory=folder, content=content, tags=tags
        )

    for permalink, title, folder, content, _ in notes_data:
        note = await read_note(permalink, project=test_project.name)
        assert content in note

    # Note: v2 API does not support glob patterns in read_note
    # For reading multiple notes, use build_context or list_directory instead


@pytest.mark.asyncio
async def test_read_note_memory_url(app, test_project):
    """Test reading a note using a memory:// URL.

    Should:
    - Handle memory:// URLs correctly
    - Normalize the URL before resolving
    - Return the note content
    """
    # First create a note
    result = await write_note(
        project=test_project.name,
        title="Memory URL Test",
        directory="test",
        content="Testing memory:// URL handling",
    )
    assert result

    # Should be able to read it with a memory:// URL
    memory_url = "memory://test/memory-url-test"
    content = await read_note(memory_url, project=test_project.name)
    assert "Testing memory:// URL handling" in content


@pytest.mark.asyncio
async def test_read_note_memory_url_with_project_prefix(app, test_project):
    """Test reading a note using a memory:// URL with explicit project prefix."""
    await write_note(
        project=test_project.name,
        title="Project Prefixed Memory URL Test",
        directory="test",
        content="Testing memory:// URL handling with project prefix",
    )

    memory_url = f"memory://{test_project.name}/test/project-prefixed-memory-url-test"
    content = await read_note(memory_url)
    assert "Testing memory:// URL handling with project prefix" in content


@pytest.mark.asyncio
async def test_read_note_skips_url_detection_when_project_id_provided(
    monkeypatch,
    app,
    test_project,
):
    """project_id is authoritative, so memory URL discovery must not run first."""
    await write_note(
        project=test_project.name,
        title="Project ID Memory URL Read",
        directory="test",
        content="Read by project id without discovery",
    )

    async def fail_if_called(*args, **kwargs):
        raise AssertionError("project_id routing should bypass URL discovery")

    import importlib

    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
    monkeypatch.setattr(read_note_module, "detect_project_from_identifier_prefix", fail_if_called)

    result = await read_note(
        f"memory://{test_project.name}/test/project-id-memory-url-read",
        project_id=test_project.external_id,
        output_format="json",
    )

    assert isinstance(result, dict)
    assert result["content"].strip() == "Read by project id without discovery"


@pytest.mark.asyncio
async def test_team_workspace_write_stores_complete_canonical_permalink(
    app,
    test_project,
    entity_repository,
):
    from basic_memory.workspace_context import workspace_permalink_context

    expected_permalink = f"team-paul/{test_project.name}/team/team-workspace-note"

    with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
        write_result = await write_note(
            project=test_project.name,
            title="Team Workspace Note",
            directory="team",
            content="Team canonical content",
        )

    assert f"permalink: {expected_permalink}" in write_result

    stored = await entity_repository.get_by_permalink(expected_permalink)
    assert stored is not None
    assert stored.permalink == expected_permalink

    read_result = await read_note(
        expected_permalink,
        project=test_project.name,
        output_format="json",
    )

    assert isinstance(read_result, dict)
    assert read_result["permalink"] == expected_permalink
    assert read_result["content"].strip() == "Team canonical content"


@pytest.mark.asyncio
async def test_team_workspace_write_stores_complete_permalink_when_project_prefixes_disabled(
    app,
    test_project,
    entity_repository,
    config_manager,
):
    from basic_memory.workspace_context import workspace_permalink_context

    config = config_manager.load_config()
    config.permalinks_include_project = False
    config_manager.save_config(config)

    expected_permalink = f"team-paul/{test_project.name}/team/team-no-project-prefix-note"

    with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
        write_result = await write_note(
            project=test_project.name,
            title="Team No Project Prefix Note",
            directory="team",
            content="Team canonical content without project-prefix config",
        )

    assert f"permalink: {expected_permalink}" in write_result

    stored = await entity_repository.get_by_permalink(expected_permalink)
    assert stored is not None
    assert stored.permalink == expected_permalink

    read_result = await read_note(
        f"memory://{expected_permalink}",
        project=test_project.name,
        output_format="json",
    )

    assert isinstance(read_result, dict)
    assert read_result["permalink"] == expected_permalink
    assert read_result["content"].strip() == "Team canonical content without project-prefix config"


@pytest.mark.asyncio
async def test_personal_workspace_write_stores_complete_canonical_permalink(
    app,
    test_project,
    entity_repository,
):
    from basic_memory.workspace_context import workspace_permalink_context

    expected_permalink = f"personal/{test_project.name}/personal/personal-workspace-note"

    with workspace_permalink_context(workspace_slug="personal", workspace_type="personal"):
        write_result = await write_note(
            project=test_project.name,
            title="Personal Workspace Note",
            directory="personal",
            content="Personal canonical content",
        )

    assert f"permalink: {expected_permalink}" in write_result

    stored = await entity_repository.get_by_permalink(expected_permalink)
    assert stored is not None
    assert stored.permalink == expected_permalink


@pytest.mark.asyncio
async def test_read_note_personal_workspace_keeps_short_project_permalink_working(
    app,
    test_project,
):
    from basic_memory.workspace_context import workspace_permalink_context

    legacy_permalink = f"{test_project.name}/personal/short-personal-note"

    await write_note(
        project=test_project.name,
        title="Short Personal Note",
        directory="personal",
        content="Short personal workspace content",
    )

    with workspace_permalink_context(workspace_slug="personal", workspace_type="personal"):
        read_result = await read_note(
            f"memory://{legacy_permalink}",
            project=test_project.name,
            output_format="json",
        )

    assert isinstance(read_result, dict)
    assert read_result["permalink"] == legacy_permalink
    assert read_result["content"].strip() == "Short personal workspace content"


@pytest.mark.asyncio
async def test_read_note_workspace_qualified_personal_url_finds_legacy_short_permalink(
    app,
    test_project,
    entity_repository,
):
    from basic_memory.workspace_context import workspace_permalink_context

    legacy_permalink = f"{test_project.name}/personal/legacy-personal-note"
    qualified_permalink = f"personal/{legacy_permalink}"

    await write_note(
        project=test_project.name,
        title="Legacy Personal Note",
        directory="personal",
        content="Legacy personal workspace content",
    )

    stored = await entity_repository.get_by_permalink(legacy_permalink)
    assert stored is not None
    assert stored.permalink == legacy_permalink

    with workspace_permalink_context(workspace_slug="personal", workspace_type="personal"):
        read_result = await read_note(
            f"memory://{qualified_permalink}",
            project=test_project.name,
            output_format="json",
        )

    assert isinstance(read_result, dict)
    assert read_result["permalink"] == legacy_permalink
    assert read_result["content"].strip() == "Legacy personal workspace content"


@pytest.mark.asyncio
async def test_read_note_workspace_qualified_memory_url_uses_complete_permalink(
    app,
    test_project,
):
    from basic_memory.workspace_context import workspace_permalink_context

    expected_permalink = f"team-paul/{test_project.name}/team/tool-read-note"

    with workspace_permalink_context(workspace_slug="team-paul", workspace_type="organization"):
        await write_note(
            project=test_project.name,
            title="Tool Read Note",
            directory="team",
            content="Workspace URL read content",
        )

    result = await read_note(
        f"memory://{expected_permalink}",
        project=test_project.name,
        output_format="json",
    )

    assert isinstance(result, dict)
    assert result["permalink"] == expected_permalink
    assert result["content"].strip() == "Workspace URL read content"


@pytest.mark.asyncio
async def test_read_note_memory_url_fallback_uses_search_tool_normalization(
    monkeypatch, app, test_project
):
    """Fallback search should go back through search_notes for memory:// normalization."""
    await write_note(
        project=test_project.name,
        title="Memory URL Fallback Note",
        directory="test",
        content="Fallback note content",
    )

    import importlib

    read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
    clients_mod = importlib.import_module("basic_memory.mcp.clients")
    OriginalKnowledgeClient = clients_mod.KnowledgeClient

    fallback_memory_url = f"memory://{test_project.name}/test/memory-url-fallback-note"
    search_calls: list[tuple[str, str, str | None]] = []

    class SelectiveKnowledgeClient(OriginalKnowledgeClient):
        async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
            if strict and identifier.endswith("test/memory-url-fallback-note"):
                raise RuntimeError("force direct lookup failure")
            return await super().resolve_entity(identifier, strict=strict)

    async def fake_search_notes_fn(*, query, search_type, project, **kwargs):
        search_calls.append((search_type, query, project))
        return {
            "results": [
                {
                    "title": "Memory URL Fallback Note",
                    "permalink": "test/memory-url-fallback-note",
                    "content": "",
                    "type": "entity",
                    "score": 1.0,
                    "file_path": "test/Memory URL Fallback Note.md",
                }
            ],
            "current_page": 1,
            "page_size": 10,
        }

    monkeypatch.setattr(clients_mod, "KnowledgeClient", SelectiveKnowledgeClient)
    monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)

    result = await read_note(fallback_memory_url)

    assert search_calls == [
        ("title", fallback_memory_url, test_project.name),
        ("text", fallback_memory_url, test_project.name),
    ]
    assert "I couldn't find an exact match" in result
    assert "Memory URL Fallback Note" in result


class TestReadNoteSecurityValidation:
    """Test read_note security validation features."""

    @pytest.mark.asyncio
    async def test_read_note_blocks_path_traversal_unix(self, app, test_project):
        """Test that Unix-style path traversal attacks are blocked in identifier parameter."""
        # Test various Unix-style path traversal patterns
        attack_identifiers = [
            "../secrets.txt",
            "../../etc/passwd",
            "../../../root/.ssh/id_rsa",
            "notes/../../../etc/shadow",
            "folder/../../outside/file.md",
            "../../../../etc/hosts",
            "../../../home/user/.env",
        ]

        for attack_identifier in attack_identifiers:
            result = await read_note(attack_identifier, project=test_project.name)

            assert isinstance(result, str)
            assert "# Error" in result
            assert "paths must stay within project boundaries" in result
            assert attack_identifier in result

    @pytest.mark.asyncio
    async def test_read_note_blocks_path_traversal_windows(self, app, test_project):
        """Test that Windows-style path traversal attacks are blocked in identifier parameter."""
        # Test various Windows-style path traversal patterns
        attack_identifiers = [
            "..\\secrets.txt",
            "..\\..\\Windows\\System32\\config\\SAM",
            "notes\\..\\..\\..\\Windows\\System32",
            "\\\\server\\share\\file.txt",
            "..\\..\\Users\\user\\.env",
            "\\\\..\\..\\Windows",
            "..\\..\\..\\Boot.ini",
        ]

        for attack_identifier in attack_identifiers:
            result = await read_note(attack_identifier, project=test_project.name)

            assert isinstance(result, str)
            assert "# Error" in result
            assert "paths must stay within project boundaries" in result
            assert attack_identifier in result

    @pytest.mark.asyncio
    async def test_read_note_blocks_absolute_paths(self, app, test_project):
        """Test that absolute paths are blocked in identifier parameter."""
        # Test various absolute path patterns
        attack_identifiers = [
            "/etc/passwd",
            "/home/user/.env",
            "/var/log/auth.log",
            "/root/.ssh/id_rsa",
            "C:\\Windows\\System32\\config\\SAM",
            "C:\\Users\\user\\.env",
            "D:\\secrets\\config.json",
            "/tmp/malicious.txt",
            "/usr/local/bin/evil",
        ]

        for attack_identifier in attack_identifiers:
            result = await read_note(project=test_project.name, identifier=attack_identifier)

            assert isinstance(result, str)
            assert "# Error" in result
            assert "paths must stay within project boundaries" in result
            assert attack_identifier in result

    @pytest.mark.asyncio
    async def test_read_note_blocks_home_directory_access(self, app, test_project):
        """Test that home directory access patterns are blocked in identifier parameter."""
        # Test various home directory access patterns
        attack_identifiers = [
            "~/secrets.txt",
            "~/.env",
            "~/.ssh/id_rsa",
            "~/Documents/passwords.txt",
            "~\\AppData\\secrets",
            "~\\Desktop\\config.ini",
            "~/.bashrc",
            "~/Library/Preferences/secret.plist",
        ]

        for attack_identifier in attack_identifiers:
            result = await read_note(project=test_project.name, identifier=attack_identifier)

            assert isinstance(result, str)
            assert "# Error" in result
            assert "paths must stay within project boundaries" in result
            assert attack_identifier in result

    @pytest.mark.asyncio
    async def test_read_note_blocks_memory_url_attacks(self, app, test_project):
        """Test that memory URLs with path traversal are blocked."""
        # Test memory URLs with attacks embedded
        attack_identifiers = [
            "memory://../../etc/passwd",
            "memory://../../../root/.ssh/id_rsa",
            "memory://~/.env",
            "memory:///etc/passwd",
            "memory://notes/../../../etc/shadow",
            "memory://..\\..\\Windows\\System32",
        ]

        for attack_identifier in attack_identifiers:
            result = await read_note(project=test_project.name, identifier=attack_identifier)

            assert isinstance(result, str)
            assert "# Error" in result
            assert "paths must stay within project boundaries" in result

    @pytest.mark.asyncio
    async def test_read_note_blocks_mixed_attack_patterns(self, app, test_project):
        """Test that mixed legitimate/attack patterns are blocked in identifier parameter."""
        # Test mixed patterns that start legitimate but contain attacks
        attack_identifiers = [
            "notes/../../../etc/passwd",
            "docs/../../.env",
            "legitimate/path/../../.ssh/id_rsa",
            "project/folder/../../../Windows/System32",
            "valid/folder/../../home/user/.bashrc",
            "assets/../../../tmp/evil.exe",
        ]

        for attack_identifier in attack_identifiers:
            result = await read_note(project=test_project.name, identifier=attack_identifier)

            assert isinstance(result, str)
            assert "# Error" in result
            assert "paths must stay within project boundaries" in result

    @pytest.mark.asyncio
    async def test_read_note_allows_safe_identifiers(self, app, test_project):
        """Test that legitimate identifiers are still allowed."""
        # Test various safe identifier patterns
        safe_identifiers = [
            "notes/meeting",
            "docs/readme",
            "projects/2025/planning",
            "archive/old-notes/backup",
            "folder/subfolder/document",
            "research/ml/algorithms",
            "meeting-notes",
            "test/simple-note",
        ]

        for safe_identifier in safe_identifiers:
            result = await read_note(project=test_project.name, identifier=safe_identifier)

            assert isinstance(result, str)
            # Should not contain security error message
            assert (
                "# Error" not in result or "paths must stay within project boundaries" not in result
            )
            # Should either succeed or fail for legitimate reasons (not found, etc.)
            # but not due to security validation

    @pytest.mark.asyncio
    async def test_read_note_allows_legitimate_titles(self, app, test_project):
        """Test that legitimate note titles work normally."""
        # Create a test note first
        await write_note(
            project=test_project.name,
            title="Security Test Note",
            directory="security-tests",
            content="# Security Test Note\nThis is a legitimate note for security testing.",
        )

        # Test reading by title (should work)
        result = await read_note("Security Test Note", project=test_project.name)

        assert isinstance(result, str)
        # Should not be a security error
        assert "# Error" not in result or "paths must stay within project boundaries" not in result
        # Should either return the note content or search results

    @pytest.mark.asyncio
    async def test_read_note_empty_identifier_security(self, app, test_project):
        """Test that empty identifier is handled securely."""
        # Empty identifier should be allowed (may return search results or error, but not security error)
        result = await read_note(identifier="", project=test_project.name)

        assert isinstance(result, str)
        # Empty identifier should not trigger security error
        assert "# Error" not in result or "paths must stay within project boundaries" not in result

    @pytest.mark.asyncio
    async def test_read_note_security_with_all_parameters(self, app, test_project):
        """Test security validation works with all read_note parameters."""
        # Test that security validation is applied even when all other parameters are provided
        result = await read_note(
            project=test_project.name,
            identifier="../../../etc/malicious",
        )

        assert isinstance(result, str)
        assert "# Error" in result
        assert "paths must stay within project boundaries" in result
        assert "../../../etc/malicious" in result

    @pytest.mark.asyncio
    async def test_read_note_security_logging(self, app, caplog, test_project):
        """Test that security violations are properly logged."""
        # Attempt path traversal attack
        result = await read_note(identifier="../../../etc/passwd", project=test_project.name)

        assert "# Error" in result
        assert "paths must stay within project boundaries" in result

        # Check that security violation was logged
        # Note: This test may need adjustment based on the actual logging setup
        # The security validation should generate a warning log entry

    @pytest.mark.asyncio
    async def test_read_note_preserves_functionality_with_security(self, app, test_project):
        """Test that security validation doesn't break normal note reading functionality."""
        # Create a note with complex content to ensure security validation doesn't interfere
        await write_note(
            project=test_project.name,
            title="Full Feature Security Test Note",
            directory="security-tests",
            content=dedent("""
                # Full Feature Security Test Note
                
                This note tests that security validation doesn't break normal functionality.
                
                ## Observations
                - [security] Path validation working correctly #security
                - [feature] All features still functional #test
                
                ## Relations
                - relates_to [[Security Implementation]]
                - depends_on [[Path Validation]]
                
                Additional content with various formatting.
            """).strip(),
            tags=["security", "test", "full-feature"],
            note_type="guide",
        )

        # Test reading by permalink
        result = await read_note(
            "security-tests/full-feature-security-test-note", project=test_project.name
        )

        # Should succeed normally (not a security error)
        assert isinstance(result, str)
        assert "# Error" not in result or "paths must stay within project boundaries" not in result
        # Should either return content or search results, but not security error


class TestReadNoteSecurityEdgeCases:
    """Test edge cases for read_note security validation."""

    @pytest.mark.asyncio
    async def test_read_note_unicode_identifier_attacks(self, app, test_project):
        """Test that Unicode-based path traversal attempts are blocked."""
        # Test Unicode path traversal attempts
        unicode_attack_identifiers = [
            "notes/文档/../../../etc/passwd",  # Chinese characters
            "docs/café/../../.env",  # Accented characters
            "files/αβγ/../../../secret.txt",  # Greek characters
        ]

        for attack_identifier in unicode_attack_identifiers:
            result = await read_note(attack_identifier, project=test_project.name)

            assert isinstance(result, str)
            assert "# Error" in result
            assert "paths must stay within project boundaries" in result

    @pytest.mark.asyncio
    async def test_read_note_very_long_attack_identifier(self, app, test_project):
        """Test handling of very long attack identifiers."""
        # Create a very long path traversal attack
        long_attack_identifier = "../" * 1000 + "etc/malicious"

        result = await read_note(long_attack_identifier, project=test_project.name)

        assert isinstance(result, str)
        assert "# Error" in result
        assert "paths must stay within project boundaries" in result

    @pytest.mark.asyncio
    async def test_read_note_case_variations_attacks(self, app, test_project):
        """Test that case variations don't bypass security."""
        # Test case variations (though case sensitivity depends on filesystem)
        case_attack_identifiers = [
            "../ETC/passwd",
            "../Etc/PASSWD",
            "..\\WINDOWS\\system32",
            "~/.SSH/id_rsa",
        ]

        for attack_identifier in case_attack_identifiers:
            result = await read_note(attack_identifier, project=test_project.name)

            assert isinstance(result, str)
            assert "# Error" in result
            assert "paths must stay within project boundaries" in result

    @pytest.mark.asyncio
    async def test_read_note_whitespace_in_attack_identifiers(self, app, test_project):
        """Test that whitespace doesn't help bypass security."""
        # Test attack identifiers with various whitespace
        whitespace_attack_identifiers = [
            " ../../../etc/passwd ",
            "\t../../../secrets\t",
            " ..\\..\\Windows ",
            "notes/ ../../ malicious",
        ]

        for attack_identifier in whitespace_attack_identifiers:
            result = await read_note(attack_identifier, project=test_project.name)

            assert isinstance(result, str)
            # The attack should still be blocked even with whitespace
            if ".." in attack_identifier.strip() or "~" in attack_identifier.strip():
                assert "# Error" in result
                assert "paths must stay within project boundaries" in result