diff --git a/atlassian/confluence/__init__.py b/atlassian/confluence/__init__.py index c84484fd2..9cb65bdb9 100644 --- a/atlassian/confluence/__init__.py +++ b/atlassian/confluence/__init__.py @@ -18,7 +18,11 @@ class Confluence(ConfluenceBase): def __init__(self, url, *args, **kwargs): # Detect which implementation to use - if ("atlassian.net" in url or "jira.com" in url) and ("/wiki" not in url): + # Priority: explicit cloud= kwarg > URL-based heuristic + is_cloud = kwargs.get("cloud") + if is_cloud is None: + is_cloud = "atlassian.net" in url or "jira.com" in url or "api.atlassian.com" in url + if is_cloud: impl = ConfluenceCloud(url, *args, **kwargs) else: impl = ConfluenceServer(url, *args, **kwargs) diff --git a/atlassian/confluence/base.py b/atlassian/confluence/base.py index b5587b575..78e98b12f 100644 --- a/atlassian/confluence/base.py +++ b/atlassian/confluence/base.py @@ -134,27 +134,39 @@ def _get_paged( yield from response.get("results", []) - if self.cloud: - url = response.get("_links", {}).get("next", {}).get("href") - if url is None: - break - # From now on we have absolute URLs with parameters - absolute = True - # Params are now provided by the url - params = {} - # Trailing should not be added as it is already part of the url - trailing = False + next_link = response.get("_links", {}).get("next") + if next_link is None: + break + if isinstance(next_link, str): + url = next_link else: - if response.get("_links", {}).get("next") is None: - break - # For server, we need to extract the next page URL from the _links.next.href - next_url = response.get("_links", {}).get("next", {}).get("href") - if next_url is None: - break - url = next_url - absolute = True - params = {} - trailing = False + url = next_link.get("href") + if url is None: + break + + if url.startswith("/"): + # Prepend base URL from self.url, stripping the API root suffix to preserve path prefix + # Example: self.url = "https://api.atlassian.com/ex/confluence/abc/wiki/rest/api" + # api_root = "wiki/rest/api" + # base = "https://api.atlassian.com/ex/confluence/abc" + # relative = "/rest/api/content?cursor=1" + # result = "https://api.atlassian.com/ex/confluence/abc/rest/api/content?cursor=1" + api_root_suffix = f"/{self.api_root}" + if self.url.endswith(api_root_suffix): + base = self.url[:-len(api_root_suffix)] + else: + # Fallback: extract scheme+netloc if api_root suffix not found + from urllib.parse import urlparse + parsed = urlparse(self.url) + base = f"{parsed.scheme}://{parsed.netloc}" + url = base + url + + # From now on we have absolute URLs with parameters + absolute = True + # Params are now provided by the url + params = {} + # Trailing should not be added as it is already part of the url + trailing = False return diff --git a/atlassian/confluence/cloud/__init__.py b/atlassian/confluence/cloud/__init__.py index d7c686072..cfb21eff7 100644 --- a/atlassian/confluence/cloud/__init__.py +++ b/atlassian/confluence/cloud/__init__.py @@ -13,10 +13,10 @@ def __init__(self, url="https://api.atlassian.com/", *args, **kwargs): if "cloud" not in kwargs: kwargs["cloud"] = True if "api_version" not in kwargs: - kwargs["api_version"] = "2" + kwargs["api_version"] = "latest" if "api_root" not in kwargs: - kwargs["api_root"] = "wiki/api/v2" - url = url.strip("/") + kwargs["api_root"] = "wiki/rest/api" + url = url.strip("/") + f"/{kwargs['api_root']}" super(Cloud, self).__init__(url, *args, **kwargs) # Content Management @@ -28,6 +28,14 @@ def get_content_by_type(self, content_type, **kwargs): """Get content by type (page, blogpost, etc.).""" return self.get("content", params={"type": content_type, **kwargs}) + def get_all_pages_from_space(self, space_key, **kwargs): + """Get all pages from space.""" + return self._get_paged("content", params={"spaceKey": space_key, "type": "page", **kwargs}) + + def get_all_blog_posts_from_space(self, space_key, **kwargs): + """Get all blog posts from space.""" + return self._get_paged("content", params={"spaceKey": space_key, "type": "blogpost", **kwargs}) + def create_content(self, data, **kwargs): """Create new content.""" return self.post("content", data=data, **kwargs) diff --git a/atlassian/confluence/cloud/base.py b/atlassian/confluence/cloud/base.py index b9fa9c6ce..ec2a081b0 100644 --- a/atlassian/confluence/cloud/base.py +++ b/atlassian/confluence/cloud/base.py @@ -24,53 +24,4 @@ def __init__(self, url, *args, **kwargs): """ super(ConfluenceCloudBase, self).__init__(url, *args, **kwargs) - def _get_paged( - self, - url, - params=None, - data=None, - flags=None, - trailing=None, - absolute=False, - ): - """ - Used to get the paged data for Confluence Cloud - - :param url: string: The url to retrieve - :param params: dict (default is None): The parameter's - :param data: dict (default is None): The data - :param flags: string[] (default is None): The flags - :param trailing: bool (default is None): If True, a trailing slash is added to the url - :param absolute: bool (default is False): If True, the url is used absolute and not relative to the root - - :return: A generator object for the data elements - """ - if params is None: - params = {} - - while True: - response = self.get( - url, - trailing=trailing, - params=params, - data=data, - flags=flags, - absolute=absolute, - ) - if "results" not in response: - return - - yield from response.get("results", []) - - # Confluence Cloud uses _links.next.href for pagination - url = response.get("_links", {}).get("next", {}).get("href") - if url is None: - break - # From now on we have absolute URLs with parameters - absolute = True - # Params are now provided by the url - params = {} - # Trailing should not be added as it is already part of the url - trailing = False - return diff --git a/atlassian/confluence/server/__init__.py b/atlassian/confluence/server/__init__.py index ed268df23..50ccc9978 100644 --- a/atlassian/confluence/server/__init__.py +++ b/atlassian/confluence/server/__init__.py @@ -62,11 +62,11 @@ def get_content_by_id(self, content_id, **kwargs): def get_all_pages_from_space(self, space_key, **kwargs): """Get all pages from space.""" - return self.get("content", params={"spaceKey": space_key, "type": "page", **kwargs}) + return self._get_paged("content", params={"spaceKey": space_key, "type": "page", **kwargs}) def get_all_blog_posts_from_space(self, space_key, **kwargs): """Get all blog posts from space.""" - return self.get("content", params={"spaceKey": space_key, "type": "blogpost", **kwargs}) + return self._get_paged("content", params={"spaceKey": space_key, "type": "blogpost", **kwargs}) def get_page_by_title(self, space_key, title, **kwargs): """Get page by title and space key.""" @@ -195,11 +195,11 @@ def remove_content_label(self, content_id, label_name, **kwargs): def get_all_pages_by_label(self, label, **kwargs): """Get all pages by label.""" - return self.get("content", params={"label": label, "type": "page", **kwargs}) + return self._get_paged("content", params={"label": label, "type": "page", **kwargs}) def get_all_blog_posts_by_label(self, label, **kwargs): """Get all blog posts by label.""" - return self.get("content", params={"label": label, "type": "blogpost", **kwargs}) + return self._get_paged("content", params={"label": label, "type": "blogpost", **kwargs}) # Attachment Management def get_attachments(self, content_id, **kwargs): @@ -293,24 +293,24 @@ def get_draft_content(self, content_id, **kwargs): def get_all_draft_pages_from_space(self, space_key, **kwargs): """Get all draft pages from space.""" - return self.get("content", params={"spaceKey": space_key, "type": "page", "status": "draft", **kwargs}) + return self._get_paged("content", params={"spaceKey": space_key, "type": "page", "status": "draft", **kwargs}) def get_all_draft_blog_posts_from_space(self, space_key, **kwargs): """Get all draft blog posts from space.""" - return self.get("content", params={"spaceKey": space_key, "type": "blogpost", "status": "draft", **kwargs}) + return self._get_paged("content", params={"spaceKey": space_key, "type": "blogpost", "status": "draft", **kwargs}) # Trash Management def get_trash_content(self, space_key, **kwargs): """Get trash content.""" - return self.get("content", params={"spaceKey": space_key, "status": "trashed", **kwargs}) + return self._get_paged("content", params={"spaceKey": space_key, "status": "trashed", **kwargs}) def get_all_pages_from_space_trash(self, space_key, **kwargs): """Get all pages from space trash.""" - return self.get("content", params={"spaceKey": space_key, "type": "page", "status": "trashed", **kwargs}) + return self._get_paged("content", params={"spaceKey": space_key, "type": "page", "status": "trashed", **kwargs}) def get_all_blog_posts_from_space_trash(self, space_key, **kwargs): """Get all blog posts from space trash.""" - return self.get("content", params={"spaceKey": space_key, "type": "blogpost", "status": "trashed", **kwargs}) + return self._get_paged("content", params={"spaceKey": space_key, "type": "blogpost", "status": "trashed", **kwargs}) # Export def export_content(self, content_id, **kwargs): diff --git a/atlassian/confluence/server/base.py b/atlassian/confluence/server/base.py index 058e8cd2c..f7451db93 100644 --- a/atlassian/confluence/server/base.py +++ b/atlassian/confluence/server/base.py @@ -24,54 +24,4 @@ def __init__(self, url, *args, **kwargs): """ super(ConfluenceServerBase, self).__init__(url, *args, **kwargs) - def _get_paged( - self, - url, - params=None, - data=None, - flags=None, - trailing=False, - absolute=False, - ): - """ - Used to get the paged data for Confluence Server - - :param url: string: The url to retrieve - :param params: dict (default is None): The parameter's - :param data: dict (default is None): The data - :param flags: string[] (default is None): The flags - :param trailing: bool (default is None): If True, a trailing slash is added to the url - :param absolute: bool (default is False): If True, the url is used absolute and not relative to the root - - :return: A generator object for the data elements - """ - if params is None: - params = {} - - while True: - response = self.get( - url, - trailing=trailing, - params=params, - data=data, - flags=flags, - absolute=absolute, - ) - if "results" not in response: - return - - yield from response.get("results", []) - - # Confluence Server uses _links.next.href for pagination - if response.get("_links", {}).get("next") is None: - break - # For server, we need to extract the next page URL from the _links.next.href - next_url = response.get("_links", {}).get("next", {}).get("href") - if next_url is None: - break - url = next_url - absolute = True - params = {} - trailing = False - return diff --git a/tests/confluence/test_confluence_cloud.py b/tests/confluence/test_confluence_cloud.py index 060135d97..77899900f 100644 --- a/tests/confluence/test_confluence_cloud.py +++ b/tests/confluence/test_confluence_cloud.py @@ -21,9 +21,10 @@ class TestConfluenceCloud: def test_init_defaults(self): """Test ConfluenceCloud client initialization with default values.""" confluence = ConfluenceCloud(url="https://test.atlassian.net", token="test-token") - assert confluence.api_version == "2" - assert confluence.api_root == "wiki/api/v2" + assert confluence.api_version == "latest" + assert confluence.api_root == "wiki/rest/api" assert confluence.cloud is True + assert confluence.url == "https://test.atlassian.net/wiki/rest/api" def test_init_custom_values(self): """Test ConfluenceCloud client initialization with custom values.""" @@ -51,6 +52,52 @@ def test_get_content_by_type(self, mock_get, confluence_cloud): mock_get.assert_called_once_with("content", params={"type": "page", **{}}) assert result == {"results": [{"id": "123", "title": "Test Page"}]} + @patch.object(ConfluenceCloud, "get") + def test_get_all_pages_from_space(self, mock_get, confluence_cloud): + """Test get_all_pages_from_space method.""" + mock_get.return_value = {"results": [{"id": "123", "title": "Page in Space"}]} + result = confluence_cloud.get_all_pages_from_space("TEST") + assert list(result) == [{"id": "123", "title": "Page in Space"}] + mock_get.assert_called_once_with( + "content", + params={"spaceKey": "TEST", "type": "page", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, + ) + + @patch.object(ConfluenceCloud, "get") + def test_get_all_pages_from_space_pagination(self, mock_get, confluence_cloud): + """Test get_all_pages_from_space paginates correctly.""" + mock_get.side_effect = [ + { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": "https://test.atlassian.net/wiki/api/v2/content?cursor=1"}, + }, + { + "results": [{"id": "2", "title": "Page 2"}], + }, + ] + result = list(confluence_cloud.get_all_pages_from_space("TEST")) + assert result == [{"id": "1", "title": "Page 1"}, {"id": "2", "title": "Page 2"}] + assert mock_get.call_count == 2 + + @patch.object(ConfluenceCloud, "get") + def test_get_all_blog_posts_from_space(self, mock_get, confluence_cloud): + """Test get_all_blog_posts_from_space method.""" + mock_get.return_value = {"results": [{"id": "456", "title": "Blog Post"}]} + result = confluence_cloud.get_all_blog_posts_from_space("TEST") + assert list(result) == [{"id": "456", "title": "Blog Post"}] + mock_get.assert_called_once_with( + "content", + params={"spaceKey": "TEST", "type": "blogpost", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, + ) + @patch.object(ConfluenceCloud, "post") def test_create_content(self, mock_post, confluence_cloud): """Test create_content method.""" @@ -449,3 +496,92 @@ def test_get_health(self, mock_get, confluence_cloud): result = confluence_cloud.get_health() mock_get.assert_called_once_with("health", **{}) assert result == {"status": "healthy"} + + # Pagination Tests for _get_paged (tested directly since Cloud has no paginated public methods yet) + @patch.object(ConfluenceCloud, "get") + def test_pagination_with_next_link_as_string(self, mock_get, confluence_cloud): + """Test multi-page pagination when _links.next is a string URL.""" + mock_get.side_effect = [ + { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": "https://test.atlassian.net/wiki/api/v2/content?cursor=1"}, + }, + { + "results": [{"id": "2", "title": "Page 2"}], + }, + ] + result = list(confluence_cloud._get_paged("content")) + assert result == [{"id": "1", "title": "Page 1"}, {"id": "2", "title": "Page 2"}] + assert mock_get.call_count == 2 + + @patch.object(ConfluenceCloud, "get") + def test_pagination_with_next_link_as_dict(self, mock_get, confluence_cloud): + """Test multi-page pagination when _links.next is a dict with href.""" + mock_get.side_effect = [ + { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": {"href": "https://test.atlassian.net/wiki/api/v2/content?cursor=1"}}, + }, + { + "results": [{"id": "2", "title": "Page 2"}], + }, + ] + result = list(confluence_cloud._get_paged("content")) + assert result == [{"id": "1", "title": "Page 1"}, {"id": "2", "title": "Page 2"}] + assert mock_get.call_count == 2 + + @patch.object(ConfluenceCloud, "get") + def test_pagination_stops_when_next_link_is_none(self, mock_get, confluence_cloud): + """Test pagination stops when _links.next is explicitly None.""" + mock_get.return_value = { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": None}, + } + result = list(confluence_cloud._get_paged("content")) + assert result == [{"id": "1", "title": "Page 1"}] + assert mock_get.call_count == 1 + + @patch.object(ConfluenceCloud, "get") + def test_pagination_stops_when_next_link_dict_missing_href(self, mock_get, confluence_cloud): + """Test pagination stops when _links.next is a dict without href.""" + mock_get.return_value = { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": {}}, + } + result = list(confluence_cloud._get_paged("content")) + assert result == [{"id": "1", "title": "Page 1"}] + assert mock_get.call_count == 1 + + @patch.object(ConfluenceCloud, "get") + def test_pagination_returns_empty_when_no_results_key(self, mock_get, confluence_cloud): + """Test _get_paged returns immediately when response has no results key.""" + mock_get.return_value = {"error": "something went wrong"} + result = list(confluence_cloud._get_paged("content")) + assert result == [] + assert mock_get.call_count == 1 + + @patch.object(ConfluenceCloud, "get") + def test_pagination_with_relative_next_link_and_base(self, mock_get, confluence_cloud): + """Test pagination with relative next link and base URL.""" + mock_get.side_effect = [ + { + "results": [{"id": "1", "title": "Page 1"}], + "_links": { + "next": "/rest/api/content?cursor=1", + "base": "https://test.atlassian.net/wiki", + }, + }, + { + "results": [{"id": "2", "title": "Page 2"}], + }, + ] + result = list(confluence_cloud._get_paged("content")) + + assert result == [{"id": "1", "title": "Page 1"}, {"id": "2", "title": "Page 2"}] + + assert mock_get.call_count == 2 + + # Verify the second call used scheme+host from self.url (preserving API gateway routing) + args, kwargs = mock_get.call_args_list[1] + assert args[0] == "https://test.atlassian.net/rest/api/content?cursor=1" + assert kwargs["absolute"] is True diff --git a/tests/confluence/test_confluence_routing.py b/tests/confluence/test_confluence_routing.py new file mode 100644 index 000000000..aa16abd4b --- /dev/null +++ b/tests/confluence/test_confluence_routing.py @@ -0,0 +1,68 @@ +# coding=utf-8 +"""Tests for legacy Confluence class URL routing.""" + +from unittest.mock import patch, MagicMock + +import pytest + +from atlassian.confluence import Confluence, ConfluenceCloud, ConfluenceServer + + +class TestConfluenceRouting: + """Test URL routing in legacy Confluence wrapper.""" + + @patch.object(ConfluenceCloud, "__init__", return_value=None) + @patch.object(ConfluenceServer, "__init__", return_value=None) + def test_atlassian_net_routes_to_cloud(self, mock_server, mock_cloud): + """Standard atlassian.net URL routes to Cloud.""" + Confluence("https://mysite.atlassian.net") + mock_cloud.assert_called_once() + mock_server.assert_not_called() + + @patch.object(ConfluenceCloud, "__init__", return_value=None) + @patch.object(ConfluenceServer, "__init__", return_value=None) + def test_atlassian_net_wiki_routes_to_cloud(self, mock_server, mock_cloud): + """atlassian.net/wiki URL should still route to Cloud.""" + Confluence("https://mysite.atlassian.net/wiki") + mock_cloud.assert_called_once() + mock_server.assert_not_called() + + @patch.object(ConfluenceCloud, "__init__", return_value=None) + @patch.object(ConfluenceServer, "__init__", return_value=None) + def test_jira_com_routes_to_cloud(self, mock_server, mock_cloud): + """jira.com URL routes to Cloud.""" + Confluence("https://mysite.jira.com") + mock_cloud.assert_called_once() + mock_server.assert_not_called() + + @patch.object(ConfluenceCloud, "__init__", return_value=None) + @patch.object(ConfluenceServer, "__init__", return_value=None) + def test_api_gateway_routes_to_cloud(self, mock_server, mock_cloud): + """OAuth2 API gateway URL routes to Cloud.""" + Confluence("https://api.atlassian.com/ex/confluence/abc123") + mock_cloud.assert_called_once() + mock_server.assert_not_called() + + @patch.object(ConfluenceCloud, "__init__", return_value=None) + @patch.object(ConfluenceServer, "__init__", return_value=None) + def test_self_hosted_routes_to_server(self, mock_server, mock_cloud): + """Self-hosted URL routes to Server.""" + Confluence("https://confluence.mycompany.com") + mock_server.assert_called_once() + mock_cloud.assert_not_called() + + @patch.object(ConfluenceCloud, "__init__", return_value=None) + @patch.object(ConfluenceServer, "__init__", return_value=None) + def test_explicit_cloud_true_overrides_url(self, mock_server, mock_cloud): + """cloud=True forces Cloud routing regardless of URL.""" + Confluence("https://confluence.mycompany.com", cloud=True) + mock_cloud.assert_called_once() + mock_server.assert_not_called() + + @patch.object(ConfluenceCloud, "__init__", return_value=None) + @patch.object(ConfluenceServer, "__init__", return_value=None) + def test_explicit_cloud_false_overrides_url(self, mock_server, mock_cloud): + """cloud=False forces Server routing regardless of URL.""" + Confluence("https://mysite.atlassian.net", cloud=False) + mock_server.assert_called_once() + mock_cloud.assert_not_called() diff --git a/tests/confluence/test_confluence_server.py b/tests/confluence/test_confluence_server.py index 600e6e32b..38ae0a628 100644 --- a/tests/confluence/test_confluence_server.py +++ b/tests/confluence/test_confluence_server.py @@ -125,16 +125,93 @@ def test_get_all_pages_from_space(self, mock_get, confluence_server): """Test get_all_pages_from_space method.""" mock_get.return_value = {"results": [{"id": "123", "title": "Page in Space"}]} result = confluence_server.get_all_pages_from_space("TEST") - mock_get.assert_called_once_with("content", params={"spaceKey": "TEST", "type": "page", **{}}) - assert result == {"results": [{"id": "123", "title": "Page in Space"}]} + assert list(result) == [{"id": "123", "title": "Page in Space"}] + mock_get.assert_called_once_with( + "content", + params={"spaceKey": "TEST", "type": "page", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, + ) + + # Pagination Tests for _get_paged (fix for issue #1598) + @patch.object(ConfluenceServer, "get") + def test_pagination_with_next_link_as_string(self, mock_get, confluence_server): + """Test multi-page pagination when _links.next is a string URL.""" + mock_get.side_effect = [ + { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": "https://test.confluence.com/rest/api/1.0/content?start=1"}, + }, + { + "results": [{"id": "2", "title": "Page 2"}], + }, + ] + result = list(confluence_server.get_all_pages_from_space("TEST")) + assert result == [{"id": "1", "title": "Page 1"}, {"id": "2", "title": "Page 2"}] + assert mock_get.call_count == 2 + + @patch.object(ConfluenceServer, "get") + def test_pagination_with_next_link_as_dict(self, mock_get, confluence_server): + """Test multi-page pagination when _links.next is a dict with href.""" + mock_get.side_effect = [ + { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": {"href": "https://test.confluence.com/rest/api/1.0/content?start=1"}}, + }, + { + "results": [{"id": "2", "title": "Page 2"}], + }, + ] + result = list(confluence_server.get_all_pages_from_space("TEST")) + assert result == [{"id": "1", "title": "Page 1"}, {"id": "2", "title": "Page 2"}] + assert mock_get.call_count == 2 + + @patch.object(ConfluenceServer, "get") + def test_pagination_stops_when_next_link_is_none(self, mock_get, confluence_server): + """Test pagination stops when _links.next is explicitly None.""" + mock_get.return_value = { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": None}, + } + result = list(confluence_server.get_all_pages_from_space("TEST")) + assert result == [{"id": "1", "title": "Page 1"}] + assert mock_get.call_count == 1 + + @patch.object(ConfluenceServer, "get") + def test_pagination_stops_when_next_link_dict_missing_href(self, mock_get, confluence_server): + """Test pagination stops when _links.next is a dict without href.""" + mock_get.return_value = { + "results": [{"id": "1", "title": "Page 1"}], + "_links": {"next": {}}, + } + result = list(confluence_server.get_all_pages_from_space("TEST")) + assert result == [{"id": "1", "title": "Page 1"}] + assert mock_get.call_count == 1 + + @patch.object(ConfluenceServer, "get") + def test_pagination_returns_empty_when_no_results_key(self, mock_get, confluence_server): + """Test _get_paged returns immediately when response has no results key.""" + mock_get.return_value = {"error": "something went wrong"} + result = list(confluence_server.get_all_pages_from_space("TEST")) + assert result == [] + assert mock_get.call_count == 1 @patch.object(ConfluenceServer, "get") def test_get_all_blog_posts_from_space(self, mock_get, confluence_server): """Test get_all_blog_posts_from_space method.""" mock_get.return_value = {"results": [{"id": "456", "title": "Blog Post in Space"}]} result = confluence_server.get_all_blog_posts_from_space("TEST") - mock_get.assert_called_once_with("content", params={"spaceKey": "TEST", "type": "blogpost", **{}}) - assert result == {"results": [{"id": "456", "title": "Blog Post in Space"}]} + assert list(result) == [{"id": "456", "title": "Blog Post in Space"}] + mock_get.assert_called_once_with( + "content", + params={"spaceKey": "TEST", "type": "blogpost", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, + ) @patch.object(ConfluenceServer, "get") def test_get_page_by_title(self, mock_get, confluence_server): @@ -391,16 +468,30 @@ def test_get_all_pages_by_label(self, mock_get, confluence_server): """Test get_all_pages_by_label method.""" mock_get.return_value = {"results": [{"id": "123", "title": "Page with Label"}]} result = confluence_server.get_all_pages_by_label("label1") - mock_get.assert_called_once_with("content", params={"label": "label1", "type": "page", **{}}) - assert result == {"results": [{"id": "123", "title": "Page with Label"}]} + assert list(result) == [{"id": "123", "title": "Page with Label"}] + mock_get.assert_called_once_with( + "content", + params={"label": "label1", "type": "page", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, + ) @patch.object(ConfluenceServer, "get") def test_get_all_blog_posts_by_label(self, mock_get, confluence_server): """Test get_all_blog_posts_by_label method.""" mock_get.return_value = {"results": [{"id": "456", "title": "Blog Post with Label"}]} result = confluence_server.get_all_blog_posts_by_label("label1") - mock_get.assert_called_once_with("content", params={"label": "label1", "type": "blogpost", **{}}) - assert result == {"results": [{"id": "456", "title": "Blog Post with Label"}]} + assert list(result) == [{"id": "456", "title": "Blog Post with Label"}] + mock_get.assert_called_once_with( + "content", + params={"label": "label1", "type": "blogpost", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, + ) # Attachment Management Tests @patch.object(ConfluenceServer, "get") @@ -587,20 +678,30 @@ def test_get_all_draft_pages_from_space(self, mock_get, confluence_server): """Test get_all_draft_pages_from_space method.""" mock_get.return_value = {"results": [{"id": "123", "title": "Draft Page"}]} result = confluence_server.get_all_draft_pages_from_space("TEST") + assert list(result) == [{"id": "123", "title": "Draft Page"}] mock_get.assert_called_once_with( - "content", params={"spaceKey": "TEST", "type": "page", "status": "draft", **{}} + "content", + params={"spaceKey": "TEST", "type": "page", "status": "draft", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, ) - assert result == {"results": [{"id": "123", "title": "Draft Page"}]} @patch.object(ConfluenceServer, "get") def test_get_all_draft_blog_posts_from_space(self, mock_get, confluence_server): """Test get_all_draft_blog_posts_from_space method.""" mock_get.return_value = {"results": [{"id": "456", "title": "Draft Blog Post"}]} result = confluence_server.get_all_draft_blog_posts_from_space("TEST") + assert list(result) == [{"id": "456", "title": "Draft Blog Post"}] mock_get.assert_called_once_with( - "content", params={"spaceKey": "TEST", "type": "blogpost", "status": "draft", **{}} + "content", + params={"spaceKey": "TEST", "type": "blogpost", "status": "draft", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, ) - assert result == {"results": [{"id": "456", "title": "Draft Blog Post"}]} # Trash Management Tests @patch.object(ConfluenceServer, "get") @@ -608,28 +709,45 @@ def test_get_trash_content(self, mock_get, confluence_server): """Test get_trash_content method.""" mock_get.return_value = {"results": [{"id": "123", "title": "Trashed Page"}]} result = confluence_server.get_trash_content("TEST") - mock_get.assert_called_once_with("content", params={"spaceKey": "TEST", "status": "trashed", **{}}) - assert result == {"results": [{"id": "123", "title": "Trashed Page"}]} + assert list(result) == [{"id": "123", "title": "Trashed Page"}] + mock_get.assert_called_once_with( + "content", + params={"spaceKey": "TEST", "status": "trashed", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, + ) @patch.object(ConfluenceServer, "get") def test_get_all_pages_from_space_trash(self, mock_get, confluence_server): """Test get_all_pages_from_space_trash method.""" mock_get.return_value = {"results": [{"id": "123", "title": "Trashed Page"}]} result = confluence_server.get_all_pages_from_space_trash("TEST") + assert list(result) == [{"id": "123", "title": "Trashed Page"}] mock_get.assert_called_once_with( - "content", params={"spaceKey": "TEST", "type": "page", "status": "trashed", **{}} + "content", + params={"spaceKey": "TEST", "type": "page", "status": "trashed", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, ) - assert result == {"results": [{"id": "123", "title": "Trashed Page"}]} @patch.object(ConfluenceServer, "get") def test_get_all_blog_posts_from_space_trash(self, mock_get, confluence_server): """Test get_all_blog_posts_from_space_trash method.""" mock_get.return_value = {"results": [{"id": "456", "title": "Trashed Blog Post"}]} result = confluence_server.get_all_blog_posts_from_space_trash("TEST") + assert list(result) == [{"id": "456", "title": "Trashed Blog Post"}] mock_get.assert_called_once_with( - "content", params={"spaceKey": "TEST", "type": "blogpost", "status": "trashed", **{}} + "content", + params={"spaceKey": "TEST", "type": "blogpost", "status": "trashed", **{}}, + trailing=None, + data=None, + flags=None, + absolute=False, ) - assert result == {"results": [{"id": "456", "title": "Trashed Blog Post"}]} # Export Tests @patch.object(ConfluenceServer, "get")