fix: add v2 API support for confluence_get_page with OAuth (sooperset#579)

This fix addresses the deprecated v1 REST API endpoint issue when using
OAuth authentication. The v1 endpoint has been removed for OAuth but
continues to work with API tokens.

- Extended ConfluenceV2Adapter with get_page method
- Updated PagesMixin to use v2 API for OAuth authentication
- Added comprehensive unit tests for OAuth flow
- Maintained backward compatibility for API token users

Reported-by: dannymyers
Github-Issue: sooperset#576

Author: sooperset

src/mcp_atlassian/confluence/pages.py

@@ -48,9 +48,25 @@ def get_page_content(
             Exception: If there is an error retrieving the page
         """
         try:
-            page = self.confluence.get_page_by_id(
-                page_id=page_id, expand="body.storage,version,space,children.attachment"
-            )
+            # Use v2 API for OAuth authentication, v1 API for token/basic auth
+            v2_adapter = self._v2_adapter
+            if v2_adapter:
+                logger.debug(
+                    f"Using v2 API for OAuth authentication to get page '{page_id}'"
+                )
+                page = v2_adapter.get_page(
+                    page_id=page_id,
+                    expand="body.storage,version,space,children.attachment",
+                )
+            else:
+                logger.debug(
+                    f"Using v1 API for token/basic authentication to get page '{page_id}'"
+                )
+                page = self.confluence.get_page_by_id(
+                    page_id=page_id,
+                    expand="body.storage,version,space,children.attachment",
+                )
+
             space_key = page.get("space", {}).get("key", "")
             content = page["body"]["storage"]["value"]
             processed_html, processed_markdown = self.preprocessor.process_html_content(

src/mcp_atlassian/confluence/v2_adapter.py

@@ -272,6 +272,74 @@ def _get_space_key_from_id(self, space_id: str) -> str:
             # Return the space_id as fallback
             return space_id
 
+    def get_page(
+        self,
+        page_id: str,
+        expand: str | None = None,
+    ) -> dict[str, Any]:
+        """Get a page using the v2 API.
+
+        Args:
+            page_id: The ID of the page to retrieve
+            expand: Fields to expand in the response (not used in v2 API, for compatibility only)
+
+        Returns:
+            The page data from the API response in v1-compatible format
+
+        Raises:
+            ValueError: If page retrieval fails
+        """
+        try:
+            # Make the v2 API call to get the page
+            url = f"{self.base_url}/api/v2/pages/{page_id}"
+
+            # Convert v1 expand parameters to v2 format
+            params = {"body-format": "storage"}
+
+            response = self.session.get(url, params=params)
+            response.raise_for_status()
+
+            v2_response = response.json()
+            logger.debug(f"Successfully retrieved page '{page_id}' with v2 API")
+
+            # Get space key from space ID
+            space_id = v2_response.get("spaceId")
+            space_key = self._get_space_key_from_id(space_id) if space_id else "unknown"
+
+            # Convert v2 response to v1-compatible format
+            v1_compatible = self._convert_v2_to_v1_format(v2_response, space_key)
+
+            # Add body.storage structure if body content exists
+            if "body" in v2_response and v2_response["body"].get("storage"):
+                storage_value = v2_response["body"]["storage"].get("value", "")
+                v1_compatible["body"] = {
+                    "storage": {"value": storage_value, "representation": "storage"}
+                }
+
+            # Add space information with more details
+            if space_id:
+                v1_compatible["space"] = {
+                    "key": space_key,
+                    "id": space_id,
+                }
+
+            # Add version information
+            if "version" in v2_response:
+                v1_compatible["version"] = {
+                    "number": v2_response["version"].get("number", 1)
+                }
+
+            return v1_compatible
+
+        except HTTPError as e:
+            logger.error(f"HTTP error getting page '{page_id}': {e}")
+            if e.response is not None:
+                logger.error(f"Response content: {e.response.text}")
+            raise ValueError(f"Failed to get page '{page_id}': {e}") from e
+        except Exception as e:
+            logger.error(f"Error getting page '{page_id}': {e}")
+            raise ValueError(f"Failed to get page '{page_id}': {e}") from e
+
     def delete_page(self, page_id: str) -> bool:
         """Delete a page using the v2 API.
 

tests/unit/confluence/test_pages.py

@@ -1029,6 +1029,61 @@ def test_update_page_oauth_uses_v2_api(self, oauth_pages_mixin):
                 assert result.id == page_id
                 assert result.title == title
 
+    def test_get_page_content_oauth_uses_v2_api(self, oauth_pages_mixin):
+        """Test that OAuth authentication uses v2 API for getting page content."""
+        # Arrange
+        page_id = "oauth_get_123"
+
+        # Mock the v2 adapter
+        with patch(
+            "mcp_atlassian.confluence.pages.ConfluenceV2Adapter"
+        ) as mock_v2_adapter_class:
+            mock_v2_adapter = MagicMock()
+            mock_v2_adapter_class.return_value = mock_v2_adapter
+
+            # Mock v2 API response
+            mock_v2_adapter.get_page.return_value = {
+                "id": page_id,
+                "title": "OAuth Test Page",
+                "body": {"storage": {"value": "<p>OAuth page content</p>"}},
+                "space": {"key": "PROJ", "name": "Project"},
+                "version": {"number": 3},
+            }
+
+            # Mock the preprocessor
+            oauth_pages_mixin.preprocessor.process_html_content.return_value = (
+                "<p>Processed HTML</p>",
+                "Processed OAuth content",
+            )
+
+            # Act
+            result = oauth_pages_mixin.get_page_content(
+                page_id, convert_to_markdown=True
+            )
+
+            # Assert that v2 API was used instead of v1
+            mock_v2_adapter.get_page.assert_called_once_with(
+                page_id=page_id, expand="body.storage,version,space,children.attachment"
+            )
+
+            # Verify v1 API was NOT called
+            oauth_pages_mixin.confluence.get_page_by_id.assert_not_called()
+
+            # Verify the preprocessor was called
+            oauth_pages_mixin.preprocessor.process_html_content.assert_called_once_with(
+                "<p>OAuth page content</p>",
+                space_key="PROJ",
+                confluence_client=oauth_pages_mixin.confluence,
+            )
+
+            # Verify result is a ConfluencePage with correct data
+            assert isinstance(result, ConfluencePage)
+            assert result.id == page_id
+            assert result.title == "OAuth Test Page"
+            assert result.content == "Processed OAuth content"
+            assert result.space.key == "PROJ"
+            assert result.version.number == 3
+
     def test_delete_page_oauth_uses_v2_api(self, oauth_pages_mixin):
         """Test that OAuth authentication uses v2 API for deleting pages."""
         # Arrange

tests/unit/confluence/test_v2_adapter.py

@@ -0,0 +1,137 @@
+"""Unit tests for ConfluenceV2Adapter class."""
+
+from unittest.mock import MagicMock, Mock
+
+import pytest
+import requests
+from requests.exceptions import HTTPError
+
+from mcp_atlassian.confluence.v2_adapter import ConfluenceV2Adapter
+
+
+class TestConfluenceV2Adapter:
+    """Test cases for ConfluenceV2Adapter."""
+
+    @pytest.fixture
+    def mock_session(self):
+        """Create a mock session."""
+        return MagicMock(spec=requests.Session)
+
+    @pytest.fixture
+    def v2_adapter(self, mock_session):
+        """Create a ConfluenceV2Adapter instance."""
+        return ConfluenceV2Adapter(
+            session=mock_session, base_url="https://example.atlassian.net/wiki"
+        )
+
+    def test_get_page_success(self, v2_adapter, mock_session):
+        """Test successful page retrieval."""
+        # Mock the v2 API response
+        mock_response = Mock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "id": "123456",
+            "status": "current",
+            "title": "Test Page",
+            "spaceId": "789",
+            "version": {"number": 5},
+            "body": {
+                "storage": {"value": "<p>Test content</p>", "representation": "storage"}
+            },
+            "_links": {"webui": "/pages/viewpage.action?pageId=123456"},
+        }
+        mock_session.get.return_value = mock_response
+
+        # Mock space key lookup
+        space_response = Mock()
+        space_response.status_code = 200
+        space_response.json.return_value = {"key": "TEST"}
+        mock_session.get.side_effect = [mock_response, space_response]
+
+        # Call the method
+        result = v2_adapter.get_page("123456")
+
+        # Verify the API call
+        assert mock_session.get.call_count == 2
+        mock_session.get.assert_any_call(
+            "https://example.atlassian.net/wiki/api/v2/pages/123456",
+            params={"body-format": "storage"},
+        )
+
+        # Verify the response format
+        assert result["id"] == "123456"
+        assert result["type"] == "page"
+        assert result["title"] == "Test Page"
+        assert result["space"]["key"] == "TEST"
+        assert result["space"]["id"] == "789"
+        assert result["version"]["number"] == 5
+        assert result["body"]["storage"]["value"] == "<p>Test content</p>"
+        assert result["body"]["storage"]["representation"] == "storage"
+
+    def test_get_page_not_found(self, v2_adapter, mock_session):
+        """Test page retrieval when page doesn't exist."""
+        # Mock a 404 response
+        mock_response = Mock()
+        mock_response.status_code = 404
+        mock_response.text = "Page not found"
+        mock_response.raise_for_status.side_effect = HTTPError(response=mock_response)
+        mock_session.get.return_value = mock_response
+
+        # Call the method and expect an exception
+        with pytest.raises(ValueError, match="Failed to get page '999999'"):
+            v2_adapter.get_page("999999")
+
+    def test_get_page_with_minimal_response(self, v2_adapter, mock_session):
+        """Test page retrieval with minimal v2 response."""
+        # Mock the v2 API response without optional fields
+        mock_response = Mock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "id": "123456",
+            "status": "current",
+            "title": "Minimal Page",
+        }
+        mock_session.get.return_value = mock_response
+
+        # Call the method
+        result = v2_adapter.get_page("123456")
+
+        # Verify the response handles missing fields gracefully
+        assert result["id"] == "123456"
+        assert result["type"] == "page"
+        assert result["title"] == "Minimal Page"
+        assert result["space"]["key"] == "unknown"  # Fallback when no spaceId
+        assert result["version"]["number"] == 1  # Default version
+
+    def test_get_page_network_error(self, v2_adapter, mock_session):
+        """Test page retrieval with network error."""
+        # Mock a network error
+        mock_session.get.side_effect = requests.RequestException("Network error")
+
+        # Call the method and expect an exception
+        with pytest.raises(ValueError, match="Failed to get page '123456'"):
+            v2_adapter.get_page("123456")
+
+    def test_get_page_with_expand_parameter(self, v2_adapter, mock_session):
+        """Test that expand parameter is accepted but not used."""
+        # Mock the v2 API response
+        mock_response = Mock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "id": "123456",
+            "status": "current",
+            "title": "Test Page",
+        }
+        mock_session.get.return_value = mock_response
+
+        # Call with expand parameter
+        result = v2_adapter.get_page("123456", expand="body.storage,version")
+
+        # Verify the API call doesn't include expand in params
+        mock_session.get.assert_called_once_with(
+            "https://example.atlassian.net/wiki/api/v2/pages/123456",
+            params={"body-format": "storage"},
+        )
+
+        # Verify we still get a result
+        assert result["id"] == "123456"