mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
phernandez
35884ef3a7
FastMCP library changes now require calling decorated functions via the .fn attribute: - Tools: @mcp.tool() functions return FunctionTool, call with tool.fn() - Prompts: @mcp.prompt() functions return FunctionPrompt, call with prompt.fn() - Resources: @mcp.resource() functions return FunctionResource, call with resource.fn() Updated core files: - view_note.py: read_note() → read_note.fn() - read_note.py: search_notes() → search_notes.fn() (2 locations) - tool.py: 6 MCP tool calls updated to use .fn - recent_activity.py: recent_activity() → recent_activity.fn() - project.py: project_info() → project_info.fn() with type ignore Updated 100+ test files systematically to use .fn attribute and fixed mock targets. All 869 tests now pass. Fixes view_note tool error in Claude Desktop. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
604 lines
23 KiB
Python
604 lines
23 KiB
Python
"""Tests for ProjectService."""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from basic_memory.schemas import (
|
|
ProjectInfoResponse,
|
|
ProjectStatistics,
|
|
ActivityMetrics,
|
|
SystemStatus,
|
|
)
|
|
from basic_memory.services.project_service import ProjectService
|
|
from basic_memory.config import ConfigManager
|
|
|
|
|
|
def test_projects_property(project_service: ProjectService):
|
|
"""Test the projects property."""
|
|
# Get the projects
|
|
projects = project_service.projects
|
|
|
|
# Assert that it returns a dictionary
|
|
assert isinstance(projects, dict)
|
|
# The test config should have at least one project
|
|
assert len(projects) > 0
|
|
|
|
|
|
def test_default_project_property(project_service: ProjectService):
|
|
"""Test the default_project property."""
|
|
# Get the default project
|
|
default_project = project_service.default_project
|
|
|
|
# Assert it's a string and has a value
|
|
assert isinstance(default_project, str)
|
|
assert default_project
|
|
|
|
|
|
def test_current_project_property(project_service: ProjectService):
|
|
"""Test the current_project property."""
|
|
# Save original environment
|
|
original_env = os.environ.get("BASIC_MEMORY_PROJECT")
|
|
|
|
try:
|
|
# Test with environment variable not set
|
|
if "BASIC_MEMORY_PROJECT" in os.environ:
|
|
del os.environ["BASIC_MEMORY_PROJECT"]
|
|
|
|
# Should return default_project when env var not set
|
|
assert project_service.current_project == project_service.default_project
|
|
|
|
# Now set the environment variable
|
|
os.environ["BASIC_MEMORY_PROJECT"] = "test-project"
|
|
|
|
# Should return env var value
|
|
assert project_service.current_project == "test-project"
|
|
finally:
|
|
# Restore original environment
|
|
if original_env is not None:
|
|
os.environ["BASIC_MEMORY_PROJECT"] = original_env
|
|
elif "BASIC_MEMORY_PROJECT" in os.environ:
|
|
del os.environ["BASIC_MEMORY_PROJECT"]
|
|
|
|
"""Test the methods of ProjectService."""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_operations_sync_methods(
|
|
app_config, project_service: ProjectService, config_manager: ConfigManager, tmp_path
|
|
):
|
|
"""Test adding, switching, and removing a project using ConfigManager directly.
|
|
|
|
This test uses the ConfigManager directly instead of the async methods.
|
|
"""
|
|
# Generate a unique project name for testing
|
|
test_project_name = f"test-project-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-project")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
try:
|
|
# Test adding a project (using ConfigManager directly)
|
|
config_manager.add_project(test_project_name, test_project_path)
|
|
|
|
# Verify it was added
|
|
assert test_project_name in project_service.projects
|
|
assert project_service.projects[test_project_name] == test_project_path
|
|
|
|
# Test setting as default
|
|
original_default = project_service.default_project
|
|
config_manager.set_default_project(test_project_name)
|
|
assert project_service.default_project == test_project_name
|
|
|
|
# Restore original default
|
|
if original_default:
|
|
config_manager.set_default_project(original_default)
|
|
|
|
# Test removing the project
|
|
config_manager.remove_project(test_project_name)
|
|
assert test_project_name not in project_service.projects
|
|
|
|
except Exception as e:
|
|
# Clean up in case of error
|
|
if test_project_name in project_service.projects:
|
|
try:
|
|
config_manager.remove_project(test_project_name)
|
|
except Exception:
|
|
pass
|
|
raise e
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_system_status(project_service: ProjectService):
|
|
"""Test getting system status."""
|
|
# Get the system status
|
|
status = project_service.get_system_status()
|
|
|
|
# Assert it returns a valid SystemStatus object
|
|
assert isinstance(status, SystemStatus)
|
|
assert status.version
|
|
assert status.database_path
|
|
assert status.database_size
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_statistics(project_service: ProjectService, test_graph, test_project):
|
|
"""Test getting statistics."""
|
|
# Get statistics
|
|
statistics = await project_service.get_statistics(test_project.id)
|
|
|
|
# Assert it returns a valid ProjectStatistics object
|
|
assert isinstance(statistics, ProjectStatistics)
|
|
assert statistics.total_entities > 0
|
|
assert "test" in statistics.entity_types
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_activity_metrics(project_service: ProjectService, test_graph, test_project):
|
|
"""Test getting activity metrics."""
|
|
# Get activity metrics
|
|
metrics = await project_service.get_activity_metrics(test_project.id)
|
|
|
|
# Assert it returns a valid ActivityMetrics object
|
|
assert isinstance(metrics, ActivityMetrics)
|
|
assert len(metrics.recently_created) > 0
|
|
assert len(metrics.recently_updated) > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_project_info(project_service: ProjectService, test_graph, test_project):
|
|
"""Test getting full project info."""
|
|
# Get project info
|
|
info = await project_service.get_project_info(test_project.name)
|
|
|
|
# Assert it returns a valid ProjectInfoResponse object
|
|
assert isinstance(info, ProjectInfoResponse)
|
|
assert info.project_name
|
|
assert info.project_path
|
|
assert info.default_project
|
|
assert isinstance(info.available_projects, dict)
|
|
assert isinstance(info.statistics, ProjectStatistics)
|
|
assert isinstance(info.activity, ActivityMetrics)
|
|
assert isinstance(info.system, SystemStatus)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_async(project_service: ProjectService, tmp_path):
|
|
"""Test adding a project with the updated async method."""
|
|
test_project_name = f"test-async-project-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-async-project")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
try:
|
|
# Test adding a project
|
|
await project_service.add_project(test_project_name, test_project_path)
|
|
|
|
# Verify it was added to config
|
|
assert test_project_name in project_service.projects
|
|
assert project_service.projects[test_project_name] == test_project_path
|
|
|
|
# Verify it was added to the database
|
|
project = await project_service.repository.get_by_name(test_project_name)
|
|
assert project is not None
|
|
assert project.name == test_project_name
|
|
assert project.path == test_project_path
|
|
|
|
finally:
|
|
# Clean up
|
|
if test_project_name in project_service.projects:
|
|
await project_service.remove_project(test_project_name)
|
|
|
|
# Ensure it was removed from both config and DB
|
|
assert test_project_name not in project_service.projects
|
|
project = await project_service.repository.get_by_name(test_project_name)
|
|
assert project is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_default_project_async(project_service: ProjectService, tmp_path):
|
|
"""Test setting a project as default with the updated async method."""
|
|
# First add a test project
|
|
test_project_name = f"test-default-project-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-default-project")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
original_default = project_service.default_project
|
|
|
|
try:
|
|
# Add the test project
|
|
await project_service.add_project(test_project_name, test_project_path)
|
|
|
|
# Set as default
|
|
await project_service.set_default_project(test_project_name)
|
|
|
|
# Verify it's set as default in config
|
|
assert project_service.default_project == test_project_name
|
|
|
|
# Verify it's set as default in database
|
|
project = await project_service.repository.get_by_name(test_project_name)
|
|
assert project is not None
|
|
assert project.is_default is True
|
|
|
|
# Make sure old default is no longer default
|
|
old_default_project = await project_service.repository.get_by_name(original_default)
|
|
if old_default_project:
|
|
assert old_default_project.is_default is not True
|
|
|
|
finally:
|
|
# Restore original default
|
|
if original_default:
|
|
await project_service.set_default_project(original_default)
|
|
|
|
# Clean up test project
|
|
if test_project_name in project_service.projects:
|
|
await project_service.remove_project(test_project_name)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_project_method(project_service: ProjectService, tmp_path):
|
|
"""Test the get_project method directly."""
|
|
test_project_name = f"test-get-project-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-get-project")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
try:
|
|
# Test getting a non-existent project
|
|
result = await project_service.get_project("non-existent-project")
|
|
assert result is None
|
|
|
|
# Add a project
|
|
await project_service.add_project(test_project_name, test_project_path)
|
|
|
|
# Test getting an existing project
|
|
result = await project_service.get_project(test_project_name)
|
|
assert result is not None
|
|
assert result.name == test_project_name
|
|
assert result.path == test_project_path
|
|
|
|
finally:
|
|
# Clean up
|
|
if test_project_name in project_service.projects:
|
|
await project_service.remove_project(test_project_name)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_default_project_config_db_mismatch(
|
|
project_service: ProjectService, config_manager: ConfigManager, tmp_path
|
|
):
|
|
"""Test set_default_project when project exists in config but not in database."""
|
|
test_project_name = f"test-mismatch-project-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-mismatch-project")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
original_default = project_service.default_project
|
|
|
|
try:
|
|
# Add project to config only (not to database)
|
|
config_manager.add_project(test_project_name, test_project_path)
|
|
|
|
# Verify it's in config but not in database
|
|
assert test_project_name in project_service.projects
|
|
db_project = await project_service.repository.get_by_name(test_project_name)
|
|
assert db_project is None
|
|
|
|
# Try to set as default - this should trigger the error log on line 142
|
|
await project_service.set_default_project(test_project_name)
|
|
|
|
# Should still update config despite database mismatch
|
|
assert project_service.default_project == test_project_name
|
|
|
|
finally:
|
|
# Restore original default
|
|
if original_default:
|
|
config_manager.set_default_project(original_default)
|
|
|
|
# Clean up
|
|
if test_project_name in project_service.projects:
|
|
config_manager.remove_project(test_project_name)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_with_set_default_true(project_service: ProjectService, tmp_path):
|
|
"""Test adding a project with set_default=True enforces single default."""
|
|
test_project_name = f"test-default-true-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-default-true")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
original_default = project_service.default_project
|
|
|
|
try:
|
|
# Get original default project from database
|
|
original_default_project = await project_service.repository.get_by_name(original_default)
|
|
|
|
# Add project with set_default=True
|
|
await project_service.add_project(test_project_name, test_project_path, set_default=True)
|
|
|
|
# Verify new project is set as default in both config and database
|
|
assert project_service.default_project == test_project_name
|
|
|
|
new_project = await project_service.repository.get_by_name(test_project_name)
|
|
assert new_project is not None
|
|
assert new_project.is_default is True
|
|
|
|
# Verify original default is no longer default in database
|
|
if original_default_project:
|
|
refreshed_original = await project_service.repository.get_by_name(original_default)
|
|
assert refreshed_original.is_default is not True
|
|
|
|
# Verify only one project has is_default=True
|
|
all_projects = await project_service.repository.find_all()
|
|
default_projects = [p for p in all_projects if p.is_default is True]
|
|
assert len(default_projects) == 1
|
|
assert default_projects[0].name == test_project_name
|
|
|
|
finally:
|
|
# Restore original default
|
|
if original_default:
|
|
await project_service.set_default_project(original_default)
|
|
|
|
# Clean up test project
|
|
if test_project_name in project_service.projects:
|
|
await project_service.remove_project(test_project_name)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_with_set_default_false(project_service: ProjectService, tmp_path):
|
|
"""Test adding a project with set_default=False doesn't change defaults."""
|
|
test_project_name = f"test-default-false-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-default-false")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
original_default = project_service.default_project
|
|
|
|
try:
|
|
# Add project with set_default=False (explicit)
|
|
await project_service.add_project(test_project_name, test_project_path, set_default=False)
|
|
|
|
# Verify default project hasn't changed
|
|
assert project_service.default_project == original_default
|
|
|
|
# Verify new project is NOT set as default
|
|
new_project = await project_service.repository.get_by_name(test_project_name)
|
|
assert new_project is not None
|
|
assert new_project.is_default is not True
|
|
|
|
# Verify original default is still default
|
|
original_default_project = await project_service.repository.get_by_name(original_default)
|
|
if original_default_project:
|
|
assert original_default_project.is_default is True
|
|
|
|
finally:
|
|
# Clean up test project
|
|
if test_project_name in project_service.projects:
|
|
await project_service.remove_project(test_project_name)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_project_default_parameter_omitted(project_service: ProjectService, tmp_path):
|
|
"""Test adding a project without set_default parameter defaults to False behavior."""
|
|
test_project_name = f"test-default-omitted-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-default-omitted")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
original_default = project_service.default_project
|
|
|
|
try:
|
|
# Add project without set_default parameter (should default to False)
|
|
await project_service.add_project(test_project_name, test_project_path)
|
|
|
|
# Verify default project hasn't changed
|
|
assert project_service.default_project == original_default
|
|
|
|
# Verify new project is NOT set as default
|
|
new_project = await project_service.repository.get_by_name(test_project_name)
|
|
assert new_project is not None
|
|
assert new_project.is_default is not True
|
|
|
|
finally:
|
|
# Clean up test project
|
|
if test_project_name in project_service.projects:
|
|
await project_service.remove_project(test_project_name)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_single_default_project_enforcement_logic(project_service: ProjectService):
|
|
"""Test that _ensure_single_default_project logic works correctly."""
|
|
# Test that the method exists and is callable
|
|
assert hasattr(project_service, "_ensure_single_default_project")
|
|
assert callable(getattr(project_service, "_ensure_single_default_project"))
|
|
|
|
# Call the enforcement method - should work without error
|
|
await project_service._ensure_single_default_project()
|
|
|
|
# Verify there is exactly one default project after enforcement
|
|
all_projects = await project_service.repository.find_all()
|
|
default_projects = [p for p in all_projects if p.is_default is True]
|
|
assert len(default_projects) == 1 # Should have exactly one default
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_synchronize_projects_calls_ensure_single_default(
|
|
project_service: ProjectService, tmp_path
|
|
):
|
|
"""Test that synchronize_projects calls _ensure_single_default_project."""
|
|
test_project_name = f"test-sync-default-{os.urandom(4).hex()}"
|
|
test_project_path = str(tmp_path / "test-sync-default")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
try:
|
|
# Add project to config only (simulating unsynchronized state)
|
|
from basic_memory.config import config_manager
|
|
|
|
config_manager.add_project(test_project_name, test_project_path)
|
|
|
|
# Verify it's in config but not in database
|
|
assert test_project_name in project_service.projects
|
|
db_project = await project_service.repository.get_by_name(test_project_name)
|
|
assert db_project is None
|
|
|
|
# Call synchronize_projects (this should call _ensure_single_default_project)
|
|
await project_service.synchronize_projects()
|
|
|
|
# Verify project is now in database
|
|
db_project = await project_service.repository.get_by_name(test_project_name)
|
|
assert db_project is not None
|
|
|
|
# Verify default project enforcement was applied
|
|
all_projects = await project_service.repository.find_all()
|
|
default_projects = [p for p in all_projects if p.is_default is True]
|
|
assert len(default_projects) <= 1 # Should be exactly 1 or 0
|
|
|
|
finally:
|
|
# Clean up test project
|
|
if test_project_name in project_service.projects:
|
|
await project_service.remove_project(test_project_name)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_synchronize_projects_normalizes_project_names(
|
|
project_service: ProjectService, tmp_path
|
|
):
|
|
"""Test that synchronize_projects normalizes project names in config to match database format."""
|
|
# Use a project name that needs normalization (uppercase, spaces)
|
|
unnormalized_name = "Test Project With Spaces"
|
|
expected_normalized_name = "test-project-with-spaces"
|
|
test_project_path = str(tmp_path / "test-project-spaces")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
# Import config manager outside try block
|
|
from basic_memory.config import config_manager
|
|
|
|
try:
|
|
# Manually add the unnormalized project name to config
|
|
|
|
# Save the original config state for potential debugging
|
|
# original_projects = config_manager.projects.copy()
|
|
|
|
# Add project with unnormalized name directly to config
|
|
config_manager.config.projects[unnormalized_name] = test_project_path
|
|
config_manager.save_config(config_manager.config)
|
|
|
|
# Verify the unnormalized name is in config
|
|
assert unnormalized_name in project_service.projects
|
|
assert project_service.projects[unnormalized_name] == test_project_path
|
|
|
|
# Call synchronize_projects - this should normalize the project name
|
|
await project_service.synchronize_projects()
|
|
|
|
# Verify the config was updated with normalized name
|
|
assert expected_normalized_name in project_service.projects
|
|
assert unnormalized_name not in project_service.projects
|
|
assert project_service.projects[expected_normalized_name] == test_project_path
|
|
|
|
# Verify the project was added to database with normalized name
|
|
db_project = await project_service.repository.get_by_name(expected_normalized_name)
|
|
assert db_project is not None
|
|
assert db_project.name == expected_normalized_name
|
|
assert db_project.path == test_project_path
|
|
assert db_project.permalink == expected_normalized_name
|
|
|
|
# Verify the unnormalized name is not in database
|
|
unnormalized_db_project = await project_service.repository.get_by_name(unnormalized_name)
|
|
assert unnormalized_db_project is None
|
|
|
|
finally:
|
|
# Clean up - remove any test projects from both config and database
|
|
current_projects = project_service.projects.copy()
|
|
for name in [unnormalized_name, expected_normalized_name]:
|
|
if name in current_projects:
|
|
try:
|
|
await project_service.remove_project(name)
|
|
except Exception:
|
|
# Try to clean up manually if remove_project fails
|
|
try:
|
|
config_manager.remove_project(name)
|
|
except Exception:
|
|
pass
|
|
|
|
# Remove from database
|
|
db_project = await project_service.repository.get_by_name(name)
|
|
if db_project:
|
|
await project_service.repository.delete(db_project.id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_synchronize_projects_handles_case_sensitivity_bug(
|
|
project_service: ProjectService, tmp_path
|
|
):
|
|
"""Test that synchronize_projects fixes the case sensitivity bug (Personal vs personal)."""
|
|
# Simulate the exact bug scenario: config has "Personal" but database expects "personal"
|
|
config_name = "Personal"
|
|
normalized_name = "personal"
|
|
test_project_path = str(tmp_path / "personal-project")
|
|
|
|
# Make sure the test directory exists
|
|
os.makedirs(test_project_path, exist_ok=True)
|
|
|
|
# Import config manager outside try block
|
|
from basic_memory.config import config_manager
|
|
|
|
try:
|
|
# Add project with uppercase name to config (simulating the bug scenario)
|
|
config_manager.config.projects[config_name] = test_project_path
|
|
config_manager.save_config(config_manager.config)
|
|
|
|
# Verify the uppercase name is in config
|
|
assert config_name in project_service.projects
|
|
assert project_service.projects[config_name] == test_project_path
|
|
|
|
# Call synchronize_projects - this should fix the case sensitivity issue
|
|
await project_service.synchronize_projects()
|
|
|
|
# Verify the config was updated to use normalized case
|
|
assert normalized_name in project_service.projects
|
|
assert config_name not in project_service.projects
|
|
assert project_service.projects[normalized_name] == test_project_path
|
|
|
|
# Verify the project exists in database with correct normalized name
|
|
db_project = await project_service.repository.get_by_name(normalized_name)
|
|
assert db_project is not None
|
|
assert db_project.name == normalized_name
|
|
assert db_project.path == test_project_path
|
|
|
|
# Verify we can now switch to this project without case sensitivity errors
|
|
# (This would have failed before the fix with "Personal" != "personal")
|
|
project_lookup = await project_service.get_project(normalized_name)
|
|
assert project_lookup is not None
|
|
assert project_lookup.name == normalized_name
|
|
|
|
finally:
|
|
# Clean up
|
|
for name in [config_name, normalized_name]:
|
|
if name in project_service.projects:
|
|
try:
|
|
await project_service.remove_project(name)
|
|
except Exception:
|
|
# Manual cleanup if needed
|
|
try:
|
|
config_manager.remove_project(name)
|
|
except Exception:
|
|
pass
|
|
|
|
db_project = await project_service.repository.get_by_name(name)
|
|
if db_project:
|
|
await project_service.repository.delete(db_project.id)
|