Compare commits

...

3 Commits

Author SHA1 Message Date
phernandez 3050fe8318 ci: replace flaky just.systems curl install with extractions/setup-just action
The just.systems install script intermittently returns HTTP 403, causing
CI jobs to fail before tests even run. The extractions/setup-just GitHub
Action downloads from GitHub releases instead, which is reliable.

Also removes the separate Windows choco install step since the action
handles all platforms natively.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>
2026-02-13 15:39:11 -06:00
phernandez b367ccbdb8 fix: update tests and fix bugs for project-prefixed permalinks
- Fix update_entity overwriting project-prefixed permalink with
  non-prefixed one during metadata merge
- Fix memory:// URL path traversal validation bypass in read_note
  and read_content tools
- Fix pyright type error in claude_projects_importer
- Update test assertions across 12 test files to use project-prefixed
  permalinks (test-project/path instead of path)
- Fix monkeypatch in search test to use resolve_project_and_path
  instead of removed get_active_project

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>
2026-02-13 15:39:11 -06:00
phernandez c6511aa745 feat: enable project-prefixed permalinks
Signed-off-by: phernandez <paul@basicmachines.co>
2026-02-13 15:39:10 -06:00
47 changed files with 1142 additions and 462 deletions
+3 -20
View File
@@ -37,18 +37,7 @@ jobs:
run: | run: |
pip install uv pip install uv
- name: Install just (Linux) - uses: extractions/setup-just@v3
if: runner.os != 'Windows'
run: |
sudo apt-get update
sudo apt-get install -y just
- name: Install just (Windows)
if: runner.os == 'Windows'
run: |
# Install just using Chocolatey (pre-installed on GitHub Actions Windows runners)
choco install just --yes
shell: pwsh
- name: Create virtual env - name: Create virtual env
run: | run: |
@@ -96,10 +85,7 @@ jobs:
run: | run: |
pip install uv pip install uv
- name: Install just - uses: extractions/setup-just@v3
run: |
sudo apt-get update
sudo apt-get install -y just
- name: Create virtual env - name: Create virtual env
run: | run: |
@@ -133,10 +119,7 @@ jobs:
run: | run: |
pip install uv pip install uv
- name: Install just - uses: extractions/setup-just@v3
run: |
sudo apt-get update
sudo apt-get install -y just
- name: Create virtual env - name: Create virtual env
run: | run: |
@@ -60,7 +60,9 @@ def import_chatgpt(
console.print(f"\nImporting chats from {conversations_json}...writing to {base_path}") console.print(f"\nImporting chats from {conversations_json}...writing to {base_path}")
# Create importer and run import # Create importer and run import
importer = ChatGPTImporter(config.home, markdown_processor, file_service) importer = ChatGPTImporter(
config.home, markdown_processor, file_service, project_name=config.name
)
with conversations_json.open("r", encoding="utf-8") as file: with conversations_json.open("r", encoding="utf-8") as file:
json_data = json.load(file) json_data = json.load(file)
result = run_with_cleanup(importer.import_data(json_data, folder)) result = run_with_cleanup(importer.import_data(json_data, folder))
@@ -57,7 +57,9 @@ def import_claude(
markdown_processor, file_service = run_with_cleanup(get_importer_dependencies()) markdown_processor, file_service = run_with_cleanup(get_importer_dependencies())
# Create the importer # Create the importer
importer = ClaudeConversationsImporter(config.home, markdown_processor, file_service) importer = ClaudeConversationsImporter(
config.home, markdown_processor, file_service, project_name=config.name
)
# Process the file # Process the file
base_path = config.home / folder base_path = config.home / folder
@@ -56,7 +56,9 @@ def import_projects(
markdown_processor, file_service = run_with_cleanup(get_importer_dependencies()) markdown_processor, file_service = run_with_cleanup(get_importer_dependencies())
# Create the importer # Create the importer
importer = ClaudeProjectsImporter(config.home, markdown_processor, file_service) importer = ClaudeProjectsImporter(
config.home, markdown_processor, file_service, project_name=config.name
)
# Process the file # Process the file
base_path = config.home / base_folder if base_folder else config.home base_path = config.home / base_folder if base_folder else config.home
@@ -55,7 +55,9 @@ def memory_json(
markdown_processor, file_service = run_with_cleanup(get_importer_dependencies()) markdown_processor, file_service = run_with_cleanup(get_importer_dependencies())
# Create the importer # Create the importer
importer = MemoryJsonImporter(config.home, markdown_processor, file_service) importer = MemoryJsonImporter(
config.home, markdown_processor, file_service, project_name=config.name
)
# Process the file # Process the file
base_path = config.home if not destination_folder else config.home / destination_folder base_path = config.home if not destination_folder else config.home / destination_folder
+5
View File
@@ -196,6 +196,11 @@ class BasicMemoryConfig(BaseSettings):
description="Disable automatic permalink generation in frontmatter. When enabled, new notes won't have permalinks added and sync won't update permalinks. Existing permalinks will still work for reading.", description="Disable automatic permalink generation in frontmatter. When enabled, new notes won't have permalinks added and sync won't update permalinks. Existing permalinks will still work for reading.",
) )
permalinks_include_project: bool = Field(
default=True,
description="When True, generated permalinks are prefixed with the project slug (e.g., 'specs/search'). Existing permalinks remain unchanged unless explicitly updated.",
)
skip_initialization_sync: bool = Field( skip_initialization_sync: bool = Field(
default=False, default=False,
description="Skip expensive initialization synchronization. Useful for cloud/stateless deployments where project reconciliation is not needed.", description="Skip expensive initialization synchronization. Useful for cloud/stateless deployments where project reconciliation is not needed.",
+72 -12
View File
@@ -41,7 +41,12 @@ async def get_chatgpt_importer(
file_service: FileServiceDep, file_service: FileServiceDep,
) -> ChatGPTImporter: ) -> ChatGPTImporter:
"""Create ChatGPTImporter with dependencies.""" """Create ChatGPTImporter with dependencies."""
return ChatGPTImporter(project_config.home, markdown_processor, file_service) return ChatGPTImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ChatGPTImporterDep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer)] ChatGPTImporterDep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer)]
@@ -53,7 +58,12 @@ async def get_chatgpt_importer_v2( # pragma: no cover
file_service: FileServiceV2Dep, file_service: FileServiceV2Dep,
) -> ChatGPTImporter: ) -> ChatGPTImporter:
"""Create ChatGPTImporter with v2 dependencies.""" """Create ChatGPTImporter with v2 dependencies."""
return ChatGPTImporter(project_config.home, markdown_processor, file_service) return ChatGPTImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ChatGPTImporterV2Dep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer_v2)] ChatGPTImporterV2Dep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer_v2)]
@@ -65,7 +75,12 @@ async def get_chatgpt_importer_v2_external(
file_service: FileServiceV2ExternalDep, file_service: FileServiceV2ExternalDep,
) -> ChatGPTImporter: ) -> ChatGPTImporter:
"""Create ChatGPTImporter with v2 external_id dependencies.""" """Create ChatGPTImporter with v2 external_id dependencies."""
return ChatGPTImporter(project_config.home, markdown_processor, file_service) return ChatGPTImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ChatGPTImporterV2ExternalDep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer_v2_external)] ChatGPTImporterV2ExternalDep = Annotated[ChatGPTImporter, Depends(get_chatgpt_importer_v2_external)]
@@ -80,7 +95,12 @@ async def get_claude_conversations_importer(
file_service: FileServiceDep, file_service: FileServiceDep,
) -> ClaudeConversationsImporter: ) -> ClaudeConversationsImporter:
"""Create ClaudeConversationsImporter with dependencies.""" """Create ClaudeConversationsImporter with dependencies."""
return ClaudeConversationsImporter(project_config.home, markdown_processor, file_service) return ClaudeConversationsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ClaudeConversationsImporterDep = Annotated[ ClaudeConversationsImporterDep = Annotated[
@@ -94,7 +114,12 @@ async def get_claude_conversations_importer_v2( # pragma: no cover
file_service: FileServiceV2Dep, file_service: FileServiceV2Dep,
) -> ClaudeConversationsImporter: ) -> ClaudeConversationsImporter:
"""Create ClaudeConversationsImporter with v2 dependencies.""" """Create ClaudeConversationsImporter with v2 dependencies."""
return ClaudeConversationsImporter(project_config.home, markdown_processor, file_service) return ClaudeConversationsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ClaudeConversationsImporterV2Dep = Annotated[ ClaudeConversationsImporterV2Dep = Annotated[
@@ -108,7 +133,12 @@ async def get_claude_conversations_importer_v2_external(
file_service: FileServiceV2ExternalDep, file_service: FileServiceV2ExternalDep,
) -> ClaudeConversationsImporter: ) -> ClaudeConversationsImporter:
"""Create ClaudeConversationsImporter with v2 external_id dependencies.""" """Create ClaudeConversationsImporter with v2 external_id dependencies."""
return ClaudeConversationsImporter(project_config.home, markdown_processor, file_service) return ClaudeConversationsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ClaudeConversationsImporterV2ExternalDep = Annotated[ ClaudeConversationsImporterV2ExternalDep = Annotated[
@@ -125,7 +155,12 @@ async def get_claude_projects_importer(
file_service: FileServiceDep, file_service: FileServiceDep,
) -> ClaudeProjectsImporter: ) -> ClaudeProjectsImporter:
"""Create ClaudeProjectsImporter with dependencies.""" """Create ClaudeProjectsImporter with dependencies."""
return ClaudeProjectsImporter(project_config.home, markdown_processor, file_service) return ClaudeProjectsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ClaudeProjectsImporterDep = Annotated[ClaudeProjectsImporter, Depends(get_claude_projects_importer)] ClaudeProjectsImporterDep = Annotated[ClaudeProjectsImporter, Depends(get_claude_projects_importer)]
@@ -137,7 +172,12 @@ async def get_claude_projects_importer_v2( # pragma: no cover
file_service: FileServiceV2Dep, file_service: FileServiceV2Dep,
) -> ClaudeProjectsImporter: ) -> ClaudeProjectsImporter:
"""Create ClaudeProjectsImporter with v2 dependencies.""" """Create ClaudeProjectsImporter with v2 dependencies."""
return ClaudeProjectsImporter(project_config.home, markdown_processor, file_service) return ClaudeProjectsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ClaudeProjectsImporterV2Dep = Annotated[ ClaudeProjectsImporterV2Dep = Annotated[
@@ -151,7 +191,12 @@ async def get_claude_projects_importer_v2_external(
file_service: FileServiceV2ExternalDep, file_service: FileServiceV2ExternalDep,
) -> ClaudeProjectsImporter: ) -> ClaudeProjectsImporter:
"""Create ClaudeProjectsImporter with v2 external_id dependencies.""" """Create ClaudeProjectsImporter with v2 external_id dependencies."""
return ClaudeProjectsImporter(project_config.home, markdown_processor, file_service) return ClaudeProjectsImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
ClaudeProjectsImporterV2ExternalDep = Annotated[ ClaudeProjectsImporterV2ExternalDep = Annotated[
@@ -168,7 +213,12 @@ async def get_memory_json_importer(
file_service: FileServiceDep, file_service: FileServiceDep,
) -> MemoryJsonImporter: ) -> MemoryJsonImporter:
"""Create MemoryJsonImporter with dependencies.""" """Create MemoryJsonImporter with dependencies."""
return MemoryJsonImporter(project_config.home, markdown_processor, file_service) return MemoryJsonImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
MemoryJsonImporterDep = Annotated[MemoryJsonImporter, Depends(get_memory_json_importer)] MemoryJsonImporterDep = Annotated[MemoryJsonImporter, Depends(get_memory_json_importer)]
@@ -180,7 +230,12 @@ async def get_memory_json_importer_v2( # pragma: no cover
file_service: FileServiceV2Dep, file_service: FileServiceV2Dep,
) -> MemoryJsonImporter: ) -> MemoryJsonImporter:
"""Create MemoryJsonImporter with v2 dependencies.""" """Create MemoryJsonImporter with v2 dependencies."""
return MemoryJsonImporter(project_config.home, markdown_processor, file_service) return MemoryJsonImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
MemoryJsonImporterV2Dep = Annotated[MemoryJsonImporter, Depends(get_memory_json_importer_v2)] MemoryJsonImporterV2Dep = Annotated[MemoryJsonImporter, Depends(get_memory_json_importer_v2)]
@@ -192,7 +247,12 @@ async def get_memory_json_importer_v2_external(
file_service: FileServiceV2ExternalDep, file_service: FileServiceV2ExternalDep,
) -> MemoryJsonImporter: ) -> MemoryJsonImporter:
"""Create MemoryJsonImporter with v2 external_id dependencies.""" """Create MemoryJsonImporter with v2 external_id dependencies."""
return MemoryJsonImporter(project_config.home, markdown_processor, file_service) return MemoryJsonImporter(
project_config.home,
markdown_processor,
file_service,
project_name=project_config.name,
)
MemoryJsonImporterV2ExternalDep = Annotated[ MemoryJsonImporterV2ExternalDep = Annotated[
+24
View File
@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, Any, Optional, TypeVar
from basic_memory.markdown.markdown_processor import MarkdownProcessor from basic_memory.markdown.markdown_processor import MarkdownProcessor
from basic_memory.markdown.schemas import EntityMarkdown from basic_memory.markdown.schemas import EntityMarkdown
from basic_memory.schemas.importer import ImportResult from basic_memory.schemas.importer import ImportResult
from basic_memory.utils import build_canonical_permalink, generate_permalink
if TYPE_CHECKING: # pragma: no cover if TYPE_CHECKING: # pragma: no cover
from basic_memory.services.file_service import FileService from basic_memory.services.file_service import FileService
@@ -29,6 +30,7 @@ class Importer[T: ImportResult]:
base_path: Path, base_path: Path,
markdown_processor: MarkdownProcessor, markdown_processor: MarkdownProcessor,
file_service: "FileService", file_service: "FileService",
project_name: Optional[str] = None,
): ):
"""Initialize the import service. """Initialize the import service.
@@ -40,6 +42,8 @@ class Importer[T: ImportResult]:
self.base_path = base_path.resolve() # Get absolute path self.base_path = base_path.resolve() # Get absolute path
self.markdown_processor = markdown_processor self.markdown_processor = markdown_processor
self.file_service = file_service self.file_service = file_service
self.project_name = project_name
self.project_permalink = generate_permalink(project_name) if project_name else None
@abstractmethod @abstractmethod
async def import_data(self, source_data, destination_folder: str, **kwargs: Any) -> T: async def import_data(self, source_data, destination_folder: str, **kwargs: Any) -> T:
@@ -73,6 +77,26 @@ class Importer[T: ImportResult]:
# FileService.write_file handles directory creation and returns checksum # FileService.write_file handles directory creation and returns checksum
return await self.file_service.write_file(file_path, content) return await self.file_service.write_file(file_path, content)
def canonical_permalink(self, path: str) -> str:
"""Build a canonical permalink for imported content."""
include_project = True
# Trigger: importer has app config with permalink prefixing flag
# Why: imported notes should align with canonical permalink format
# Outcome: include project prefix when enabled
if self.file_service.app_config is not None:
include_project = self.file_service.app_config.permalinks_include_project
return build_canonical_permalink(
self.project_permalink,
path,
include_project=include_project,
)
def build_import_paths(self, path: str) -> tuple[str, str]:
"""Return (permalink, file_path) for an imported entity."""
permalink = self.canonical_permalink(path)
return permalink, f"{path}.md"
async def ensure_folder_exists(self, folder: str) -> None: async def ensure_folder_exists(self, folder: str) -> None:
"""Ensure folder exists using FileService. """Ensure folder exists using FileService.
+13 -8
View File
@@ -51,11 +51,20 @@ class ChatGPTImporter(Importer[ChatImportResult]):
chats_imported = 0 chats_imported = 0
for chat in conversations: for chat in conversations:
created_at = chat["create_time"]
date_prefix = datetime.fromtimestamp(created_at).astimezone().strftime("%Y%m%d")
clean_title = clean_filename(chat["title"])
relative_path = (
f"{destination_folder}/{date_prefix}-{clean_title}"
if destination_folder
else f"{date_prefix}-{clean_title}"
)
permalink, file_path = self.build_import_paths(relative_path)
# Convert to entity # Convert to entity
entity = self._format_chat_content(destination_folder, chat) entity = self._format_chat_content(chat, permalink)
# Write file using relative path - FileService handles base_path # Write file using relative path - FileService handles base_path
file_path = f"{entity.frontmatter.metadata['permalink']}.md"
await self.write_entity(entity, file_path) await self.write_entity(entity, file_path)
# Count messages # Count messages
@@ -83,7 +92,7 @@ class ChatGPTImporter(Importer[ChatImportResult]):
return self.handle_error("Failed to import ChatGPT conversations", e) return self.handle_error("Failed to import ChatGPT conversations", e)
def _format_chat_content( def _format_chat_content(
self, folder: str, conversation: Dict[str, Any] self, conversation: Dict[str, Any], permalink: str
) -> EntityMarkdown: # pragma: no cover ) -> EntityMarkdown: # pragma: no cover
"""Convert chat conversation to Basic Memory entity. """Convert chat conversation to Basic Memory entity.
@@ -105,10 +114,6 @@ class ChatGPTImporter(Importer[ChatImportResult]):
root_id = node_id root_id = node_id
break break
# Generate permalink
date_prefix = datetime.fromtimestamp(created_at).astimezone().strftime("%Y%m%d")
clean_title = clean_filename(conversation["title"])
# Format content # Format content
content = self._format_chat_markdown( content = self._format_chat_markdown(
title=conversation["title"], title=conversation["title"],
@@ -126,7 +131,7 @@ class ChatGPTImporter(Importer[ChatImportResult]):
"title": conversation["title"], "title": conversation["title"],
"created": format_timestamp(created_at), "created": format_timestamp(created_at),
"modified": format_timestamp(modified_at), "modified": format_timestamp(modified_at),
"permalink": f"{folder}/{date_prefix}-{clean_title}", "permalink": permalink,
} }
), ),
content=content, content=content,
@@ -54,18 +54,27 @@ class ClaudeConversationsImporter(Importer[ChatImportResult]):
for chat in conversations: for chat in conversations:
# Get name, providing default for unnamed conversations # Get name, providing default for unnamed conversations
chat_name = chat.get("name") or f"Conversation {chat.get('uuid', 'untitled')}" chat_name = chat.get("name") or f"Conversation {chat.get('uuid', 'untitled')}"
date_prefix = datetime.fromisoformat(chat["created_at"].replace("Z", "+00:00")).strftime(
"%Y%m%d"
)
clean_title = clean_filename(chat_name)
relative_path = (
f"{destination_folder}/{date_prefix}-{clean_title}"
if destination_folder
else f"{date_prefix}-{clean_title}"
)
permalink, file_path = self.build_import_paths(relative_path)
# Convert to entity # Convert to entity
entity = self._format_chat_content( entity = self._format_chat_content(
folder=destination_folder,
name=chat_name, name=chat_name,
messages=chat["chat_messages"], messages=chat["chat_messages"],
created_at=chat["created_at"], created_at=chat["created_at"],
modified_at=chat["updated_at"], modified_at=chat["updated_at"],
permalink=permalink,
) )
# Write file using relative path - FileService handles base_path # Write file using relative path - FileService handles base_path
file_path = f"{entity.frontmatter.metadata['permalink']}.md"
await self.write_entity(entity, file_path) await self.write_entity(entity, file_path)
chats_imported += 1 chats_imported += 1
@@ -84,11 +93,11 @@ class ClaudeConversationsImporter(Importer[ChatImportResult]):
def _format_chat_content( def _format_chat_content(
self, self,
folder: str,
name: str, name: str,
messages: List[Dict[str, Any]], messages: List[Dict[str, Any]],
created_at: str, created_at: str,
modified_at: str, modified_at: str,
permalink: str,
) -> EntityMarkdown: ) -> EntityMarkdown:
"""Convert chat messages to Basic Memory entity format. """Convert chat messages to Basic Memory entity format.
@@ -102,11 +111,6 @@ class ClaudeConversationsImporter(Importer[ChatImportResult]):
Returns: Returns:
EntityMarkdown instance representing the conversation. EntityMarkdown instance representing the conversation.
""" """
# Generate permalink using folder name (relative path)
date_prefix = datetime.fromisoformat(created_at.replace("Z", "+00:00")).strftime("%Y%m%d")
clean_title = clean_filename(name)
permalink = f"{folder}/{date_prefix}-{clean_title}"
# Format content # Format content
content = self._format_chat_markdown( content = self._format_chat_markdown(
name=name, name=name,
@@ -63,17 +63,28 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
await self.file_service.ensure_directory(docs_dir) await self.file_service.ensure_directory(docs_dir)
# Import prompt template if it exists # Import prompt template if it exists
if prompt_entity := self._format_prompt_markdown(project, destination_folder): if project.get("prompt_template"):
# Write file using relative path - FileService handles base_path prompt_path = (
file_path = f"{prompt_entity.frontmatter.metadata['permalink']}.md" f"{destination_folder}/{project_dir}/prompt-template"
await self.write_entity(prompt_entity, file_path) if destination_folder
else f"{project_dir}/prompt-template"
)
permalink, file_path = self.build_import_paths(prompt_path)
prompt_entity = self._format_prompt_markdown(project, permalink)
if prompt_entity:
await self.write_entity(prompt_entity, file_path)
prompts_imported += 1 prompts_imported += 1
# Import project documents # Import project documents
for doc in project.get("docs", []): for doc in project.get("docs", []):
entity = self._format_project_markdown(project, doc, destination_folder) doc_file = clean_filename(doc["filename"])
# Write file using relative path - FileService handles base_path doc_path = (
file_path = f"{entity.frontmatter.metadata['permalink']}.md" f"{destination_folder}/{project_dir}/docs/{doc_file}"
if destination_folder
else f"{project_dir}/docs/{doc_file}"
)
permalink, file_path = self.build_import_paths(doc_path)
entity = self._format_project_markdown(project, doc, permalink)
await self.write_entity(entity, file_path) await self.write_entity(entity, file_path)
docs_imported += 1 docs_imported += 1
@@ -89,7 +100,7 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
return self.handle_error("Failed to import Claude projects", e) return self.handle_error("Failed to import Claude projects", e)
def _format_project_markdown( def _format_project_markdown(
self, project: Dict[str, Any], doc: Dict[str, Any], destination_folder: str = "" self, project: Dict[str, Any], doc: Dict[str, Any], permalink: str
) -> EntityMarkdown: ) -> EntityMarkdown:
"""Format a project document as a Basic Memory entity. """Format a project document as a Basic Memory entity.
@@ -105,17 +116,6 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
created_at = doc.get("created_at") or project["created_at"] created_at = doc.get("created_at") or project["created_at"]
modified_at = project["updated_at"] modified_at = project["updated_at"]
# Generate clean names for organization
project_dir = clean_filename(project["name"])
doc_file = clean_filename(doc["filename"])
# Build permalink with optional destination folder prefix
permalink = (
f"{destination_folder}/{project_dir}/docs/{doc_file}"
if destination_folder
else f"{project_dir}/docs/{doc_file}"
)
# Create entity # Create entity
entity = EntityMarkdown( entity = EntityMarkdown(
frontmatter=EntityFrontmatter( frontmatter=EntityFrontmatter(
@@ -136,7 +136,7 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
return entity return entity
def _format_prompt_markdown( def _format_prompt_markdown(
self, project: Dict[str, Any], destination_folder: str = "" self, project: Dict[str, Any], permalink: str
) -> Optional[EntityMarkdown]: ) -> Optional[EntityMarkdown]:
"""Format project prompt template as a Basic Memory entity. """Format project prompt template as a Basic Memory entity.
@@ -155,16 +155,6 @@ class ClaudeProjectsImporter(Importer[ProjectImportResult]):
created_at = project["created_at"] created_at = project["created_at"]
modified_at = project["updated_at"] modified_at = project["updated_at"]
# Generate clean project directory name
project_dir = clean_filename(project["name"])
# Build permalink with optional destination folder prefix
permalink = (
f"{destination_folder}/{project_dir}/prompt-template"
if destination_folder
else f"{project_dir}/prompt-template"
)
# Create entity # Create entity
entity = EntityMarkdown( entity = EntityMarkdown(
frontmatter=EntityFrontmatter( frontmatter=EntityFrontmatter(
@@ -80,11 +80,12 @@ class MemoryJsonImporter(Importer[EntityImportResult]):
entity_type = entity_data.get("entityType") or entity_data.get("type") or "entity" entity_type = entity_data.get("entityType") or entity_data.get("type") or "entity"
# Build permalink with optional destination folder prefix # Build permalink with optional destination folder prefix
permalink = ( relative_path = (
f"{destination_folder}/{entity_type}/{name}" f"{destination_folder}/{entity_type}/{name}"
if destination_folder if destination_folder
else f"{entity_type}/{name}" else f"{entity_type}/{name}"
) )
permalink, file_path = self.build_import_paths(relative_path)
# Ensure entity type directory exists using FileService with relative path # Ensure entity type directory exists using FileService with relative path
entity_type_dir = ( entity_type_dir = (
@@ -109,7 +110,6 @@ class MemoryJsonImporter(Importer[EntityImportResult]):
) )
# Write file using relative path - FileService handles base_path # Write file using relative path - FileService handles base_path
file_path = f"{entity.frontmatter.metadata['permalink']}.md"
await self.write_entity(entity, file_path) await self.write_entity(entity, file_path)
entities_created += 1 entities_created += 1
+4 -2
View File
@@ -1,6 +1,8 @@
"""Markdown-it plugins for Basic Memory markdown parsing.""" """Markdown-it plugins for Basic Memory markdown parsing."""
from typing import List, Any, Dict from typing import List, Any, Dict
from basic_memory.utils import normalize_project_reference
from markdown_it import MarkdownIt from markdown_it import MarkdownIt
from markdown_it.token import Token from markdown_it.token import Token
@@ -114,7 +116,7 @@ def parse_relation(token: Token) -> Dict[str, Any] | None:
rel_type = before rel_type = before
# Get target # Get target
target = content[start + 2 : end].strip() target = normalize_project_reference(content[start + 2 : end].strip())
# Look for context after # Look for context after
after = content[end + 2 :].strip() after = content[end + 2 :].strip()
@@ -160,7 +162,7 @@ def parse_inline_relations(content: str) -> List[Dict[str, Any]]:
# No matching ]] found # No matching ]] found
break break
target = content[start + 2 : end].strip() target = normalize_project_reference(content[start + 2 : end].strip())
if target: if target:
relations.append({"type": "links_to", "target": target, "context": None}) relations.append({"type": "links_to", "target": target, "context": None})
+95
View File
@@ -17,11 +17,14 @@ from httpx._types import (
) )
from loguru import logger from loguru import logger
from fastmcp import Context from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.config import ConfigManager from basic_memory.config import ConfigManager
from basic_memory.project_resolver import ProjectResolver from basic_memory.project_resolver import ProjectResolver
from basic_memory.schemas.project_info import ProjectItem, ProjectList from basic_memory.schemas.project_info import ProjectItem, ProjectList
from basic_memory.schemas.v2 import ProjectResolveResponse from basic_memory.schemas.v2 import ProjectResolveResponse
from basic_memory.schemas.memory import memory_url_path
from basic_memory.utils import generate_permalink, normalize_project_reference
async def resolve_project_parameter( async def resolve_project_parameter(
@@ -150,6 +153,98 @@ async def get_active_project(
return active_project return active_project
def _split_project_prefix(path: str) -> tuple[Optional[str], str]:
"""Split a possible project prefix from a memory URL path."""
if "/" not in path:
return None, path
project_prefix, remainder = path.split("/", 1)
if not project_prefix or not remainder:
return None, path
if "*" in project_prefix:
return None, path
return project_prefix, remainder
async def resolve_project_and_path(
client: AsyncClient,
identifier: str,
project: Optional[str] = None,
context: Optional[Context] = None,
headers: HeaderTypes | None = None,
) -> tuple[ProjectItem, str, bool]:
"""Resolve project and normalized path for memory:// identifiers.
Returns:
Tuple of (active_project, normalized_path, is_memory_url)
"""
is_memory_url = identifier.strip().startswith("memory://")
if not is_memory_url:
active_project = await get_active_project(client, project, context, headers)
return active_project, identifier, False
normalized_path = normalize_project_reference(memory_url_path(identifier))
project_prefix, remainder = _split_project_prefix(normalized_path)
include_project = ConfigManager().config.permalinks_include_project
# Trigger: memory URL begins with a potential project segment
# Why: allow project-scoped memory URLs without requiring a separate project parameter
# Outcome: attempt to resolve the prefix as a project and route to it
if project_prefix:
try:
from basic_memory.mcp.tools.utils import call_post
response = await call_post(
client,
"/v2/projects/resolve",
json={"identifier": project_prefix},
headers=headers,
)
resolved = ProjectResolveResponse.model_validate(response.json())
except ToolError as exc:
if "project not found" not in str(exc).lower():
raise
else:
resolved_project = await resolve_project_parameter(project_prefix)
if resolved_project and generate_permalink(resolved_project) != generate_permalink(
project_prefix
):
raise ValueError(
f"Project is constrained to '{resolved_project}', cannot use '{project_prefix}'."
)
active_project = ProjectItem(
id=resolved.project_id,
external_id=resolved.external_id,
name=resolved.name,
path=resolved.path,
is_default=resolved.is_default,
)
if context:
context.set_state("active_project", active_project)
resolved_path = (
f"{resolved.permalink}/{remainder}" if include_project else remainder
)
return active_project, resolved_path, True
# Trigger: no resolvable project prefix in the memory URL
# Why: preserve existing memory URL behavior within the active project
# Outcome: use the active project and normalize the path for lookup
active_project = await get_active_project(client, project, context, headers)
resolved_path = normalized_path
if include_project:
# Trigger: project-prefixed permalinks are enabled and the path lacks a prefix
# Why: ensure memory URL lookups align with canonical permalinks
# Outcome: prefix the path with the active project's permalink
project_prefix = active_project.permalink
if resolved_path != project_prefix and not resolved_path.startswith(f"{project_prefix}/"):
resolved_path = f"{project_prefix}/{resolved_path}"
return active_project, resolved_path, True
def add_project_metadata(result: str, project_name: str) -> str: def add_project_metadata(result: str, project_name: str) -> str:
"""Add project context as metadata footer for assistant session tracking. """Add project context as metadata footer for assistant session tracking.
+8 -7
View File
@@ -5,14 +5,10 @@ from typing import Optional
from loguru import logger from loguru import logger
from fastmcp import Context from fastmcp import Context
from basic_memory.mcp.project_context import get_project_client from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.server import mcp from basic_memory.mcp.server import mcp
from basic_memory.schemas.base import TimeFrame from basic_memory.schemas.base import TimeFrame
from basic_memory.schemas.memory import ( from basic_memory.schemas.memory import GraphContext, MemoryUrl
GraphContext,
MemoryUrl,
memory_url_path,
)
@mcp.tool( @mcp.tool(
@@ -100,13 +96,18 @@ async def build_context(
# URL is already validated and normalized by MemoryUrl type annotation # URL is already validated and normalized by MemoryUrl type annotation
async with get_project_client(project, context) as (client, active_project): async with get_project_client(project, context) as (client, active_project):
# Resolve memory:// identifier with project-prefix awareness
_, resolved_path, _ = await resolve_project_and_path(
client, url, project, context
)
# Import here to avoid circular import # Import here to avoid circular import
from basic_memory.mcp.clients import MemoryClient from basic_memory.mcp.clients import MemoryClient
# Use typed MemoryClient for API calls # Use typed MemoryClient for API calls
memory_client = MemoryClient(client, active_project.external_id) memory_client = MemoryClient(client, active_project.external_id)
return await memory_client.build_context( return await memory_client.build_context(
memory_url_path(url), resolved_path,
depth=depth or 1, depth=depth or 1,
timeframe=timeframe, timeframe=timeframe,
page=page, page=page,
+9 -3
View File
@@ -15,7 +15,7 @@ from PIL import Image as PILImage
from fastmcp import Context from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.project_context import get_project_client from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.server import mcp from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import call_get, resolve_entity_id from basic_memory.mcp.tools.utils import call_get, resolve_entity_id
from basic_memory.schemas.memory import memory_url_path from basic_memory.schemas.memory import memory_url_path
@@ -202,11 +202,17 @@ async def read_content(
logger.info("Reading file", path=path, project=project) logger.info("Reading file", path=path, project=project)
async with get_project_client(project, context) as (client, active_project): async with get_project_client(project, context) as (client, active_project):
url = memory_url_path(path) # Resolve path with project-prefix awareness for memory:// URLs
_, url, _ = await resolve_project_and_path(client, path, project, context)
# Validate path to prevent path traversal attacks # Validate path to prevent path traversal attacks
# For memory:// URLs, validate the extracted path (not the raw URL which
# has a scheme prefix that confuses path validation)
raw_path = memory_url_path(path) if path.startswith("memory://") else path
project_path = active_project.home project_path = active_project.home
if not validate_project_path(url, project_path): if not validate_project_path(raw_path, project_path) or not validate_project_path(
url, project_path
):
logger.warning( logger.warning(
"Attempted path traversal attack blocked", "Attempted path traversal attack blocked",
path=path, path=path,
+19 -7
View File
@@ -6,7 +6,7 @@ from typing import Optional, Literal
from loguru import logger from loguru import logger
from fastmcp import Context from fastmcp import Context
from basic_memory.mcp.project_context import get_project_client from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.server import mcp from basic_memory.mcp.server import mcp
from basic_memory.mcp.formatting import format_note_preview_ascii from basic_memory.mcp.formatting import format_note_preview_ascii
from basic_memory.mcp.tools.search import search_notes from basic_memory.mcp.tools.search import search_notes
@@ -82,12 +82,19 @@ async def read_note(
including related notes, search commands, and note creation templates. including related notes, search commands, and note creation templates.
""" """
async with get_project_client(project, context) as (client, active_project): async with get_project_client(project, context) as (client, active_project):
# Resolve identifier with project-prefix awareness for memory:// URLs
_, entity_path, _ = await resolve_project_and_path(
client, identifier, project, context
)
# Validate identifier to prevent path traversal attacks # Validate identifier to prevent path traversal attacks
# We need to check both the raw identifier and the processed path # For memory:// URLs, validate the extracted path (not the raw URL which
processed_path = memory_url_path(identifier) # has a scheme prefix that confuses path validation)
raw_path = memory_url_path(identifier) if identifier.startswith("memory://") else identifier
processed_path = entity_path
project_path = active_project.home project_path = active_project.home
if not validate_project_path(identifier, project_path) or not validate_project_path( if not validate_project_path(raw_path, project_path) or not validate_project_path(
processed_path, project_path processed_path, project_path
): ):
logger.warning( logger.warning(
@@ -99,7 +106,6 @@ async def read_note(
return f"# Error\n\nIdentifier '{identifier}' is not allowed - paths must stay within project boundaries" return f"# Error\n\nIdentifier '{identifier}' is not allowed - paths must stay within project boundaries"
# Get the file via REST API - first try direct identifier resolution # Get the file via REST API - first try direct identifier resolution
entity_path = memory_url_path(identifier)
logger.info( logger.info(
f"Attempting to read note from Project: {active_project.name} identifier: {entity_path}" f"Attempting to read note from Project: {active_project.name} identifier: {entity_path}"
) )
@@ -135,7 +141,10 @@ async def read_note(
# Fallback 1: Try title search via API # Fallback 1: Try title search via API
logger.info(f"Search title for: {identifier}") logger.info(f"Search title for: {identifier}")
title_results = await search_notes.fn( title_results = await search_notes.fn(
query=identifier, search_type="title", project=project, context=context query=identifier,
search_type="title",
project=active_project.name,
context=context,
) )
# Handle both SearchResponse object and error strings # Handle both SearchResponse object and error strings
@@ -170,7 +179,10 @@ async def read_note(
# Fallback 2: Text search as a last resort # Fallback 2: Text search as a last resort
logger.info(f"Title search failed, trying text search for: {identifier}") logger.info(f"Title search failed, trying text search for: {identifier}")
text_results = await search_notes.fn( text_results = await search_notes.fn(
query=identifier, search_type="text", project=project, context=context query=identifier,
search_type="text",
project=active_project.name,
context=context,
) )
# We didn't find a direct match, construct a helpful error message # We didn't find a direct match, construct a helpful error message
+44 -36
View File
@@ -6,7 +6,7 @@ from typing import List, Optional, Dict, Any, Literal
from loguru import logger from loguru import logger
from fastmcp import Context from fastmcp import Context
from basic_memory.mcp.project_context import get_project_client from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
from basic_memory.mcp.formatting import format_search_results_ascii from basic_memory.mcp.formatting import format_search_results_ascii
from basic_memory.mcp.server import mcp from basic_memory.mcp.server import mcp
from basic_memory.schemas.search import ( from basic_memory.schemas.search import (
@@ -399,42 +399,50 @@ async def search_notes(
types = types or [] types = types or []
entity_types = entity_types or [] entity_types = entity_types or []
# Create a SearchQuery object based on the parameters
search_query = SearchQuery()
# Set the appropriate search field based on search_type
if search_type == "text":
search_query.text = query
elif search_type == "vector":
search_query.text = query
search_query.retrieval_mode = SearchRetrievalMode.VECTOR
elif search_type == "hybrid":
search_query.text = query
search_query.retrieval_mode = SearchRetrievalMode.HYBRID
elif search_type == "title":
search_query.title = query
elif search_type == "permalink" and "*" in query:
search_query.permalink_match = query
elif search_type == "permalink":
search_query.permalink = query
else: # pragma: no cover
search_query.text = query # Default to text search
# Add optional filters if provided (empty lists are treated as no filter)
if entity_types:
search_query.entity_types = [SearchItemType(t) for t in entity_types]
if types:
search_query.types = types
if after_date:
search_query.after_date = after_date
if metadata_filters:
search_query.metadata_filters = metadata_filters
if tags:
search_query.tags = tags
if status:
search_query.status = status
async with get_project_client(project, context) as (client, active_project): async with get_project_client(project, context) as (client, active_project):
# Handle memory:// URLs by resolving to permalink search
_, resolved_query, is_memory_url = await resolve_project_and_path(
client, query, project, context
)
if is_memory_url:
query = resolved_query
search_type = "permalink"
# Create a SearchQuery object based on the parameters
search_query = SearchQuery()
# Set the appropriate search field based on search_type
if search_type == "text":
search_query.text = query
elif search_type == "vector":
search_query.text = query
search_query.retrieval_mode = SearchRetrievalMode.VECTOR
elif search_type == "hybrid":
search_query.text = query
search_query.retrieval_mode = SearchRetrievalMode.HYBRID
elif search_type == "title":
search_query.title = query
elif search_type == "permalink" and "*" in query:
search_query.permalink_match = query
elif search_type == "permalink":
search_query.permalink = query
else: # pragma: no cover
search_query.text = query # Default to text search
# Add optional filters if provided (empty lists are treated as no filter)
if entity_types:
search_query.entity_types = [SearchItemType(t) for t in entity_types]
if types:
search_query.types = types
if after_date:
search_query.after_date = after_date
if metadata_filters:
search_query.metadata_filters = metadata_filters
if tags:
search_query.tags = tags
if status:
search_query.status = status
logger.info(f"Searching for {search_query} in project {active_project.name}") logger.info(f"Searching for {search_query} in project {active_project.name}")
try: try:
@@ -539,9 +539,7 @@ class ContextService:
{relation_date_filter} {relation_date_filter}
{relation_project_filter} {relation_project_filter}
) )
LEFT JOIN entity e_to ON (r.to_id = e_to.id)
WHERE eg.depth < :max_depth WHERE eg.depth < :max_depth
AND (r.to_id IS NULL OR e_to.project_id = :project_id)
UNION ALL UNION ALL
+41 -5
View File
@@ -24,6 +24,7 @@ from basic_memory.models import Entity as EntityModel
from basic_memory.models import Observation, Relation from basic_memory.models import Observation, Relation
from basic_memory.models.knowledge import Entity from basic_memory.models.knowledge import Entity
from basic_memory.repository import ObservationRepository, RelationRepository from basic_memory.repository import ObservationRepository, RelationRepository
from basic_memory.repository.project_repository import ProjectRepository
from basic_memory.repository.entity_repository import EntityRepository from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.schemas import Entity as EntitySchema from basic_memory.schemas import Entity as EntitySchema
from basic_memory.schemas.base import Permalink from basic_memory.schemas.base import Permalink
@@ -41,7 +42,7 @@ from basic_memory.services.exceptions import (
) )
from basic_memory.services.link_resolver import LinkResolver from basic_memory.services.link_resolver import LinkResolver
from basic_memory.services.search_service import SearchService from basic_memory.services.search_service import SearchService
from basic_memory.utils import generate_permalink from basic_memory.utils import build_canonical_permalink
class EntityService(BaseService[EntityModel]): class EntityService(BaseService[EntityModel]):
@@ -66,6 +67,7 @@ class EntityService(BaseService[EntityModel]):
self.link_resolver = link_resolver self.link_resolver = link_resolver
self.search_service = search_service self.search_service = search_service
self.app_config = app_config self.app_config = app_config
self._project_permalink: Optional[str] = None
async def detect_file_path_conflicts( async def detect_file_path_conflicts(
self, file_path: str, skip_check: bool = False self, file_path: str, skip_check: bool = False
@@ -159,7 +161,23 @@ class EntityService(BaseService[EntityModel]):
if markdown and markdown.frontmatter.permalink: if markdown and markdown.frontmatter.permalink:
desired_permalink = markdown.frontmatter.permalink desired_permalink = markdown.frontmatter.permalink
else: else:
desired_permalink = generate_permalink(file_path_str) # Trigger: generating a permalink for a new file
# Why: canonical permalinks may require project prefix for global addressing
# Outcome: include project slug when enabled in config
include_project = True
if self.app_config:
include_project = self.app_config.permalinks_include_project
project_permalink = None
# Trigger: project-prefixed permalinks are enabled
# Why: we need the project slug to build the canonical permalink
# Outcome: fetch and cache the project's permalink
if include_project:
project_permalink = await self._get_project_permalink()
desired_permalink = build_canonical_permalink(
project_permalink, file_path_str, include_project=include_project
)
# Make unique if needed - enhanced to handle character conflicts # Make unique if needed - enhanced to handle character conflicts
# Use lightweight existence check instead of loading full entity # Use lightweight existence check instead of loading full entity
@@ -172,6 +190,21 @@ class EntityService(BaseService[EntityModel]):
return permalink return permalink
async def _get_project_permalink(self) -> Optional[str]:
"""Get and cache the current project's permalink."""
if self._project_permalink is not None:
return self._project_permalink
project_id = self.repository.project_id
if project_id is None: # pragma: no cover
return None # pragma: no cover
project_repository = ProjectRepository(self.repository.session_maker)
project = await project_repository.get_by_id(project_id)
if project:
self._project_permalink = project.permalink
return self._project_permalink
def _build_frontmatter_markdown( def _build_frontmatter_markdown(
self, title: str, entity_type: str, permalink: str self, title: str, entity_type: str, permalink: str
) -> EntityMarkdown: ) -> EntityMarkdown:
@@ -313,9 +346,12 @@ class EntityService(BaseService[EntityModel]):
# Merge new metadata with existing metadata # Merge new metadata with existing metadata
existing_markdown.frontmatter.metadata.update(post.metadata) existing_markdown.frontmatter.metadata.update(post.metadata)
# Ensure the permalink in the metadata is the resolved one # Always ensure the permalink in the metadata is the canonical one from the database.
if new_permalink != entity.permalink: # The schema_to_markdown call above uses EntitySchema.permalink which computes a
existing_markdown.frontmatter.metadata["permalink"] = new_permalink # non-prefixed permalink (e.g., "test/note"). The metadata merge on the previous line
# would overwrite the project-prefixed permalink (e.g., "project/test/note") stored
# in the existing file. Setting it unconditionally preserves the correct value.
existing_markdown.frontmatter.metadata["permalink"] = new_permalink
# Create a new post with merged metadata # Create a new post with merged metadata
merged_post = frontmatter.Post(post.content, **existing_markdown.frontmatter.metadata) merged_post = frontmatter.Post(post.content, **existing_markdown.frontmatter.metadata)
+295 -102
View File
@@ -1,14 +1,21 @@
"""Service for resolving markdown links to permalinks.""" """Service for resolving markdown links to permalinks."""
from typing import Optional, Tuple from typing import Optional, Tuple, Dict
from loguru import logger from loguru import logger
from basic_memory.models import Entity from basic_memory.config import BasicMemoryConfig, ConfigManager
from basic_memory.models import Entity, Project
from basic_memory.repository.entity_repository import EntityRepository from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.project_repository import ProjectRepository
from basic_memory.repository.search_repository import create_search_repository
from basic_memory.schemas.search import SearchQuery, SearchItemType from basic_memory.schemas.search import SearchQuery, SearchItemType
from basic_memory.services.search_service import SearchService from basic_memory.services.search_service import SearchService
from basic_memory.utils import (
build_canonical_permalink,
generate_permalink,
normalize_project_reference,
)
class LinkResolver: class LinkResolver:
@@ -26,6 +33,12 @@ class LinkResolver:
"""Initialize with repositories.""" """Initialize with repositories."""
self.entity_repository = entity_repository self.entity_repository = entity_repository
self.search_service = search_service self.search_service = search_service
self._project_repository = ProjectRepository(entity_repository.session_maker)
self._app_config: BasicMemoryConfig = ConfigManager().config
self._project_permalink: Optional[str] = None
self._project_cache_by_identifier: Dict[str, Project] = {}
self._entity_repository_cache: Dict[int, EntityRepository] = {}
self._search_service_cache: Dict[int, SearchService] = {}
async def resolve_link( async def resolve_link(
self, self,
@@ -47,111 +60,69 @@ class LinkResolver:
# Clean link text and extract any alias # Clean link text and extract any alias
clean_text, alias = self._normalize_link_text(link_text) clean_text, alias = self._normalize_link_text(link_text)
explicit_project_reference = "::" in clean_text
clean_text = normalize_project_reference(clean_text)
# --- Path Resolution --- # Trigger: link uses project namespace syntax (project::note)
# Note: All paths in Basic Memory are stored as POSIX strings (forward slashes) # Why: treat it as an explicit cross-project reference
# for cross-platform compatibility. See entity_repository.py which normalizes # Outcome: resolve only within the referenced project scope
# paths using Path().as_posix(). This allows consistent path operations here. if explicit_project_reference:
project_prefix, remainder = self._split_project_prefix(clean_text)
if not project_prefix:
return None
# --- Relative Path Resolution --- project_resources = await self._get_project_resources(project_prefix)
# Trigger: source_path is provided AND link contains "/" if not project_resources:
# Why: Resolve paths like [[nested/deep-note]] relative to source folder first return None
# Outcome: [[nested/deep-note]] from testing/link-test.md → testing/nested/deep-note.md
if source_path and "/" in clean_text:
source_folder = source_path.rsplit("/", 1)[0] if "/" in source_path else ""
if source_folder:
# Construct relative path from source folder
relative_path = f"{source_folder}/{clean_text}"
# Try with .md extension project, entity_repository, search_service = project_resources
if not relative_path.endswith(".md"): return await self._resolve_in_project(
relative_path_md = f"{relative_path}.md" entity_repository=entity_repository,
entity = await self.entity_repository.get_by_file_path(relative_path_md) search_service=search_service,
if entity: link_text=remainder,
return entity use_search=use_search,
strict=strict,
# Try as-is (already has extension or is a permalink) source_path=None,
entity = await self.entity_repository.get_by_file_path(relative_path) project_permalink=project.permalink,
if entity:
return entity
# When source_path is provided, use context-aware resolution:
# Check both permalink and title matches, prefer closest to source.
# Example: [[testing]] from folder/note.md prefers folder/testing.md
# over a root testing.md with permalink "testing".
if source_path:
# Gather all potential matches
candidates: list[Entity] = []
# Check permalink match
permalink_entity = await self.entity_repository.get_by_permalink(clean_text)
if permalink_entity:
candidates.append(permalink_entity)
# Check title matches
title_entities = await self.entity_repository.get_by_title(clean_text)
for entity in title_entities:
# Avoid duplicates (permalink match might also be in title matches)
if entity.id not in [c.id for c in candidates]:
candidates.append(entity)
if candidates:
if len(candidates) == 1:
return candidates[0]
else:
# Multiple candidates - pick closest to source
return self._find_closest_entity(candidates, source_path)
# Standard resolution (no source context): permalink first, then title
# 1. Try exact permalink match first (most efficient)
entity = await self.entity_repository.get_by_permalink(clean_text)
if entity:
logger.debug(f"Found exact permalink match: {entity.permalink}")
return entity
# 2. Try exact title match
found = await self.entity_repository.get_by_title(clean_text)
if found:
# Return first match (shortest path) if no source context
entity = found[0]
logger.debug(f"Found title match: {entity.title}")
return entity
# 3. Try file path
found_path = await self.entity_repository.get_by_file_path(clean_text)
if found_path:
logger.debug(f"Found entity with path: {found_path.file_path}")
return found_path
# 4. Try file path with .md extension if not already present
if not clean_text.endswith(".md") and "/" in clean_text:
file_path_with_md = f"{clean_text}.md"
found_path_md = await self.entity_repository.get_by_file_path(file_path_with_md)
if found_path_md:
logger.debug(f"Found entity with path (with .md): {found_path_md.file_path}")
return found_path_md
# In strict mode, don't try fuzzy search - return None if no exact match found
if strict:
return None
# 5. Fall back to search for fuzzy matching (only if not in strict mode)
if use_search and "*" not in clean_text:
results = await self.search_service.search(
query=SearchQuery(text=clean_text, entity_types=[SearchItemType.ENTITY]),
) )
if results: current_project_permalink = await self._get_current_project_permalink()
# Look for best match resolved = await self._resolve_in_project(
best_match = min(results, key=lambda x: x.score) # pyright: ignore entity_repository=self.entity_repository,
logger.trace( search_service=self.search_service,
f"Selected best match from {len(results)} results: {best_match.permalink}" link_text=clean_text,
) use_search=use_search,
if best_match.permalink: strict=strict,
return await self.entity_repository.get_by_permalink(best_match.permalink) source_path=source_path,
project_permalink=current_project_permalink,
)
if resolved:
return resolved
# if we couldn't find anything then return None # Trigger: local resolution failed and identifier looks like project/path
return None # Why: allow explicit project path references without namespace syntax
# Outcome: attempt resolution in the referenced project if it exists
project_prefix, remainder = self._split_project_prefix(clean_text)
if not project_prefix:
return None
project_resources = await self._get_project_resources(project_prefix)
if not project_resources:
return None
project, entity_repository, search_service = project_resources
if project.id == self.entity_repository.project_id:
return None
return await self._resolve_in_project(
entity_repository=entity_repository,
search_service=search_service,
link_text=remainder,
use_search=use_search,
strict=strict,
source_path=None,
project_permalink=project.permalink,
)
def _normalize_link_text(self, link_text: str) -> Tuple[str, Optional[str]]: def _normalize_link_text(self, link_text: str) -> Tuple[str, Optional[str]]:
"""Normalize link text and extract alias if present. """Normalize link text and extract alias if present.
@@ -181,6 +152,228 @@ class LinkResolver:
return text, alias return text, alias
async def _resolve_in_project(
self,
*,
entity_repository: EntityRepository,
search_service: SearchService,
link_text: str,
use_search: bool,
strict: bool,
source_path: Optional[str],
project_permalink: Optional[str],
) -> Optional[Entity]:
"""Resolve a link within a specific project scope."""
clean_text = link_text
include_project = self._include_project_permalinks()
canonical_permalink: Optional[str] = None
legacy_permalink: Optional[str] = None
# Trigger: permalinks include project slug and project permalink is known
# Why: support globally addressable permalinks while keeping legacy links resolvable
# Outcome: include canonical and legacy candidates for resolution
if include_project and project_permalink:
canonical_permalink = build_canonical_permalink(
project_permalink, clean_text, include_project=True
)
if clean_text.startswith(f"{project_permalink}/"):
legacy_candidate = clean_text.removeprefix(f"{project_permalink}/")
if legacy_candidate:
legacy_permalink = legacy_candidate
permalink_candidates = []
for candidate in (clean_text, canonical_permalink, legacy_permalink):
if candidate and candidate not in permalink_candidates:
permalink_candidates.append(candidate)
# --- Path Resolution ---
# Note: All paths in Basic Memory are stored as POSIX strings (forward slashes)
# for cross-platform compatibility. See entity_repository.py which normalizes
# paths using Path().as_posix(). This allows consistent path operations here.
# --- Relative Path Resolution ---
# Trigger: source_path is provided AND link contains "/"
# Why: Resolve paths like [[nested/deep-note]] relative to source folder first
# Outcome: [[nested/deep-note]] from testing/link-test.md → testing/nested/deep-note.md
if source_path and "/" in clean_text:
if not (
include_project
and project_permalink
and clean_text.startswith(f"{project_permalink}/")
):
source_folder = source_path.rsplit("/", 1)[0] if "/" in source_path else ""
if source_folder:
# Construct relative path from source folder
relative_path = f"{source_folder}/{clean_text}"
# Try with .md extension
if not relative_path.endswith(".md"):
relative_path_md = f"{relative_path}.md"
entity = await entity_repository.get_by_file_path(relative_path_md)
if entity:
return entity
# Try as-is (already has extension or is a permalink)
entity = await entity_repository.get_by_file_path(relative_path)
if entity:
return entity
# When source_path is provided, use context-aware resolution:
# Check both permalink and title matches, prefer closest to source.
# Example: [[testing]] from folder/note.md prefers folder/testing.md
# over a root testing.md with permalink "testing".
if source_path:
# Gather all potential matches
candidates: list[Entity] = []
# Check permalink match
for candidate_permalink in permalink_candidates:
permalink_entity = await entity_repository.get_by_permalink(candidate_permalink)
if permalink_entity and permalink_entity.id not in [c.id for c in candidates]:
candidates.append(permalink_entity)
# Check title matches
title_entities = await entity_repository.get_by_title(clean_text)
for entity in title_entities:
# Avoid duplicates (permalink match might also be in title matches)
if entity.id not in [c.id for c in candidates]:
candidates.append(entity)
if candidates:
if len(candidates) == 1:
return candidates[0]
else:
# Multiple candidates - pick closest to source
return self._find_closest_entity(candidates, source_path)
# Standard resolution (no source context): permalink first, then title
# 1. Try exact permalink match first (most efficient)
for candidate_permalink in permalink_candidates:
entity = await entity_repository.get_by_permalink(candidate_permalink)
if entity:
logger.debug(f"Found exact permalink match: {entity.permalink}")
return entity
# 2. Try exact title match
found = await entity_repository.get_by_title(clean_text)
if found:
# Return first match (shortest path) if no source context
entity = found[0]
logger.debug(f"Found title match: {entity.title}")
return entity
# 3. Try file path
found_path = await entity_repository.get_by_file_path(clean_text)
if found_path:
logger.debug(f"Found entity with path: {found_path.file_path}")
return found_path
# 4. Try file path with .md extension if not already present
if not clean_text.endswith(".md") and "/" in clean_text:
file_path_with_md = f"{clean_text}.md"
found_path_md = await entity_repository.get_by_file_path(file_path_with_md)
if found_path_md:
logger.debug(f"Found entity with path (with .md): {found_path_md.file_path}")
return found_path_md
# In strict mode, don't try fuzzy search - return None if no exact match found
if strict:
return None
# 5. Fall back to search for fuzzy matching (only if not in strict mode)
if use_search and "*" not in clean_text:
results = await search_service.search(
query=SearchQuery(text=clean_text, entity_types=[SearchItemType.ENTITY]),
)
if results:
# Look for best match
best_match = min(results, key=lambda x: x.score) # pyright: ignore
logger.trace(
f"Selected best match from {len(results)} results: {best_match.permalink}"
)
if best_match.permalink:
return await entity_repository.get_by_permalink(best_match.permalink)
# if we couldn't find anything then return None
return None
def _include_project_permalinks(self) -> bool:
"""Return True when permalinks should include the project slug."""
return self._app_config.permalinks_include_project
async def _get_current_project_permalink(self) -> Optional[str]:
"""Get and cache the current project's permalink."""
if self._project_permalink is not None:
return self._project_permalink
project_id = self.entity_repository.project_id
if project_id is None: # pragma: no cover
return None # pragma: no cover
project = await self._project_repository.get_by_id(project_id)
if project:
self._project_permalink = project.permalink
return self._project_permalink
async def _get_project_by_identifier(self, identifier: str) -> Optional[Project]:
"""Resolve project by name or permalink."""
cache_key = identifier.strip().lower()
if cache_key in self._project_cache_by_identifier:
return self._project_cache_by_identifier[cache_key]
project = await self._project_repository.get_by_name(identifier)
if not project:
project = await self._project_repository.get_by_name_case_insensitive(identifier)
if not project:
project = await self._project_repository.get_by_permalink(generate_permalink(identifier))
if project:
self._project_cache_by_identifier[cache_key] = project
return project
async def _get_project_resources(
self, project_identifier: str
) -> Optional[Tuple[Project, EntityRepository, SearchService]]:
"""Fetch repositories and services scoped to a project."""
project = await self._get_project_by_identifier(project_identifier)
if not project:
return None
entity_repository = self._entity_repository_cache.get(project.id)
if not entity_repository:
entity_repository = EntityRepository(
self.entity_repository.session_maker, project_id=project.id
)
self._entity_repository_cache[project.id] = entity_repository
search_service = self._search_service_cache.get(project.id)
if not search_service:
search_repository = create_search_repository(
self.entity_repository.session_maker,
project_id=project.id,
database_backend=self._app_config.database_backend,
)
search_service = SearchService(
search_repository,
entity_repository,
self.search_service.file_service,
)
self._search_service_cache[project.id] = search_service
return project, entity_repository, search_service
def _split_project_prefix(self, identifier: str) -> Tuple[Optional[str], str]:
"""Split project prefix from a path-like identifier."""
if "/" not in identifier:
return None, identifier
project_prefix, remainder = identifier.split("/", 1)
if not project_prefix or not remainder:
return None, identifier
return project_prefix, remainder
def _find_closest_entity(self, entities: list[Entity], source_path: str) -> Entity: def _find_closest_entity(self, entities: list[Entity], source_path: str) -> Entity:
"""Find the entity closest to the source file path. """Find the entity closest to the source file path.
+44 -1
View File
@@ -7,7 +7,7 @@ import re
import sys import sys
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Protocol, Union, runtime_checkable, List from typing import Protocol, Union, runtime_checkable, List, Optional
from loguru import logger from loguru import logger
from unidecode import unidecode from unidecode import unidecode
@@ -202,6 +202,49 @@ def generate_permalink(file_path: Union[Path, str, PathLike], split_extension: b
return return_val return return_val
def normalize_project_reference(identifier: str) -> str:
"""Normalize project-prefixed references.
Converts project namespace syntax ("project::note") to path syntax ("project/note").
Leaves non-namespaced identifiers unchanged.
"""
if "::" not in identifier:
return identifier
project, remainder = identifier.split("::", 1)
remainder = remainder.lstrip("/")
return f"{project}/{remainder}"
def build_canonical_permalink(
project_permalink: Optional[str],
file_path: Union[Path, str, PathLike],
include_project: bool = True,
) -> str:
"""Build a canonical permalink, optionally prefixed with project slug.
Args:
project_permalink: URL-friendly project identifier (slug). If None, no prefix is added.
file_path: Original file path or permalink-like string.
include_project: When True, prefix with project slug.
Returns:
Canonical permalink string.
"""
normalized_path = generate_permalink(file_path)
if not include_project or not project_permalink:
return normalized_path
normalized_project = generate_permalink(project_permalink)
if normalized_path == normalized_project or normalized_path.startswith(
f"{normalized_project}/"
):
return normalized_path
return f"{normalized_project}/{normalized_path}"
def setup_logging( def setup_logging(
log_level: str = "INFO", log_level: str = "INFO",
log_to_file: bool = False, log_to_file: bool = False,
@@ -283,7 +283,7 @@ async def test_read_content_empty_file(mcp_server, app, test_project):
# Should still have frontmatter even with empty content # Should still have frontmatter even with empty content
assert "title: Empty Test" in content assert "title: Empty Test" in content
assert "permalink: test/empty-test" in content assert f"permalink: {test_project.name}/test/empty-test" in content
@pytest.mark.asyncio @pytest.mark.asyncio
+3 -3
View File
@@ -45,7 +45,7 @@ async def test_read_note_after_write(mcp_server, app, test_project):
# Should contain the note content and metadata # Should contain the note content and metadata
assert "# Test Note" in result_text assert "# Test Note" in result_text
assert "This is test content." in result_text assert "This is test content." in result_text
assert "test/test-note" in result_text # permalink assert f"{test_project.name}/test/test-note" in result_text # permalink
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -78,7 +78,7 @@ async def test_read_note_underscored_folder_by_permalink(mcp_server, app, test_p
assert "_archive/articles/Example Note.md" in write_text assert "_archive/articles/Example Note.md" in write_text
# Verify the permalink has underscores stripped (this is the expected behavior) # Verify the permalink has underscores stripped (this is the expected behavior)
assert "archive/articles/example-note" in write_text assert f"{test_project.name}/archive/articles/example-note" in write_text
# Now try to read the note using the permalink (without underscores) # Now try to read the note using the permalink (without underscores)
# This is the exact scenario from the bug report - using the permalink # This is the exact scenario from the bug report - using the permalink
@@ -99,4 +99,4 @@ async def test_read_note_underscored_folder_by_permalink(mcp_server, app, test_p
# Should contain the note content # Should contain the note content
assert "# Example Note" in result_text assert "# Example Note" in result_text
assert "This is a test note in an underscored folder." in result_text assert "This is a test note in an underscored folder." in result_text
assert "archive/articles/example-note" in result_text # permalink assert f"{test_project.name}/archive/articles/example-note" in result_text # permalink
+2 -2
View File
@@ -224,7 +224,7 @@ async def test_search_permalink_exact(mcp_server, app, test_project):
"search_notes", "search_notes",
{ {
"project": test_project.name, "project": test_project.name,
"query": "api/api-documentation", "query": f"{test_project.name}/api/api-documentation",
"search_type": "permalink", "search_type": "permalink",
}, },
) )
@@ -278,7 +278,7 @@ async def test_search_permalink_pattern(mcp_server, app, test_project):
"search_notes", "search_notes",
{ {
"project": test_project.name, "project": test_project.name,
"query": "meetings/*", "query": f"{test_project.name}/meetings/*",
"search_type": "permalink", "search_type": "permalink",
}, },
) )
+10 -10
View File
@@ -38,7 +38,7 @@ async def test_write_note_basic_creation(mcp_server, app, test_project):
assert "# Created note" in response_text assert "# Created note" in response_text
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert "file_path: basic/Simple Note.md" in response_text assert "file_path: basic/Simple Note.md" in response_text
assert "permalink: basic/simple-note" in response_text assert f"permalink: {test_project.name}/basic/simple-note" in response_text
assert "## Tags" in response_text assert "## Tags" in response_text
assert "- simple, test" in response_text assert "- simple, test" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -65,7 +65,7 @@ async def test_write_note_no_tags(mcp_server, app, test_project):
assert "# Created note" in response_text assert "# Created note" in response_text
assert "file_path: test/No Tags Note.md" in response_text assert "file_path: test/No Tags Note.md" in response_text
assert "permalink: test/no-tags-note" in response_text assert f"permalink: {test_project.name}/test/no-tags-note" in response_text
# Should not have tags section when no tags provided # Should not have tags section when no tags provided
@@ -107,7 +107,7 @@ async def test_write_note_update_existing(mcp_server, app, test_project):
assert "# Updated note" in response_text assert "# Updated note" in response_text
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert "file_path: test/Update Test.md" in response_text assert "file_path: test/Update Test.md" in response_text
assert "permalink: test/update-test" in response_text assert f"permalink: {test_project.name}/test/update-test" in response_text
assert "- updated, modified" in response_text assert "- updated, modified" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -136,7 +136,7 @@ async def test_write_note_tag_array(mcp_server, app, test_project):
assert "# Created note" in response_text assert "# Created note" in response_text
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert "file_path: test/Array Tags Test.md" in response_text assert "file_path: test/Array Tags Test.md" in response_text
assert "permalink: test/array-tags-test" in response_text assert f"permalink: {test_project.name}/test/array-tags-test" in response_text
assert "## Tags" in response_text assert "## Tags" in response_text
assert "python" in response_text assert "python" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -206,7 +206,7 @@ async def test_write_note_unicode_content(mcp_server, app, test_project):
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert "file_path: test/Unicode Test 🌟.md" in response_text assert "file_path: test/Unicode Test 🌟.md" in response_text
# Permalink should be sanitized # Permalink should be sanitized
assert "permalink: test/unicode-test" in response_text assert f"permalink: {test_project.name}/test/unicode-test" in response_text
assert "## Tags" in response_text assert "## Tags" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -256,7 +256,7 @@ async def test_write_note_complex_content_with_observations_relations(
assert "# Created note" in response_text assert "# Created note" in response_text
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert "file_path: knowledge/Complex Knowledge Note.md" in response_text assert "file_path: knowledge/Complex Knowledge Note.md" in response_text
assert "permalink: knowledge/complex-knowledge-note" in response_text assert f"permalink: {test_project.name}/knowledge/complex-knowledge-note" in response_text
# Should show observation and relation counts # Should show observation and relation counts
assert "## Observations" in response_text assert "## Observations" in response_text
@@ -309,7 +309,7 @@ async def test_write_note_preserve_frontmatter(mcp_server, app, test_project):
assert "# Created note" in response_text assert "# Created note" in response_text
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert "file_path: test/Frontmatter Note.md" in response_text assert "file_path: test/Frontmatter Note.md" in response_text
assert "permalink: test/frontmatter-note" in response_text assert f"permalink: {test_project.name}/test/frontmatter-note" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -338,7 +338,7 @@ async def test_write_note_kebab_filenames_basic(mcp_server, app, test_project, a
# File path and permalink should be kebab-case and sanitized # File path and permalink should be kebab-case and sanitized
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert "file_path: my-folder/my-note-with-invalid-chars.md" in response_text assert "file_path: my-folder/my-note-with-invalid-chars.md" in response_text
assert "permalink: my-folder/my-note-with-invalid-chars" in response_text assert f"permalink: {test_project.name}/my-folder/my-note-with-invalid-chars" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -366,7 +366,7 @@ async def test_write_note_kebab_filenames_repeat_invalid(mcp_server, app, test_p
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert "file_path: my-folder/crazy-note-name.md" in response_text assert "file_path: my-folder/crazy-note-name.md" in response_text
assert "permalink: my-folder/crazy-note-name" in response_text assert f"permalink: {test_project.name}/my-folder/crazy-note-name" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text assert f"[Session: Using project '{test_project.name}']" in response_text
@@ -416,7 +416,7 @@ async def test_write_note_file_path_os_path_join(mcp_server, app, test_project,
assert f"project: {test_project.name}" in response_text assert f"project: {test_project.name}" in response_text
assert f"file_path: {expected_path}" in response_text assert f"file_path: {expected_path}" in response_text
assert f"permalink: {expected_permalink}" in response_text assert f"permalink: {test_project.name}/{expected_permalink}" in response_text
assert f"[Session: Using project '{test_project.name}']" in response_text assert f"[Session: Using project '{test_project.name}']" in response_text
+1 -1
View File
@@ -171,7 +171,7 @@ async def test_create_entity(client: AsyncClient, file_service, v2_project_url):
assert isinstance(entity.id, int) assert isinstance(entity.id, int)
assert entity.api_version == "v2" assert entity.api_version == "v2"
assert entity.permalink == "test/test-v2-entity" assert entity.permalink == "test-project/test/test-v2-entity"
assert entity.file_path == "test/TestV2Entity.md" assert entity.file_path == "test/TestV2Entity.md"
assert entity.entity_type == data["entity_type"] assert entity.entity_type == data["entity_type"]
+15 -4
View File
@@ -38,7 +38,9 @@ async def test_imported_conversations_have_correct_permalink_and_title(
file_service = FileService(base_path, processor) file_service = FileService(base_path, processor)
# Create importer # Create importer
importer = ClaudeConversationsImporter(base_path, processor, file_service) importer = ClaudeConversationsImporter(
base_path, processor, file_service, project_name=project_config.name
)
# Sample conversation data # Sample conversation data
conversations = [ conversations = [
@@ -80,7 +82,10 @@ async def test_imported_conversations_have_correct_permalink_and_title(
content = conv_path.read_text() content = conv_path.read_text()
assert "---" in content, "File should have frontmatter markers" assert "---" in content, "File should have frontmatter markers"
assert "title: My Test Conversation Title" in content, "File should have title in frontmatter" assert "title: My Test Conversation Title" in content, "File should have title in frontmatter"
assert "permalink: conversations/20250115-My_Test_Conversation_Title" in content, ( assert (
f"permalink: {project_config.name}/conversations/20250115-my-test-conversation-title"
in content
), (
"File should have permalink in frontmatter" "File should have permalink in frontmatter"
) )
@@ -97,7 +102,10 @@ async def test_imported_conversations_have_correct_permalink_and_title(
assert entity.title == "My Test Conversation Title", ( assert entity.title == "My Test Conversation Title", (
f"Title should be from frontmatter, got: {entity.title}" f"Title should be from frontmatter, got: {entity.title}"
) )
assert entity.permalink == "conversations/20250115-My_Test_Conversation_Title", ( assert (
entity.permalink
== f"{project_config.name}/conversations/20250115-my-test-conversation-title"
), (
f"Permalink should be from frontmatter, got: {entity.permalink}" f"Permalink should be from frontmatter, got: {entity.permalink}"
) )
@@ -111,6 +119,9 @@ async def test_imported_conversations_have_correct_permalink_and_title(
assert search_result.title == "My Test Conversation Title", ( assert search_result.title == "My Test Conversation Title", (
f"Search title should be from frontmatter, got: {search_result.title}" f"Search title should be from frontmatter, got: {search_result.title}"
) )
assert search_result.permalink == "conversations/20250115-My_Test_Conversation_Title", ( assert (
search_result.permalink
== f"{project_config.name}/conversations/20250115-my-test-conversation-title"
), (
f"Search permalink should not be null, got: {search_result.permalink}" f"Search permalink should not be null, got: {search_result.permalink}"
) )
+8
View File
@@ -176,6 +176,14 @@ def test_relation_plugin():
assert rel["target"] == "Component" assert rel["target"] == "Component"
assert rel["context"] == "with context" assert rel["context"] == "with context"
# Test project namespace normalization
content = dedent("""
- relates_to [[other-project::Component]]
""")
token = [t for t in md.parse(content) if t.type == "inline"][0]
rel = token.meta["relations"][0]
assert rel["target"] == "other-project/Component"
# Test implicit relations in text # Test implicit relations in text
content = "Some text with a [[Link]] and [[Another Link]]" content = "Some text with a [[Link]] and [[Another Link]]"
token = [t for t in md.parse(content) if t.type == "inline"][0] token = [t for t in md.parse(content) if t.type == "inline"][0]
@@ -22,6 +22,7 @@ from basic_memory.mcp.tools import write_note, read_note
from basic_memory.sync.sync_service import SyncService from basic_memory.sync.sync_service import SyncService
from basic_memory.config import ProjectConfig from basic_memory.config import ProjectConfig
from basic_memory.services import EntityService from basic_memory.services import EntityService
from basic_memory.utils import generate_permalink
async def force_full_scan(sync_service: SyncService) -> None: async def force_full_scan(sync_service: SyncService) -> None:
@@ -70,7 +71,7 @@ async def test_permalink_collision_should_not_overwrite_different_file(app, test
assert "# Created note" in result_a assert "# Created note" in result_a
assert "file_path: edge-cases/Node A.md" in result_a assert "file_path: edge-cases/Node A.md" in result_a
assert "permalink: edge-cases/node-a" in result_a assert f"permalink: {test_project.name}/edge-cases/node-a" in result_a
# Verify Node A content via read # Verify Node A content via read
content_a = await read_note.fn("edge-cases/node-a", project=test_project.name) content_a = await read_note.fn("edge-cases/node-a", project=test_project.name)
@@ -87,7 +88,7 @@ async def test_permalink_collision_should_not_overwrite_different_file(app, test
assert "# Created note" in result_b assert "# Created note" in result_b
assert "file_path: edge-cases/Node B.md" in result_b assert "file_path: edge-cases/Node B.md" in result_b
assert "permalink: edge-cases/node-b" in result_b assert f"permalink: {test_project.name}/edge-cases/node-b" in result_b
# Step 3: Create third note "Node C" (this is where the bug occurs) # Step 3: Create third note "Node C" (this is where the bug occurs)
result_c = await write_note.fn( result_c = await write_note.fn(
@@ -99,7 +100,7 @@ async def test_permalink_collision_should_not_overwrite_different_file(app, test
assert "# Created note" in result_c assert "# Created note" in result_c
assert "file_path: edge-cases/Node C.md" in result_c assert "file_path: edge-cases/Node C.md" in result_c
assert "permalink: edge-cases/node-c" in result_c assert f"permalink: {test_project.name}/edge-cases/node-c" in result_c
# CRITICAL CHECK: Verify Node A still has its original content # CRITICAL CHECK: Verify Node A still has its original content
# This is where the bug manifests - Node A.md gets overwritten with Node C content # This is where the bug manifests - Node A.md gets overwritten with Node C content
@@ -258,6 +259,7 @@ async def test_sync_permalink_collision_file_overwrite_bug(
project_dir = project_config.home project_dir = project_config.home
edge_cases_dir = project_dir / "edge-cases" edge_cases_dir = project_dir / "edge-cases"
edge_cases_dir.mkdir(parents=True, exist_ok=True) edge_cases_dir.mkdir(parents=True, exist_ok=True)
project_prefix = generate_permalink(project_config.name)
# Step 1: Create Node A file # Step 1: Create Node A file
node_a_content = dedent(""" node_a_content = dedent("""
@@ -284,7 +286,7 @@ async def test_sync_permalink_collision_file_overwrite_bug(
await sync_service.sync(project_dir) await sync_service.sync(project_dir)
# Verify Node A is in database # Verify Node A is in database
node_a = await entity_service.get_by_permalink("edge-cases/node-a") node_a = await entity_service.get_by_permalink(f"{project_prefix}/edge-cases/node-a")
assert node_a is not None assert node_a is not None
assert node_a.title == "Node A" assert node_a.title == "Node A"
@@ -375,8 +377,8 @@ async def test_sync_permalink_collision_file_overwrite_bug(
assert "Content for Node C" in node_c_after_sync assert "Content for Node C" in node_c_after_sync
# Verify database has both entities correctly # Verify database has both entities correctly
node_a_db = await entity_service.get_by_permalink("edge-cases/node-a") node_a_db = await entity_service.get_by_permalink(f"{project_prefix}/edge-cases/node-a")
node_c_db = await entity_service.get_by_permalink("edge-cases/node-c") node_c_db = await entity_service.get_by_permalink(f"{project_prefix}/edge-cases/node-c")
assert node_a_db is not None, "Node A should exist in database" assert node_a_db is not None, "Node A should exist in database"
assert node_a_db.title == "Node A", "Node A database entry should have correct title" assert node_a_db.title == "Node A", "Node A database entry should have correct title"
+6 -3
View File
@@ -18,11 +18,11 @@ async def test_get_basic_discussion_context(client, test_graph, test_project):
assert isinstance(context, GraphContext) assert isinstance(context, GraphContext)
assert len(context.results) == 1 assert len(context.results) == 1
assert context.results[0].primary_result.permalink == "test/root" assert context.results[0].primary_result.permalink == f"{test_project.name}/test/root"
assert len(context.results[0].related_results) > 0 assert len(context.results[0].related_results) > 0
# Verify metadata # Verify metadata
assert context.metadata.uri == "test/root" assert context.metadata.uri == f"{test_project.name}/test/root"
assert context.metadata.depth == 1 # default depth assert context.metadata.depth == 1 # default depth
assert context.metadata.timeframe is not None assert context.metadata.timeframe is not None
assert isinstance(context.metadata.generated_at, datetime) assert isinstance(context.metadata.generated_at, datetime)
@@ -38,7 +38,10 @@ async def test_get_discussion_context_pattern(client, test_graph, test_project):
assert isinstance(context, GraphContext) assert isinstance(context, GraphContext)
assert len(context.results) > 1 # Should match multiple test/* paths assert len(context.results) > 1 # Should match multiple test/* paths
assert all("test/" in item.primary_result.permalink for item in context.results) # pyright: ignore [reportOperatorIssue] assert all(
f"{test_project.name}/test/" in item.primary_result.permalink
for item in context.results
) # pyright: ignore [reportOperatorIssue]
assert context.metadata.depth == 1 assert context.metadata.depth == 1
+4 -4
View File
@@ -29,7 +29,7 @@ async def test_edit_note_append_operation(client, test_project):
assert "Edited note (append)" in result assert "Edited note (append)" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result assert "file_path: test/Test Note.md" in result
assert "permalink: test/test-note" in result assert f"permalink: {test_project.name}/test/test-note" in result
assert "Added 3 lines to end of note" in result assert "Added 3 lines to end of note" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
@@ -57,7 +57,7 @@ async def test_edit_note_prepend_operation(client, test_project):
assert "Edited note (prepend)" in result assert "Edited note (prepend)" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: meetings/Meeting Notes.md" in result assert "file_path: meetings/Meeting Notes.md" in result
assert "permalink: meetings/meeting-notes" in result assert f"permalink: {test_project.name}/meetings/meeting-notes" in result
assert "Added 3 lines to beginning of note" in result assert "Added 3 lines to beginning of note" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
@@ -439,7 +439,7 @@ async def test_edit_note_preserves_permalink_when_frontmatter_missing(client, te
) )
assert isinstance(first_result, str) assert isinstance(first_result, str)
assert "permalink: test/test-note" in first_result assert f"permalink: {test_project.name}/test/test-note" in first_result
# Perform another edit - this should preserve the permalink even if the # Perform another edit - this should preserve the permalink even if the
# file doesn't have a permalink in its frontmatter # file doesn't have a permalink in its frontmatter
@@ -453,6 +453,6 @@ async def test_edit_note_preserves_permalink_when_frontmatter_missing(client, te
assert isinstance(second_result, str) assert isinstance(second_result, str)
assert "Edited note (append)" in second_result assert "Edited note (append)" in second_result
assert f"project: {test_project.name}" in second_result assert f"project: {test_project.name}" in second_result
assert "permalink: test/test-note" in second_result assert f"permalink: {test_project.name}/test/test-note" in second_result
assert f"[Session: Using project '{test_project.name}']" in second_result assert f"[Session: Using project '{test_project.name}']" in second_result
# The edit should succeed without validation errors # The edit should succeed without validation errors
+3 -3
View File
@@ -67,7 +67,7 @@ async def test_move_note_success(app, client, test_project):
content = await read_note.fn("target/moved-note", project=test_project.name) content = await read_note.fn("target/moved-note", project=test_project.name)
assert "# Test Note" in content assert "# Test Note" in content
assert "Original content here" in content assert "Original content here" in content
assert "permalink: target/moved-note" in content assert f"permalink: {test_project.name}/target/moved-note" in content
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -426,7 +426,7 @@ async def test_move_note_rename_only(client, test_project):
content = await read_note.fn("test/new-name", project=test_project.name) content = await read_note.fn("test/new-name", project=test_project.name)
assert "# OriginalName" in content # Title in content remains same assert "# OriginalName" in content # Title in content remains same
assert "Content to rename" in content assert "Content to rename" in content
assert "permalink: test/new-name" in content assert f"permalink: {test_project.name}/test/new-name" in content
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -587,7 +587,7 @@ async def test_move_note_preserves_frontmatter(app, client, test_project):
content = await read_note.fn("target/moved-custom-note", project=test_project.name) content = await read_note.fn("target/moved-custom-note", project=test_project.name)
assert "title: Custom Frontmatter Note" in content assert "title: Custom Frontmatter Note" in content
assert "type: note" in content assert "type: note" in content
assert "permalink: target/moved-custom-note" in content assert f"permalink: {test_project.name}/target/moved-custom-note" in content
assert "# Custom Frontmatter Note" in content assert "# Custom Frontmatter Note" in content
assert "Content with custom metadata" in content assert "Content with custom metadata" in content
+16 -1
View File
@@ -119,7 +119,7 @@ async def test_note_unicode_content(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: test/Unicode Test.md" in result assert "file_path: test/Unicode Test.md" in result
assert "permalink: test/unicode-test" in result assert f"permalink: {test_project.name}/test/unicode-test" in result
assert "checksum:" in result # Checksum exists but may be "unknown" assert "checksum:" in result # Checksum exists but may be "unknown"
# Read back should preserve unicode # Read back should preserve unicode
@@ -201,6 +201,21 @@ async def test_read_note_memory_url(app, test_project):
assert "Testing memory:// URL handling" in content 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.fn(
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.fn(memory_url)
assert "Testing memory:// URL handling with project prefix" in content
class TestReadNoteSecurityValidation: class TestReadNoteSecurityValidation:
"""Test read_note security validation features.""" """Test read_note security validation features."""
+17
View File
@@ -150,6 +150,23 @@ async def test_read_file_memory_url(app, synced_files, test_project):
assert "Testing memory:// URL handling for resources" in response["text"] assert "Testing memory:// URL handling for resources" in response["text"]
@pytest.mark.asyncio
async def test_read_file_memory_url_with_project_prefix(app, synced_files, test_project):
"""Test reading a resource using a memory:// URL with explicit project prefix."""
await write_note.fn(
project=test_project.name,
title="Project Prefixed Resource URL Test",
directory="test",
content="Testing memory:// URL handling for resources with project prefix",
)
memory_url = f"memory://{test_project.name}/test/project-prefixed-resource-url-test"
response = await read_content.fn(memory_url)
assert response["type"] == "text"
assert "Testing memory:// URL handling for resources with project prefix" in response["text"]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_image_optimization_functions(app): async def test_image_optimization_functions(app):
"""Test the image optimization helper functions.""" """Test the image optimization helper functions."""
+70 -14
View File
@@ -1,6 +1,7 @@
"""Tests for search MCP tools.""" """Tests for search MCP tools."""
import pytest import pytest
from contextlib import asynccontextmanager
from datetime import datetime, timedelta from datetime import datetime, timedelta
from basic_memory.mcp.tools import write_note from basic_memory.mcp.tools import write_note
@@ -28,7 +29,10 @@ async def test_search_text(client, test_project):
if isinstance(response, SearchResponse): if isinstance(response, SearchResponse):
# Success case - verify SearchResponse # Success case - verify SearchResponse
assert len(response.results) > 0 assert len(response.results) > 0
assert any(r.permalink == "test/test-search-note" for r in response.results) assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
else: else:
# If search failed and returned error message, test should fail with informative message # If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}") pytest.fail(f"Search failed with error: {response}")
@@ -59,7 +63,10 @@ async def test_search_title(client, test_project):
else: else:
# Success case - verify SearchResponse # Success case - verify SearchResponse
assert len(response.results) > 0 assert len(response.results) > 0
assert any(r.permalink == "test/test-search-note" for r in response.results) assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -77,14 +84,19 @@ async def test_search_permalink(client, test_project):
# Search for it # Search for it
response = await search_notes.fn( response = await search_notes.fn(
project=test_project.name, query="test/test-search-note", search_type="permalink" project=test_project.name,
query=f"{test_project.name}/test/test-search-note",
search_type="permalink",
) )
# Verify results - handle both success and error cases # Verify results - handle both success and error cases
if isinstance(response, SearchResponse): if isinstance(response, SearchResponse):
# Success case - verify SearchResponse # Success case - verify SearchResponse
assert len(response.results) > 0 assert len(response.results) > 0
assert any(r.permalink == "test/test-search-note" for r in response.results) assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
else: else:
# If search failed and returned error message, test should fail with informative message # If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}") pytest.fail(f"Search failed with error: {response}")
@@ -105,19 +117,49 @@ async def test_search_permalink_match(client, test_project):
# Search for it # Search for it
response = await search_notes.fn( response = await search_notes.fn(
project=test_project.name, query="test/test-search-*", search_type="permalink" project=test_project.name,
query=f"{test_project.name}/test/test-search-*",
search_type="permalink",
) )
# Verify results - handle both success and error cases # Verify results - handle both success and error cases
if isinstance(response, SearchResponse): if isinstance(response, SearchResponse):
# Success case - verify SearchResponse # Success case - verify SearchResponse
assert len(response.results) > 0 assert len(response.results) > 0
assert any(r.permalink == "test/test-search-note" for r in response.results) assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
else: else:
# If search failed and returned error message, test should fail with informative message # If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}") pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio
async def test_search_memory_url_with_project_prefix(client, test_project):
"""Test searching with a memory:// URL that includes the project prefix."""
result = await write_note.fn(
project=test_project.name,
title="Memory URL Search Note",
directory="test",
content="# Memory URL Search\nThis note should be found via memory URL search",
)
assert result
response = await search_notes.fn(
query=f"memory://{test_project.name}/test/memory-url-search-note"
)
if isinstance(response, SearchResponse):
assert len(response.results) > 0
assert any(
r.permalink == f"{test_project.name}/test/memory-url-search-note"
for r in response.results
)
else:
pytest.fail(f"Search failed with error: {response}")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_search_pagination(client, test_project): async def test_search_pagination(client, test_project):
"""Test basic search functionality.""" """Test basic search functionality."""
@@ -140,7 +182,10 @@ async def test_search_pagination(client, test_project):
if isinstance(response, SearchResponse): if isinstance(response, SearchResponse):
# Success case - verify SearchResponse # Success case - verify SearchResponse
assert len(response.results) == 1 assert len(response.results) == 1
assert any(r.permalink == "test/test-search-note" for r in response.results) assert any(
r.permalink == f"{test_project.name}/test/test-search-note"
for r in response.results
)
else: else:
# If search failed and returned error message, test should fail with informative message # If search failed and returned error message, test should fail with informative message
pytest.fail(f"Search failed with error: {response}") pytest.fail(f"Search failed with error: {response}")
@@ -315,21 +360,23 @@ class TestSearchToolErrorHandling:
async def test_search_notes_exception_handling(self, monkeypatch): async def test_search_notes_exception_handling(self, monkeypatch):
"""Test exception handling in search_notes.""" """Test exception handling in search_notes."""
import importlib import importlib
from contextlib import asynccontextmanager
search_mod = importlib.import_module("basic_memory.mcp.tools.search") search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients") clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject: class StubProject:
project_url = "http://test"
name = "test-project" name = "test-project"
id = 1
external_id = "test-external-id" external_id = "test-external-id"
@asynccontextmanager @asynccontextmanager
async def fake_get_project_client(*args, **kwargs): async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject()) yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
# Mock SearchClient to raise an exception # Mock SearchClient to raise an exception
class MockSearchClient: class MockSearchClient:
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
@@ -339,6 +386,7 @@ class TestSearchToolErrorHandling:
raise Exception("syntax error") raise Exception("syntax error")
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client) monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
# Patch at the clients module level where the import happens # Patch at the clients module level where the import happens
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient) monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
@@ -350,21 +398,23 @@ class TestSearchToolErrorHandling:
async def test_search_notes_permission_error(self, monkeypatch): async def test_search_notes_permission_error(self, monkeypatch):
"""Test search_notes with permission error.""" """Test search_notes with permission error."""
import importlib import importlib
from contextlib import asynccontextmanager
search_mod = importlib.import_module("basic_memory.mcp.tools.search") search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients") clients_mod = importlib.import_module("basic_memory.mcp.clients")
class StubProject: class StubProject:
project_url = "http://test"
name = "test-project" name = "test-project"
id = 1
external_id = "test-external-id" external_id = "test-external-id"
@asynccontextmanager @asynccontextmanager
async def fake_get_project_client(*args, **kwargs): async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject()) yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
# Mock SearchClient to raise a permission error # Mock SearchClient to raise a permission error
class MockSearchClient: class MockSearchClient:
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
@@ -374,6 +424,7 @@ class TestSearchToolErrorHandling:
raise Exception("permission denied") raise Exception("permission denied")
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client) monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
# Patch at the clients module level where the import happens # Patch at the clients module level where the import happens
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient) monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
@@ -387,7 +438,6 @@ class TestSearchToolErrorHandling:
async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch, search_type): async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch, search_type):
"""Vector/hybrid search types should populate retrieval_mode in API payload.""" """Vector/hybrid search types should populate retrieval_mode in API payload."""
import importlib import importlib
from contextlib import asynccontextmanager
search_mod = importlib.import_module("basic_memory.mcp.tools.search") search_mod = importlib.import_module("basic_memory.mcp.tools.search")
clients_mod = importlib.import_module("basic_memory.mcp.clients") clients_mod = importlib.import_module("basic_memory.mcp.clients")
@@ -402,6 +452,11 @@ async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch,
async def fake_get_project_client(*args, **kwargs): async def fake_get_project_client(*args, **kwargs):
yield (object(), StubProject()) yield (object(), StubProject())
async def fake_resolve_project_and_path(
client, identifier, project=None, context=None, headers=None
):
return StubProject(), identifier, False
captured_payload: dict = {} captured_payload: dict = {}
class MockSearchClient: class MockSearchClient:
@@ -413,6 +468,7 @@ async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch,
return SearchResponse(results=[], current_page=page, page_size=page_size) return SearchResponse(results=[], current_page=page, page_size=page_size)
monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client) monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient) monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
result = await search_mod.search_notes.fn( result = await search_mod.search_notes.fn(
+42 -43
View File
@@ -29,31 +29,29 @@ async def test_write_note(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result assert "file_path: test/Test Note.md" in result
assert "permalink: test/test-note" in result assert f"permalink: {test_project.name}/test/test-note" in result
assert "## Tags" in result assert "## Tags" in result
assert "- test, documentation" in result assert "- test, documentation" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
# Try reading it back via permalink # Try reading it back via permalink
content = await read_note.fn("test/test-note", project=test_project.name) content = await read_note.fn("test/test-note", project=test_project.name)
assert ( expected = normalize_newlines(
normalize_newlines( dedent("""
dedent("""
--- ---
title: Test Note title: Test Note
type: note type: note
permalink: test/test-note permalink: {permalink}
tags: tags:
- test - test
- documentation - documentation
--- ---
# Test # Test
This is a test note This is a test note
""").strip() """).format(permalink=f"{test_project.name}/test/test-note").strip()
)
in content
) )
assert expected in content
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -67,24 +65,22 @@ async def test_write_note_no_tags(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: test/Simple Note.md" in result assert "file_path: test/Simple Note.md" in result
assert "permalink: test/simple-note" in result assert f"permalink: {test_project.name}/test/simple-note" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
# Should be able to read it back # Should be able to read it back
content = await read_note.fn("test/simple-note", project=test_project.name) content = await read_note.fn("test/simple-note", project=test_project.name)
assert ( expected = normalize_newlines(
normalize_newlines( dedent("""
dedent("""
--- ---
title: Simple Note title: Simple Note
type: note type: note
permalink: test/simple-note permalink: {permalink}
--- ---
Just some text Just some text
""").strip() """).format(permalink=f"{test_project.name}/test/simple-note").strip()
)
in content
) )
assert expected in content
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -109,7 +105,7 @@ async def test_write_note_update_existing(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result assert "file_path: test/Test Note.md" in result
assert "permalink: test/test-note" in result assert f"permalink: {test_project.name}/test/test-note" in result
assert "## Tags" in result assert "## Tags" in result
assert "- test, documentation" in result assert "- test, documentation" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
@@ -124,7 +120,7 @@ async def test_write_note_update_existing(app, test_project):
assert "# Updated note" in result assert "# Updated note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result assert "file_path: test/Test Note.md" in result
assert "permalink: test/test-note" in result assert f"permalink: {test_project.name}/test/test-note" in result
assert "## Tags" in result assert "## Tags" in result
assert "- test, documentation" in result assert "- test, documentation" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
@@ -138,16 +134,16 @@ async def test_write_note_update_existing(app, test_project):
--- ---
title: Test Note title: Test Note
type: note type: note
permalink: test/test-note permalink: {permalink}
tags: tags:
- test - test
- documentation - documentation
--- ---
# Test # Test
This is an updated note This is an updated note
""" """
).strip() ).format(permalink=f"{test_project.name}/test/test-note").strip()
) )
== content == content
) )
@@ -294,7 +290,7 @@ async def test_write_note_with_tag_array_from_bug_report(app, test_project):
assert result assert result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "permalink: folder/title" in result assert f"permalink: {test_project.name}/folder/title" in result
assert "Tags" in result assert "Tags" in result
assert "hipporag" in result assert "hipporag" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
@@ -327,7 +323,7 @@ async def test_write_note_verbose(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: test/Test Note.md" in result assert "file_path: test/Test Note.md" in result
assert "permalink: test/test-note" in result assert f"permalink: {test_project.name}/test/test-note" in result
assert "## Observations" in result assert "## Observations" in result
assert "- note: 1" in result assert "- note: 1" in result
assert "## Relations" in result assert "## Relations" in result
@@ -441,7 +437,7 @@ async def test_write_note_preserves_content_frontmatter(app, test_project):
--- ---
title: Test Note title: Test Note
type: note type: note
permalink: test/test-note permalink: {permalink}
version: 1.0 version: 1.0
author: name author: name
tags: tags:
@@ -453,7 +449,7 @@ async def test_write_note_preserves_content_frontmatter(app, test_project):
This is a test note This is a test note
""" """
).strip() ).format(permalink=f"{test_project.name}/test/test-note").strip()
) )
in content in content
) )
@@ -480,7 +476,7 @@ async def test_write_note_permalink_collision_fix_issue_139(app, test_project):
) )
assert "# Created note" in result1 assert "# Created note" in result1
assert f"project: {test_project.name}" in result1 assert f"project: {test_project.name}" in result1
assert "permalink: test/note-1" in result1 assert f"permalink: {test_project.name}/test/note-1" in result1
# Step 2: Create second note with different title # Step 2: Create second note with different title
result2 = await write_note.fn( result2 = await write_note.fn(
@@ -488,7 +484,7 @@ async def test_write_note_permalink_collision_fix_issue_139(app, test_project):
) )
assert "# Created note" in result2 assert "# Created note" in result2
assert f"project: {test_project.name}" in result2 assert f"project: {test_project.name}" in result2
assert "permalink: test/note-2" in result2 assert f"permalink: {test_project.name}/test/note-2" in result2
# Step 3: Try to create/replace first note again # Step 3: Try to create/replace first note again
# This scenario would trigger the UNIQUE constraint failure before the fix # This scenario would trigger the UNIQUE constraint failure before the fix
@@ -509,10 +505,13 @@ async def test_write_note_permalink_collision_fix_issue_139(app, test_project):
assert "Updated note" in result3 or "Created note" in result3 assert "Updated note" in result3 or "Created note" in result3
# The result should contain either the original permalink or a unique one # The result should contain either the original permalink or a unique one
assert "permalink: test/note-1" in result3 or "permalink: test/note-1-1" in result3 assert (
f"permalink: {test_project.name}/test/note-1" in result3
or f"permalink: {test_project.name}/test/note-1-1" in result3
)
# Verify we can read back the content # Verify we can read back the content
if "permalink: test/note-1" in result3: if f"permalink: {test_project.name}/test/note-1" in result3:
# Updated existing note case # Updated existing note case
content = await read_note.fn("test/note-1", project=test_project.name) content = await read_note.fn("test/note-1", project=test_project.name)
assert "Replacement content for note 1" in content assert "Replacement content for note 1" in content
@@ -545,20 +544,19 @@ async def test_write_note_with_custom_entity_type(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: guides/Test Guide.md" in result assert "file_path: guides/Test Guide.md" in result
assert "permalink: guides/test-guide" in result assert f"permalink: {test_project.name}/guides/test-guide" in result
assert "## Tags" in result assert "## Tags" in result
assert "- guide, documentation" in result assert "- guide, documentation" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
# Verify the entity type is correctly set in the frontmatter # Verify the entity type is correctly set in the frontmatter
content = await read_note.fn("guides/test-guide", project=test_project.name) content = await read_note.fn("guides/test-guide", project=test_project.name)
assert ( expected = normalize_newlines(
normalize_newlines( dedent("""
dedent("""
--- ---
title: Test Guide title: Test Guide
type: guide type: guide
permalink: guides/test-guide permalink: {permalink}
tags: tags:
- guide - guide
- documentation - documentation
@@ -566,10 +564,9 @@ async def test_write_note_with_custom_entity_type(app, test_project):
# Guide Content # Guide Content
This is a guide This is a guide
""").strip() """).format(permalink=f"{test_project.name}/guides/test-guide").strip()
)
in content
) )
assert expected in content
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -588,7 +585,7 @@ async def test_write_note_with_report_entity_type(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: reports/Monthly Report.md" in result assert "file_path: reports/Monthly Report.md" in result
assert "permalink: reports/monthly-report" in result assert f"permalink: {test_project.name}/reports/monthly-report" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
# Verify the entity type is correctly set in the frontmatter # Verify the entity type is correctly set in the frontmatter
@@ -612,7 +609,7 @@ async def test_write_note_with_config_entity_type(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: config/System Config.md" in result assert "file_path: config/System Config.md" in result
assert "permalink: config/system-config" in result assert f"permalink: {test_project.name}/config/system-config" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
# Verify the entity type is correctly set in the frontmatter # Verify the entity type is correctly set in the frontmatter
@@ -640,7 +637,7 @@ async def test_write_note_entity_type_default_behavior(app, test_project):
assert "# Created note" in result assert "# Created note" in result
assert f"project: {test_project.name}" in result assert f"project: {test_project.name}" in result
assert "file_path: test/Default Type Test.md" in result assert "file_path: test/Default Type Test.md" in result
assert "permalink: test/default-type-test" in result assert f"permalink: {test_project.name}/test/default-type-test" in result
assert f"[Session: Using project '{test_project.name}']" in result assert f"[Session: Using project '{test_project.name}']" in result
# Verify the entity type defaults to "note" # Verify the entity type defaults to "note"
@@ -1035,7 +1032,9 @@ class TestWriteNoteSecurityValidation:
assert "paths must stay within project boundaries" not in result assert "paths must stay within project boundaries" not in result
assert "# Created note" in result assert "# Created note" in result
assert "file_path: security-tests/Full Feature Security Test.md" in result assert "file_path: security-tests/Full Feature Security Test.md" in result
assert "permalink: security-tests/full-feature-security-test" in result assert (
f"permalink: {test_project.name}/security-tests/full-feature-security-test" in result
)
# Should process observations and relations # Should process observations and relations
assert "## Observations" in result assert "## Observations" in result
@@ -37,7 +37,7 @@ async def test_write_note_spaces_to_hyphens(app, test_project, app_config):
) )
assert "file_path: test/my-awesome-note.md" in result assert "file_path: test/my-awesome-note.md" in result
assert "permalink: test/my-awesome-note" in result assert f"permalink: {test_project.name}/test/my-awesome-note" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -53,7 +53,7 @@ async def test_write_note_underscores_to_hyphens(app, test_project, app_config):
) )
assert "file_path: test/my-note-with-underscores.md" in result assert "file_path: test/my-note-with-underscores.md" in result
assert "permalink: test/my-note-with-underscores" in result assert f"permalink: {test_project.name}/test/my-note-with-underscores" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -69,7 +69,7 @@ async def test_write_note_camelcase_to_kebab(app, test_project, app_config):
) )
assert "file_path: test/my-awesome-feature.md" in result assert "file_path: test/my-awesome-feature.md" in result
assert "permalink: test/my-awesome-feature" in result assert f"permalink: {test_project.name}/test/my-awesome-feature" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -85,7 +85,7 @@ async def test_write_note_mixed_case_to_lowercase(app, test_project, app_config)
) )
assert "file_path: test/mixed-case-example.md" in result assert "file_path: test/mixed-case-example.md" in result
assert "permalink: test/mixed-case-example" in result assert f"permalink: {test_project.name}/test/mixed-case-example" in result
# ============================================================================= # =============================================================================
@@ -110,7 +110,7 @@ async def test_write_note_single_period_preserved(app, test_project, app_config)
) )
assert "file_path: test/test-3.0-version.md" in result assert "file_path: test/test-3.0-version.md" in result
assert "permalink: test/test-3.0-version" in result assert f"permalink: {test_project.name}/test/test-3.0-version" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -126,7 +126,7 @@ async def test_write_note_multiple_periods_preserved(app, test_project, app_conf
) )
assert "file_path: test/version-1.2.3-release.md" in result assert "file_path: test/version-1.2.3-release.md" in result
assert "permalink: test/version-1.2.3-release" in result assert f"permalink: {test_project.name}/test/version-1.2.3-release" in result
# ============================================================================= # =============================================================================
@@ -147,7 +147,7 @@ async def test_write_note_special_chars_to_hyphens(app, test_project, app_config
) )
assert "file_path: test/test-2.0-new-feature.md" in result assert "file_path: test/test-2.0-new-feature.md" in result
assert "permalink: test/test-2.0-new-feature" in result assert f"permalink: {test_project.name}/test/test-2.0-new-feature" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -163,7 +163,7 @@ async def test_write_note_parentheses_removed(app, test_project, app_config):
) )
assert "file_path: test/feature-v2.0-update.md" in result assert "file_path: test/feature-v2.0-update.md" in result
assert "permalink: test/feature-v2.0-update" in result assert f"permalink: {test_project.name}/test/feature-v2.0-update" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -179,7 +179,7 @@ async def test_write_note_apostrophes_removed(app, test_project, app_config):
) )
assert "file_path: test/users-guide.md" in result assert "file_path: test/users-guide.md" in result
assert "permalink: test/users-guide" in result assert f"permalink: {test_project.name}/test/users-guide" in result
# ============================================================================= # =============================================================================
@@ -200,7 +200,10 @@ async def test_write_note_all_transformations_combined(app, test_project, app_co
) )
assert "file_path: test/my-project-v3.0-feature-update-draft.md" in result assert "file_path: test/my-project-v3.0-feature-update-draft.md" in result
assert "permalink: test/my-project-v3.0-feature-update-draft" in result assert (
f"permalink: {test_project.name}/test/my-project-v3.0-feature-update-draft"
in result
)
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -217,7 +220,7 @@ async def test_write_note_consecutive_special_chars_collapsed(app, test_project,
# Multiple underscores/hyphens should collapse to single hyphen # Multiple underscores/hyphens should collapse to single hyphen
assert "file_path: test/test-multiple-separators.md" in result assert "file_path: test/test-multiple-separators.md" in result
assert "permalink: test/test-multiple-separators" in result assert f"permalink: {test_project.name}/test/test-multiple-separators" in result
# ============================================================================= # =============================================================================
@@ -238,7 +241,7 @@ async def test_write_note_leading_trailing_hyphens_trimmed(app, test_project, ap
) )
assert "file_path: test/test-note.md" in result assert "file_path: test/test-note.md" in result
assert "permalink: test/test-note" in result assert f"permalink: {test_project.name}/test/test-note" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -254,7 +257,7 @@ async def test_write_note_all_special_chars_becomes_valid_filename(app, test_pro
) )
assert "file_path: test/test.md" in result assert "file_path: test/test.md" in result
assert "permalink: test/test" in result assert f"permalink: {test_project.name}/test/test" in result
# ============================================================================= # =============================================================================
@@ -276,7 +279,7 @@ async def test_write_note_folder_path_unaffected(app, test_project, app_config):
# Folder paths should be preserved (sanitized but not kebab-cased) # Folder paths should be preserved (sanitized but not kebab-cased)
assert "file_path: My_Folder/Sub Folder/test-note.md" in result assert "file_path: My_Folder/Sub Folder/test-note.md" in result
assert "permalink: my-folder/sub-folder/test-note" in result assert f"permalink: {test_project.name}/my-folder/sub-folder/test-note" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -292,7 +295,7 @@ async def test_write_note_root_folder_with_kebab(app, test_project, app_config):
) )
assert "file_path: test-3.0-note.md" in result assert "file_path: test-3.0-note.md" in result
assert "permalink: test-3.0-note" in result assert f"permalink: {test_project.name}/test-3.0-note" in result
# ============================================================================= # =============================================================================
@@ -315,7 +318,7 @@ async def test_write_note_kebab_disabled_preserves_original(app, test_project, a
# Periods and spaces should be preserved # Periods and spaces should be preserved
assert "file_path: test/Test 3.0 Version.md" in result assert "file_path: test/Test 3.0 Version.md" in result
# Permalinks are ALWAYS kebab-case regardless of setting, and preserve periods # Permalinks are ALWAYS kebab-case regardless of setting, and preserve periods
assert "permalink: test/test-3.0-version" in result assert f"permalink: {test_project.name}/test/test-3.0-version" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -331,7 +334,7 @@ async def test_write_note_kebab_disabled_preserves_underscores(app, test_project
) )
assert "file_path: test/my_note_example.md" in result assert "file_path: test/my_note_example.md" in result
assert "permalink: test/my-note-example" in result assert f"permalink: {test_project.name}/test/my-note-example" in result
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -347,7 +350,7 @@ async def test_write_note_kebab_disabled_preserves_case(app, test_project, app_c
) )
assert "file_path: test/MyAwesomeNote.md" in result assert "file_path: test/MyAwesomeNote.md" in result
assert "permalink: test/my-awesome-note" in result assert f"permalink: {test_project.name}/test/my-awesome-note" in result
# ============================================================================= # =============================================================================
@@ -375,7 +378,7 @@ async def test_permalinks_always_kebab_case(app, test_project, app_config):
# Filename preserves original, permalink is kebab-case # Filename preserves original, permalink is kebab-case
assert "file_path: test/Test Note 1.md" in result1 assert "file_path: test/Test Note 1.md" in result1
assert "permalink: test/test-note-1" in result1 assert f"permalink: {test_project.name}/test/test-note-1" in result1
# Test with kebab enabled # Test with kebab enabled
ConfigManager().config.kebab_filenames = True ConfigManager().config.kebab_filenames = True
@@ -389,4 +392,4 @@ async def test_permalinks_always_kebab_case(app, test_project, app_config):
# Both filename and permalink are kebab-case # Both filename and permalink are kebab-case
assert "file_path: test/test-note-2.md" in result2 assert "file_path: test/test-note-2.md" in result2
assert "permalink: test/test-note-2" in result2 assert f"permalink: {test_project.name}/test/test-note-2" in result2
+2 -2
View File
@@ -38,8 +38,8 @@ async def test_search_successful_results(client, test_project):
assert content["query"] == "test content" assert content["query"] == "test content"
# Verify individual result format # Verify individual result format
assert any(r["id"] == "docs/test-document-1" for r in content["results"]) assert any(r["id"] == f"{test_project.name}/docs/test-document-1" for r in content["results"])
assert any(r["id"] == "docs/test-document-2" for r in content["results"]) assert any(r["id"] == f"{test_project.name}/docs/test-document-2" for r in content["results"])
@pytest.mark.asyncio @pytest.mark.asyncio
+5 -5
View File
@@ -137,7 +137,7 @@ async def test_find_connected_timeframe(
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_build_context(context_service, test_graph): async def test_build_context(context_service, test_graph):
"""Test exact permalink lookup.""" """Test exact permalink lookup."""
url = memory_url.validate_strings("memory://test/root") url = memory_url.validate_strings("memory://test-project/test/root")
context_result = await context_service.build_context(url) context_result = await context_service.build_context(url)
# Check metadata # Check metadata
@@ -156,7 +156,7 @@ async def test_build_context(context_service, test_graph):
assert primary_result.id == test_graph["root"].id assert primary_result.id == test_graph["root"].id
assert primary_result.type == "entity" assert primary_result.type == "entity"
assert primary_result.title == "Root" assert primary_result.title == "Root"
assert primary_result.permalink == "test/root" assert primary_result.permalink == "test-project/test/root"
assert primary_result.file_path == "test/Root.md" assert primary_result.file_path == "test/Root.md"
assert primary_result.created_at is not None assert primary_result.created_at is not None
@@ -185,7 +185,7 @@ async def test_build_context_with_observations(context_service, test_graph):
# Let's use those existing observations # Let's use those existing observations
# Build context # Build context
url = memory_url.validate_strings("memory://test/root") url = memory_url.validate_strings("memory://test-project/test/root")
context_result = await context_service.build_context(url, include_observations=True) context_result = await context_service.build_context(url, include_observations=True)
# Check the metadata # Check the metadata
@@ -220,9 +220,9 @@ async def test_build_context_not_found(context_service):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_context_metadata(context_service, test_graph): async def test_context_metadata(context_service, test_graph):
"""Test metadata is correctly populated.""" """Test metadata is correctly populated."""
context = await context_service.build_context("memory://test/root", depth=2) context = await context_service.build_context("memory://test-project/test/root", depth=2)
metadata = context.metadata metadata = context.metadata
assert metadata.uri == "test/root" assert metadata.uri == "test-project/test/root"
assert metadata.depth == 2 assert metadata.depth == 2
assert metadata.generated_at is not None assert metadata.generated_at is not None
assert metadata.primary_count > 0 assert metadata.primary_count > 0
+1 -1
View File
@@ -53,7 +53,7 @@ async def test_directory_tree(directory_service: DirectoryService, test_graph):
assert node_file.entity_id == 1 assert node_file.entity_id == 1
assert node_file.entity_type == "deeper" assert node_file.entity_type == "deeper"
assert node_file.title == "Deeper Entity" assert node_file.title == "Deeper Entity"
assert node_file.permalink == "test/deeper-entity" assert node_file.permalink == "test-project/test/deeper-entity"
assert node_file.directory_path == "/test/Deeper Entity.md" assert node_file.directory_path == "/test/Deeper Entity.md"
assert node_file.file_path == "test/Deeper Entity.md" assert node_file.file_path == "test/Deeper Entity.md"
assert node_file.has_children is False assert node_file.has_children is False
+24 -10
View File
@@ -20,27 +20,31 @@ from basic_memory.utils import generate_permalink
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_create_entity(entity_service: EntityService, file_service: FileService): async def test_create_entity(
entity_service: EntityService, file_service: FileService, project_config: ProjectConfig
):
"""Test successful entity creation.""" """Test successful entity creation."""
entity_data = EntitySchema( entity_data = EntitySchema(
title="Test Entity", title="Test Entity",
directory="", directory="",
entity_type="test", entity_type="test",
) )
# Save expected permalink before create_entity mutates entity_data._permalink
expected_permalink = f"{generate_permalink(project_config.name)}/{entity_data.permalink}"
# Act # Act
entity = await entity_service.create_entity(entity_data) entity = await entity_service.create_entity(entity_data)
# Assert Entity # Assert Entity
assert isinstance(entity, EntityModel) assert isinstance(entity, EntityModel)
assert entity.permalink == entity_data.permalink assert entity.permalink == expected_permalink
assert entity.file_path == entity_data.file_path assert entity.file_path == entity_data.file_path
assert entity.entity_type == "test" assert entity.entity_type == "test"
assert entity.created_at is not None assert entity.created_at is not None
assert len(entity.relations) == 0 assert len(entity.relations) == 0
# Verify we can retrieve it using permalink # Verify we can retrieve it using permalink
retrieved = await entity_service.get_by_permalink(entity_data.permalink) retrieved = await entity_service.get_by_permalink(entity.permalink)
assert retrieved.title == "Test Entity" assert retrieved.title == "Test Entity"
assert retrieved.entity_type == "test" assert retrieved.entity_type == "test"
assert retrieved.created_at is not None assert retrieved.created_at is not None
@@ -59,7 +63,9 @@ async def test_create_entity(entity_service: EntityService, file_service: FileSe
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_create_entity_file_exists(entity_service: EntityService, file_service: FileService): async def test_create_entity_file_exists(
entity_service: EntityService, file_service: FileService, project_config: ProjectConfig
):
"""Test successful entity creation.""" """Test successful entity creation."""
entity_data = EntitySchema( entity_data = EntitySchema(
title="Test Entity", title="Test Entity",
@@ -77,7 +83,8 @@ async def test_create_entity_file_exists(entity_service: EntityService, file_ser
file_content, _ = await file_service.read_file(file_path) file_content, _ = await file_service.read_file(file_path)
assert ( assert (
"---\ntitle: Test Entity\ntype: test\npermalink: test-entity\n---\n\nfirst" == file_content f"---\ntitle: Test Entity\ntype: test\npermalink: {generate_permalink(project_config.name)}/test-entity\n---\n\nfirst"
== file_content
) )
entity_data = EntitySchema( entity_data = EntitySchema(
@@ -108,7 +115,9 @@ async def test_create_entity_unique_permalink(
entity = await entity_service.create_entity(entity_data) entity = await entity_service.create_entity(entity_data)
# default permalink # default permalink
assert entity.permalink == generate_permalink(entity.file_path) assert entity.permalink == (
f"{generate_permalink(project_config.name)}/{generate_permalink(entity.file_path)}"
)
# move file # move file
file_path = file_service.get_entity_path(entity) file_path = file_service.get_entity_path(entity)
@@ -536,7 +545,11 @@ async def test_create_with_content(entity_service: EntityService, file_service:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_update_with_content(entity_service: EntityService, file_service: FileService): async def test_update_with_content(
entity_service: EntityService,
file_service: FileService,
project_config: ProjectConfig,
):
content = """# Git Workflow Guide""" content = """# Git Workflow Guide"""
# Create test entity # Create test entity
@@ -560,13 +573,14 @@ async def test_update_with_content(entity_service: EntityService, file_service:
file_content, _ = await file_service.read_file(file_path) file_content, _ = await file_service.read_file(file_path)
# assert content is in file # assert content is in file
project_prefix = generate_permalink(project_config.name)
assert ( assert (
dedent( dedent(
""" f"""
--- ---
title: Git Workflow Guide title: Git Workflow Guide
type: test type: test
permalink: test/git-workflow-guide permalink: {project_prefix}/test/git-workflow-guide
--- ---
# Git Workflow Guide # Git Workflow Guide
@@ -1414,7 +1428,7 @@ async def test_move_entity_success(
app_config = BasicMemoryConfig(update_permalinks_on_move=False) app_config = BasicMemoryConfig(update_permalinks_on_move=False)
# Move entity # Move entity
assert entity.permalink == "original/test-note" assert entity.permalink == f"{generate_permalink(project_config.name)}/original/test-note"
await entity_service.move_entity( await entity_service.move_entity(
identifier=entity.permalink, identifier=entity.permalink,
destination_path="moved/test-note.md", destination_path="moved/test-note.md",
+82 -34
View File
@@ -6,9 +6,10 @@ import pytest
import pytest_asyncio import pytest_asyncio
from basic_memory.models.knowledge import Entity as EntityModel
from basic_memory.repository import EntityRepository
from basic_memory.schemas.base import Entity as EntitySchema from basic_memory.schemas.base import Entity as EntitySchema
from basic_memory.services.link_resolver import LinkResolver from basic_memory.services.link_resolver import LinkResolver
from basic_memory.models.knowledge import Entity as EntityModel
@pytest_asyncio.fixture @pytest_asyncio.fixture
@@ -110,40 +111,46 @@ async def link_resolver(entity_repository, search_service, test_entities):
return LinkResolver(entity_repository, search_service) return LinkResolver(entity_repository, search_service)
@pytest.fixture
def project_prefix(test_entities) -> str:
"""Project permalink prefix for expected permalinks."""
return test_entities[0].permalink.split("/", 1)[0]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_exact_permalink_match(link_resolver, test_entities): async def test_exact_permalink_match(link_resolver, test_entities, project_prefix):
"""Test resolving a link that exactly matches a permalink.""" """Test resolving a link that exactly matches a permalink."""
entity = await link_resolver.resolve_link("components/core-service") entity = await link_resolver.resolve_link("components/core-service")
assert entity.permalink == "components/core-service" assert entity.permalink == f"{project_prefix}/components/core-service"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_exact_title_match(link_resolver, test_entities): async def test_exact_title_match(link_resolver, test_entities, project_prefix):
"""Test resolving a link that matches an entity title.""" """Test resolving a link that matches an entity title."""
entity = await link_resolver.resolve_link("Core Service") entity = await link_resolver.resolve_link("Core Service")
assert entity.permalink == "components/core-service" assert entity.permalink == f"{project_prefix}/components/core-service"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_duplicate_title_match(link_resolver, test_entities): async def test_duplicate_title_match(link_resolver, test_entities, project_prefix):
"""Test resolving a link that matches an entity title.""" """Test resolving a link that matches an entity title."""
entity = await link_resolver.resolve_link("Core Service") entity = await link_resolver.resolve_link("Core Service")
assert entity.permalink == "components/core-service" assert entity.permalink == f"{project_prefix}/components/core-service"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_fuzzy_title_partial_match(link_resolver): async def test_fuzzy_title_partial_match(link_resolver, project_prefix):
# Test partial match # Test partial match
result = await link_resolver.resolve_link("Auth Serv") result = await link_resolver.resolve_link("Auth Serv")
assert result is not None, "Did not find partial match" assert result is not None, "Did not find partial match"
assert result.permalink == "components/auth-service" assert result.permalink == f"{project_prefix}/components/auth-service"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_fuzzy_title_exact_match(link_resolver): async def test_fuzzy_title_exact_match(link_resolver, project_prefix):
# Test partial match # Test partial match
result = await link_resolver.resolve_link("auth-service") result = await link_resolver.resolve_link("auth-service")
assert result.permalink == "components/auth-service" assert result.permalink == f"{project_prefix}/components/auth-service"
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -183,7 +190,7 @@ async def test_resolve_file(link_resolver):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_folder_title_pattern_with_md_extension(link_resolver, test_entities): async def test_folder_title_pattern_with_md_extension(link_resolver, test_entities, project_prefix):
"""Test resolving folder/title patterns that need .md extension added. """Test resolving folder/title patterns that need .md extension added.
This tests the new logic added in step 4 of resolve_link that handles This tests the new logic added in step 4 of resolve_link that handles
@@ -193,25 +200,25 @@ async def test_folder_title_pattern_with_md_extension(link_resolver, test_entiti
# "components/Core Service" should resolve to file path "components/Core Service.md" # "components/Core Service" should resolve to file path "components/Core Service.md"
entity = await link_resolver.resolve_link("components/Core Service") entity = await link_resolver.resolve_link("components/Core Service")
assert entity is not None assert entity is not None
assert entity.permalink == "components/core-service" assert entity.permalink == f"{project_prefix}/components/core-service"
assert entity.file_path == "components/Core Service.md" assert entity.file_path == "components/Core Service.md"
# Test with different entity # Test with different entity
entity = await link_resolver.resolve_link("config/Service Config") entity = await link_resolver.resolve_link("config/Service Config")
assert entity is not None assert entity is not None
assert entity.permalink == "config/service-config" assert entity.permalink == f"{project_prefix}/config/service-config"
assert entity.file_path == "config/Service Config.md" assert entity.file_path == "config/Service Config.md"
# Test with nested folder structure # Test with nested folder structure
entity = await link_resolver.resolve_link("specs/subspec/Sub Features 1") entity = await link_resolver.resolve_link("specs/subspec/Sub Features 1")
assert entity is not None assert entity is not None
assert entity.permalink == "specs/subspec/sub-features-1" assert entity.permalink == f"{project_prefix}/specs/subspec/sub-features-1"
assert entity.file_path == "specs/subspec/Sub Features 1.md" assert entity.file_path == "specs/subspec/Sub Features 1.md"
# Test that it doesn't try to add .md to things that already have it # Test that it doesn't try to add .md to things that already have it
entity = await link_resolver.resolve_link("components/Core Service.md") entity = await link_resolver.resolve_link("components/Core Service.md")
assert entity is not None assert entity is not None
assert entity.permalink == "components/core-service" assert entity.permalink == f"{project_prefix}/components/core-service"
# Test that it doesn't try to add .md to single words (no slash) # Test that it doesn't try to add .md to single words (no slash)
entity = await link_resolver.resolve_link("NonExistent") entity = await link_resolver.resolve_link("NonExistent")
@@ -220,12 +227,12 @@ async def test_folder_title_pattern_with_md_extension(link_resolver, test_entiti
# Test that it doesn't interfere with exact permalink matches # Test that it doesn't interfere with exact permalink matches
entity = await link_resolver.resolve_link("components/core-service") entity = await link_resolver.resolve_link("components/core-service")
assert entity is not None assert entity is not None
assert entity.permalink == "components/core-service" assert entity.permalink == f"{project_prefix}/components/core-service"
# Tests for strict mode parameter combinations # Tests for strict mode parameter combinations
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_strict_mode_parameter_combinations(link_resolver, test_entities): async def test_strict_mode_parameter_combinations(link_resolver, test_entities, project_prefix):
"""Test all combinations of use_search and strict parameters.""" """Test all combinations of use_search and strict parameters."""
# Test queries # Test queries
@@ -236,11 +243,11 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities):
# Case 1: use_search=True, strict=False (default behavior - fuzzy matching allowed) # Case 1: use_search=True, strict=False (default behavior - fuzzy matching allowed)
result = await link_resolver.resolve_link(exact_match, use_search=True, strict=False) result = await link_resolver.resolve_link(exact_match, use_search=True, strict=False)
assert result is not None assert result is not None
assert result.permalink == "components/auth-service" assert result.permalink == f"{project_prefix}/components/auth-service"
result = await link_resolver.resolve_link(fuzzy_match, use_search=True, strict=False) result = await link_resolver.resolve_link(fuzzy_match, use_search=True, strict=False)
assert result is not None # Should find "Auth Service" via fuzzy matching assert result is not None # Should find "Auth Service" via fuzzy matching
assert result.permalink == "components/auth-service" assert result.permalink == f"{project_prefix}/components/auth-service"
result = await link_resolver.resolve_link(non_existent, use_search=True, strict=False) result = await link_resolver.resolve_link(non_existent, use_search=True, strict=False)
assert result is None assert result is None
@@ -248,7 +255,7 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities):
# Case 2: use_search=True, strict=True (exact matches only, even with search enabled) # Case 2: use_search=True, strict=True (exact matches only, even with search enabled)
result = await link_resolver.resolve_link(exact_match, use_search=True, strict=True) result = await link_resolver.resolve_link(exact_match, use_search=True, strict=True)
assert result is not None assert result is not None
assert result.permalink == "components/auth-service" assert result.permalink == f"{project_prefix}/components/auth-service"
result = await link_resolver.resolve_link(fuzzy_match, use_search=True, strict=True) result = await link_resolver.resolve_link(fuzzy_match, use_search=True, strict=True)
assert result is None # Should NOT find via fuzzy matching in strict mode assert result is None # Should NOT find via fuzzy matching in strict mode
@@ -259,7 +266,7 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities):
# Case 3: use_search=False, strict=False (no search, exact repository matches only) # Case 3: use_search=False, strict=False (no search, exact repository matches only)
result = await link_resolver.resolve_link(exact_match, use_search=False, strict=False) result = await link_resolver.resolve_link(exact_match, use_search=False, strict=False)
assert result is not None assert result is not None
assert result.permalink == "components/auth-service" assert result.permalink == f"{project_prefix}/components/auth-service"
result = await link_resolver.resolve_link(fuzzy_match, use_search=False, strict=False) result = await link_resolver.resolve_link(fuzzy_match, use_search=False, strict=False)
assert result is None # No search means no fuzzy matching assert result is None # No search means no fuzzy matching
@@ -270,7 +277,7 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities):
# Case 4: use_search=False, strict=True (redundant but should work same as case 3) # Case 4: use_search=False, strict=True (redundant but should work same as case 3)
result = await link_resolver.resolve_link(exact_match, use_search=False, strict=True) result = await link_resolver.resolve_link(exact_match, use_search=False, strict=True)
assert result is not None assert result is not None
assert result.permalink == "components/auth-service" assert result.permalink == f"{project_prefix}/components/auth-service"
result = await link_resolver.resolve_link(fuzzy_match, use_search=False, strict=True) result = await link_resolver.resolve_link(fuzzy_match, use_search=False, strict=True)
assert result is None # No search means no fuzzy matching assert result is None # No search means no fuzzy matching
@@ -280,28 +287,28 @@ async def test_strict_mode_parameter_combinations(link_resolver, test_entities):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_exact_match_types_in_strict_mode(link_resolver, test_entities): async def test_exact_match_types_in_strict_mode(link_resolver, test_entities, project_prefix):
"""Test that all types of exact matches work in strict mode.""" """Test that all types of exact matches work in strict mode."""
# 1. Exact permalink match # 1. Exact permalink match
result = await link_resolver.resolve_link("components/core-service", strict=True) result = await link_resolver.resolve_link("components/core-service", strict=True)
assert result is not None assert result is not None
assert result.permalink == "components/core-service" assert result.permalink == f"{project_prefix}/components/core-service"
# 2. Exact title match # 2. Exact title match
result = await link_resolver.resolve_link("Core Service", strict=True) result = await link_resolver.resolve_link("Core Service", strict=True)
assert result is not None assert result is not None
assert result.permalink == "components/core-service" assert result.permalink == f"{project_prefix}/components/core-service"
# 3. Exact file path match # 3. Exact file path match
result = await link_resolver.resolve_link("components/Core Service.md", strict=True) result = await link_resolver.resolve_link("components/Core Service.md", strict=True)
assert result is not None assert result is not None
assert result.permalink == "components/core-service" assert result.permalink == f"{project_prefix}/components/core-service"
# 4. Folder/title pattern with .md extension added # 4. Folder/title pattern with .md extension added
result = await link_resolver.resolve_link("components/Core Service", strict=True) result = await link_resolver.resolve_link("components/Core Service", strict=True)
assert result is not None assert result is not None
assert result.permalink == "components/core-service" assert result.permalink == f"{project_prefix}/components/core-service"
# 5. Non-markdown file (Image.png) # 5. Non-markdown file (Image.png)
result = await link_resolver.resolve_link("Image.png", strict=True) result = await link_resolver.resolve_link("Image.png", strict=True)
@@ -329,14 +336,14 @@ async def test_fuzzy_matching_blocked_in_strict_mode(link_resolver, test_entitie
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_link_normalization_with_strict_mode(link_resolver, test_entities): async def test_link_normalization_with_strict_mode(link_resolver, test_entities, project_prefix):
"""Test that link normalization still works in strict mode.""" """Test that link normalization still works in strict mode."""
# Test bracket removal and alias handling in strict mode # Test bracket removal and alias handling in strict mode
queries_and_expected = [ queries_and_expected = [
("[[Core Service]]", "components/core-service"), ("[[Core Service]]", f"{project_prefix}/components/core-service"),
("[[Core Service|Main]]", "components/core-service"), # Alias should be ignored ("[[Core Service|Main]]", f"{project_prefix}/components/core-service"),
(" [[ Core Service ]] ", "components/core-service"), # Extra whitespace (" [[ Core Service ]] ", f"{project_prefix}/components/core-service"),
] ]
for query, expected_permalink in queries_and_expected: for query, expected_permalink in queries_and_expected:
@@ -346,7 +353,7 @@ async def test_link_normalization_with_strict_mode(link_resolver, test_entities)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_duplicate_title_handling_in_strict_mode(link_resolver, test_entities): async def test_duplicate_title_handling_in_strict_mode(link_resolver, test_entities, project_prefix):
"""Test how duplicate titles are handled in strict mode.""" """Test how duplicate titles are handled in strict mode."""
# "Core Service" appears twice in test data (components/core-service and components2/core-service) # "Core Service" appears twice in test data (components/core-service and components2/core-service)
@@ -356,7 +363,48 @@ async def test_duplicate_title_handling_in_strict_mode(link_resolver, test_entit
result = await link_resolver.resolve_link("Core Service", strict=True) result = await link_resolver.resolve_link("Core Service", strict=True)
assert result is not None assert result is not None
# Should return the first match (components/core-service based on test fixture order) # Should return the first match (components/core-service based on test fixture order)
assert result.permalink == "components/core-service" assert result.permalink == f"{project_prefix}/components/core-service"
@pytest.mark.asyncio
async def test_cross_project_link_resolution(
session_maker, entity_repository, search_service, tmp_path
):
"""Test resolving explicit cross-project links."""
from basic_memory.repository.project_repository import ProjectRepository
project_repo = ProjectRepository(session_maker)
other_project = await project_repo.create(
{
"name": "other-project",
"description": "Secondary project",
"path": str(tmp_path / "other-project"),
"is_active": True,
"is_default": False,
}
)
now = datetime.now(timezone.utc)
other_entity_repo = EntityRepository(session_maker, project_id=other_project.id)
target = await other_entity_repo.add(
EntityModel(
title="Cross Project Note",
entity_type="note",
content_type="text/markdown",
file_path="docs/Cross Project Note.md",
permalink=f"{other_project.permalink}/docs/cross-project-note",
created_at=now,
updated_at=now,
project_id=other_project.id,
)
)
resolver = LinkResolver(entity_repository, search_service)
resolved = await resolver.resolve_link("other-project::Cross Project Note", strict=True)
assert resolved is not None
assert resolved.id == target.id
assert resolved.project_id == other_project.id
# ============================================================================ # ============================================================================
+31 -23
View File
@@ -12,28 +12,32 @@ from basic_memory.schemas.search import SearchQuery, SearchItemType
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_search_permalink(search_service, test_graph): async def test_search_permalink(search_service, test_graph):
"""Exact permalink""" """Exact permalink"""
results = await search_service.search(SearchQuery(permalink="test/root")) results = await search_service.search(SearchQuery(permalink="test-project/test/root"))
assert len(results) == 1 assert len(results) == 1
for r in results: for r in results:
assert "test/root" in r.permalink assert "test-project/test/root" in r.permalink
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_search_limit_offset(search_service, test_graph): async def test_search_limit_offset(search_service, test_graph):
"""Exact permalink""" """Exact permalink"""
results = await search_service.search(SearchQuery(permalink_match="test/*")) results = await search_service.search(SearchQuery(permalink_match="test-project/test/*"))
assert len(results) > 1 assert len(results) > 1
results = await search_service.search(SearchQuery(permalink_match="test/*"), limit=1) results = await search_service.search(
SearchQuery(permalink_match="test-project/test/*"), limit=1
)
assert len(results) == 1 assert len(results) == 1
results = await search_service.search(SearchQuery(permalink_match="test/*"), limit=100) results = await search_service.search(
SearchQuery(permalink_match="test-project/test/*"), limit=100
)
num_results = len(results) num_results = len(results)
# assert offset # assert offset
offset_results = await search_service.search( offset_results = await search_service.search(
SearchQuery(permalink_match="test/*"), limit=100, offset=1 SearchQuery(permalink_match="test-project/test/*"), limit=100, offset=1
) )
assert len(offset_results) == num_results - 1 assert len(offset_results) == num_results - 1
@@ -41,20 +45,24 @@ async def test_search_limit_offset(search_service, test_graph):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_search_permalink_observations_wildcard(search_service, test_graph): async def test_search_permalink_observations_wildcard(search_service, test_graph):
"""Pattern matching""" """Pattern matching"""
results = await search_service.search(SearchQuery(permalink_match="test/root/observations/*")) results = await search_service.search(
SearchQuery(permalink_match="test-project/test/root/observations/*")
)
assert len(results) == 2 assert len(results) == 2
permalinks = {r.permalink for r in results} permalinks = {r.permalink for r in results}
assert "test/root/observations/note/root-note-1" in permalinks assert "test-project/test/root/observations/note/root-note-1" in permalinks
assert "test/root/observations/tech/root-tech-note" in permalinks assert "test-project/test/root/observations/tech/root-tech-note" in permalinks
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_search_permalink_relation_wildcard(search_service, test_graph): async def test_search_permalink_relation_wildcard(search_service, test_graph):
"""Pattern matching""" """Pattern matching"""
results = await search_service.search(SearchQuery(permalink_match="test/root/connects-to/*")) results = await search_service.search(
SearchQuery(permalink_match="test-project/test/root/connects-to/*")
)
assert len(results) == 1 assert len(results) == 1
permalinks = {r.permalink for r in results} permalinks = {r.permalink for r in results}
assert "test/root/connects-to/test/connected-entity-1" in permalinks assert "test-project/test/root/connects-to/test-project/test/connected-entity-1" in permalinks
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -62,13 +70,13 @@ async def test_search_permalink_wildcard2(search_service, test_graph):
"""Pattern matching""" """Pattern matching"""
results = await search_service.search( results = await search_service.search(
SearchQuery( SearchQuery(
permalink_match="test/connected*", permalink_match="test-project/test/connected*",
) )
) )
assert len(results) >= 2 assert len(results) >= 2
permalinks = {r.permalink for r in results} permalinks = {r.permalink for r in results}
assert "test/connected-entity-1" in permalinks assert "test-project/test/connected-entity-1" in permalinks
assert "test/connected-entity-2" in permalinks assert "test-project/test/connected-entity-2" in permalinks
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -78,7 +86,7 @@ async def test_search_text(search_service, test_graph):
SearchQuery(text="Root Entity", entity_types=[SearchItemType.ENTITY]) SearchQuery(text="Root Entity", entity_types=[SearchItemType.ENTITY])
) )
assert len(results) >= 1 assert len(results) >= 1
assert results[0].permalink == "test/root" assert results[0].permalink == "test-project/test/root"
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -88,7 +96,7 @@ async def test_search_title(search_service, test_graph):
SearchQuery(title="Root", entity_types=[SearchItemType.ENTITY]) SearchQuery(title="Root", entity_types=[SearchItemType.ENTITY])
) )
assert len(results) >= 1 assert len(results) >= 1
assert results[0].permalink == "test/root" assert results[0].permalink == "test-project/test/root"
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -96,7 +104,7 @@ async def test_text_search_case_insensitive(search_service, test_graph):
"""Test text search functionality.""" """Test text search functionality."""
# Case insensitive # Case insensitive
results = await search_service.search(SearchQuery(text="ENTITY")) results = await search_service.search(SearchQuery(text="ENTITY"))
assert any("test/root" in r.permalink for r in results) assert any("test-project/test/root" in r.permalink for r in results)
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -115,16 +123,16 @@ async def test_text_search_multiple_terms(search_service, test_graph):
# Multiple terms # Multiple terms
results = await search_service.search(SearchQuery(text="root note")) results = await search_service.search(SearchQuery(text="root note"))
assert any("test/root" in r.permalink for r in results) assert any("test-project/test/root" in r.permalink for r in results)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_pattern_matching(search_service, test_graph): async def test_pattern_matching(search_service, test_graph):
"""Test pattern matching with various wildcards.""" """Test pattern matching with various wildcards."""
# Test wildcards # Test wildcards
results = await search_service.search(SearchQuery(permalink_match="test/*")) results = await search_service.search(SearchQuery(permalink_match="test-project/test/*"))
for r in results: for r in results:
assert "test/" in r.permalink assert "test-project/test/" in r.permalink
# Test start wildcards # Test start wildcards
results = await search_service.search(SearchQuery(permalink_match="*/observations")) results = await search_service.search(SearchQuery(permalink_match="*/observations"))
@@ -132,9 +140,9 @@ async def test_pattern_matching(search_service, test_graph):
assert "/observations" in r.permalink assert "/observations" in r.permalink
# Test permalink partial match # Test permalink partial match
results = await search_service.search(SearchQuery(permalink_match="test")) results = await search_service.search(SearchQuery(permalink_match="test-project/test"))
for r in results: for r in results:
assert "test/" in r.permalink assert "test-project/test/" in r.permalink
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -328,7 +336,7 @@ async def test_boolean_or_search(search_service, test_graph):
connected_found = False connected_found = False
for result in results: for result in results:
if result.permalink == "test/root": if result.permalink == "test-project/test/root":
root_found = True root_found = True
elif "connected" in result.permalink.lower(): elif "connected" in result.permalink.lower():
connected_found = True connected_found = True
+30 -20
View File
@@ -14,6 +14,7 @@ from basic_memory.schemas.search import SearchQuery
from basic_memory.services import EntityService, FileService from basic_memory.services import EntityService, FileService
from basic_memory.services.search_service import SearchService from basic_memory.services.search_service import SearchService
from basic_memory.sync.sync_service import SyncService from basic_memory.sync.sync_service import SyncService
from basic_memory.utils import generate_permalink
async def create_test_file(path: Path, content: str = "test content") -> None: async def create_test_file(path: Path, content: str = "test content") -> None:
@@ -74,7 +75,8 @@ type: knowledge
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
# Verify forward reference # Verify forward reference
source = await entity_service.get_by_permalink("source") project_prefix = generate_permalink(project_config.name)
source = await entity_service.get_by_permalink(f"{project_prefix}/source")
assert len(source.relations) == 1 assert len(source.relations) == 1
assert source.relations[0].to_id is None assert source.relations[0].to_id is None
assert source.relations[0].to_name == "target-doc" assert source.relations[0].to_name == "target-doc"
@@ -98,8 +100,8 @@ Target content
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
# Verify reference is now resolved # Verify reference is now resolved
source = await entity_service.get_by_permalink("source") source = await entity_service.get_by_permalink(f"{project_prefix}/source")
target = await entity_service.get_by_permalink("target-doc") target = await entity_service.get_by_permalink(f"{project_prefix}/target-doc")
assert len(source.relations) == 1 assert len(source.relations) == 1
assert source.relations[0].to_id == target.id assert source.relations[0].to_id == target.id
assert source.relations[0].to_name == target.title assert source.relations[0].to_name == target.title
@@ -144,8 +146,9 @@ Content
# Sync to create both entities # Sync to create both entities
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
source = await entity_service.get_by_permalink("source") project_prefix = generate_permalink(project_config.name)
target = await entity_service.get_by_permalink("target") source = await entity_service.get_by_permalink(f"{project_prefix}/source")
target = await entity_service.get_by_permalink(f"{project_prefix}/target")
# Create a resolved relation (already exists) that the unresolved one would become. # Create a resolved relation (already exists) that the unresolved one would become.
resolved_relation = Relation( resolved_relation = Relation(
@@ -167,7 +170,7 @@ Content
unresolved_id = unresolved_relation.id unresolved_id = unresolved_relation.id
# Verify we have the unresolved relation # Verify we have the unresolved relation
source = await entity_service.get_by_permalink("source") source = await entity_service.get_by_permalink(f"{project_prefix}/source")
unresolved_outgoing = [r for r in source.outgoing_relations if r.to_id is None] unresolved_outgoing = [r for r in source.outgoing_relations if r.to_id is None]
assert len(unresolved_outgoing) == 1 assert len(unresolved_outgoing) == 1
assert unresolved_outgoing[0].id == unresolved_id assert unresolved_outgoing[0].id == unresolved_id
@@ -186,7 +189,7 @@ Content
assert len(unresolved) == 0 assert len(unresolved) == 0
# Verify only the resolved relation remains # Verify only the resolved relation remains
source = await entity_service.get_by_permalink("source") source = await entity_service.get_by_permalink(f"{project_prefix}/source")
assert len(source.outgoing_relations) == 1 assert len(source.outgoing_relations) == 1
assert source.outgoing_relations[0].to_id == target.id assert source.outgoing_relations[0].to_id == target.id
@@ -642,6 +645,7 @@ async def test_permalink_formatting(
sync_service: SyncService, project_config: ProjectConfig, entity_service: EntityService sync_service: SyncService, project_config: ProjectConfig, entity_service: EntityService
): ):
"""Test that permalinks are properly formatted during sync.""" """Test that permalinks are properly formatted during sync."""
project_prefix = generate_permalink(project_config.name)
# Test cases with different filename formats # Test cases with different filename formats
test_files = { test_files = {
@@ -675,8 +679,9 @@ Testing permalink generation.
for filename, expected_permalink in test_files.items(): for filename, expected_permalink in test_files.items():
# Find entity for this file # Find entity for this file
entity = next(e for e in entities if e.file_path == filename) entity = next(e for e in entities if e.file_path == filename)
assert entity.permalink == expected_permalink, ( expected_full_permalink = f"{project_prefix}/{expected_permalink}"
f"File {filename} should have permalink {expected_permalink}" assert entity.permalink == expected_full_permalink, (
f"File {filename} should have permalink {expected_full_permalink}"
) )
@@ -743,12 +748,13 @@ Testing file timestamps
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
# Check explicit frontmatter dates # Check explicit frontmatter dates
explicit_entity = await entity_service.get_by_permalink("explicit-dates") project_prefix = generate_permalink(project_config.name)
explicit_entity = await entity_service.get_by_permalink(f"{project_prefix}/explicit-dates")
assert explicit_entity.created_at is not None assert explicit_entity.created_at is not None
assert explicit_entity.updated_at is not None assert explicit_entity.updated_at is not None
# Check file timestamps # Check file timestamps
file_entity = await entity_service.get_by_permalink("file-dates3") file_entity = await entity_service.get_by_permalink(f"{project_prefix}/file-dates3")
file_stats = file_path.stat() file_stats = file_path.stat()
# Compare using epoch timestamps to handle timezone differences correctly # Compare using epoch timestamps to handle timezone differences correctly
@@ -793,7 +799,8 @@ Initial content for timestamp test
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
# Get initial entity and timestamps # Get initial entity and timestamps
entity_before = await entity_service.get_by_permalink("timestamp-test") project_prefix = generate_permalink(project_config.name)
entity_before = await entity_service.get_by_permalink(f"{project_prefix}/timestamp-test")
initial_updated_at = entity_before.updated_at initial_updated_at = entity_before.updated_at
# Modify the file content and update mtime to be newer than watermark # Modify the file content and update mtime to be newer than watermark
@@ -824,7 +831,7 @@ Modified content for timestamp test
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
# Get entity after re-sync # Get entity after re-sync
entity_after = await entity_service.get_by_permalink("timestamp-test") entity_after = await entity_service.get_by_permalink(f"{project_prefix}/timestamp-test")
# Verify that updated_at changed # Verify that updated_at changed
assert entity_after.updated_at != initial_updated_at, ( assert entity_after.updated_at != initial_updated_at, (
@@ -938,6 +945,7 @@ async def test_sync_permalink_resolved(
): ):
"""Test that we resolve duplicate permalinks on sync .""" """Test that we resolve duplicate permalinks on sync ."""
project_dir = project_config.home project_dir = project_config.home
project_prefix = generate_permalink(project_config.name)
# Create initial file # Create initial file
content = """ content = """
@@ -967,13 +975,13 @@ Content for move test
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
file_content, _ = await file_service.read_file(new_path) file_content, _ = await file_service.read_file(new_path)
assert "permalink: new/moved-file" in file_content assert f"permalink: {project_prefix}/new/moved-file" in file_content
# Create another that has the same permalink # Create another that has the same permalink
content = """ content = f"""
--- ---
type: knowledge type: knowledge
permalink: new/moved-file permalink: {project_prefix}/new/moved-file
--- ---
# Test Move # Test Move
Content for move test Content for move test
@@ -991,7 +999,7 @@ Content for move test
# assert permalink is unique # assert permalink is unique
file_content, _ = await file_service.read_file(old_path) file_content, _ = await file_service.read_file(old_path)
assert "permalink: new/moved-file-1" in file_content assert f"permalink: {project_prefix}/new/moved-file-1" in file_content
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -1124,6 +1132,7 @@ async def test_sync_permalink_updated_on_move(
): ):
"""Test that we update a permalink on a file move if set in config .""" """Test that we update a permalink on a file move if set in config ."""
project_dir = project_config.home project_dir = project_config.home
project_prefix = generate_permalink(project_config.name)
# Create initial file # Create initial file
content = dedent( content = dedent(
@@ -1145,7 +1154,7 @@ async def test_sync_permalink_updated_on_move(
# verify permalink # verify permalink
old_content, _ = await file_service.read_file(old_path) old_content, _ = await file_service.read_file(old_path)
assert "permalink: old/test-move" in old_content assert f"permalink: {project_prefix}/old/test-move" in old_content
# Move the file # Move the file
new_path = project_dir / "new" / "moved_file.md" new_path = project_dir / "new" / "moved_file.md"
@@ -1160,7 +1169,7 @@ async def test_sync_permalink_updated_on_move(
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
file_content, _ = await file_service.read_file(new_path) file_content, _ = await file_service.read_file(new_path)
assert "permalink: new/moved-file" in file_content assert f"permalink: {project_prefix}/new/moved-file" in file_content
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -1297,6 +1306,7 @@ async def test_sync_relation_to_non_markdown_file(
): ):
"""Test that sync resolves permalink conflicts on update.""" """Test that sync resolves permalink conflicts on update."""
project_dir = project_config.home project_dir = project_config.home
project_prefix = generate_permalink(project_config.name)
content = f""" content = f"""
--- ---
@@ -1321,7 +1331,7 @@ tags: []
title: a note title: a note
type: note type: note
tags: [] tags: []
permalink: note permalink: {project_prefix}/note
--- ---
- relates_to [[{test_files["pdf"].name}]] - relates_to [[{test_files["pdf"].name}]]
+10
View File
@@ -215,6 +215,16 @@ class TestConfigManager:
config = BasicMemoryConfig(disable_permalinks=True) config = BasicMemoryConfig(disable_permalinks=True)
assert config.disable_permalinks is True assert config.disable_permalinks is True
def test_permalinks_include_project_flag_default(self):
"""Test that permalinks_include_project defaults to True."""
config = BasicMemoryConfig()
assert config.permalinks_include_project is True
def test_permalinks_include_project_flag_can_be_disabled(self):
"""Test that permalinks_include_project can be set to False."""
config = BasicMemoryConfig(permalinks_include_project=False)
assert config.permalinks_include_project is False
def test_config_manager_respects_custom_config_dir(self, monkeypatch): def test_config_manager_respects_custom_config_dir(self, monkeypatch):
"""Test that ConfigManager respects BASIC_MEMORY_CONFIG_DIR environment variable.""" """Test that ConfigManager respects BASIC_MEMORY_CONFIG_DIR environment variable."""
with tempfile.TemporaryDirectory() as temp_dir: with tempfile.TemporaryDirectory() as temp_dir:
+6 -3
View File
@@ -61,11 +61,14 @@ Testing permalink generation.
# Run sync # Run sync
await sync_service.sync(project_config.home) await sync_service.sync(project_config.home)
# Verify permalinks # Verify permalinks - with project-prefixed permalinks enabled,
# auto-generated permalinks include the project slug prefix
project_prefix = generate_permalink(project_config.name)
for filename, expected_permalink in test_cases: for filename, expected_permalink in test_cases:
entity = await entity_service.repository.get_by_file_path(filename) entity = await entity_service.repository.get_by_file_path(filename)
assert entity.permalink == expected_permalink, ( expected_full = f"{project_prefix}/{expected_permalink}"
f"File {filename} should have permalink {expected_permalink}" assert entity.permalink == expected_full, (
f"File {filename} should have permalink {expected_full}"
) )