Skip to main content
Glama
test_batch_operations.py24.2 kB
""" Tests for batch operations tools. """ from unittest.mock import AsyncMock, patch import pytest from mcp_outline.features.documents.common import OutlineClientError # Mock FastMCP for registering tools class MockMCP: def __init__(self): self.tools = {} def tool(self, **kwargs): def decorator(func): self.tools[func.__name__] = func return func return decorator # Sample document data SAMPLE_DOCUMENT = { "id": "doc123", "title": "Test Document", "text": "Sample content", } SAMPLE_DOCUMENTS = [ {"id": "doc1", "title": "Document 1"}, {"id": "doc2", "title": "Document 2"}, {"id": "doc3", "title": "Document 3"}, ] @pytest.fixture def mcp(): """Fixture to provide mock MCP instance.""" return MockMCP() @pytest.fixture def register_batch_tools(mcp): """Fixture to register batch operation tools.""" from mcp_outline.features.documents.batch_operations import ( register_tools, ) register_tools(mcp) return mcp class TestBatchArchiveDocuments: """Tests for batch_archive_documents tool.""" @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_archive_all_success( self, mock_get_client, register_batch_tools ): """Test batch_archive_documents with all successes.""" mock_client = AsyncMock() mock_client.archive_document.side_effect = [ {"id": "doc1", "title": "Document 1"}, {"id": "doc2", "title": "Document 2"}, {"id": "doc3", "title": "Document 3"}, ] mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_archive_documents"]( ["doc1", "doc2", "doc3"] ) assert "Total: 3" in result assert "Succeeded: 3" in result assert "Failed: 0" in result assert "✓" in result assert mock_client.archive_document.call_count == 3 @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_archive_all_failures( self, mock_get_client, register_batch_tools ): """Test batch_archive_documents with all failures.""" mock_client = AsyncMock() mock_client.archive_document.side_effect = OutlineClientError( "API error" ) mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_archive_documents"]( ["doc1", "doc2", "doc3"] ) assert "Total: 3" in result assert "Succeeded: 0" in result assert "Failed: 3" in result assert "✗" in result assert "API error" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_archive_partial_success( self, mock_get_client, register_batch_tools ): """Test batch_archive_documents with mixed results.""" mock_client = AsyncMock() mock_client.archive_document.side_effect = [ {"id": "doc1", "title": "Document 1"}, OutlineClientError("Not found"), {"id": "doc3", "title": "Document 3"}, ] mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_archive_documents"]( ["doc1", "doc2", "doc3"] ) assert "Total: 3" in result assert "Succeeded: 2" in result assert "Failed: 1" in result assert "Document 1" in result assert "Document 3" in result assert "Not found" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_archive_empty_list( self, mock_get_client, register_batch_tools ): """Test batch_archive_documents with empty list.""" mock_client = AsyncMock() mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_archive_documents"]( [] ) assert "Error: No document IDs provided" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_archive_single_document( self, mock_get_client, register_batch_tools ): """Test batch_archive_documents with single document.""" mock_client = AsyncMock() mock_client.archive_document.return_value = { "id": "doc1", "title": "Single Document", } mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_archive_documents"]( ["doc1"] ) assert "Total: 1" in result assert "Succeeded: 1" in result assert "Failed: 0" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_archive_no_document_returned( self, mock_get_client, register_batch_tools ): """Test batch_archive_documents when API returns None.""" mock_client = AsyncMock() mock_client.archive_document.return_value = None mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_archive_documents"]( ["doc1"] ) assert "Failed: 1" in result assert "No document returned from API" in result class TestBatchMoveDocuments: """Tests for batch_move_documents tool.""" @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_move_all_success( self, mock_get_client, register_batch_tools ): """Test batch_move_documents with all successes.""" mock_client = AsyncMock() mock_client.post.return_value = { "data": {"id": "doc1", "title": "Moved Doc"} } mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_move_documents"]( ["doc1", "doc2"], collection_id="col123" ) assert "Total: 2" in result assert "Succeeded: 2" in result assert "Failed: 0" in result assert mock_client.post.call_count == 2 @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_move_no_destination( self, mock_get_client, register_batch_tools ): """Test batch_move_documents without collection or parent.""" result = await register_batch_tools.tools["batch_move_documents"]( ["doc1", "doc2"] ) assert "Error" in result assert "collection_id or parent_document_id" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_move_with_parent( self, mock_get_client, register_batch_tools ): """Test batch_move_documents with parent document.""" mock_client = AsyncMock() mock_client.post.return_value = { "data": {"id": "doc1", "title": "Moved Doc"} } mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_move_documents"]( ["doc1"], parent_document_id="parent123" ) assert "Succeeded: 1" in result mock_client.post.assert_called_once() call_args = mock_client.post.call_args[0] assert call_args[0] == "documents.move" assert call_args[1]["parentDocumentId"] == "parent123" @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_move_partial_success( self, mock_get_client, register_batch_tools ): """Test batch_move_documents with mixed results.""" mock_client = AsyncMock() mock_client.post.side_effect = [ {"data": {"id": "doc1", "title": "Doc 1"}}, OutlineClientError("Permission denied"), {"data": {"id": "doc3", "title": "Doc 3"}}, ] mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_move_documents"]( ["doc1", "doc2", "doc3"], collection_id="col123" ) assert "Total: 3" in result assert "Succeeded: 2" in result assert "Failed: 1" in result assert "Permission denied" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_move_empty_list( self, mock_get_client, register_batch_tools ): """Test batch_move_documents with empty list.""" result = await register_batch_tools.tools["batch_move_documents"]( [], collection_id="col123" ) assert "Error: No document IDs provided" in result class TestBatchDeleteDocuments: """Tests for batch_delete_documents tool.""" @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_delete_to_trash_success( self, mock_get_client, register_batch_tools ): """Test batch_delete_documents moving to trash.""" mock_client = AsyncMock() mock_client.get_document.side_effect = [ {"id": "doc1", "title": "Document 1"}, {"id": "doc2", "title": "Document 2"}, ] mock_client.post.return_value = {"success": True} mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_delete_documents"]( ["doc1", "doc2"], permanent=False ) assert "Total: 2" in result assert "Succeeded: 2" in result assert "Failed: 0" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_delete_permanent_success( self, mock_get_client, register_batch_tools ): """Test batch_delete_documents with permanent deletion.""" mock_client = AsyncMock() mock_client.permanently_delete_document.return_value = True mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_delete_documents"]( ["doc1", "doc2"], permanent=True ) assert "Total: 2" in result assert "Succeeded: 2" in result assert mock_client.permanently_delete_document.call_count == 2 @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_delete_partial_success( self, mock_get_client, register_batch_tools ): """Test batch_delete_documents with mixed results.""" mock_client = AsyncMock() mock_client.get_document.side_effect = [ {"id": "doc1", "title": "Document 1"}, OutlineClientError("Not found"), ] mock_client.post.return_value = {"success": True} mock_get_client.return_value = mock_client result = await register_batch_tools.tools["batch_delete_documents"]( ["doc1", "doc2"], permanent=False ) assert "Total: 2" in result assert "Succeeded: 1" in result assert "Failed: 1" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_delete_empty_list( self, mock_get_client, register_batch_tools ): """Test batch_delete_documents with empty list.""" result = await register_batch_tools.tools["batch_delete_documents"]([]) assert "Error: No document IDs provided" in result class TestBatchUpdateDocuments: """Tests for batch_update_documents tool.""" @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_update_all_success( self, mock_get_client, register_batch_tools ): """Test batch_update_documents with all successes.""" mock_client = AsyncMock() mock_client.post.return_value = { "data": {"id": "doc1", "title": "Updated Title"} } mock_get_client.return_value = mock_client updates = [ {"id": "doc1", "title": "New Title 1"}, {"id": "doc2", "text": "New content"}, {"id": "doc3", "title": "Title 3", "text": "Content 3"}, ] result = await register_batch_tools.tools["batch_update_documents"]( updates ) assert "Total: 3" in result assert "Succeeded: 3" in result assert "Failed: 0" in result assert mock_client.post.call_count == 3 @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_update_with_append( self, mock_get_client, register_batch_tools ): """Test batch_update_documents with append flag.""" mock_client = AsyncMock() mock_client.post.return_value = { "data": {"id": "doc1", "title": "Document"} } mock_get_client.return_value = mock_client updates = [{"id": "doc1", "text": "Appended content", "append": True}] result = await register_batch_tools.tools["batch_update_documents"]( updates ) assert "Succeeded: 1" in result call_args = mock_client.post.call_args[0] assert call_args[1]["append"] is True @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_update_missing_id( self, mock_get_client, register_batch_tools ): """Test batch_update_documents with missing document ID.""" mock_client = AsyncMock() mock_get_client.return_value = mock_client updates = [{"title": "No ID"}] result = await register_batch_tools.tools["batch_update_documents"]( updates ) assert "Failed: 1" in result assert "Missing document ID" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_update_partial_success( self, mock_get_client, register_batch_tools ): """Test batch_update_documents with mixed results.""" mock_client = AsyncMock() mock_client.post.side_effect = [ {"data": {"id": "doc1", "title": "Updated"}}, OutlineClientError("Permission denied"), ] mock_get_client.return_value = mock_client updates = [ {"id": "doc1", "title": "Title 1"}, {"id": "doc2", "title": "Title 2"}, ] result = await register_batch_tools.tools["batch_update_documents"]( updates ) assert "Total: 2" in result assert "Succeeded: 1" in result assert "Failed: 1" in result assert "Permission denied" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_update_empty_list( self, mock_get_client, register_batch_tools ): """Test batch_update_documents with empty list.""" result = await register_batch_tools.tools["batch_update_documents"]([]) assert "Error: No updates provided" in result class TestBatchCreateDocuments: """Tests for batch_create_documents tool.""" @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_create_all_success( self, mock_get_client, register_batch_tools ): """Test batch_create_documents with all successes.""" mock_client = AsyncMock() mock_client.post.side_effect = [ {"data": {"id": "new1", "title": "Document 1"}}, {"data": {"id": "new2", "title": "Document 2"}}, {"data": {"id": "new3", "title": "Document 3"}}, ] mock_get_client.return_value = mock_client documents = [ {"title": "Doc 1", "collection_id": "col1"}, {"title": "Doc 2", "collection_id": "col1", "text": "Content"}, { "title": "Doc 3", "collection_id": "col1", "publish": False, }, ] result = await register_batch_tools.tools["batch_create_documents"]( documents ) assert "Total: 3" in result assert "Succeeded: 3" in result assert "Failed: 0" in result assert "new1" in result assert "new2" in result assert "new3" in result assert mock_client.post.call_count == 3 @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_create_missing_title( self, mock_get_client, register_batch_tools ): """Test batch_create_documents with missing title.""" mock_client = AsyncMock() mock_get_client.return_value = mock_client documents = [{"collection_id": "col1"}] result = await register_batch_tools.tools["batch_create_documents"]( documents ) assert "Failed: 1" in result assert "Missing required field: title" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_create_missing_collection_id( self, mock_get_client, register_batch_tools ): """Test batch_create_documents with missing collection_id.""" mock_client = AsyncMock() mock_get_client.return_value = mock_client documents = [{"title": "Document"}] result = await register_batch_tools.tools["batch_create_documents"]( documents ) assert "Failed: 1" in result assert "Missing required field: collection_id" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_create_with_parent( self, mock_get_client, register_batch_tools ): """Test batch_create_documents with parent document.""" mock_client = AsyncMock() mock_client.post.return_value = { "data": {"id": "new1", "title": "Child Doc"} } mock_get_client.return_value = mock_client documents = [ { "title": "Child", "collection_id": "col1", "parent_document_id": "parent123", } ] result = await register_batch_tools.tools["batch_create_documents"]( documents ) assert "Succeeded: 1" in result call_args = mock_client.post.call_args[0] assert call_args[1]["parentDocumentId"] == "parent123" @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_create_partial_success( self, mock_get_client, register_batch_tools ): """Test batch_create_documents with mixed results.""" mock_client = AsyncMock() mock_client.post.side_effect = [ {"data": {"id": "new1", "title": "Doc 1"}}, OutlineClientError("Quota exceeded"), {"data": {"id": "new3", "title": "Doc 3"}}, ] mock_get_client.return_value = mock_client documents = [ {"title": "Doc 1", "collection_id": "col1"}, {"title": "Doc 2", "collection_id": "col1"}, {"title": "Doc 3", "collection_id": "col1"}, ] result = await register_batch_tools.tools["batch_create_documents"]( documents ) assert "Total: 3" in result assert "Succeeded: 2" in result assert "Failed: 1" in result assert "Quota exceeded" in result assert "new1" in result assert "new3" in result @pytest.mark.asyncio @patch( "mcp_outline.features.documents.batch_operations.get_outline_client" ) async def test_batch_create_empty_list( self, mock_get_client, register_batch_tools ): """Test batch_create_documents with empty list.""" result = await register_batch_tools.tools["batch_create_documents"]([]) assert "Error: No documents provided" in result class TestHelperFunctions: """Tests for helper functions.""" def test_create_result_entry_success(self): """Test _create_result_entry for successful operation.""" from mcp_outline.features.documents.batch_operations import ( _create_result_entry, ) result = _create_result_entry( "doc123", "success", title="Test Document" ) assert result["id"] == "doc123" assert result["status"] == "success" assert result["title"] == "Test Document" assert "error" not in result def test_create_result_entry_failure(self): """Test _create_result_entry for failed operation.""" from mcp_outline.features.documents.batch_operations import ( _create_result_entry, ) result = _create_result_entry("doc123", "failed", error="Not found") assert result["id"] == "doc123" assert result["status"] == "failed" assert result["error"] == "Not found" assert "title" not in result def test_format_batch_results_all_success(self): """Test _format_batch_results with all successes.""" from mcp_outline.features.documents.batch_operations import ( _format_batch_results, ) results = [ {"id": "doc1", "status": "success", "title": "Doc 1"}, {"id": "doc2", "status": "success", "title": "Doc 2"}, ] output = _format_batch_results("archive", 2, 2, 0, results) assert "Total: 2" in output assert "Succeeded: 2" in output assert "Failed: 0" in output assert "✓ All 2 documents archived successfully" in output def test_format_batch_results_all_failure(self): """Test _format_batch_results with all failures.""" from mcp_outline.features.documents.batch_operations import ( _format_batch_results, ) results = [ {"id": "doc1", "status": "failed", "error": "Error 1"}, {"id": "doc2", "status": "failed", "error": "Error 2"}, ] output = _format_batch_results("delete", 2, 0, 2, results) assert "Total: 2" in output assert "Succeeded: 0" in output assert "Failed: 2" in output assert "✗ All 2 operations failed" in output def test_format_batch_results_mixed(self): """Test _format_batch_results with mixed results.""" from mcp_outline.features.documents.batch_operations import ( _format_batch_results, ) results = [ {"id": "doc1", "status": "success", "title": "Doc 1"}, {"id": "doc2", "status": "failed", "error": "Not found"}, {"id": "doc3", "status": "success", "title": "Doc 3"}, ] output = _format_batch_results("move", 3, 2, 1, results) assert "Total: 3" in output assert "Succeeded: 2" in output assert "Failed: 1" in output assert "✓ doc1 - Doc 1" in output assert "✓ doc3 - Doc 3" in output assert "✗ doc2 - Error: Not found" in output

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/Vortiago/mcp-outline'

If you have feedback or need assistance with the MCP directory API, please join our Discord server