mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Paul Hernandez
fb5e9e1d77
Signed-off-by: phernandez <paul@basicmachines.co> Co-authored-by: Claude <noreply@anthropic.com>
298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""Tests for the ProjectRepository."""
|
|
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
|
|
from basic_memory import db
|
|
from basic_memory.models.project import Project
|
|
from basic_memory.repository.project_repository import ProjectRepository
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def sample_project(project_repository: ProjectRepository) -> Project:
|
|
"""Create a sample project for testing."""
|
|
project_data = {
|
|
"name": "Sample Project",
|
|
"description": "A sample project",
|
|
"path": "/sample/project/path",
|
|
"is_active": True,
|
|
"is_default": False,
|
|
"created_at": datetime.now(timezone.utc),
|
|
"updated_at": datetime.now(timezone.utc),
|
|
}
|
|
return await project_repository.create(project_data)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_project(project_repository: ProjectRepository):
|
|
"""Test creating a new project."""
|
|
project_data = {
|
|
"name": "Sample Project",
|
|
"description": "A sample project",
|
|
"path": "/sample/project/path",
|
|
"is_active": True,
|
|
"is_default": False,
|
|
}
|
|
project = await project_repository.create(project_data)
|
|
|
|
# Verify returned object
|
|
assert project.id is not None
|
|
assert project.name == "Sample Project"
|
|
assert project.description == "A sample project"
|
|
assert project.path == "/sample/project/path"
|
|
assert project.is_active is True
|
|
assert project.is_default is False
|
|
assert isinstance(project.created_at, datetime)
|
|
assert isinstance(project.updated_at, datetime)
|
|
|
|
# Verify permalink was generated correctly
|
|
assert project.permalink == "sample-project"
|
|
|
|
# Verify in database
|
|
found = await project_repository.find_by_id(project.id)
|
|
assert found is not None
|
|
assert found.id == project.id
|
|
assert found.name == project.name
|
|
assert found.description == project.description
|
|
assert found.path == project.path
|
|
assert found.permalink == "sample-project"
|
|
assert found.is_active is True
|
|
assert found.is_default is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_name(project_repository: ProjectRepository, sample_project: Project):
|
|
"""Test getting a project by name."""
|
|
# Test exact match
|
|
found = await project_repository.get_by_name(sample_project.name)
|
|
assert found is not None
|
|
assert found.id == sample_project.id
|
|
assert found.name == sample_project.name
|
|
|
|
# Test non-existent name
|
|
found = await project_repository.get_by_name("Non-existent Project")
|
|
assert found is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_permalink(project_repository: ProjectRepository, sample_project: Project):
|
|
"""Test getting a project by permalink."""
|
|
# Verify the permalink value
|
|
assert sample_project.permalink == "sample-project"
|
|
|
|
# Test exact match
|
|
found = await project_repository.get_by_permalink(sample_project.permalink)
|
|
assert found is not None
|
|
assert found.id == sample_project.id
|
|
assert found.permalink == sample_project.permalink
|
|
|
|
# Test non-existent permalink
|
|
found = await project_repository.get_by_permalink("non-existent-project")
|
|
assert found is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_path(project_repository: ProjectRepository, sample_project: Project):
|
|
"""Test getting a project by path."""
|
|
# Test exact match
|
|
found = await project_repository.get_by_path(sample_project.path)
|
|
assert found is not None
|
|
assert found.id == sample_project.id
|
|
assert found.path == sample_project.path
|
|
|
|
# Test with Path object
|
|
found = await project_repository.get_by_path(Path(sample_project.path))
|
|
assert found is not None
|
|
assert found.id == sample_project.id
|
|
assert found.path == sample_project.path
|
|
|
|
# Test non-existent path
|
|
found = await project_repository.get_by_path("/non/existent/path")
|
|
assert found is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_default_project(project_repository: ProjectRepository, test_project: Project):
|
|
"""Test getting the default project."""
|
|
# We already have a default project from the test_project fixture
|
|
# So just create a non-default project
|
|
non_default_project_data = {
|
|
"name": "Non-Default Project",
|
|
"description": "A non-default project",
|
|
"path": "/non-default/project/path",
|
|
"is_active": True,
|
|
"is_default": None, # Not the default project
|
|
}
|
|
|
|
await project_repository.create(non_default_project_data)
|
|
|
|
# Get default project
|
|
default_project = await project_repository.get_default_project()
|
|
assert default_project is not None
|
|
assert default_project.is_default is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_active_projects(project_repository: ProjectRepository):
|
|
"""Test getting all active projects."""
|
|
# Create active and inactive projects
|
|
active_project_data = {
|
|
"name": "Active Project",
|
|
"description": "An active project",
|
|
"path": "/active/project/path",
|
|
"is_active": True,
|
|
}
|
|
inactive_project_data = {
|
|
"name": "Inactive Project",
|
|
"description": "An inactive project",
|
|
"path": "/inactive/project/path",
|
|
"is_active": False,
|
|
}
|
|
|
|
await project_repository.create(active_project_data)
|
|
await project_repository.create(inactive_project_data)
|
|
|
|
# Get active projects
|
|
active_projects = await project_repository.get_active_projects()
|
|
assert len(active_projects) >= 1 # Could be more from other tests
|
|
|
|
# Verify that all returned projects are active
|
|
for project in active_projects:
|
|
assert project.is_active is True
|
|
|
|
# Verify active project is included
|
|
active_names = [p.name for p in active_projects]
|
|
assert "Active Project" in active_names
|
|
|
|
# Verify inactive project is not included
|
|
assert "Inactive Project" not in active_names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_as_default(project_repository: ProjectRepository, test_project: Project):
|
|
"""Test setting a project as default."""
|
|
# The test_project fixture is already the default
|
|
# Create a non-default project
|
|
project2_data = {
|
|
"name": "Project 2",
|
|
"description": "Project 2",
|
|
"path": "/project2/path",
|
|
"is_active": True,
|
|
"is_default": None, # Not default
|
|
}
|
|
|
|
# Get the existing default project
|
|
project1 = test_project
|
|
project2 = await project_repository.create(project2_data)
|
|
|
|
# Verify initial state
|
|
assert project1.is_default is True
|
|
assert project2.is_default is None
|
|
|
|
# Set project2 as default
|
|
updated_project2 = await project_repository.set_as_default(project2.id)
|
|
assert updated_project2 is not None
|
|
assert updated_project2.is_default is True
|
|
|
|
# Verify project1 is no longer default
|
|
project1_updated = await project_repository.find_by_id(project1.id)
|
|
assert project1_updated is not None
|
|
assert project1_updated.is_default is None
|
|
|
|
# Verify project2 is now default
|
|
project2_updated = await project_repository.find_by_id(project2.id)
|
|
assert project2_updated is not None
|
|
assert project2_updated.is_default is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_project(project_repository: ProjectRepository, sample_project: Project):
|
|
"""Test updating a project."""
|
|
# Update project
|
|
updated_data = {
|
|
"name": "Updated Project Name",
|
|
"description": "Updated description",
|
|
"path": "/updated/path",
|
|
}
|
|
updated_project = await project_repository.update(sample_project.id, updated_data)
|
|
|
|
# Verify returned object
|
|
assert updated_project is not None
|
|
assert updated_project.id == sample_project.id
|
|
assert updated_project.name == "Updated Project Name"
|
|
assert updated_project.description == "Updated description"
|
|
assert updated_project.path == "/updated/path"
|
|
|
|
# Verify permalink was updated based on new name
|
|
assert updated_project.permalink == "updated-project-name"
|
|
|
|
# Verify in database
|
|
found = await project_repository.find_by_id(sample_project.id)
|
|
assert found is not None
|
|
assert found.name == "Updated Project Name"
|
|
assert found.description == "Updated description"
|
|
assert found.path == "/updated/path"
|
|
assert found.permalink == "updated-project-name"
|
|
|
|
# Verify we can find by the new permalink
|
|
found_by_permalink = await project_repository.get_by_permalink("updated-project-name")
|
|
assert found_by_permalink is not None
|
|
assert found_by_permalink.id == sample_project.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_project(project_repository: ProjectRepository, sample_project: Project):
|
|
"""Test deleting a project."""
|
|
# Delete project
|
|
result = await project_repository.delete(sample_project.id)
|
|
assert result is True
|
|
|
|
# Verify deletion
|
|
deleted = await project_repository.find_by_id(sample_project.id)
|
|
assert deleted is None
|
|
|
|
# Verify with direct database query
|
|
async with db.scoped_session(project_repository.session_maker) as session:
|
|
query = select(Project).filter(Project.id == sample_project.id)
|
|
result = await session.execute(query)
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_nonexistent_project(project_repository: ProjectRepository):
|
|
"""Test deleting a project that doesn't exist."""
|
|
result = await project_repository.delete(999) # Non-existent ID
|
|
assert result is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_path(project_repository: ProjectRepository, sample_project: Project):
|
|
"""Test updating a project's path."""
|
|
new_path = "/new/project/path"
|
|
|
|
# Update the project path
|
|
updated_project = await project_repository.update_path(sample_project.id, new_path)
|
|
|
|
# Verify returned object
|
|
assert updated_project is not None
|
|
assert updated_project.id == sample_project.id
|
|
assert updated_project.path == new_path
|
|
assert updated_project.name == sample_project.name # Other fields unchanged
|
|
|
|
# Verify in database
|
|
found = await project_repository.find_by_id(sample_project.id)
|
|
assert found is not None
|
|
assert found.path == new_path
|
|
assert found.name == sample_project.name
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_path_nonexistent_project(project_repository: ProjectRepository):
|
|
"""Test updating path for a project that doesn't exist."""
|
|
result = await project_repository.update_path(999, "/some/path") # Non-existent ID
|
|
assert result is None
|