mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Paul Hernandez
634486107b
Signed-off-by: phernandez <paul@basicmachines.co> Co-authored-by: Claude <noreply@anthropic.com>
280 lines
9.2 KiB
Python
280 lines
9.2 KiB
Python
"""Tests for the directory router API endpoints."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from basic_memory.schemas.directory import DirectoryNode
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_directory_tree_endpoint(test_graph, client, project_url):
|
|
"""Test the get_directory_tree endpoint returns correctly structured data."""
|
|
# Call the endpoint
|
|
response = await client.get(f"{project_url}/directory/tree")
|
|
|
|
# Verify response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Check that the response is a valid directory tree
|
|
assert "name" in data
|
|
assert "directory_path" in data
|
|
assert "children" in data
|
|
assert "type" in data
|
|
|
|
# The root node should have children
|
|
assert isinstance(data["children"], list)
|
|
|
|
# Root name should be the project name or similar
|
|
assert data["name"]
|
|
|
|
# Root directory_path should be a string
|
|
assert isinstance(data["directory_path"], str)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_directory_tree_structure(test_graph, client, project_url):
|
|
"""Test the structure of the directory tree returned by the endpoint."""
|
|
# Call the endpoint
|
|
response = await client.get(f"{project_url}/directory/tree")
|
|
|
|
# Verify response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Function to recursively check each node in the tree
|
|
def check_node_structure(node):
|
|
assert "name" in node
|
|
assert "directory_path" in node
|
|
assert "children" in node
|
|
assert "type" in node
|
|
assert isinstance(node["children"], list)
|
|
|
|
# Check each child recursively
|
|
for child in node["children"]:
|
|
check_node_structure(child)
|
|
|
|
# Check the entire tree structure
|
|
check_node_structure(data)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_directory_tree_mocked(client, project_url):
|
|
"""Test the get_directory_tree endpoint with a mocked service."""
|
|
# Create a mock directory tree
|
|
mock_tree = DirectoryNode(
|
|
name="root",
|
|
directory_path="/test",
|
|
type="directory",
|
|
children=[
|
|
DirectoryNode(
|
|
name="folder1",
|
|
directory_path="/test/folder1",
|
|
type="directory",
|
|
children=[
|
|
DirectoryNode(
|
|
name="subfolder",
|
|
directory_path="/test/folder1/subfolder",
|
|
type="directory",
|
|
children=[],
|
|
)
|
|
],
|
|
),
|
|
DirectoryNode(
|
|
name="folder2", directory_path="/test/folder2", type="directory", children=[]
|
|
),
|
|
],
|
|
)
|
|
|
|
# Patch the directory service
|
|
with patch(
|
|
"basic_memory.services.directory_service.DirectoryService.get_directory_tree",
|
|
return_value=mock_tree,
|
|
):
|
|
# Call the endpoint
|
|
response = await client.get(f"{project_url}/directory/tree")
|
|
|
|
# Verify response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Check structure matches our mock
|
|
assert data["name"] == "root"
|
|
assert data["directory_path"] == "/test"
|
|
assert data["type"] == "directory"
|
|
assert len(data["children"]) == 2
|
|
|
|
# Check first child
|
|
folder1 = data["children"][0]
|
|
assert folder1["name"] == "folder1"
|
|
assert folder1["directory_path"] == "/test/folder1"
|
|
assert folder1["type"] == "directory"
|
|
assert len(folder1["children"]) == 1
|
|
|
|
# Check subfolder
|
|
subfolder = folder1["children"][0]
|
|
assert subfolder["name"] == "subfolder"
|
|
assert subfolder["directory_path"] == "/test/folder1/subfolder"
|
|
assert subfolder["type"] == "directory"
|
|
assert subfolder["children"] == []
|
|
|
|
# Check second child
|
|
folder2 = data["children"][1]
|
|
assert folder2["name"] == "folder2"
|
|
assert folder2["directory_path"] == "/test/folder2"
|
|
assert folder2["type"] == "directory"
|
|
assert folder2["children"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_directory_endpoint_default(test_graph, client, project_url):
|
|
"""Test the list_directory endpoint with default parameters."""
|
|
# Call the endpoint with default parameters
|
|
response = await client.get(f"{project_url}/directory/list")
|
|
|
|
# Verify response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Should return a list
|
|
assert isinstance(data, list)
|
|
|
|
# With test_graph, should return the "test" directory
|
|
assert len(data) == 1
|
|
assert data[0]["name"] == "test"
|
|
assert data[0]["type"] == "directory"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_directory_endpoint_specific_path(test_graph, client, project_url):
|
|
"""Test the list_directory endpoint with specific directory path."""
|
|
# Call the endpoint with /test directory
|
|
response = await client.get(f"{project_url}/directory/list?dir_name=/test")
|
|
|
|
# Verify response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Should return list of files in test directory
|
|
assert isinstance(data, list)
|
|
assert len(data) == 5
|
|
|
|
# All should be files (no subdirectories in test_graph)
|
|
for item in data:
|
|
assert item["type"] == "file"
|
|
assert item["name"].endswith(".md")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_directory_endpoint_with_glob(test_graph, client, project_url):
|
|
"""Test the list_directory endpoint with glob filtering."""
|
|
# Call the endpoint with glob filter
|
|
response = await client.get(
|
|
f"{project_url}/directory/list?dir_name=/test&file_name_glob=*Connected*"
|
|
)
|
|
|
|
# Verify response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Should return only Connected Entity files
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
|
|
file_names = {item["name"] for item in data}
|
|
assert file_names == {"Connected Entity 1.md", "Connected Entity 2.md"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_directory_endpoint_with_depth(test_graph, client, project_url):
|
|
"""Test the list_directory endpoint with depth control."""
|
|
# Test depth=1 (default)
|
|
response_depth_1 = await client.get(f"{project_url}/directory/list?dir_name=/&depth=1")
|
|
assert response_depth_1.status_code == 200
|
|
data_depth_1 = response_depth_1.json()
|
|
assert len(data_depth_1) == 1 # Just the test directory
|
|
|
|
# Test depth=2 (should include files in test directory)
|
|
response_depth_2 = await client.get(f"{project_url}/directory/list?dir_name=/&depth=2")
|
|
assert response_depth_2.status_code == 200
|
|
data_depth_2 = response_depth_2.json()
|
|
assert len(data_depth_2) == 6 # test directory + 5 files
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_directory_endpoint_nonexistent_path(test_graph, client, project_url):
|
|
"""Test the list_directory endpoint with nonexistent directory."""
|
|
# Call the endpoint with nonexistent directory
|
|
response = await client.get(f"{project_url}/directory/list?dir_name=/nonexistent")
|
|
|
|
# Verify response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Should return empty list
|
|
assert isinstance(data, list)
|
|
assert len(data) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_directory_endpoint_validation_errors(client, project_url):
|
|
"""Test the list_directory endpoint with invalid parameters."""
|
|
# Test depth too low
|
|
response = await client.get(f"{project_url}/directory/list?depth=0")
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
# Test depth too high
|
|
response = await client.get(f"{project_url}/directory/list?depth=11")
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_directory_endpoint_mocked(client, project_url):
|
|
"""Test the list_directory endpoint with mocked service."""
|
|
# Create mock directory nodes
|
|
mock_nodes = [
|
|
DirectoryNode(
|
|
name="folder1",
|
|
directory_path="/folder1",
|
|
type="directory",
|
|
),
|
|
DirectoryNode(
|
|
name="file1.md",
|
|
directory_path="/file1.md",
|
|
file_path="file1.md",
|
|
type="file",
|
|
title="File 1",
|
|
permalink="file-1",
|
|
),
|
|
]
|
|
|
|
# Patch the directory service
|
|
with patch(
|
|
"basic_memory.services.directory_service.DirectoryService.list_directory",
|
|
return_value=mock_nodes,
|
|
):
|
|
# Call the endpoint
|
|
response = await client.get(f"{project_url}/directory/list?dir_name=/test")
|
|
|
|
# Verify response
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Check structure matches our mock
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
|
|
# Check directory
|
|
folder = next(item for item in data if item["type"] == "directory")
|
|
assert folder["name"] == "folder1"
|
|
assert folder["directory_path"] == "/folder1"
|
|
|
|
# Check file
|
|
file_item = next(item for item in data if item["type"] == "file")
|
|
assert file_item["name"] == "file1.md"
|
|
assert file_item["directory_path"] == "/file1.md"
|
|
assert file_item["file_path"] == "file1.md"
|
|
assert file_item["title"] == "File 1"
|
|
assert file_item["permalink"] == "file-1"
|