mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Drew Cain
510 lines
16 KiB
Python
510 lines
16 KiB
Python
"""Integration coverage for local, cloud, and team permalink routing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Iterable
|
|
from contextlib import asynccontextmanager, contextmanager
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastmcp import Client
|
|
from httpx import ASGITransport, AsyncClient as HttpxAsyncClient
|
|
|
|
from basic_memory.config import BasicMemoryConfig, ConfigManager, ProjectEntry
|
|
from basic_memory.mcp import async_client
|
|
from basic_memory.mcp import project_context
|
|
from basic_memory.models import Project
|
|
from basic_memory.repository.project_repository import ProjectRepository
|
|
from basic_memory.schemas.cloud import WorkspaceInfo
|
|
from basic_memory.workspace_context import workspace_permalink_headers
|
|
|
|
|
|
def _json_content(tool_result) -> Any:
|
|
"""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) # pyright: ignore [reportAttributeAccessIssue]
|
|
|
|
|
|
def _workspace(
|
|
*,
|
|
tenant_id: str,
|
|
slug: str,
|
|
workspace_type: str,
|
|
is_default: bool = True,
|
|
) -> WorkspaceInfo:
|
|
return WorkspaceInfo(
|
|
tenant_id=tenant_id,
|
|
name=slug.replace("-", " ").title(),
|
|
workspace_type=workspace_type,
|
|
slug=slug,
|
|
role="owner",
|
|
organization_id=None,
|
|
is_default=is_default,
|
|
has_active_subscription=True,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def personal_workspace() -> WorkspaceInfo:
|
|
return _workspace(
|
|
tenant_id="personal-tenant",
|
|
slug="personal",
|
|
workspace_type="personal",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def team_workspace() -> WorkspaceInfo:
|
|
return _workspace(
|
|
tenant_id="team-tenant",
|
|
slug="team-paul",
|
|
workspace_type="organization",
|
|
is_default=False,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def route_workspaces(app):
|
|
@contextmanager
|
|
def route(*workspaces: WorkspaceInfo, forward_permalink_headers: bool = True):
|
|
with _workspace_routing(
|
|
app,
|
|
workspaces,
|
|
forward_permalink_headers=forward_permalink_headers,
|
|
):
|
|
yield
|
|
|
|
return route
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def alternate_project(
|
|
config_home,
|
|
engine_factory,
|
|
app_config: BasicMemoryConfig,
|
|
config_manager: ConfigManager,
|
|
) -> Project:
|
|
"""Create a second project so no-project lookups cannot pass via the default."""
|
|
alternate_path = Path(config_home) / "alternate-project"
|
|
alternate_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
_, session_maker = engine_factory
|
|
project_repository = ProjectRepository(session_maker)
|
|
project = await project_repository.create(
|
|
{
|
|
"name": "alternate-project",
|
|
"description": "Non-default project for prefix routing tests",
|
|
"path": str(alternate_path),
|
|
"is_active": True,
|
|
"is_default": False,
|
|
}
|
|
)
|
|
|
|
app_config.projects[project.name] = ProjectEntry(path=str(alternate_path))
|
|
config_manager.save_config(app_config)
|
|
return project
|
|
|
|
|
|
def _save_permalink_config(
|
|
app_config: BasicMemoryConfig,
|
|
*,
|
|
include_project: bool,
|
|
default_project: str | None,
|
|
) -> None:
|
|
app_config.permalinks_include_project = include_project
|
|
app_config.default_project = default_project
|
|
ConfigManager().save_config(app_config)
|
|
|
|
|
|
@contextmanager
|
|
def _workspace_routing(
|
|
app,
|
|
workspaces: Iterable[WorkspaceInfo],
|
|
*,
|
|
forward_permalink_headers: bool = True,
|
|
):
|
|
"""Route MCP tool HTTP calls through an ASGI-backed cloud workspace seam."""
|
|
workspace_list = tuple(workspaces)
|
|
workspace_ids = {workspace.tenant_id for workspace in workspace_list}
|
|
|
|
async def workspace_provider():
|
|
return list(workspace_list)
|
|
|
|
@asynccontextmanager
|
|
async def factory(workspace: str | None = None):
|
|
assert workspace is None or workspace in workspace_ids
|
|
headers = workspace_permalink_headers() if forward_permalink_headers else {}
|
|
async with HttpxAsyncClient(
|
|
transport=ASGITransport(app=app),
|
|
base_url="http://test",
|
|
headers=headers,
|
|
) as inner:
|
|
yield inner
|
|
|
|
original_factory = async_client._client_factory
|
|
original_workspace_provider = project_context._workspace_provider
|
|
async_client.set_client_factory(factory)
|
|
project_context.set_workspace_provider(workspace_provider)
|
|
try:
|
|
yield
|
|
finally:
|
|
async_client._client_factory = original_factory
|
|
project_context._workspace_provider = original_workspace_provider
|
|
|
|
|
|
async def _call_json(mcp_server, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
|
|
async with Client(mcp_server) as client:
|
|
result = await client.call_tool(tool_name, arguments)
|
|
|
|
payload = _json_content(result)
|
|
assert isinstance(payload, dict)
|
|
return payload
|
|
|
|
|
|
async def _write_json(
|
|
mcp_server,
|
|
*,
|
|
project: str,
|
|
title: str,
|
|
directory: str = "permalink-suite",
|
|
) -> dict[str, Any]:
|
|
return await _call_json(
|
|
mcp_server,
|
|
"write_note",
|
|
{
|
|
"project": project,
|
|
"title": title,
|
|
"directory": directory,
|
|
"content": f"# {title}\n\nUnique body for {title}.",
|
|
"output_format": "json",
|
|
},
|
|
)
|
|
|
|
|
|
async def _read_json(
|
|
mcp_server,
|
|
*,
|
|
identifier: str,
|
|
project: str | None = None,
|
|
) -> dict[str, Any]:
|
|
arguments = {
|
|
"identifier": identifier,
|
|
"output_format": "json",
|
|
}
|
|
if project is not None:
|
|
arguments["project"] = project
|
|
return await _call_json(mcp_server, "read_note", arguments)
|
|
|
|
|
|
async def _search_single_permalink(
|
|
mcp_server,
|
|
*,
|
|
title: str,
|
|
project: str | None = None,
|
|
) -> str:
|
|
arguments = {
|
|
"query": title,
|
|
"search_type": "text",
|
|
"output_format": "json",
|
|
}
|
|
if project is not None:
|
|
arguments["project"] = project
|
|
|
|
payload = await _call_json(mcp_server, "search_notes", arguments)
|
|
matching_results = [
|
|
item for item in payload["results"] if isinstance(item, dict) and item.get("title") == title
|
|
]
|
|
assert len(matching_results) == 1
|
|
permalink = matching_results[0].get("permalink")
|
|
assert isinstance(permalink, str)
|
|
return permalink
|
|
|
|
|
|
async def _search_permalink_exact(
|
|
mcp_server,
|
|
*,
|
|
permalink: str,
|
|
title: str,
|
|
project: str | None = None,
|
|
) -> str:
|
|
arguments = {
|
|
"query": permalink,
|
|
"search_type": "permalink",
|
|
"output_format": "json",
|
|
}
|
|
if project is not None:
|
|
arguments["project"] = project
|
|
|
|
payload = await _call_json(mcp_server, "search_notes", arguments)
|
|
matching_results = [
|
|
item for item in payload["results"] if isinstance(item, dict) and item.get("title") == title
|
|
]
|
|
assert len(matching_results) == 1
|
|
result_permalink = matching_results[0].get("permalink")
|
|
assert isinstance(result_permalink, str)
|
|
return result_permalink
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_short_permalink_round_trips_when_project_is_supplied(
|
|
mcp_server,
|
|
test_project,
|
|
app_config,
|
|
):
|
|
"""Local short IDs need the project argument for write/search/read routing."""
|
|
_save_permalink_config(app_config, include_project=False, default_project=None)
|
|
|
|
title = "Local Short Permalink"
|
|
expected_permalink = "permalink-suite/local-short-permalink"
|
|
|
|
write_payload = await _write_json(mcp_server, project=test_project.name, title=title)
|
|
assert write_payload["permalink"] == expected_permalink
|
|
|
|
assert (
|
|
await _search_single_permalink(mcp_server, project=test_project.name, title=title)
|
|
== expected_permalink
|
|
)
|
|
|
|
short_read = await _read_json(
|
|
mcp_server,
|
|
project=test_project.name,
|
|
identifier=expected_permalink,
|
|
)
|
|
assert short_read["permalink"] == expected_permalink
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_project_permalink_routes_without_project_argument(
|
|
mcp_server,
|
|
alternate_project,
|
|
app_config,
|
|
):
|
|
"""Local project-qualified IDs carry enough route context for search/read."""
|
|
_save_permalink_config(app_config, include_project=True, default_project=None)
|
|
|
|
title = "Local Project Permalink"
|
|
short_permalink = "permalink-suite/local-project-permalink"
|
|
expected_permalink = f"{alternate_project.name}/{short_permalink}"
|
|
|
|
write_payload = await _write_json(mcp_server, project=alternate_project.name, title=title)
|
|
assert write_payload["permalink"] == expected_permalink
|
|
|
|
assert (
|
|
await _search_permalink_exact(
|
|
mcp_server,
|
|
permalink=expected_permalink,
|
|
title=title,
|
|
)
|
|
== expected_permalink
|
|
)
|
|
|
|
project_read = await _read_json(mcp_server, identifier=expected_permalink)
|
|
assert project_read["permalink"] == expected_permalink
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_personal_cloud_short_permalink_round_trips_when_project_is_supplied(
|
|
mcp_server,
|
|
test_project,
|
|
app_config,
|
|
personal_workspace,
|
|
route_workspaces,
|
|
):
|
|
"""Legacy short IDs remain readable/searchable in a personal cloud workspace."""
|
|
_save_permalink_config(app_config, include_project=False, default_project=None)
|
|
|
|
title = "Personal Cloud Short Permalink"
|
|
expected_permalink = "permalink-suite/personal-cloud-short-permalink"
|
|
|
|
write_payload = await _write_json(mcp_server, project=test_project.name, title=title)
|
|
assert write_payload["permalink"] == expected_permalink
|
|
|
|
with route_workspaces(personal_workspace):
|
|
assert (
|
|
await _search_single_permalink(
|
|
mcp_server,
|
|
project=test_project.name,
|
|
title=title,
|
|
)
|
|
== expected_permalink
|
|
)
|
|
|
|
short_read = await _read_json(
|
|
mcp_server,
|
|
project=test_project.name,
|
|
identifier=expected_permalink,
|
|
)
|
|
assert short_read["permalink"] == expected_permalink
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_personal_cloud_project_permalink_routes_without_project_argument(
|
|
mcp_server,
|
|
alternate_project,
|
|
app_config,
|
|
personal_workspace,
|
|
route_workspaces,
|
|
):
|
|
"""Legacy project-qualified cloud IDs carry enough route context."""
|
|
_save_permalink_config(app_config, include_project=True, default_project=None)
|
|
|
|
title = "Personal Cloud Project Permalink"
|
|
short_permalink = "permalink-suite/personal-cloud-project-permalink"
|
|
expected_permalink = f"{alternate_project.name}/{short_permalink}"
|
|
|
|
write_payload = await _write_json(mcp_server, project=alternate_project.name, title=title)
|
|
assert write_payload["permalink"] == expected_permalink
|
|
|
|
with route_workspaces(personal_workspace):
|
|
assert (
|
|
await _search_permalink_exact(
|
|
mcp_server,
|
|
permalink=expected_permalink,
|
|
title=title,
|
|
)
|
|
== expected_permalink
|
|
)
|
|
|
|
project_read = await _read_json(mcp_server, identifier=expected_permalink)
|
|
assert project_read["permalink"] == expected_permalink
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_team_short_permalink_round_trips_when_project_is_supplied(
|
|
mcp_server,
|
|
test_project,
|
|
app_config,
|
|
team_workspace,
|
|
route_workspaces,
|
|
):
|
|
"""Team short IDs need the qualified project argument for search/read routing."""
|
|
team_project = f"{team_workspace.slug}/{test_project.name}"
|
|
|
|
_save_permalink_config(app_config, include_project=False, default_project=None)
|
|
title = "Team Short Permalink"
|
|
expected_permalink = "permalink-suite/team-short-permalink"
|
|
|
|
write_payload = await _write_json(mcp_server, project=test_project.name, title=title)
|
|
assert write_payload["permalink"] == expected_permalink
|
|
|
|
with route_workspaces(team_workspace):
|
|
assert (
|
|
await _search_single_permalink(
|
|
mcp_server,
|
|
project=team_project,
|
|
title=title,
|
|
)
|
|
== expected_permalink
|
|
)
|
|
short_read = await _read_json(
|
|
mcp_server,
|
|
project=team_project,
|
|
identifier=expected_permalink,
|
|
)
|
|
assert short_read["permalink"] == expected_permalink
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_team_project_permalink_routes_without_project_argument(
|
|
mcp_server,
|
|
alternate_project,
|
|
app_config,
|
|
team_workspace,
|
|
route_workspaces,
|
|
):
|
|
"""Team project-qualified IDs carry enough route context for search/read."""
|
|
_save_permalink_config(app_config, include_project=True, default_project=None)
|
|
|
|
title = "Team Project Permalink"
|
|
short_permalink = "permalink-suite/team-project-permalink"
|
|
expected_permalink = f"{alternate_project.name}/{short_permalink}"
|
|
|
|
write_payload = await _write_json(mcp_server, project=alternate_project.name, title=title)
|
|
assert write_payload["permalink"] == expected_permalink
|
|
|
|
with route_workspaces(team_workspace):
|
|
assert (
|
|
await _search_permalink_exact(
|
|
mcp_server,
|
|
permalink=expected_permalink,
|
|
title=title,
|
|
)
|
|
== expected_permalink
|
|
)
|
|
project_read = await _read_json(mcp_server, identifier=expected_permalink)
|
|
assert project_read["permalink"] == expected_permalink
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_team_workspace_permalink_routes_to_specific_workspace(
|
|
mcp_server,
|
|
test_project,
|
|
app_config,
|
|
personal_workspace,
|
|
team_workspace,
|
|
route_workspaces,
|
|
):
|
|
"""Workspace-qualified IDs route to the project in that workspace."""
|
|
_save_permalink_config(app_config, include_project=True, default_project=None)
|
|
team_project = f"{team_workspace.slug}/{test_project.name}"
|
|
|
|
title = "Team Workspace Permalink"
|
|
short_permalink = "permalink-suite/team-workspace-permalink"
|
|
expected_permalink = f"{team_workspace.slug}/{test_project.name}/{short_permalink}"
|
|
|
|
with route_workspaces(personal_workspace, team_workspace):
|
|
write_payload = await _write_json(mcp_server, project=team_project, title=title)
|
|
assert write_payload["permalink"] == expected_permalink
|
|
|
|
assert (
|
|
await _search_permalink_exact(
|
|
mcp_server,
|
|
permalink=expected_permalink,
|
|
title=title,
|
|
)
|
|
== expected_permalink
|
|
)
|
|
|
|
workspace_read = await _read_json(mcp_server, identifier=expected_permalink)
|
|
assert workspace_read["permalink"] == expected_permalink
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_note_by_project_id_qualifies_permalink_when_headers_not_forwarded(
|
|
mcp_server,
|
|
test_project,
|
|
app_config,
|
|
team_workspace,
|
|
route_workspaces,
|
|
):
|
|
"""MCP writes should return self-routing IDs even if the API omits slug headers."""
|
|
_save_permalink_config(app_config, include_project=True, default_project=None)
|
|
|
|
title = "Project Id Workspace Permalink"
|
|
short_permalink = "permalink-suite/project-id-workspace-permalink"
|
|
expected_permalink = f"{team_workspace.slug}/{test_project.name}/{short_permalink}"
|
|
|
|
with route_workspaces(team_workspace, forward_permalink_headers=False):
|
|
write_payload = await _call_json(
|
|
mcp_server,
|
|
"write_note",
|
|
{
|
|
"project_id": test_project.external_id,
|
|
"title": title,
|
|
"directory": "permalink-suite",
|
|
"content": f"# {title}\n\nProject ID workspace body.",
|
|
"output_format": "json",
|
|
},
|
|
)
|
|
assert write_payload["permalink"] == expected_permalink
|
|
|
|
workspace_read = await _read_json(
|
|
mcp_server,
|
|
identifier=f"memory://{write_payload['permalink']}",
|
|
)
|
|
assert workspace_read["title"] == title
|