admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: prevent nested project paths to avoid data conflicts (#338)

Browse Source 
Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Paul Hernandez
2025-10-07 09:44:39 -05:00
committed by GitHub
parent 07e304ce8e
commit 795e339333
8 changed files with 1310 additions and 954 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/services/project_service.py
+57
View File
@@ -88,6 +88,40 @@ class ProjectService:
name
)
def _check_nested_paths(self, path1: str, path2: str) -> bool:
"""Check if two paths are nested (one is a prefix of the other).
Args:
path1: First path to compare
path2: Second path to compare
Returns:
True if one path is nested within the other, False otherwise
Examples:
_check_nested_paths("/foo", "/foo/bar") # True (child under parent)
_check_nested_paths("/foo/bar", "/foo") # True (parent over child)
_check_nested_paths("/foo", "/bar") # False (siblings)
"""
# Normalize paths to ensure proper comparison
p1 = Path(path1).resolve()
p2 = Path(path2).resolve()
# Check if either path is a parent of the other
try:
# Check if p2 is under p1
p2.relative_to(p1)
return True
except ValueError:
# Not nested in this direction, check the other
try:
# Check if p1 is under p2
p1.relative_to(p2)
return True
except ValueError:
# Not nested in either direction
return False
async def add_project(self, name: str, path: str, set_default: bool = False) -> None:
"""Add a new project to the configuration and database.
@@ -142,6 +176,29 @@ class ProjectService:
else:
resolved_path = Path(os.path.abspath(os.path.expanduser(path))).as_posix()
# Check for nested paths with existing projects
existing_projects = await self.list_projects()
for existing in existing_projects:
if self._check_nested_paths(resolved_path, existing.path):
# Determine which path is nested within which for appropriate error message
p_new = Path(resolved_path).resolve()
p_existing = Path(existing.path).resolve()
# Check if new path is nested under existing project
if p_new.is_relative_to(p_existing):
raise ValueError(
f"Cannot create project at '{resolved_path}': "
f"path is nested within existing project '{existing.name}' at '{existing.path}'. "
f"Projects cannot share directory trees."
)
else:
# Existing project is nested under new path
raise ValueError(
f"Cannot create project at '{resolved_path}': "
f"existing project '{existing.name}' at '{existing.path}' is nested within this path. "
f"Projects cannot share directory trees."
)
# First add to config file (this will validate the project doesn't exist)
project_config = self.config_manager.add_project(name, resolved_path)
test-int/cli/test_project_commands_integration.py
+54 -45
View File
@@ -1,5 +1,8 @@
"""Integration tests for project CLI commands."""
import tempfile
from pathlib import Path
from typer.testing import CliRunner
from basic_memory.cli.main import app
@@ -52,61 +55,67 @@ def test_project_info_json(app_config, test_project, config_manager):
assert "system" in data
def test_project_add_and_remove(app_config, tmp_path, config_manager):
def test_project_add_and_remove(app_config, config_manager):
"""Test adding and removing a project."""
runner = CliRunner()
new_project_path = tmp_path / "new-project"
new_project_path.mkdir()
# Add project
result = runner.invoke(app, ["project", "add", "new-project", str(new_project_path)])
# Use a separate temporary directory to avoid nested path conflicts
with tempfile.TemporaryDirectory() as temp_dir:
new_project_path = Path(temp_dir) / "new-project"
new_project_path.mkdir()
if result.exit_code != 0:
print(f"STDOUT: {result.stdout}")
print(f"STDERR: {result.stderr}")
assert result.exit_code == 0
assert (
"Project 'new-project' added successfully" in result.stdout
or "added" in result.stdout.lower()
)
# Add project
result = runner.invoke(app, ["project", "add", "new-project", str(new_project_path)])
# Verify it shows up in list
result = runner.invoke(app, ["project", "list"])
assert result.exit_code == 0
assert "new-project" in result.stdout
if result.exit_code != 0:
print(f"STDOUT: {result.stdout}")
print(f"STDERR: {result.stderr}")
assert result.exit_code == 0
assert (
"Project 'new-project' added successfully" in result.stdout
or "added" in result.stdout.lower()
)
# Remove project
result = runner.invoke(app, ["project", "remove", "new-project"])
assert result.exit_code == 0
assert "removed" in result.stdout.lower() or "deleted" in result.stdout.lower()
# Verify it shows up in list
result = runner.invoke(app, ["project", "list"])
assert result.exit_code == 0
assert "new-project" in result.stdout
# Remove project
result = runner.invoke(app, ["project", "remove", "new-project"])
assert result.exit_code == 0
assert "removed" in result.stdout.lower() or "deleted" in result.stdout.lower()
def test_project_set_default(app_config, tmp_path, config_manager):
def test_project_set_default(app_config, config_manager):
"""Test setting default project."""
runner = CliRunner()
new_project_path = tmp_path / "another-project"
new_project_path.mkdir()
# Add a second project
result = runner.invoke(app, ["project", "add", "another-project", str(new_project_path)])
if result.exit_code != 0:
print(f"STDOUT: {result.stdout}")
print(f"STDERR: {result.stderr}")
assert result.exit_code == 0
# Use a separate temporary directory to avoid nested path conflicts
with tempfile.TemporaryDirectory() as temp_dir:
new_project_path = Path(temp_dir) / "another-project"
new_project_path.mkdir()
# Set as default
result = runner.invoke(app, ["project", "default", "another-project"])
if result.exit_code != 0:
print(f"STDOUT: {result.stdout}")
print(f"STDERR: {result.stderr}")
assert result.exit_code == 0
assert "default" in result.stdout.lower()
# Add a second project
result = runner.invoke(app, ["project", "add", "another-project", str(new_project_path)])
if result.exit_code != 0:
print(f"STDOUT: {result.stdout}")
print(f"STDERR: {result.stderr}")
assert result.exit_code == 0
# Verify in list
result = runner.invoke(app, ["project", "list"])
assert result.exit_code == 0
# The new project should have the checkmark now
lines = result.stdout.split("\n")
for line in lines:
if "another-project" in line:
assert "✓" in line
# Set as default
result = runner.invoke(app, ["project", "default", "another-project"])
if result.exit_code != 0:
print(f"STDOUT: {result.stdout}")
print(f"STDERR: {result.stderr}")
assert result.exit_code == 0
assert "default" in result.stdout.lower()
# Verify in list
result = runner.invoke(app, ["project", "list"])
assert result.exit_code == 0
# The new project should have the checkmark now
lines = result.stdout.split("\n")
for line in lines:
if "another-project" in line:
assert "✓" in line
test-int/mcp/test_project_management_integration.py
+39
View File
@@ -512,3 +512,42 @@ async def test_case_preservation_in_project_list(mcp_server, app, test_project):
# Clean up - delete test projects
for project_name in test_projects:
await client.call_tool("delete_project", {"project_name": project_name})
@pytest.mark.asyncio
async def test_nested_project_paths_rejected(mcp_server, app, test_project):
"""Test that creating nested project paths is rejected with clear error message."""
async with Client(mcp_server) as client:
# Create a parent project
parent_name = "parent-project"
parent_path = "/tmp/nested-test/parent"
await client.call_tool(
"create_memory_project",
{
"project_name": parent_name,
"project_path": parent_path,
},
)
# Try to create a child project nested under the parent
child_name = "child-project"
child_path = "/tmp/nested-test/parent/child"
with pytest.raises(Exception) as exc_info:
await client.call_tool(
"create_memory_project",
{
"project_name": child_name,
"project_path": child_path,
},
)
# Verify error message mentions nested paths
error_message = str(exc_info.value)
assert "nested" in error_message.lower()
assert parent_name in error_message or parent_path in error_message
# Clean up parent project
await client.call_tool("delete_project", {"project_name": parent_name})
tests/api/test_project_router.py
+79 -74
View File
@@ -1,5 +1,8 @@
"""Tests for the project router API endpoints."""
import tempfile
from pathlib import Path
import pytest
from basic_memory.schemas.project_info import ProjectItem
@@ -210,58 +213,58 @@ async def test_set_default_project_endpoint(test_config, client, project_service
@pytest.mark.asyncio
async def test_update_project_path_endpoint(
test_config, client, project_service, project_url, tmp_path
):
async def test_update_project_path_endpoint(test_config, client, project_service, project_url):
"""Test the update project endpoint for changing project path."""
# Create a test project to update
test_project_name = "test-update-project"
old_path = (tmp_path / "old-location").as_posix()
new_path = (tmp_path / "new-location").as_posix()
with tempfile.TemporaryDirectory() as temp_dir:
test_root = Path(temp_dir)
old_path = (test_root / "old-location").as_posix()
new_path = (test_root / "new-location").as_posix()
await project_service.add_project(test_project_name, old_path)
await project_service.add_project(test_project_name, old_path)
try:
# Verify initial state
project = await project_service.get_project(test_project_name)
assert project is not None
assert project.path == old_path
# Update the project path
response = await client.patch(
f"{project_url}/project/{test_project_name}", json={"path": new_path}
)
# Verify response
assert response.status_code == 200
data = response.json()
# Check response structure
assert "message" in data
assert "status" in data
assert data["status"] == "success"
assert "old_project" in data
assert "new_project" in data
# Check old project data
assert data["old_project"]["name"] == test_project_name
assert data["old_project"]["path"] == old_path
# Check new project data
assert data["new_project"]["name"] == test_project_name
assert data["new_project"]["path"] == new_path
# Verify project was actually updated in database
updated_project = await project_service.get_project(test_project_name)
assert updated_project is not None
assert updated_project.path == new_path
finally:
# Clean up
try:
await project_service.remove_project(test_project_name)
except Exception:
pass
# Verify initial state
project = await project_service.get_project(test_project_name)
assert project is not None
assert project.path == old_path
# Update the project path
response = await client.patch(
f"{project_url}/project/{test_project_name}", json={"path": new_path}
)
# Verify response
assert response.status_code == 200
data = response.json()
# Check response structure
assert "message" in data
assert "status" in data
assert data["status"] == "success"
assert "old_project" in data
assert "new_project" in data
# Check old project data
assert data["old_project"]["name"] == test_project_name
assert data["old_project"]["path"] == old_path
# Check new project data
assert data["new_project"]["name"] == test_project_name
assert data["new_project"]["path"] == new_path
# Verify project was actually updated in database
updated_project = await project_service.get_project(test_project_name)
assert updated_project is not None
assert updated_project.path == new_path
finally:
# Clean up
try:
await project_service.remove_project(test_project_name)
except Exception:
pass
@pytest.mark.asyncio
@@ -299,41 +302,43 @@ async def test_update_project_is_active_endpoint(test_config, client, project_se
@pytest.mark.asyncio
async def test_update_project_both_params_endpoint(
test_config, client, project_service, project_url, tmp_path
test_config, client, project_service, project_url
):
"""Test the update project endpoint with both path and is_active parameters."""
# Create a test project to update
test_project_name = "test-update-both-project"
old_path = (tmp_path / "old-location").as_posix()
new_path = (tmp_path / "new-location").as_posix()
with tempfile.TemporaryDirectory() as temp_dir:
test_root = Path(temp_dir)
old_path = (test_root / "old-location").as_posix()
new_path = (test_root / "new-location").as_posix()
await project_service.add_project(test_project_name, old_path)
await project_service.add_project(test_project_name, old_path)
try:
# Update both path and is_active (path should take precedence)
response = await client.patch(
f"{project_url}/project/{test_project_name}",
json={"path": new_path, "is_active": False},
)
# Verify response
assert response.status_code == 200
data = response.json()
# Check that path update was performed (takes precedence)
assert data["new_project"]["path"] == new_path
# Verify project was actually updated in database
updated_project = await project_service.get_project(test_project_name)
assert updated_project is not None
assert updated_project.path == new_path
finally:
# Clean up
try:
await project_service.remove_project(test_project_name)
except Exception:
pass
# Update both path and is_active (path should take precedence)
response = await client.patch(
f"{project_url}/project/{test_project_name}",
json={"path": new_path, "is_active": False},
)
# Verify response
assert response.status_code == 200
data = response.json()
# Check that path update was performed (takes precedence)
assert data["new_project"]["path"] == new_path
# Verify project was actually updated in database
updated_project = await project_service.get_project(test_project_name)
assert updated_project is not None
assert updated_project.path == new_path
finally:
# Clean up
try:
await project_service.remove_project(test_project_name)
except Exception:
pass
@pytest.mark.asyncio
tests/db/test_issue_254_foreign_key_constraints.py
+107 -95
View File
@@ -14,7 +14,9 @@ This test file verifies that the fix works correctly in production databases
that have had the migration applied.
"""
import tempfile
from datetime import datetime, timezone
from pathlib import Path
import pytest
@@ -23,7 +25,7 @@ from basic_memory.services.project_service import ProjectService
# @pytest.mark.skip(reason="Issue #254 not fully resolved yet - foreign key constraint errors still occur")
@pytest.mark.asyncio
async def test_issue_254_foreign_key_constraint_fix(project_service: ProjectService, tmp_path):
async def test_issue_254_foreign_key_constraint_fix(project_service: ProjectService):
"""Test to verify issue #254 is fixed: project removal with foreign key constraints.
This test reproduces the exact scenario from issue #254:
@@ -36,123 +38,133 @@ async def test_issue_254_foreign_key_constraint_fix(project_service: ProjectServ
Once issue #254 is fully fixed, remove the @pytest.mark.skip decorator.
"""
test_project_name = "issue-254-verification"
test_project_path = str(tmp_path / "issue-254-verification")
with tempfile.TemporaryDirectory() as temp_dir:
test_root = Path(temp_dir)
test_project_path = str(test_root / "issue-254-verification")
# Step 1: Create test project
await project_service.add_project(test_project_name, test_project_path)
project = await project_service.get_project(test_project_name)
assert project is not None, "Project should be created successfully"
# Step 1: Create test project
await project_service.add_project(test_project_name, test_project_path)
project = await project_service.get_project(test_project_name)
assert project is not None, "Project should be created successfully"
# Step 2: Create related entities that would cause foreign key constraint issues
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.observation_repository import ObservationRepository
from basic_memory.repository.relation_repository import RelationRepository
# Step 2: Create related entities that would cause foreign key constraint issues
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.observation_repository import ObservationRepository
from basic_memory.repository.relation_repository import RelationRepository
entity_repo = EntityRepository(project_service.repository.session_maker, project_id=project.id)
obs_repo = ObservationRepository(
project_service.repository.session_maker, project_id=project.id
)
rel_repo = RelationRepository(project_service.repository.session_maker, project_id=project.id)
entity_repo = EntityRepository(
project_service.repository.session_maker, project_id=project.id
)
obs_repo = ObservationRepository(
project_service.repository.session_maker, project_id=project.id
)
rel_repo = RelationRepository(
project_service.repository.session_maker, project_id=project.id
)
# Create entity
entity_data = {
"title": "Issue 254 Test Entity",
"entity_type": "note",
"content_type": "text/markdown",
"project_id": project.id,
"permalink": "issue-254-entity",
"file_path": "issue-254-entity.md",
"checksum": "issue254test",
"created_at": datetime.now(timezone.utc),
"updated_at": datetime.now(timezone.utc),
}
entity = await entity_repo.create(entity_data)
# Create entity
entity_data = {
"title": "Issue 254 Test Entity",
"entity_type": "note",
"content_type": "text/markdown",
"project_id": project.id,
"permalink": "issue-254-entity",
"file_path": "issue-254-entity.md",
"checksum": "issue254test",
"created_at": datetime.now(timezone.utc),
"updated_at": datetime.now(timezone.utc),
}
entity = await entity_repo.create(entity_data)
# Create observation linked to entity
observation_data = {
"entity_id": entity.id,
"content": "This observation should be cascade deleted",
"category": "test",
}
observation = await obs_repo.create(observation_data)
# Create observation linked to entity
observation_data = {
"entity_id": entity.id,
"content": "This observation should be cascade deleted",
"category": "test",
}
observation = await obs_repo.create(observation_data)
# Create relation involving the entity
relation_data = {
"from_id": entity.id,
"to_name": "some-other-entity",
"relation_type": "relates-to",
}
relation = await rel_repo.create(relation_data)
# Create relation involving the entity
relation_data = {
"from_id": entity.id,
"to_name": "some-other-entity",
"relation_type": "relates-to",
}
relation = await rel_repo.create(relation_data)
# Step 3: Attempt to remove the project
# This is where issue #254 manifested - should NOT raise "FOREIGN KEY constraint failed"
try:
await project_service.remove_project(test_project_name)
except Exception as e:
if "FOREIGN KEY constraint failed" in str(e):
pytest.fail(
f"Issue #254 not fixed - foreign key constraint error still occurs: {e}. "
f"The migration a1b2c3d4e5f6 may not have been applied correctly or "
f"the CASCADE DELETE constraint is not working as expected."
)
else:
# Re-raise unexpected errors
raise
# Step 3: Attempt to remove the project
# This is where issue #254 manifested - should NOT raise "FOREIGN KEY constraint failed"
try:
await project_service.remove_project(test_project_name)
except Exception as e:
if "FOREIGN KEY constraint failed" in str(e):
pytest.fail(
f"Issue #254 not fixed - foreign key constraint error still occurs: {e}. "
f"The migration a1b2c3d4e5f6 may not have been applied correctly or "
f"the CASCADE DELETE constraint is not working as expected."
)
else:
# Re-raise unexpected errors
raise
# Step 4: Verify project was successfully removed
removed_project = await project_service.get_project(test_project_name)
assert removed_project is None, "Project should have been removed"
# Step 4: Verify project was successfully removed
removed_project = await project_service.get_project(test_project_name)
assert removed_project is None, "Project should have been removed"
# Step 5: Verify related data was cascade deleted
remaining_entity = await entity_repo.find_by_id(entity.id)
assert remaining_entity is None, "Entity should have been cascade deleted"
# Step 5: Verify related data was cascade deleted
remaining_entity = await entity_repo.find_by_id(entity.id)
assert remaining_entity is None, "Entity should have been cascade deleted"
remaining_observation = await obs_repo.find_by_id(observation.id)
assert remaining_observation is None, "Observation should have been cascade deleted"
remaining_observation = await obs_repo.find_by_id(observation.id)
assert remaining_observation is None, "Observation should have been cascade deleted"
remaining_relation = await rel_repo.find_by_id(relation.id)
assert remaining_relation is None, "Relation should have been cascade deleted"
remaining_relation = await rel_repo.find_by_id(relation.id)
assert remaining_relation is None, "Relation should have been cascade deleted"
@pytest.mark.asyncio
async def test_issue_254_reproduction(project_service: ProjectService, tmp_path):
async def test_issue_254_reproduction(project_service: ProjectService):
"""Test that reproduces issue #254 to document the current state.
This test demonstrates the current behavior and will fail until the issue is fixed.
It serves as documentation of what the problem was.
"""
test_project_name = "issue-254-reproduction"
test_project_path = str(tmp_path / "issue-254-reproduction")
with tempfile.TemporaryDirectory() as temp_dir:
test_root = Path(temp_dir)
test_project_path = str(test_root / "issue-254-reproduction")
# Create project and entity
await project_service.add_project(test_project_name, test_project_path)
project = await project_service.get_project(test_project_name)
# Create project and entity
await project_service.add_project(test_project_name, test_project_path)
project = await project_service.get_project(test_project_name)
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.entity_repository import EntityRepository
entity_repo = EntityRepository(project_service.repository.session_maker, project_id=project.id)
entity_repo = EntityRepository(
project_service.repository.session_maker, project_id=project.id
)
entity_data = {
"title": "Reproduction Entity",
"entity_type": "note",
"content_type": "text/markdown",
"project_id": project.id,
"permalink": "reproduction-entity",
"file_path": "reproduction-entity.md",
"checksum": "repro123",
"created_at": datetime.now(timezone.utc),
"updated_at": datetime.now(timezone.utc),
}
await entity_repo.create(entity_data)
entity_data = {
"title": "Reproduction Entity",
"entity_type": "note",
"content_type": "text/markdown",
"project_id": project.id,
"permalink": "reproduction-entity",
"file_path": "reproduction-entity.md",
"checksum": "repro123",
"created_at": datetime.now(timezone.utc),
"updated_at": datetime.now(timezone.utc),
}
await entity_repo.create(entity_data)
# This should eventually work without errors once issue #254 is fixed
# with pytest.raises(Exception) as exc_info:
await project_service.remove_project(test_project_name)
# This should eventually work without errors once issue #254 is fixed
# with pytest.raises(Exception) as exc_info:
await project_service.remove_project(test_project_name)
# Document the current error for tracking
# error_message = str(exc_info.value)
# assert any(keyword in error_message for keyword in [
# "FOREIGN KEY constraint failed",
# "constraint",
# "integrity"
# ]), f"Expected foreign key or integrity constraint error, got: {error_message}"
# Document the current error for tracking
# error_message = str(exc_info.value)
# assert any(keyword in error_message for keyword in [
# "FOREIGN KEY constraint failed",
# "constraint",
# "integrity"
# ]), f"Expected foreign key or integrity constraint error, got: {error_message}"
tests/services/test_project_removal_bug.py
+108 -104
View File
@@ -1,7 +1,9 @@
"""Test for project removal bug #254."""
import os
import tempfile
from datetime import timezone, datetime
from pathlib import Path
import pytest
@@ -9,7 +11,7 @@ from basic_memory.services.project_service import ProjectService
@pytest.mark.asyncio
async def test_remove_project_with_related_entities(project_service: ProjectService, tmp_path):
async def test_remove_project_with_related_entities(project_service: ProjectService):
"""Test removing a project that has related entities (reproduces issue #254).
This test verifies that projects with related entities (entities, observations, relations)
@@ -19,116 +21,118 @@ async def test_remove_project_with_related_entities(project_service: ProjectServ
the project table was recreated in migration 647e7a75e2cd.
"""
test_project_name = f"test-remove-with-entities-{os.urandom(4).hex()}"
test_project_path = str(tmp_path / "test-remove-with-entities")
with tempfile.TemporaryDirectory() as temp_dir:
test_root = Path(temp_dir)
test_project_path = str(test_root / "test-remove-with-entities")
# Make sure the test directory exists
os.makedirs(test_project_path, exist_ok=True)
# Make sure the test directory exists
os.makedirs(test_project_path, exist_ok=True)
try:
# Step 1: Add the test project
await project_service.add_project(test_project_name, test_project_path)
try:
# Step 1: Add the test project
await project_service.add_project(test_project_name, test_project_path)
# Verify project exists
project = await project_service.get_project(test_project_name)
assert project is not None
# Verify project exists
project = await project_service.get_project(test_project_name)
assert project is not None
# Step 2: Create related entities for this project
from basic_memory.repository.entity_repository import EntityRepository
# Step 2: Create related entities for this project
from basic_memory.repository.entity_repository import EntityRepository
entity_repo = EntityRepository(
project_service.repository.session_maker, project_id=project.id
)
entity_data = {
"title": "Test Entity for Deletion",
"entity_type": "note",
"content_type": "text/markdown",
"project_id": project.id,
"permalink": "test-deletion-entity",
"file_path": "test-deletion-entity.md",
"checksum": "test123",
"created_at": datetime.now(timezone.utc),
"updated_at": datetime.now(timezone.utc),
}
entity = await entity_repo.create(entity_data)
assert entity is not None
# Step 3: Create observations for the entity
from basic_memory.repository.observation_repository import ObservationRepository
obs_repo = ObservationRepository(
project_service.repository.session_maker, project_id=project.id
)
observation_data = {
"entity_id": entity.id,
"content": "This is a test observation",
"category": "note",
}
observation = await obs_repo.create(observation_data)
assert observation is not None
# Step 4: Create relations involving the entity
from basic_memory.repository.relation_repository import RelationRepository
rel_repo = RelationRepository(
project_service.repository.session_maker, project_id=project.id
)
relation_data = {
"from_id": entity.id,
"to_name": "some-target-entity",
"relation_type": "relates-to",
}
relation = await rel_repo.create(relation_data)
assert relation is not None
# Step 5: Attempt to remove the project
# This should work with proper cascade delete, or fail with foreign key constraint
await project_service.remove_project(test_project_name)
# Step 6: Verify everything was properly deleted
# Project should be gone
removed_project = await project_service.get_project(test_project_name)
assert removed_project is None, "Project should have been removed"
# Related entities should be cascade deleted
remaining_entity = await entity_repo.find_by_id(entity.id)
assert remaining_entity is None, "Entity should have been cascade deleted"
# Observations should be cascade deleted
remaining_obs = await obs_repo.find_by_id(observation.id)
assert remaining_obs is None, "Observation should have been cascade deleted"
# Relations should be cascade deleted
remaining_rel = await rel_repo.find_by_id(relation.id)
assert remaining_rel is None, "Relation should have been cascade deleted"
except Exception as e:
# Check if this is the specific foreign key constraint error from the bug report
if "FOREIGN KEY constraint failed" in str(e):
pytest.fail(
f"Bug #254 reproduced: {e}. "
"This indicates missing foreign key constraints with CASCADE DELETE. "
"Run migration a1b2c3d4e5f6_fix_project_foreign_keys.py to fix this."
entity_repo = EntityRepository(
project_service.repository.session_maker, project_id=project.id
)
else:
# Re-raise other unexpected errors
raise e
finally:
# Clean up - remove project if it still exists
if test_project_name in project_service.projects:
try:
await project_service.remove_project(test_project_name)
except Exception:
# Manual cleanup if remove_project fails
entity_data = {
"title": "Test Entity for Deletion",
"entity_type": "note",
"content_type": "text/markdown",
"project_id": project.id,
"permalink": "test-deletion-entity",
"file_path": "test-deletion-entity.md",
"checksum": "test123",
"created_at": datetime.now(timezone.utc),
"updated_at": datetime.now(timezone.utc),
}
entity = await entity_repo.create(entity_data)
assert entity is not None
# Step 3: Create observations for the entity
from basic_memory.repository.observation_repository import ObservationRepository
obs_repo = ObservationRepository(
project_service.repository.session_maker, project_id=project.id
)
observation_data = {
"entity_id": entity.id,
"content": "This is a test observation",
"category": "note",
}
observation = await obs_repo.create(observation_data)
assert observation is not None
# Step 4: Create relations involving the entity
from basic_memory.repository.relation_repository import RelationRepository
rel_repo = RelationRepository(
project_service.repository.session_maker, project_id=project.id
)
relation_data = {
"from_id": entity.id,
"to_name": "some-target-entity",
"relation_type": "relates-to",
}
relation = await rel_repo.create(relation_data)
assert relation is not None
# Step 5: Attempt to remove the project
# This should work with proper cascade delete, or fail with foreign key constraint
await project_service.remove_project(test_project_name)
# Step 6: Verify everything was properly deleted
# Project should be gone
removed_project = await project_service.get_project(test_project_name)
assert removed_project is None, "Project should have been removed"
# Related entities should be cascade deleted
remaining_entity = await entity_repo.find_by_id(entity.id)
assert remaining_entity is None, "Entity should have been cascade deleted"
# Observations should be cascade deleted
remaining_obs = await obs_repo.find_by_id(observation.id)
assert remaining_obs is None, "Observation should have been cascade deleted"
# Relations should be cascade deleted
remaining_rel = await rel_repo.find_by_id(relation.id)
assert remaining_rel is None, "Relation should have been cascade deleted"
except Exception as e:
# Check if this is the specific foreign key constraint error from the bug report
if "FOREIGN KEY constraint failed" in str(e):
pytest.fail(
f"Bug #254 reproduced: {e}. "
"This indicates missing foreign key constraints with CASCADE DELETE. "
"Run migration a1b2c3d4e5f6_fix_project_foreign_keys.py to fix this."
)
else:
# Re-raise other unexpected errors
raise e
finally:
# Clean up - remove project if it still exists
if test_project_name in project_service.projects:
try:
project_service.config_manager.remove_project(test_project_name)
await project_service.remove_project(test_project_name)
except Exception:
pass
# Manual cleanup if remove_project fails
try:
project_service.config_manager.remove_project(test_project_name)
except Exception:
pass
project = await project_service.get_project(test_project_name)
if project:
await project_service.repository.delete(project.id)
project = await project_service.get_project(test_project_name)
if project:
await project_service.repository.delete(project.id)
tests/services/test_project_service.py
+793 -571
View File
File diff suppressed because it is too large Load Diff
tests/services/test_project_service_operations.py
+73 -65
View File
@@ -1,6 +1,8 @@
"""Additional tests for ProjectService operations."""
import os
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
@@ -9,95 +11,101 @@ from basic_memory.services.project_service import ProjectService
@pytest.mark.asyncio
async def test_get_project_from_database(project_service: ProjectService, tmp_path):
async def test_get_project_from_database(project_service: ProjectService):
"""Test getting projects from the database."""
# Generate unique project name for testing
test_project_name = f"test-project-{os.urandom(4).hex()}"
test_path = str(tmp_path / "test-project")
with tempfile.TemporaryDirectory() as temp_dir:
test_root = Path(temp_dir)
test_path = str(test_root / "test-project")
# Make sure directory exists
os.makedirs(test_path, exist_ok=True)
# Make sure directory exists
os.makedirs(test_path, exist_ok=True)
try:
# Add a project to the database
project_data = {
"name": test_project_name,
"path": test_path,
"permalink": test_project_name.lower().replace(" ", "-"),
"is_active": True,
"is_default": False,
}
await project_service.repository.create(project_data)
try:
# Add a project to the database
project_data = {
"name": test_project_name,
"path": test_path,
"permalink": test_project_name.lower().replace(" ", "-"),
"is_active": True,
"is_default": False,
}
await project_service.repository.create(project_data)
# Verify we can get the project
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_path
# Verify we can get the project
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_path
finally:
# Clean up
project = await project_service.repository.get_by_name(test_project_name)
if project:
await project_service.repository.delete(project.id)
finally:
# Clean up
project = await project_service.repository.get_by_name(test_project_name)
if project:
await project_service.repository.delete(project.id)
@pytest.mark.asyncio
async def test_add_project_to_config(project_service: ProjectService, tmp_path, config_manager):
async def test_add_project_to_config(project_service: ProjectService, config_manager):
"""Test adding a project to the config manager."""
# Generate unique project name for testing
test_project_name = f"config-project-{os.urandom(4).hex()}"
test_path = (tmp_path / "config-project").as_posix()
with tempfile.TemporaryDirectory() as temp_dir:
test_root = Path(temp_dir)
test_path = (test_root / "config-project").as_posix()
# Make sure directory exists
os.makedirs(test_path, exist_ok=True)
# Make sure directory exists
os.makedirs(test_path, exist_ok=True)
try:
# Add a project to config only (using ConfigManager directly)
config_manager.add_project(test_project_name, test_path)
try:
# Add a project to config only (using ConfigManager directly)
config_manager.add_project(test_project_name, test_path)
# Verify it's in the config
assert test_project_name in project_service.projects
assert project_service.projects[test_project_name] == test_path
# Verify it's in the config
assert test_project_name in project_service.projects
assert project_service.projects[test_project_name] == test_path
finally:
# Clean up
if test_project_name in project_service.projects:
config_manager.remove_project(test_project_name)
finally:
# Clean up
if test_project_name in project_service.projects:
config_manager.remove_project(test_project_name)
@pytest.mark.asyncio
async def test_update_project_path(project_service: ProjectService, tmp_path, config_manager):
async def test_update_project_path(project_service: ProjectService, config_manager):
"""Test updating a project's path."""
# Create a test project
test_project = f"path-update-test-project-{os.urandom(4).hex()}"
original_path = (tmp_path / "original-path").as_posix()
new_path = (tmp_path / "new-path").as_posix()
with tempfile.TemporaryDirectory() as temp_dir:
test_root = Path(temp_dir)
original_path = (test_root / "original-path").as_posix()
new_path = (test_root / "new-path").as_posix()
# Make sure directories exist
os.makedirs(original_path, exist_ok=True)
os.makedirs(new_path, exist_ok=True)
# Make sure directories exist
os.makedirs(original_path, exist_ok=True)
os.makedirs(new_path, exist_ok=True)
try:
# Add the project
await project_service.add_project(test_project, original_path)
try:
# Add the project
await project_service.add_project(test_project, original_path)
# Mock the update_project method to avoid issues with complex DB updates
with patch.object(project_service, "update_project"):
# Just check if the project exists
project = await project_service.repository.get_by_name(test_project)
assert project is not None
assert project.path == original_path
# Since we mock the update_project method, we skip verifying path updates
finally:
# Clean up
if test_project in project_service.projects:
try:
# Mock the update_project method to avoid issues with complex DB updates
with patch.object(project_service, "update_project"):
# Just check if the project exists
project = await project_service.repository.get_by_name(test_project)
if project:
await project_service.repository.delete(project.id)
config_manager.remove_project(test_project)
except Exception:
pass
assert project is not None
assert project.path == original_path
# Since we mock the update_project method, we skip verifying path updates
finally:
# Clean up
if test_project in project_service.projects:
try:
project = await project_service.repository.get_by_name(test_project)
if project:
await project_service.repository.delete(project.id)
config_manager.remove_project(test_project)
except Exception:
pass