From 4b2c30bbe07a6679f29b063fedd049e41f80ac68 Mon Sep 17 00:00:00 2001 From: LUIS NOVO Date: Mon, 15 Dec 2025 21:36:46 -0300 Subject: [PATCH] feat: add rename_page and get_page_backlinks tools Add two new MCP tools for page management: - rename_page: Rename a page and update all references throughout the graph - Pre-validates source exists and target doesn't exist - Uses logseq.Editor.renamePage API - get_page_backlinks: Get all pages/blocks that reference a page - Groups results by referencing page - Optional include_content parameter for block details - Uses logseq.Editor.getPageLinkedReferences API Includes comprehensive unit tests for both tools (15 new tests). Total tools now: 10 --- pyproject.toml | 2 +- src/mcp_logseq/logseq.py | 107 +++++++++ src/mcp_logseq/server.py | 4 + src/mcp_logseq/tools.py | 267 ++++++++++++++++++++++ tests/conftest.py | 81 ++++++- tests/integration/test_mcp_server.py | 10 +- tests/unit/test_logseq_api.py | 159 ++++++++++++- tests/unit/test_tool_handlers.py | 325 ++++++++++++++++++++++++++- 8 files changed, 942 insertions(+), 13 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d454330..b572036 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mcp-logseq" -version = "1.0.1" +version = "1.1.0" description = "MCP server to work with LogSeq via the local HTTP server" readme = "README.md" requires-python = ">=3.11" diff --git a/src/mcp_logseq/logseq.py b/src/mcp_logseq/logseq.py index 014d184..a6992eb 100644 --- a/src/mcp_logseq/logseq.py +++ b/src/mcp_logseq/logseq.py @@ -295,3 +295,110 @@ def update_page(self, page_name: str, content: str = None, properties: dict = No except Exception as e: logger.error(f"Error updating page '{page_name}': {str(e)}") raise + + def get_pages_from_namespace(self, namespace: str) -> Any: + """Get all pages within a namespace (flat list).""" + url = self.get_base_url() + logger.info(f"Getting pages from namespace '{namespace}'") + + try: + response = requests.post( + url, + headers=self._get_headers(), + json={ + "method": "logseq.Editor.getPagesFromNamespace", + "args": [namespace] + }, + verify=self.verify_ssl, + timeout=self.timeout + ) + response.raise_for_status() + return response.json() + + except Exception as e: + logger.error(f"Error getting pages from namespace: {str(e)}") + raise + + def get_pages_tree_from_namespace(self, namespace: str) -> Any: + """Get pages within a namespace as a tree structure.""" + url = self.get_base_url() + logger.info(f"Getting pages tree from namespace '{namespace}'") + + try: + response = requests.post( + url, + headers=self._get_headers(), + json={ + "method": "logseq.Editor.getPagesTreeFromNamespace", + "args": [namespace] + }, + verify=self.verify_ssl, + timeout=self.timeout + ) + response.raise_for_status() + return response.json() + + except Exception as e: + logger.error(f"Error getting pages tree from namespace: {str(e)}") + raise + + def rename_page(self, old_name: str, new_name: str) -> Any: + """Rename a page and update all references.""" + url = self.get_base_url() + logger.info(f"Renaming page '{old_name}' to '{new_name}'") + + try: + # Validate old page exists + existing_pages = self.list_pages() + page_names = [p.get("originalName") or p.get("name") for p in existing_pages] + + if old_name not in page_names: + raise ValueError(f"Page '{old_name}' does not exist") + + if new_name in page_names: + raise ValueError(f"Page '{new_name}' already exists") + + response = requests.post( + url, + headers=self._get_headers(), + json={ + "method": "logseq.Editor.renamePage", + "args": [old_name, new_name] + }, + verify=self.verify_ssl, + timeout=self.timeout + ) + response.raise_for_status() + # renamePage returns null on success + if response.text and response.text.strip() and response.text.strip() != 'null': + return response.json() + return None + + except ValueError: + raise + except Exception as e: + logger.error(f"Error renaming page: {str(e)}") + raise + + def get_page_linked_references(self, page_name: str) -> Any: + """Get all pages and blocks that reference this page (backlinks).""" + url = self.get_base_url() + logger.info(f"Getting backlinks for page '{page_name}'") + + try: + response = requests.post( + url, + headers=self._get_headers(), + json={ + "method": "logseq.Editor.getPageLinkedReferences", + "args": [page_name] + }, + verify=self.verify_ssl, + timeout=self.timeout + ) + response.raise_for_status() + return response.json() + + except Exception as e: + logger.error(f"Error getting backlinks: {str(e)}") + raise diff --git a/src/mcp_logseq/server.py b/src/mcp_logseq/server.py index e78c525..58ae727 100644 --- a/src/mcp_logseq/server.py +++ b/src/mcp_logseq/server.py @@ -78,6 +78,10 @@ def get_tool_handler(name: str) -> tools.ToolHandler | None: add_tool_handler(tools.DeletePageToolHandler()) add_tool_handler(tools.UpdatePageToolHandler()) add_tool_handler(tools.SearchToolHandler()) +add_tool_handler(tools.GetPagesFromNamespaceToolHandler()) +add_tool_handler(tools.GetPagesTreeFromNamespaceToolHandler()) +add_tool_handler(tools.RenamePageToolHandler()) +add_tool_handler(tools.GetPageBacklinksToolHandler()) logger.info("Tool handlers registration complete") @app.list_tools() diff --git a/src/mcp_logseq/tools.py b/src/mcp_logseq/tools.py index 13e61e0..3d109f4 100644 --- a/src/mcp_logseq/tools.py +++ b/src/mcp_logseq/tools.py @@ -486,3 +486,270 @@ def run_tool(self, args: dict) -> list[TextContent]: type="text", text=f"❌ Search failed: {str(e)}" )] + + +class GetPagesFromNamespaceToolHandler(ToolHandler): + def __init__(self): + super().__init__("get_pages_from_namespace") + + def get_tool_description(self): + return Tool( + name=self.name, + description="Get all pages within a namespace hierarchy (flat list). Use this to discover subpages of a parent page.", + inputSchema={ + "type": "object", + "properties": { + "namespace": { + "type": "string", + "description": "The namespace to query (e.g., 'Customer', 'Projects/2024')" + } + }, + "required": ["namespace"] + } + ) + + def run_tool(self, args: dict) -> list[TextContent]: + if "namespace" not in args: + raise RuntimeError("namespace argument required") + + try: + api = logseq.LogSeq(api_key=api_key) + result = api.get_pages_from_namespace(args["namespace"]) + + if not result: + return [TextContent( + type="text", + text=f"No pages found in namespace '{args['namespace']}'" + )] + + # Format pages for display + pages_info = [] + for page in result: + name = page.get('originalName') or page.get('name', '') + pages_info.append(f"- {name}") + + pages_info.sort() + + response = f"Pages in namespace '{args['namespace']}':\n\n" + response += "\n".join(pages_info) + response += f"\n\nTotal: {len(pages_info)} pages" + + return [TextContent(type="text", text=response)] + + except Exception as e: + logger.error(f"Failed to get pages from namespace: {str(e)}") + raise + + +class GetPagesTreeFromNamespaceToolHandler(ToolHandler): + def __init__(self): + super().__init__("get_pages_tree_from_namespace") + + def get_tool_description(self): + return Tool( + name=self.name, + description="Get pages within a namespace as a hierarchical tree structure. Useful for understanding the full page hierarchy.", + inputSchema={ + "type": "object", + "properties": { + "namespace": { + "type": "string", + "description": "The root namespace to build tree from (e.g., 'Projects')" + } + }, + "required": ["namespace"] + } + ) + + def run_tool(self, args: dict) -> list[TextContent]: + if "namespace" not in args: + raise RuntimeError("namespace argument required") + + try: + api = logseq.LogSeq(api_key=api_key) + result = api.get_pages_tree_from_namespace(args["namespace"]) + + if not result: + return [TextContent( + type="text", + text=f"No pages found in namespace '{args['namespace']}'" + )] + + # Format as tree structure + def format_tree(pages, prefix="", is_last_list=None): + if is_last_list is None: + is_last_list = [] + lines = [] + for i, page in enumerate(pages): + is_last = i == len(pages) - 1 + name = page.get('originalName') or page.get('name', '') + + # Build the prefix for this line + if prefix == "": + lines.append(name) + else: + connector = "└── " if is_last else "├── " + lines.append(f"{prefix}{connector}{name}") + + # Handle children if present + children = page.get('children', []) + if children: + # Build prefix for children + if prefix == "": + child_prefix = "" + else: + child_prefix = prefix + (" " if is_last else "│ ") + lines.extend(format_tree(children, child_prefix, is_last_list + [is_last])) + return lines + + tree_lines = format_tree(result) + + response = f"Page tree for namespace '{args['namespace']}':\n\n" + response += "\n".join(tree_lines) + + return [TextContent(type="text", text=response)] + + except Exception as e: + logger.error(f"Failed to get pages tree: {str(e)}") + raise + + +class RenamePageToolHandler(ToolHandler): + def __init__(self): + super().__init__("rename_page") + + def get_tool_description(self): + return Tool( + name=self.name, + description="Rename an existing page. All references throughout the graph will be automatically updated.", + inputSchema={ + "type": "object", + "properties": { + "old_name": { + "type": "string", + "description": "Current name of the page" + }, + "new_name": { + "type": "string", + "description": "New name for the page" + } + }, + "required": ["old_name", "new_name"] + } + ) + + def run_tool(self, args: dict) -> list[TextContent]: + if "old_name" not in args or "new_name" not in args: + raise RuntimeError("old_name and new_name arguments required") + + old_name = args["old_name"] + new_name = args["new_name"] + + try: + api = logseq.LogSeq(api_key=api_key) + api.rename_page(old_name, new_name) + + return [TextContent( + type="text", + text=f"Successfully renamed page '{old_name}' to '{new_name}'\n" + f"All references in the graph have been updated." + )] + except ValueError as e: + return [TextContent( + type="text", + text=f"Error: {str(e)}" + )] + except Exception as e: + logger.error(f"Failed to rename page: {str(e)}") + return [TextContent( + type="text", + text=f"Failed to rename page: {str(e)}" + )] + + +class GetPageBacklinksToolHandler(ToolHandler): + def __init__(self): + super().__init__("get_page_backlinks") + + def get_tool_description(self): + return Tool( + name=self.name, + description="Get all pages and blocks that link to a specific page (backlinks/linked references).", + inputSchema={ + "type": "object", + "properties": { + "page_name": { + "type": "string", + "description": "Name of the page to find backlinks for" + }, + "include_content": { + "type": "boolean", + "description": "Whether to include the content of referencing blocks", + "default": True + } + }, + "required": ["page_name"] + } + ) + + def run_tool(self, args: dict) -> list[TextContent]: + if "page_name" not in args: + raise RuntimeError("page_name argument required") + + page_name = args["page_name"] + include_content = args.get("include_content", True) + + try: + api = logseq.LogSeq(api_key=api_key) + result = api.get_page_linked_references(page_name) + + if not result: + return [TextContent( + type="text", + text=f"No backlinks found for page '{page_name}'" + )] + + # Format results + # API returns: [[PageEntity, [BlockEntity, ...]], ...] + content_parts = [] + content_parts.append(f"# Backlinks for '{page_name}'\n") + + total_refs = 0 + + for item in result: + if not isinstance(item, list) or len(item) < 2: + continue + + page_info, blocks = item[0], item[1] + + # Get page name + ref_page_name = page_info.get('originalName') or page_info.get('name', '') + block_count = len(blocks) if blocks else 0 + total_refs += block_count + + content_parts.append(f"**{ref_page_name}** ({block_count} reference{'s' if block_count != 1 else ''})") + + # Include block content if requested + if include_content and blocks: + for block in blocks: + block_content = block.get('content', '').strip() + if block_content: + # Truncate long content + if len(block_content) > 150: + block_content = block_content[:150] + "..." + content_parts.append(f" - {block_content}") + + content_parts.append("") + + # Summary + page_count = len(result) + content_parts.append(f"---\n**Total: {page_count} page{'s' if page_count != 1 else ''}, {total_refs} reference{'s' if total_refs != 1 else ''}**") + + return [TextContent(type="text", text="\n".join(content_parts))] + + except Exception as e: + logger.error(f"Failed to get backlinks: {str(e)}") + return [TextContent( + type="text", + text=f"Failed to get backlinks: {str(e)}" + )] diff --git a/tests/conftest.py b/tests/conftest.py index c18915e..33ca07b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,11 +4,15 @@ from mcp_logseq.logseq import LogSeq from mcp_logseq.tools import ( CreatePageToolHandler, - ListPagesToolHandler, + ListPagesToolHandler, GetPageContentToolHandler, DeletePageToolHandler, UpdatePageToolHandler, - SearchToolHandler + SearchToolHandler, + GetPagesFromNamespaceToolHandler, + GetPagesTreeFromNamespaceToolHandler, + RenamePageToolHandler, + GetPageBacklinksToolHandler ) @pytest.fixture @@ -76,7 +80,70 @@ def mock_logseq_responses(): ], "files": [], "has-more?": False - } + }, + "get_pages_from_namespace_success": [ + { + "id": "page-1", + "name": "customer/insideout", + "originalName": "Customer/InsideOut" + }, + { + "id": "page-2", + "name": "customer/orienteme", + "originalName": "Customer/Orienteme" + } + ], + "get_pages_tree_from_namespace_success": [ + { + "id": "page-1", + "name": "projects/2024", + "originalName": "Projects/2024", + "children": [ + { + "id": "page-2", + "name": "projects/2024/clienta", + "originalName": "Projects/2024/ClientA", + "children": [] + }, + { + "id": "page-3", + "name": "projects/2024/clientb", + "originalName": "Projects/2024/ClientB", + "children": [] + } + ] + }, + { + "id": "page-4", + "name": "projects/archive", + "originalName": "Projects/Archive", + "children": [] + } + ], + "rename_page_success": None, + "get_page_linked_references_success": [ + [ + { + "id": "page-1", + "name": "dec 15th, 2024", + "originalName": "Dec 15th, 2024" + }, + [ + {"content": "session [[Customer/Orienteme]]"}, + {"content": "followup with [[Customer/Orienteme]] team"} + ] + ], + [ + { + "id": "page-2", + "name": "projects/ai consulting", + "originalName": "Projects/AI Consulting" + }, + [ + {"content": "Active client: [[Customer/Orienteme]]"} + ] + ] + ] } @pytest.fixture @@ -86,9 +153,13 @@ def tool_handlers(): "create_page": CreatePageToolHandler(), "list_pages": ListPagesToolHandler(), "get_page_content": GetPageContentToolHandler(), - "delete_page": DeletePageToolHandler(), + "delete_page": DeletePageToolHandler(), "update_page": UpdatePageToolHandler(), - "search": SearchToolHandler() + "search": SearchToolHandler(), + "get_pages_from_namespace": GetPagesFromNamespaceToolHandler(), + "get_pages_tree_from_namespace": GetPagesTreeFromNamespaceToolHandler(), + "rename_page": RenamePageToolHandler(), + "get_page_backlinks": GetPageBacklinksToolHandler() } @pytest.fixture diff --git a/tests/integration/test_mcp_server.py b/tests/integration/test_mcp_server.py index f5a9dae..6d671b8 100644 --- a/tests/integration/test_mcp_server.py +++ b/tests/integration/test_mcp_server.py @@ -39,13 +39,15 @@ def test_get_tool_handler_non_existing(self): def test_list_tools_handler_count(self): """Test that we have the expected number of tool handlers.""" - # We should have 6 registered tool handlers - assert len(tool_handlers) == 6 - + # We should have 10 registered tool handlers + assert len(tool_handlers) == 10 + # Verify specific tool names are present expected_names = [ "create_page", "list_pages", "get_page_content", - "delete_page", "update_page", "search" + "delete_page", "update_page", "search", + "get_pages_from_namespace", "get_pages_tree_from_namespace", + "rename_page", "get_page_backlinks" ] for name in expected_names: assert name in tool_handlers diff --git a/tests/unit/test_logseq_api.py b/tests/unit/test_logseq_api.py index a2dc4f4..e16c081 100644 --- a/tests/unit/test_logseq_api.py +++ b/tests/unit/test_logseq_api.py @@ -296,4 +296,161 @@ def test_update_page_success(self, logseq_client, mock_logseq_responses): assert result == expected # Verify all calls were made - assert len(responses.calls) == 3 \ No newline at end of file + assert len(responses.calls) == 3 + + @responses.activate + def test_get_pages_from_namespace_success(self, logseq_client, mock_logseq_responses): + """Test successful namespace pages retrieval.""" + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=mock_logseq_responses["get_pages_from_namespace_success"], + status=200 + ) + + result = logseq_client.get_pages_from_namespace("Customer") + assert result == mock_logseq_responses["get_pages_from_namespace_success"] + + # Verify the request + request_data = json.loads(responses.calls[0].request.body) + assert request_data["method"] == "logseq.Editor.getPagesFromNamespace" + assert request_data["args"] == ["Customer"] + + @responses.activate + def test_get_pages_from_namespace_empty(self, logseq_client): + """Test namespace pages retrieval with no results.""" + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=[], + status=200 + ) + + result = logseq_client.get_pages_from_namespace("EmptyNamespace") + assert result == [] + + @responses.activate + def test_get_pages_tree_from_namespace_success(self, logseq_client, mock_logseq_responses): + """Test successful namespace tree retrieval.""" + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=mock_logseq_responses["get_pages_tree_from_namespace_success"], + status=200 + ) + + result = logseq_client.get_pages_tree_from_namespace("Projects") + assert result == mock_logseq_responses["get_pages_tree_from_namespace_success"] + + # Verify the request + request_data = json.loads(responses.calls[0].request.body) + assert request_data["method"] == "logseq.Editor.getPagesTreeFromNamespace" + assert request_data["args"] == ["Projects"] + + @responses.activate + def test_get_pages_tree_from_namespace_empty(self, logseq_client): + """Test namespace tree retrieval with no results.""" + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=[], + status=200 + ) + + result = logseq_client.get_pages_tree_from_namespace("EmptyNamespace") + assert result == [] + + @responses.activate + def test_rename_page_success(self, logseq_client, mock_logseq_responses): + """Test successful page rename.""" + # Mock list_pages for validation + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=[ + {"originalName": "OldPage"}, + {"originalName": "OtherPage"} + ], + status=200 + ) + + # Mock rename call (returns null on success) + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + body='null', + status=200, + content_type='application/json' + ) + + result = logseq_client.rename_page("OldPage", "NewPage") + assert result is None + + # Verify the rename request + request_data = json.loads(responses.calls[1].request.body) + assert request_data["method"] == "logseq.Editor.renamePage" + assert request_data["args"] == ["OldPage", "NewPage"] + + @responses.activate + def test_rename_page_source_not_found(self, logseq_client): + """Test rename with non-existent source page.""" + # Mock list_pages - source doesn't exist + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=[ + {"originalName": "OtherPage"} + ], + status=200 + ) + + with pytest.raises(ValueError, match="does not exist"): + logseq_client.rename_page("NonExistent", "NewPage") + + @responses.activate + def test_rename_page_target_exists(self, logseq_client): + """Test rename to existing page name.""" + # Mock list_pages - target already exists + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=[ + {"originalName": "OldPage"}, + {"originalName": "ExistingPage"} + ], + status=200 + ) + + with pytest.raises(ValueError, match="already exists"): + logseq_client.rename_page("OldPage", "ExistingPage") + + @responses.activate + def test_get_page_linked_references_success(self, logseq_client, mock_logseq_responses): + """Test successful backlinks retrieval.""" + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=mock_logseq_responses["get_page_linked_references_success"], + status=200 + ) + + result = logseq_client.get_page_linked_references("Customer/Orienteme") + assert result == mock_logseq_responses["get_page_linked_references_success"] + + # Verify the request + request_data = json.loads(responses.calls[0].request.body) + assert request_data["method"] == "logseq.Editor.getPageLinkedReferences" + assert request_data["args"] == ["Customer/Orienteme"] + + @responses.activate + def test_get_page_linked_references_empty(self, logseq_client): + """Test backlinks retrieval with no results.""" + responses.add( + responses.POST, + "http://127.0.0.1:12315/api", + json=[], + status=200 + ) + + result = logseq_client.get_page_linked_references("OrphanPage") + assert result == [] \ No newline at end of file diff --git a/tests/unit/test_tool_handlers.py b/tests/unit/test_tool_handlers.py index 80056c4..aee330a 100644 --- a/tests/unit/test_tool_handlers.py +++ b/tests/unit/test_tool_handlers.py @@ -7,7 +7,11 @@ GetPageContentToolHandler, DeletePageToolHandler, UpdatePageToolHandler, - SearchToolHandler + SearchToolHandler, + GetPagesFromNamespaceToolHandler, + GetPagesTreeFromNamespaceToolHandler, + RenamePageToolHandler, + GetPageBacklinksToolHandler ) class TestCreatePageToolHandler: @@ -387,4 +391,321 @@ def test_run_tool_with_options(self, mock_logseq_class): }) # Verify API was called with correct options - mock_api.search_content.assert_called_once_with("test", {"limit": 5}) \ No newline at end of file + mock_api.search_content.assert_called_once_with("test", {"limit": 5}) + + +class TestGetPagesFromNamespaceToolHandler: + """Test cases for GetPagesFromNamespaceToolHandler.""" + + def test_get_tool_description(self): + """Test tool description schema.""" + handler = GetPagesFromNamespaceToolHandler() + tool = handler.get_tool_description() + + assert tool.name == "get_pages_from_namespace" + assert "namespace" in tool.description.lower() + assert "namespace" in tool.inputSchema["properties"] + assert "namespace" in tool.inputSchema["required"] + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_success(self, mock_logseq_class): + """Test successful namespace pages retrieval.""" + # Setup mock + mock_api = Mock() + mock_api.get_pages_from_namespace.return_value = [ + {"originalName": "Customer/InsideOut"}, + {"originalName": "Customer/Orienteme"} + ] + mock_logseq_class.return_value = mock_api + + handler = GetPagesFromNamespaceToolHandler() + result = handler.run_tool({"namespace": "Customer"}) + + # Verify API was called correctly + mock_api.get_pages_from_namespace.assert_called_once_with("Customer") + + # Verify result + assert len(result) == 1 + assert isinstance(result[0], TextContent) + text = result[0].text + assert "Customer/InsideOut" in text + assert "Customer/Orienteme" in text + assert "Total: 2 pages" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_empty_namespace(self, mock_logseq_class): + """Test namespace with no pages.""" + # Setup mock + mock_api = Mock() + mock_api.get_pages_from_namespace.return_value = [] + mock_logseq_class.return_value = mock_api + + handler = GetPagesFromNamespaceToolHandler() + result = handler.run_tool({"namespace": "EmptyNamespace"}) + + # Verify result + text = result[0].text + assert "No pages found in namespace 'EmptyNamespace'" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + def test_run_tool_missing_args(self): + """Test tool with missing namespace argument.""" + handler = GetPagesFromNamespaceToolHandler() + + with pytest.raises(RuntimeError, match="namespace argument required"): + handler.run_tool({}) + + +class TestGetPagesTreeFromNamespaceToolHandler: + """Test cases for GetPagesTreeFromNamespaceToolHandler.""" + + def test_get_tool_description(self): + """Test tool description schema.""" + handler = GetPagesTreeFromNamespaceToolHandler() + tool = handler.get_tool_description() + + assert tool.name == "get_pages_tree_from_namespace" + assert "tree" in tool.description.lower() + assert "namespace" in tool.inputSchema["properties"] + assert "namespace" in tool.inputSchema["required"] + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_success(self, mock_logseq_class): + """Test successful namespace tree retrieval.""" + # Setup mock with hierarchical data + mock_api = Mock() + mock_api.get_pages_tree_from_namespace.return_value = [ + { + "originalName": "Projects/2024", + "children": [ + {"originalName": "Projects/2024/ClientA", "children": []}, + {"originalName": "Projects/2024/ClientB", "children": []} + ] + }, + { + "originalName": "Projects/Archive", + "children": [] + } + ] + mock_logseq_class.return_value = mock_api + + handler = GetPagesTreeFromNamespaceToolHandler() + result = handler.run_tool({"namespace": "Projects"}) + + # Verify API was called correctly + mock_api.get_pages_tree_from_namespace.assert_called_once_with("Projects") + + # Verify result + assert len(result) == 1 + assert isinstance(result[0], TextContent) + text = result[0].text + assert "Projects/2024" in text + assert "Projects/2024/ClientA" in text + assert "Projects/2024/ClientB" in text + assert "Projects/Archive" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_empty_namespace(self, mock_logseq_class): + """Test namespace tree with no pages.""" + # Setup mock + mock_api = Mock() + mock_api.get_pages_tree_from_namespace.return_value = [] + mock_logseq_class.return_value = mock_api + + handler = GetPagesTreeFromNamespaceToolHandler() + result = handler.run_tool({"namespace": "EmptyNamespace"}) + + # Verify result + text = result[0].text + assert "No pages found in namespace 'EmptyNamespace'" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + def test_run_tool_missing_args(self): + """Test tool with missing namespace argument.""" + handler = GetPagesTreeFromNamespaceToolHandler() + + with pytest.raises(RuntimeError, match="namespace argument required"): + handler.run_tool({}) + + +class TestRenamePageToolHandler: + """Test cases for RenamePageToolHandler.""" + + def test_get_tool_description(self): + """Test tool description schema.""" + handler = RenamePageToolHandler() + tool = handler.get_tool_description() + + assert tool.name == "rename_page" + assert "rename" in tool.description.lower() + assert "old_name" in tool.inputSchema["properties"] + assert "new_name" in tool.inputSchema["properties"] + assert "old_name" in tool.inputSchema["required"] + assert "new_name" in tool.inputSchema["required"] + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_success(self, mock_logseq_class): + """Test successful page rename.""" + # Setup mock + mock_api = Mock() + mock_api.rename_page.return_value = None + mock_logseq_class.return_value = mock_api + + handler = RenamePageToolHandler() + result = handler.run_tool({"old_name": "OldPage", "new_name": "NewPage"}) + + # Verify API was called correctly + mock_api.rename_page.assert_called_once_with("OldPage", "NewPage") + + # Verify result + assert len(result) == 1 + assert isinstance(result[0], TextContent) + text = result[0].text + assert "Successfully renamed" in text + assert "OldPage" in text + assert "NewPage" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_source_not_found(self, mock_logseq_class): + """Test rename with non-existent source page.""" + # Setup mock to raise ValueError + mock_api = Mock() + mock_api.rename_page.side_effect = ValueError("Page 'NonExistent' does not exist") + mock_logseq_class.return_value = mock_api + + handler = RenamePageToolHandler() + result = handler.run_tool({"old_name": "NonExistent", "new_name": "NewPage"}) + + # Verify error message + text = result[0].text + assert "does not exist" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_target_exists(self, mock_logseq_class): + """Test rename to existing page name.""" + # Setup mock to raise ValueError + mock_api = Mock() + mock_api.rename_page.side_effect = ValueError("Page 'ExistingPage' already exists") + mock_logseq_class.return_value = mock_api + + handler = RenamePageToolHandler() + result = handler.run_tool({"old_name": "OldPage", "new_name": "ExistingPage"}) + + # Verify error message + text = result[0].text + assert "already exists" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + def test_run_tool_missing_args(self): + """Test tool with missing arguments.""" + handler = RenamePageToolHandler() + + with pytest.raises(RuntimeError, match="old_name and new_name arguments required"): + handler.run_tool({"old_name": "OnlyOld"}) + + +class TestGetPageBacklinksToolHandler: + """Test cases for GetPageBacklinksToolHandler.""" + + def test_get_tool_description(self): + """Test tool description schema.""" + handler = GetPageBacklinksToolHandler() + tool = handler.get_tool_description() + + assert tool.name == "get_page_backlinks" + assert "backlink" in tool.description.lower() + assert "page_name" in tool.inputSchema["properties"] + assert "include_content" in tool.inputSchema["properties"] + assert "page_name" in tool.inputSchema["required"] + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_success(self, mock_logseq_class): + """Test successful backlinks retrieval.""" + # Setup mock with backlinks data + mock_api = Mock() + mock_api.get_page_linked_references.return_value = [ + [ + {"originalName": "Dec 15th, 2024"}, + [ + {"content": "session [[Customer/Orienteme]]"}, + {"content": "followup with [[Customer/Orienteme]] team"} + ] + ], + [ + {"originalName": "Projects/AI Consulting"}, + [ + {"content": "Active client: [[Customer/Orienteme]]"} + ] + ] + ] + mock_logseq_class.return_value = mock_api + + handler = GetPageBacklinksToolHandler() + result = handler.run_tool({"page_name": "Customer/Orienteme"}) + + # Verify API was called correctly + mock_api.get_page_linked_references.assert_called_once_with("Customer/Orienteme") + + # Verify result + assert len(result) == 1 + assert isinstance(result[0], TextContent) + text = result[0].text + assert "Dec 15th, 2024" in text + assert "Projects/AI Consulting" in text + assert "2 references" in text + assert "1 reference" in text + assert "Total: 2 pages, 3 references" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_no_backlinks(self, mock_logseq_class): + """Test page with no backlinks.""" + # Setup mock + mock_api = Mock() + mock_api.get_page_linked_references.return_value = [] + mock_logseq_class.return_value = mock_api + + handler = GetPageBacklinksToolHandler() + result = handler.run_tool({"page_name": "OrphanPage"}) + + # Verify result + text = result[0].text + assert "No backlinks found for page 'OrphanPage'" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + @patch('mcp_logseq.tools.logseq.LogSeq') + def test_run_tool_without_content(self, mock_logseq_class): + """Test backlinks without including block content.""" + # Setup mock + mock_api = Mock() + mock_api.get_page_linked_references.return_value = [ + [ + {"originalName": "Source Page"}, + [{"content": "Reference to [[Target]]"}] + ] + ] + mock_logseq_class.return_value = mock_api + + handler = GetPageBacklinksToolHandler() + result = handler.run_tool({"page_name": "Target", "include_content": False}) + + # Verify result shows page but not detailed content + text = result[0].text + assert "Source Page" in text + assert "1 reference" in text + + @patch.dict('os.environ', {'LOGSEQ_API_TOKEN': 'test_token'}) + def test_run_tool_missing_args(self): + """Test tool with missing page_name argument.""" + handler = GetPageBacklinksToolHandler() + + with pytest.raises(RuntimeError, match="page_name argument required"): + handler.run_tool({}) \ No newline at end of file