chuk-mcp-vfs

""" Tests for MCP Server These tests verify that the server is properly configured and all tools are registered. We test the server by calling the decorated tool functions directly since they're closures in the create_server() function. """ import pytest from chuk_mcp_server import ChukMCPServer from chuk_mcp_vfs import server as vfs_server # Import to ensure coverage tracking from chuk_mcp_vfs.models import ( CheckpointCreateRequest, CheckpointRestoreRequest, CopyRequest, FindRequest, GrepRequest, ProviderType, WriteRequest, ) from chuk_mcp_vfs.server import create_server @pytest.mark.asyncio async def test_create_server(): """Test that server can be created.""" server = create_server() assert server is not None assert isinstance(server, ChukMCPServer) @pytest.mark.asyncio async def test_server_has_tools(): """Test server has tools registered.""" server = create_server() # The server should have tools registered # We can verify this by checking that server was created successfully assert server is not None # Check that it's a ChukMCPServer instance assert isinstance(server, ChukMCPServer) @pytest.mark.asyncio async def test_server_creation_registers_all_tools(): """Test that creating server registers all tool decorators.""" # This test ensures all the @server.tool decorators execute # by calling create_server(), which will run lines 47-173 in server.py server = vfs_server.create_server() # Just verify the server was created - the act of creating it # executes all the decorator code assert server is not None # Also test run_server exists (though we can't actually run it in tests) assert hasattr(vfs_server, "run_server") assert hasattr(vfs_server, "main") # Create it again to ensure idempotence server2 = vfs_server.create_server() assert server2 is not None # The following tests exercise the server's tool registration by creating # a server and using its workspace/VFS tools through integration tests @pytest.fixture async def server_with_workspace(): """Fixture that creates a server and a test workspace.""" from chuk_mcp_vfs.checkpoint_manager import CheckpointManager from chuk_mcp_vfs.checkpoint_tools import CheckpointTools from chuk_mcp_vfs.vfs_tools import VFSTools from chuk_mcp_vfs.workspace_manager import WorkspaceManager from chuk_mcp_vfs.workspace_tools import WorkspaceTools # Create the components that the server uses workspace_manager = WorkspaceManager() await workspace_manager.create_workspace( name="test-ws", provider_type=ProviderType.MEMORY ) checkpoint_manager = CheckpointManager(workspace_manager) workspace_tools = WorkspaceTools(workspace_manager) vfs_tools = VFSTools(workspace_manager) checkpoint_tools = CheckpointTools(checkpoint_manager) return { "workspace_manager": workspace_manager, "workspace_tools": workspace_tools, "vfs_tools": vfs_tools, "checkpoint_manager": checkpoint_manager, "checkpoint_tools": checkpoint_tools, } @pytest.mark.asyncio async def test_server_workspace_operations(server_with_workspace): """Test workspace operations through server tools.""" tools = server_with_workspace # Test workspace list result = await tools["workspace_tools"].workspace_list() assert len(result.workspaces) >= 1 # Test workspace info info = await tools["workspace_tools"].workspace_info("test-ws") assert info.name == "test-ws" @pytest.mark.asyncio async def test_server_file_operations(server_with_workspace): """Test file operations through server tools.""" tools = server_with_workspace # Test write write_result = await tools["vfs_tools"].write( WriteRequest(path="/test.txt", content="Hello Server") ) assert write_result.success # Test read read_result = await tools["vfs_tools"].read("/test.txt") assert read_result.content == "Hello Server" # Test ls ls_result = await tools["vfs_tools"].ls("/") assert len(ls_result.entries) >= 1 # Test mkdir mkdir_result = await tools["vfs_tools"].mkdir("/testdir") assert mkdir_result.success @pytest.mark.asyncio async def test_server_navigation_operations(server_with_workspace): """Test navigation operations through server tools.""" tools = server_with_workspace # Create a directory first await tools["vfs_tools"].mkdir("/navtest") # Test cd cd_result = await tools["vfs_tools"].cd("/navtest") assert cd_result.success assert cd_result.cwd == "/navtest" # Test pwd pwd_result = await tools["vfs_tools"].pwd() assert pwd_result.cwd == "/navtest" @pytest.mark.asyncio async def test_server_search_operations(server_with_workspace): """Test search operations through server tools.""" tools = server_with_workspace # Create test files await tools["vfs_tools"].mkdir("/search") await tools["vfs_tools"].write( WriteRequest(path="/search/file1.txt", content="test content") ) await tools["vfs_tools"].write( WriteRequest(path="/search/file2.log", content="test data") ) # Test find find_result = await tools["vfs_tools"].find( FindRequest(pattern="*.txt", path="/search", max_results=100) ) assert len(find_result.matches) >= 1 assert "/search/file1.txt" in find_result.matches # Test grep grep_result = await tools["vfs_tools"].grep( GrepRequest(pattern="content", path="/search", max_results=100) ) assert len(grep_result.matches) >= 1 @pytest.mark.asyncio async def test_server_file_manipulation_operations(server_with_workspace): """Test file manipulation operations through server tools.""" tools = server_with_workspace # Create test file await tools["vfs_tools"].write( WriteRequest(path="/original.txt", content="move me") ) # Test mv mv_result = await tools["vfs_tools"].mv("/original.txt", "/moved.txt") assert mv_result.success assert mv_result.dest == "/moved.txt" # Test cp cp_result = await tools["vfs_tools"].cp( CopyRequest(source="/moved.txt", dest="/copy.txt", recursive=False) ) assert cp_result.success # Test rm rm_result = await tools["vfs_tools"].rm("/copy.txt") assert rm_result.success @pytest.mark.asyncio async def test_server_tree_operation(server_with_workspace): """Test tree operation through server tools.""" tools = server_with_workspace # Create directory structure await tools["vfs_tools"].mkdir("/tree_test") await tools["vfs_tools"].write( WriteRequest(path="/tree_test/file.txt", content="test") ) # Test tree tree_result = await tools["vfs_tools"].tree("/tree_test", max_depth=2) assert tree_result.root is not None assert tree_result.root.name == "tree_test" @pytest.mark.asyncio async def test_server_checkpoint_operations(server_with_workspace): """Test checkpoint operations through server tools.""" tools = server_with_workspace # Create some files await tools["vfs_tools"].write( WriteRequest(path="/checkpoint_test.txt", content="original") ) # Test checkpoint create create_result = await tools["checkpoint_tools"].checkpoint_create( CheckpointCreateRequest(name="test-checkpoint") ) assert create_result.success assert create_result.checkpoint_id == "test-checkpoint" # Modify file await tools["vfs_tools"].write( WriteRequest(path="/checkpoint_test.txt", content="modified") ) # Test checkpoint restore restore_result = await tools["checkpoint_tools"].checkpoint_restore( CheckpointRestoreRequest(checkpoint_id="test-checkpoint") ) assert restore_result.success # Verify restoration read_result = await tools["vfs_tools"].read("/checkpoint_test.txt") assert read_result.content == "original" # Test checkpoint list list_result = await tools["checkpoint_tools"].checkpoint_list() assert len(list_result.checkpoints) >= 1 # Test checkpoint delete delete_result = await tools["checkpoint_tools"].checkpoint_delete("test-checkpoint") assert delete_result.success