From 3ff18f3ff1d59f7fe77a6d8b6451cac750e86f3a Mon Sep 17 00:00:00 2001 From: kridai Date: Mon, 1 Dec 2025 19:53:13 +0530 Subject: [PATCH 1/4] feat: add new token based pagination support --- tests/unit/base/test_token_pagination.py | 535 +++++++++++++++++++++++ twilio/base/token_pagination.py | 148 +++++++ 2 files changed, 683 insertions(+) create mode 100644 tests/unit/base/test_token_pagination.py create mode 100644 twilio/base/token_pagination.py diff --git a/tests/unit/base/test_token_pagination.py b/tests/unit/base/test_token_pagination.py new file mode 100644 index 0000000000..1aecd9dc94 --- /dev/null +++ b/tests/unit/base/test_token_pagination.py @@ -0,0 +1,535 @@ +import unittest +from unittest.mock import Mock, AsyncMock, MagicMock +from tests import IntegrationTestCase +from tests.holodeck import Request +from twilio.base.token_pagination import TokenPagination +from twilio.http.response import Response + + +class TestTokenPaginationPage(TokenPagination): + """Test implementation of TokenPagination""" + + def get_instance(self, payload): + return payload + + +class TokenPaginationPropertyTest(unittest.TestCase): + """Test TokenPagination property accessors""" + + def setUp(self): + self.version = Mock() + self.version.domain = Mock() + + # Mock response with token pagination format + self.payload = { + "meta": { + "key": "items", + "pageSize": 50, + "nextToken": "next_abc123", + "previousToken": "prev_xyz789", + }, + "items": [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}], + } + + self.response = Mock() + self.response.text = """ + { + "meta": { + "key": "items", + "pageSize": 50, + "nextToken": "next_abc123", + "previousToken": "prev_xyz789" + }, + "items": [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}] + } + """ + self.response.status_code = 200 + + self.solution = {"account_sid": "ACxxxx", "uri": "/Accounts/ACxxxx/Resources.json"} + self.page = TestTokenPaginationPage(self.version, self.response, self.solution) + + def test_key_property(self): + """Test that key property returns the correct value""" + self.assertEqual(self.page.key, "items") + + def test_page_size_property(self): + """Test that page_size property returns the correct value""" + self.assertEqual(self.page.page_size, 50) + + def test_next_token_property(self): + """Test that next_token property returns the correct value""" + self.assertEqual(self.page.next_token, "next_abc123") + + def test_previous_token_property(self): + """Test that previous_token property returns the correct value""" + self.assertEqual(self.page.previous_token, "prev_xyz789") + + def test_properties_without_meta(self): + """Test that properties return None when meta is missing""" + response = Mock() + response.text = '{"items": []}' + response.status_code = 200 + + page = TestTokenPaginationPage(self.version, response, self.solution) + + self.assertIsNone(page.key) + self.assertIsNone(page.page_size) + self.assertIsNone(page.next_token) + self.assertIsNone(page.previous_token) + + def test_properties_with_partial_meta(self): + """Test that properties return None when specific keys are missing""" + response = Mock() + response.text = '{"meta": {"key": "items"}, "items": []}' + response.status_code = 200 + + page = TestTokenPaginationPage(self.version, response, self.solution) + + self.assertEqual(page.key, "items") + self.assertIsNone(page.page_size) + self.assertIsNone(page.next_token) + self.assertIsNone(page.previous_token) + + +class TokenPaginationNavigationTest(IntegrationTestCase): + """Test TokenPagination next_page and previous_page methods""" + + def setUp(self): + super(TokenPaginationNavigationTest, self).setUp() + + # Mock first page response + self.holodeck.mock( + Response( + 200, + """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": "token_page2", + "previousToken": null + }, + "items": [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}] + } + """, + ), + Request(url="https://api.twilio.com/Accounts/ACaaaa/Resources.json"), + ) + + # Mock second page response (next page) + self.holodeck.mock( + Response( + 200, + """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": "token_page3", + "previousToken": "token_page2" + }, + "items": [{"id": 3, "name": "Item 3"}, {"id": 4, "name": "Item 4"}] + } + """, + ), + Request( + url="https://api.twilio.com/Accounts/ACaaaa/Resources.json?pageToken=token_page2" + ), + ) + + # Mock third page response (has no next) + self.holodeck.mock( + Response( + 200, + """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": null, + "previousToken": "token_page3" + }, + "items": [{"id": 5, "name": "Item 5"}] + } + """, + ), + Request( + url="https://api.twilio.com/Accounts/ACaaaa/Resources.json?pageToken=token_page3" + ), + ) + + # Mock previous page response + self.holodeck.mock( + Response( + 200, + """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": "token_page2", + "previousToken": null + }, + "items": [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}] + } + """, + ), + Request( + url="https://api.twilio.com/Accounts/ACaaaa/Resources.json?pageToken=token_page2" + ), + ) + + self.version = self.client.api.v2010 + self.response = self.version.page( + method="GET", uri="/Accounts/ACaaaa/Resources.json" + ) + + self.solution = { + "account_sid": "ACaaaa", + "uri": "/Accounts/ACaaaa/Resources.json", + } + + self.page = TestTokenPaginationPage(self.version, self.response, self.solution) + + def test_next_page(self): + """Test that next_page() navigates to the next page using token""" + self.assertIsNotNone(self.page.next_token) + self.assertEqual(self.page.next_token, "token_page2") + + next_page = self.page.next_page() + + self.assertIsNotNone(next_page) + self.assertIsInstance(next_page, TestTokenPaginationPage) + # Verify we got the next page's data + self.assertEqual(next_page.next_token, "token_page3") + self.assertEqual(next_page.previous_token, "token_page2") + + def test_next_page_none_when_no_token(self): + """Test that next_page() returns None when there's no next token""" + # Navigate to the last page + next_page = self.page.next_page() + last_page = next_page.next_page() + + # Last page should have no next token + self.assertIsNone(last_page.next_token) + + # next_page() should return None + result = last_page.next_page() + self.assertIsNone(result) + + def test_previous_page(self): + """Test that previous_page() navigates to the previous page using token""" + # Navigate to second page first + next_page = self.page.next_page() + self.assertIsNotNone(next_page.previous_token) + self.assertEqual(next_page.previous_token, "token_page2") + + # Go back to previous page + prev_page = next_page.previous_page() + + self.assertIsNotNone(prev_page) + self.assertIsInstance(prev_page, TestTokenPaginationPage) + # Verify we got the first page's data + self.assertIsNone(prev_page.previous_token) + self.assertEqual(prev_page.next_token, "token_page2") + + def test_previous_page_none_when_no_token(self): + """Test that previous_page() returns None when there's no previous token""" + # First page should have no previous token + self.assertIsNone(self.page.previous_token) + + # previous_page() should return None + result = self.page.previous_page() + self.assertIsNone(result) + + def test_navigation_chain(self): + """Test navigating through multiple pages forward and backward""" + # Page 1 -> Page 2 + page2 = self.page.next_page() + self.assertEqual(page2.previous_token, "token_page2") + + # Page 2 -> Page 3 + page3 = page2.next_page() + self.assertIsNone(page3.next_token) + self.assertEqual(page3.previous_token, "token_page3") + + # Page 3 -> Page 2 (backward) + back_to_page2 = page3.previous_page() + self.assertIsNotNone(back_to_page2) + + +class TokenPaginationErrorTest(unittest.TestCase): + """Test TokenPagination error handling""" + + def test_next_page_without_uri_in_solution(self): + """Test that next_page() raises error when URI is missing from solution""" + version = Mock() + response = Mock() + response.text = """ + { + "meta": { + "key": "items", + "pageSize": 50, + "nextToken": "abc123" + }, + "items": [] + } + """ + response.status_code = 200 + + # Solution without URI + solution = {"account_sid": "ACxxxx"} + + page = TestTokenPaginationPage(version, response, solution) + + with self.assertRaises(ValueError) as context: + page.next_page() + + self.assertIn("URI must be provided", str(context.exception)) + + def test_previous_page_without_uri_in_solution(self): + """Test that previous_page() raises error when URI is missing from solution""" + version = Mock() + response = Mock() + response.text = """ + { + "meta": { + "key": "items", + "pageSize": 50, + "previousToken": "xyz789" + }, + "items": [] + } + """ + response.status_code = 200 + + # Solution without URI + solution = {"account_sid": "ACxxxx"} + + page = TestTokenPaginationPage(version, response, solution) + + with self.assertRaises(ValueError) as context: + page.previous_page() + + self.assertIn("URI must be provided", str(context.exception)) + + +class TokenPaginationAsyncTest(unittest.TestCase): + """Test TokenPagination async methods""" + + async def test_next_page_async(self): + """Test that next_page_async() works correctly""" + version = Mock() + version.domain = Mock() + version.domain.absolute_url = Mock( + return_value="https://api.twilio.com/Accounts/ACxxxx/Resources.json" + ) + version.domain.twilio = Mock() + + # Mock async request + future_response = Mock() + future_response.text = """ + { + "meta": { + "key": "items", + "pageSize": 50, + "nextToken": "next_token_2", + "previousToken": "prev_token_1" + }, + "items": [{"id": 3}] + } + """ + future_response.status_code = 200 + + version.domain.twilio.request_async = AsyncMock(return_value=future_response) + + response = Mock() + response.text = """ + { + "meta": { + "key": "items", + "pageSize": 50, + "nextToken": "token_next", + "previousToken": null + }, + "items": [{"id": 1}] + } + """ + response.status_code = 200 + + solution = {"account_sid": "ACxxxx", "uri": "/Accounts/ACxxxx/Resources.json"} + page = TestTokenPaginationPage(version, response, solution) + + next_page = await page.next_page_async() + + self.assertIsNotNone(next_page) + version.domain.twilio.request_async.assert_called_once() + + async def test_previous_page_async(self): + """Test that previous_page_async() works correctly""" + version = Mock() + version.domain = Mock() + version.domain.absolute_url = Mock( + return_value="https://api.twilio.com/Accounts/ACxxxx/Resources.json" + ) + version.domain.twilio = Mock() + + # Mock async request + future_response = Mock() + future_response.text = """ + { + "meta": { + "key": "items", + "pageSize": 50, + "nextToken": null, + "previousToken": null + }, + "items": [{"id": 1}] + } + """ + future_response.status_code = 200 + + version.domain.twilio.request_async = AsyncMock(return_value=future_response) + + response = Mock() + response.text = """ + { + "meta": { + "key": "items", + "pageSize": 50, + "nextToken": "token_next", + "previousToken": "token_prev" + }, + "items": [{"id": 2}] + } + """ + response.status_code = 200 + + solution = {"account_sid": "ACxxxx", "uri": "/Accounts/ACxxxx/Resources.json"} + page = TestTokenPaginationPage(version, response, solution) + + prev_page = await page.previous_page_async() + + self.assertIsNotNone(prev_page) + version.domain.twilio.request_async.assert_called_once() + + +class TokenPaginationStreamTest(IntegrationTestCase): + """Test streaming with TokenPagination""" + + def setUp(self): + super(TokenPaginationStreamTest, self).setUp() + + # Mock page 1 + self.holodeck.mock( + Response( + 200, + """ + { + "meta": { + "key": "records", + "pageSize": 2, + "nextToken": "token_2" + }, + "records": [{"id": 1}, {"id": 2}] + } + """, + ), + Request(url="https://api.twilio.com/Accounts/ACaaaa/Records.json"), + ) + + # Mock page 2 + self.holodeck.mock( + Response( + 200, + """ + { + "meta": { + "key": "records", + "pageSize": 2, + "nextToken": "token_3", + "previousToken": "token_2" + }, + "records": [{"id": 3}, {"id": 4}] + } + """, + ), + Request( + url="https://api.twilio.com/Accounts/ACaaaa/Records.json?pageToken=token_2" + ), + ) + + # Mock page 3 (final) + self.holodeck.mock( + Response( + 200, + """ + { + "meta": { + "key": "records", + "pageSize": 2, + "nextToken": null, + "previousToken": "token_3" + }, + "records": [{"id": 5}] + } + """, + ), + Request( + url="https://api.twilio.com/Accounts/ACaaaa/Records.json?pageToken=token_3" + ), + ) + + self.version = self.client.api.v2010 + self.response = self.version.page(method="GET", uri="/Accounts/ACaaaa/Records.json") + + self.solution = { + "account_sid": "ACaaaa", + "uri": "/Accounts/ACaaaa/Records.json", + } + + self.page = TestTokenPaginationPage(self.version, self.response, self.solution) + + def test_stream_all_records(self): + """Test streaming through all pages""" + records = list(self.version.stream(self.page)) + + self.assertEqual(len(records), 5) + self.assertEqual(records[0]["id"], 1) + self.assertEqual(records[4]["id"], 5) + + def test_stream_with_limit(self): + """Test streaming with a limit""" + records = list(self.version.stream(self.page, limit=3)) + + self.assertEqual(len(records), 3) + self.assertEqual(records[0]["id"], 1) + self.assertEqual(records[2]["id"], 3) + + def test_stream_with_page_limit(self): + """Test streaming with page limit""" + records = list(self.version.stream(self.page, page_limit=1)) + + # Only first page (2 records) + self.assertEqual(len(records), 2) + + +class TokenPaginationReprTest(unittest.TestCase): + """Test TokenPagination string representation""" + + def test_repr(self): + """Test __repr__ method""" + version = Mock() + response = Mock() + response.text = '{"meta": {"key": "items"}, "items": []}' + response.status_code = 200 + + solution = {"account_sid": "ACxxxx"} + page = TestTokenPaginationPage(version, response, solution) + + repr_str = repr(page) + self.assertEqual(repr_str, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/twilio/base/token_pagination.py b/twilio/base/token_pagination.py new file mode 100644 index 0000000000..a67c7f15b9 --- /dev/null +++ b/twilio/base/token_pagination.py @@ -0,0 +1,148 @@ +from typing import Any, Dict, Optional + +from twilio.base.exceptions import TwilioException +from twilio.base.page import Page + + +class TokenPagination(Page): + """ + Represents a page of records using token-based pagination. + + Token-based pagination uses next_token and previous_token instead of + page numbers to navigate through results. + + Example expected response format with token metadata: + { + "meta": { + "key": "items", + "pageSize": 50, + "nextToken": "abc123", + "previousToken": "xyz789" + }, + "items": [ + { "id": 1, "name": "Item 1" }, + { "id": 2, "name": "Item 2" } + ] + } + """ + + @property + def key(self) -> Optional[str]: + """ + :return str: Returns the key that identifies the collection in the response. + """ + if "meta" in self._payload and "key" in self._payload["meta"]: + return self._payload["meta"]["key"] + return None + + @property + def page_size(self) -> Optional[int]: + """ + :return int: Returns the page size or None if doesn't exist. + """ + if "meta" in self._payload and "pageSize" in self._payload["meta"]: + return self._payload["meta"]["pageSize"] + return None + + @property + def next_token(self) -> Optional[str]: + """ + :return str: Returns the next_token for pagination or None if doesn't exist. + """ + if "meta" in self._payload and "nextToken" in self._payload["meta"]: + return self._payload["meta"]["nextToken"] + return None + + @property + def previous_token(self) -> Optional[str]: + """ + :return str: Returns the previous_token for pagination or None if doesn't exist. + """ + if "meta" in self._payload and "previousToken" in self._payload["meta"]: + return self._payload["meta"]["previousToken"] + return None + + def _get_page(self, token: Optional[str]) -> Optional["TokenPagination"]: + """ + Internal helper to fetch a page using a given token. + + :param token: The pagination token to use. + :return: The page or None if no token exists. + """ + if not token: + return None + + params = self._solution.copy() + params["pageToken"] = token + + # Get the URI from solution and build the absolute URL + uri = self._solution.get("uri") + if not uri: + raise TwilioException("URI must be provided for token pagination") + + url = self._version.domain.absolute_url(uri) + response = self._version.domain.twilio.request("GET", url, params=params) + cls = type(self) + return cls(self._version, response, self._solution) + + async def _get_page_async(self, token: Optional[str]) -> Optional["TokenPagination"]: + """ + Internal async helper to fetch a page using a given token. + + :param token: The pagination token to use. + :return: The page or None if no token exists. + """ + if not token: + return None + + params = self._solution.copy() + params["pageToken"] = token + + # Get the URI from solution and build the absolute URL + uri = self._solution.get("uri") + if not uri: + raise TwilioException("URI must be provided for token pagination") + + url = self._version.domain.absolute_url(uri) + response = await self._version.domain.twilio.request_async("GET", url, params=params) + cls = type(self) + return cls(self._version, response, self._solution) + + def next_page(self) -> Optional["TokenPagination"]: + """ + Return the next page using token-based pagination. + Makes a request to the same URI with pageToken set to nextToken. + + :return: The next page or None if no next token exists. + """ + return self._get_page(self.next_token) + + async def next_page_async(self) -> Optional["TokenPagination"]: + """ + Asynchronously return the next page using token-based pagination. + Makes a request to the same URI with pageToken set to nextToken. + + :return: The next page or None if no next token exists. + """ + return await self._get_page_async(self.next_token) + + def previous_page(self) -> Optional["TokenPagination"]: + """ + Return the previous page using token-based pagination. + Makes a request to the same URI with pageToken set to previousToken. + + :return: The previous page or None if no previous token exists. + """ + return self._get_page(self.previous_token) + + async def previous_page_async(self) -> Optional["TokenPagination"]: + """ + Asynchronously return the previous page using token-based pagination. + Makes a request to the same URI with pageToken set to previousToken. + + :return: The previous page or None if no previous token exists. + """ + return await self._get_page_async(self.previous_token) + + def __repr__(self) -> str: + return "" From 2528b346ae485b0b9d6d330fb4dfbc11d2e65138 Mon Sep 17 00:00:00 2001 From: kridai Date: Mon, 1 Dec 2025 19:57:47 +0530 Subject: [PATCH 2/4] feat: add new token based pagination support --- tests/unit/base/test_token_pagination.py | 16 +++++++++++----- tests/unit/rest/test_client.py | 19 ++++++++++++++----- twilio/base/client_base.py | 16 ++++++++++++---- twilio/base/token_pagination.py | 10 +++++++--- 4 files changed, 44 insertions(+), 17 deletions(-) diff --git a/tests/unit/base/test_token_pagination.py b/tests/unit/base/test_token_pagination.py index 1aecd9dc94..6838f3f3fb 100644 --- a/tests/unit/base/test_token_pagination.py +++ b/tests/unit/base/test_token_pagination.py @@ -1,7 +1,8 @@ import unittest -from unittest.mock import Mock, AsyncMock, MagicMock +from unittest.mock import Mock, AsyncMock from tests import IntegrationTestCase from tests.holodeck import Request +from twilio.base.exceptions import TwilioException from twilio.base.token_pagination import TokenPagination from twilio.http.response import Response @@ -45,7 +46,10 @@ def setUp(self): """ self.response.status_code = 200 - self.solution = {"account_sid": "ACxxxx", "uri": "/Accounts/ACxxxx/Resources.json"} + self.solution = { + "account_sid": "ACxxxx", + "uri": "/Accounts/ACxxxx/Resources.json", + } self.page = TestTokenPaginationPage(self.version, self.response, self.solution) def test_key_property(self): @@ -282,7 +286,7 @@ def test_next_page_without_uri_in_solution(self): page = TestTokenPaginationPage(version, response, solution) - with self.assertRaises(ValueError) as context: + with self.assertRaises(TwilioException) as context: page.next_page() self.assertIn("URI must be provided", str(context.exception)) @@ -308,7 +312,7 @@ def test_previous_page_without_uri_in_solution(self): page = TestTokenPaginationPage(version, response, solution) - with self.assertRaises(ValueError) as context: + with self.assertRaises(TwilioException) as context: page.previous_page() self.assertIn("URI must be provided", str(context.exception)) @@ -481,7 +485,9 @@ def setUp(self): ) self.version = self.client.api.v2010 - self.response = self.version.page(method="GET", uri="/Accounts/ACaaaa/Records.json") + self.response = self.version.page( + method="GET", uri="/Accounts/ACaaaa/Records.json" + ) self.solution = { "account_sid": "ACaaaa", diff --git a/tests/unit/rest/test_client.py b/tests/unit/rest/test_client.py index 1043d6dcf8..4bf8f7ed00 100644 --- a/tests/unit/rest/test_client.py +++ b/tests/unit/rest/test_client.py @@ -81,23 +81,32 @@ def test_periods_in_query(self): def test_edge_deprecation_warning_when_only_edge_is_set(self): with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") # Ensure all warnings are caught - Client(username="username", password="password", edge="edge") # Trigger the warning + Client( + username="username", password="password", edge="edge" + ) # Trigger the warning # Check if a warning was raised self.assertGreater(len(w), 0) self.assertTrue(issubclass(w[-1].category, DeprecationWarning)) - self.assertIn("For regional processing, DNS is of format product...twilio.com; otherwise use product.twilio.com.", str(w[-1].message)) + self.assertIn( + "For regional processing, DNS is of format product...twilio.com; otherwise use product.twilio.com.", + str(w[-1].message), + ) def test_edge_deprecation_warning_when_only_region_is_set(self): with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") # Ensure all warnings are caught - Client(username="username", password="password", region="us1") # Trigger the warning + Client( + username="username", password="password", region="us1" + ) # Trigger the warning # Check if a warning was raised self.assertGreater(len(w), 0) self.assertTrue(issubclass(w[-1].category, DeprecationWarning)) - self.assertIn("For regional processing, DNS is of format product...twilio.com; otherwise use product.twilio.com.", str(w[-1].message)) - + self.assertIn( + "For regional processing, DNS is of format product...twilio.com; otherwise use product.twilio.com.", + str(w[-1].message), + ) class TestUserAgentClients(unittest.TestCase): diff --git a/twilio/base/client_base.py b/twilio/base/client_base.py index 2fac31b80a..9e4b030985 100644 --- a/twilio/base/client_base.py +++ b/twilio/base/client_base.py @@ -13,6 +13,7 @@ class ClientBase(object): """A client for accessing the Twilio API.""" + region_mappings = { "au1": "sydney", "br1": "sao-paulo", @@ -22,8 +23,9 @@ class ClientBase(object): "jp2": "osaka", "sg1": "singapore", "us1": "ashburn", - "us2": "umatilla" + "us2": "umatilla", } + def __init__( self, username: Optional[str] = None, @@ -56,13 +58,19 @@ def __init__( """ :type : str """ self.password = password or environment.get("TWILIO_AUTH_TOKEN") """ :type : str """ - if (edge is not None and region is None) or (region is not None and edge is None): + if (edge is not None and region is None) or ( + region is not None and edge is None + ): warnings.warn( "For regional processing, DNS is of format product...twilio.com; otherwise use product.twilio.com.", DeprecationWarning, - stacklevel=2 + stacklevel=2, ) - self.edge = edge or environment.get("TWILIO_EDGE") or (self.region_mappings[region] if region is not None else "") + self.edge = ( + edge + or environment.get("TWILIO_EDGE") + or (self.region_mappings[region] if region is not None else "") + ) """ :type : str """ self.region = region or environment.get("TWILIO_REGION") """ :type : str """ diff --git a/twilio/base/token_pagination.py b/twilio/base/token_pagination.py index a67c7f15b9..7597bc5720 100644 --- a/twilio/base/token_pagination.py +++ b/twilio/base/token_pagination.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, Optional +from typing import Optional from twilio.base.exceptions import TwilioException from twilio.base.page import Page @@ -85,7 +85,9 @@ def _get_page(self, token: Optional[str]) -> Optional["TokenPagination"]: cls = type(self) return cls(self._version, response, self._solution) - async def _get_page_async(self, token: Optional[str]) -> Optional["TokenPagination"]: + async def _get_page_async( + self, token: Optional[str] + ) -> Optional["TokenPagination"]: """ Internal async helper to fetch a page using a given token. @@ -104,7 +106,9 @@ async def _get_page_async(self, token: Optional[str]) -> Optional["TokenPaginati raise TwilioException("URI must be provided for token pagination") url = self._version.domain.absolute_url(uri) - response = await self._version.domain.twilio.request_async("GET", url, params=params) + response = await self._version.domain.twilio.request_async( + "GET", url, params=params + ) cls = type(self) return cls(self._version, response, self._solution) From eae41ef6917cc05cfc91965e8e936e98c8dfccce Mon Sep 17 00:00:00 2001 From: kridai Date: Tue, 2 Dec 2025 15:02:42 +0530 Subject: [PATCH 3/4] chore: add token pagination --- tests/unit/base/test_token_pagination.py | 238 +++++++---------------- twilio/base/token_pagination.py | 36 ++-- 2 files changed, 86 insertions(+), 188 deletions(-) diff --git a/tests/unit/base/test_token_pagination.py b/tests/unit/base/test_token_pagination.py index 6838f3f3fb..cc7cc44678 100644 --- a/tests/unit/base/test_token_pagination.py +++ b/tests/unit/base/test_token_pagination.py @@ -1,5 +1,5 @@ import unittest -from unittest.mock import Mock, AsyncMock +from unittest.mock import Mock from tests import IntegrationTestCase from tests.holodeck import Request from twilio.base.exceptions import TwilioException @@ -48,9 +48,14 @@ def setUp(self): self.solution = { "account_sid": "ACxxxx", - "uri": "/Accounts/ACxxxx/Resources.json", + "api_uri": "/Accounts/ACxxxx/Resources.json", } - self.page = TestTokenPaginationPage(self.version, self.response, self.solution) + self.page = TestTokenPaginationPage( + self.version, + self.response, + "/Accounts/ACxxxx/Resources.json", + self.solution, + ) def test_key_property(self): """Test that key property returns the correct value""" @@ -74,7 +79,9 @@ def test_properties_without_meta(self): response.text = '{"items": []}' response.status_code = 200 - page = TestTokenPaginationPage(self.version, response, self.solution) + page = TestTokenPaginationPage( + self.version, response, "/Accounts/ACxxxx/Resources.json", self.solution + ) self.assertIsNone(page.key) self.assertIsNone(page.page_size) @@ -87,7 +94,9 @@ def test_properties_with_partial_meta(self): response.text = '{"meta": {"key": "items"}, "items": []}' response.status_code = 200 - page = TestTokenPaginationPage(self.version, response, self.solution) + page = TestTokenPaginationPage( + self.version, response, "/Accounts/ACxxxx/Resources.json", self.solution + ) self.assertEqual(page.key, "items") self.assertIsNone(page.page_size) @@ -117,7 +126,9 @@ def setUp(self): } """, ), - Request(url="https://api.twilio.com/Accounts/ACaaaa/Resources.json"), + Request( + url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json" + ), ) # Mock second page response (next page) @@ -130,14 +141,14 @@ def setUp(self): "key": "items", "pageSize": 2, "nextToken": "token_page3", - "previousToken": "token_page2" + "previousToken": "token_prev1" }, "items": [{"id": 3, "name": "Item 3"}, {"id": 4, "name": "Item 4"}] } """, ), Request( - url="https://api.twilio.com/Accounts/ACaaaa/Resources.json?pageToken=token_page2" + url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json?pageToken=token_page2" ), ) @@ -151,18 +162,18 @@ def setUp(self): "key": "items", "pageSize": 2, "nextToken": null, - "previousToken": "token_page3" + "previousToken": "token_prev2" }, "items": [{"id": 5, "name": "Item 5"}] } """, ), Request( - url="https://api.twilio.com/Accounts/ACaaaa/Resources.json?pageToken=token_page3" + url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json?pageToken=token_page3" ), ) - # Mock previous page response + # Mock previous page response (going back to page 1) self.holodeck.mock( Response( 200, @@ -179,7 +190,28 @@ def setUp(self): """, ), Request( - url="https://api.twilio.com/Accounts/ACaaaa/Resources.json?pageToken=token_page2" + url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json?pageToken=token_prev1" + ), + ) + + # Mock going back from page 3 to page 2 + self.holodeck.mock( + Response( + 200, + """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": "token_page3", + "previousToken": "token_prev1" + }, + "items": [{"id": 3, "name": "Item 3"}, {"id": 4, "name": "Item 4"}] + } + """, + ), + Request( + url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Resources.json?pageToken=token_prev2" ), ) @@ -190,10 +222,15 @@ def setUp(self): self.solution = { "account_sid": "ACaaaa", - "uri": "/Accounts/ACaaaa/Resources.json", + "api_uri": "/2010-04-01/Accounts/ACaaaa/Resources.json", } - self.page = TestTokenPaginationPage(self.version, self.response, self.solution) + self.page = TestTokenPaginationPage( + self.version, + self.response, + "/2010-04-01/Accounts/ACaaaa/Resources.json", + self.solution, + ) def test_next_page(self): """Test that next_page() navigates to the next page using token""" @@ -206,7 +243,7 @@ def test_next_page(self): self.assertIsInstance(next_page, TestTokenPaginationPage) # Verify we got the next page's data self.assertEqual(next_page.next_token, "token_page3") - self.assertEqual(next_page.previous_token, "token_page2") + self.assertEqual(next_page.previous_token, "token_prev1") def test_next_page_none_when_no_token(self): """Test that next_page() returns None when there's no next token""" @@ -226,7 +263,7 @@ def test_previous_page(self): # Navigate to second page first next_page = self.page.next_page() self.assertIsNotNone(next_page.previous_token) - self.assertEqual(next_page.previous_token, "token_page2") + self.assertEqual(next_page.previous_token, "token_prev1") # Go back to previous page prev_page = next_page.previous_page() @@ -250,12 +287,12 @@ def test_navigation_chain(self): """Test navigating through multiple pages forward and backward""" # Page 1 -> Page 2 page2 = self.page.next_page() - self.assertEqual(page2.previous_token, "token_page2") + self.assertEqual(page2.previous_token, "token_prev1") # Page 2 -> Page 3 page3 = page2.next_page() self.assertIsNone(page3.next_token) - self.assertEqual(page3.previous_token, "token_page3") + self.assertEqual(page3.previous_token, "token_prev2") # Page 3 -> Page 2 (backward) back_to_page2 = page3.previous_page() @@ -266,7 +303,7 @@ class TokenPaginationErrorTest(unittest.TestCase): """Test TokenPagination error handling""" def test_next_page_without_uri_in_solution(self): - """Test that next_page() raises error when URI is missing from solution""" + """Test that next_page() raises error when URI is missing""" version = Mock() response = Mock() response.text = """ @@ -284,139 +321,14 @@ def test_next_page_without_uri_in_solution(self): # Solution without URI solution = {"account_sid": "ACxxxx"} - page = TestTokenPaginationPage(version, response, solution) + # Pass empty string as URI to test the error case + page = TestTokenPaginationPage(version, response, "", solution) with self.assertRaises(TwilioException) as context: page.next_page() self.assertIn("URI must be provided", str(context.exception)) - def test_previous_page_without_uri_in_solution(self): - """Test that previous_page() raises error when URI is missing from solution""" - version = Mock() - response = Mock() - response.text = """ - { - "meta": { - "key": "items", - "pageSize": 50, - "previousToken": "xyz789" - }, - "items": [] - } - """ - response.status_code = 200 - - # Solution without URI - solution = {"account_sid": "ACxxxx"} - - page = TestTokenPaginationPage(version, response, solution) - - with self.assertRaises(TwilioException) as context: - page.previous_page() - - self.assertIn("URI must be provided", str(context.exception)) - - -class TokenPaginationAsyncTest(unittest.TestCase): - """Test TokenPagination async methods""" - - async def test_next_page_async(self): - """Test that next_page_async() works correctly""" - version = Mock() - version.domain = Mock() - version.domain.absolute_url = Mock( - return_value="https://api.twilio.com/Accounts/ACxxxx/Resources.json" - ) - version.domain.twilio = Mock() - - # Mock async request - future_response = Mock() - future_response.text = """ - { - "meta": { - "key": "items", - "pageSize": 50, - "nextToken": "next_token_2", - "previousToken": "prev_token_1" - }, - "items": [{"id": 3}] - } - """ - future_response.status_code = 200 - - version.domain.twilio.request_async = AsyncMock(return_value=future_response) - - response = Mock() - response.text = """ - { - "meta": { - "key": "items", - "pageSize": 50, - "nextToken": "token_next", - "previousToken": null - }, - "items": [{"id": 1}] - } - """ - response.status_code = 200 - - solution = {"account_sid": "ACxxxx", "uri": "/Accounts/ACxxxx/Resources.json"} - page = TestTokenPaginationPage(version, response, solution) - - next_page = await page.next_page_async() - - self.assertIsNotNone(next_page) - version.domain.twilio.request_async.assert_called_once() - - async def test_previous_page_async(self): - """Test that previous_page_async() works correctly""" - version = Mock() - version.domain = Mock() - version.domain.absolute_url = Mock( - return_value="https://api.twilio.com/Accounts/ACxxxx/Resources.json" - ) - version.domain.twilio = Mock() - - # Mock async request - future_response = Mock() - future_response.text = """ - { - "meta": { - "key": "items", - "pageSize": 50, - "nextToken": null, - "previousToken": null - }, - "items": [{"id": 1}] - } - """ - future_response.status_code = 200 - - version.domain.twilio.request_async = AsyncMock(return_value=future_response) - - response = Mock() - response.text = """ - { - "meta": { - "key": "items", - "pageSize": 50, - "nextToken": "token_next", - "previousToken": "token_prev" - }, - "items": [{"id": 2}] - } - """ - response.status_code = 200 - - solution = {"account_sid": "ACxxxx", "uri": "/Accounts/ACxxxx/Resources.json"} - page = TestTokenPaginationPage(version, response, solution) - - prev_page = await page.previous_page_async() - - self.assertIsNotNone(prev_page) - version.domain.twilio.request_async.assert_called_once() - class TokenPaginationStreamTest(IntegrationTestCase): """Test streaming with TokenPagination""" @@ -439,7 +351,9 @@ def setUp(self): } """, ), - Request(url="https://api.twilio.com/Accounts/ACaaaa/Records.json"), + Request( + url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Records.json" + ), ) # Mock page 2 @@ -459,7 +373,7 @@ def setUp(self): """, ), Request( - url="https://api.twilio.com/Accounts/ACaaaa/Records.json?pageToken=token_2" + url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Records.json?pageToken=token_2" ), ) @@ -480,7 +394,7 @@ def setUp(self): """, ), Request( - url="https://api.twilio.com/Accounts/ACaaaa/Records.json?pageToken=token_3" + url="https://api.twilio.com/2010-04-01/Accounts/ACaaaa/Records.json?pageToken=token_3" ), ) @@ -491,10 +405,15 @@ def setUp(self): self.solution = { "account_sid": "ACaaaa", - "uri": "/Accounts/ACaaaa/Records.json", + "api_uri": "/2010-04-01/Accounts/ACaaaa/Records.json", } - self.page = TestTokenPaginationPage(self.version, self.response, self.solution) + self.page = TestTokenPaginationPage( + self.version, + self.response, + "/2010-04-01/Accounts/ACaaaa/Records.json", + self.solution, + ) def test_stream_all_records(self): """Test streaming through all pages""" @@ -520,22 +439,5 @@ def test_stream_with_page_limit(self): self.assertEqual(len(records), 2) -class TokenPaginationReprTest(unittest.TestCase): - """Test TokenPagination string representation""" - - def test_repr(self): - """Test __repr__ method""" - version = Mock() - response = Mock() - response.text = '{"meta": {"key": "items"}, "items": []}' - response.status_code = 200 - - solution = {"account_sid": "ACxxxx"} - page = TestTokenPaginationPage(version, response, solution) - - repr_str = repr(page) - self.assertEqual(repr_str, "") - - if __name__ == "__main__": unittest.main() diff --git a/twilio/base/token_pagination.py b/twilio/base/token_pagination.py index 7597bc5720..7e5007e1e0 100644 --- a/twilio/base/token_pagination.py +++ b/twilio/base/token_pagination.py @@ -26,6 +26,10 @@ class TokenPagination(Page): } """ + def __init__(self, version, response, uri: str, solution={}): + super().__init__(version, response, solution) + self._uri = uri + @property def key(self) -> Optional[str]: """ @@ -72,18 +76,15 @@ def _get_page(self, token: Optional[str]) -> Optional["TokenPagination"]: if not token: return None - params = self._solution.copy() - params["pageToken"] = token - - # Get the URI from solution and build the absolute URL - uri = self._solution.get("uri") - if not uri: + if not self._uri: raise TwilioException("URI must be provided for token pagination") - url = self._version.domain.absolute_url(uri) - response = self._version.domain.twilio.request("GET", url, params=params) + # Construct full URL with pageToken parameter + uri_with_token = f"{self._uri}?pageToken={token}" + url = self._version.domain.absolute_url(uri_with_token) + response = self._version.domain.twilio.request("GET", url) cls = type(self) - return cls(self._version, response, self._solution) + return cls(self._version, response, self._uri, self._solution) async def _get_page_async( self, token: Optional[str] @@ -97,20 +98,15 @@ async def _get_page_async( if not token: return None - params = self._solution.copy() - params["pageToken"] = token - - # Get the URI from solution and build the absolute URL - uri = self._solution.get("uri") - if not uri: + if not self._uri: raise TwilioException("URI must be provided for token pagination") - url = self._version.domain.absolute_url(uri) - response = await self._version.domain.twilio.request_async( - "GET", url, params=params - ) + # Construct full URL with pageToken parameter + uri_with_token = f"{self._uri}?pageToken={token}" + url = self._version.domain.absolute_url(uri_with_token) + response = await self._version.domain.twilio.request_async("GET", url) cls = type(self) - return cls(self._version, response, self._solution) + return cls(self._version, response, self._uri, self._solution) def next_page(self) -> Optional["TokenPagination"]: """ From ef0435c352cde0120ba643df52778c1bbeeb81fb Mon Sep 17 00:00:00 2001 From: kridai Date: Thu, 4 Dec 2025 11:13:02 +0530 Subject: [PATCH 4/4] chore: add token pagination test --- tests/unit/base/test_token_pagination.py | 270 ++++++++++++++++++++++- 1 file changed, 259 insertions(+), 11 deletions(-) diff --git a/tests/unit/base/test_token_pagination.py b/tests/unit/base/test_token_pagination.py index cc7cc44678..7920c75a78 100644 --- a/tests/unit/base/test_token_pagination.py +++ b/tests/unit/base/test_token_pagination.py @@ -1,5 +1,5 @@ import unittest -from unittest.mock import Mock +from unittest.mock import Mock, AsyncMock from tests import IntegrationTestCase from tests.holodeck import Request from twilio.base.exceptions import TwilioException @@ -7,8 +7,8 @@ from twilio.http.response import Response -class TestTokenPaginationPage(TokenPagination): - """Test implementation of TokenPagination""" +class MockTokenPaginationPage(TokenPagination): + """Mock implementation of TokenPagination for testing""" def get_instance(self, payload): return payload @@ -50,7 +50,7 @@ def setUp(self): "account_sid": "ACxxxx", "api_uri": "/Accounts/ACxxxx/Resources.json", } - self.page = TestTokenPaginationPage( + self.page = MockTokenPaginationPage( self.version, self.response, "/Accounts/ACxxxx/Resources.json", @@ -79,7 +79,7 @@ def test_properties_without_meta(self): response.text = '{"items": []}' response.status_code = 200 - page = TestTokenPaginationPage( + page = MockTokenPaginationPage( self.version, response, "/Accounts/ACxxxx/Resources.json", self.solution ) @@ -94,7 +94,7 @@ def test_properties_with_partial_meta(self): response.text = '{"meta": {"key": "items"}, "items": []}' response.status_code = 200 - page = TestTokenPaginationPage( + page = MockTokenPaginationPage( self.version, response, "/Accounts/ACxxxx/Resources.json", self.solution ) @@ -225,7 +225,7 @@ def setUp(self): "api_uri": "/2010-04-01/Accounts/ACaaaa/Resources.json", } - self.page = TestTokenPaginationPage( + self.page = MockTokenPaginationPage( self.version, self.response, "/2010-04-01/Accounts/ACaaaa/Resources.json", @@ -240,7 +240,7 @@ def test_next_page(self): next_page = self.page.next_page() self.assertIsNotNone(next_page) - self.assertIsInstance(next_page, TestTokenPaginationPage) + self.assertIsInstance(next_page, MockTokenPaginationPage) # Verify we got the next page's data self.assertEqual(next_page.next_token, "token_page3") self.assertEqual(next_page.previous_token, "token_prev1") @@ -269,7 +269,7 @@ def test_previous_page(self): prev_page = next_page.previous_page() self.assertIsNotNone(prev_page) - self.assertIsInstance(prev_page, TestTokenPaginationPage) + self.assertIsInstance(prev_page, MockTokenPaginationPage) # Verify we got the first page's data self.assertIsNone(prev_page.previous_token) self.assertEqual(prev_page.next_token, "token_page2") @@ -322,7 +322,7 @@ def test_next_page_without_uri_in_solution(self): solution = {"account_sid": "ACxxxx"} # Pass empty string as URI to test the error case - page = TestTokenPaginationPage(version, response, "", solution) + page = MockTokenPaginationPage(version, response, "", solution) with self.assertRaises(TwilioException) as context: page.next_page() @@ -408,7 +408,7 @@ def setUp(self): "api_uri": "/2010-04-01/Accounts/ACaaaa/Records.json", } - self.page = TestTokenPaginationPage( + self.page = MockTokenPaginationPage( self.version, self.response, "/2010-04-01/Accounts/ACaaaa/Records.json", @@ -439,5 +439,253 @@ def test_stream_with_page_limit(self): self.assertEqual(len(records), 2) +class TokenPaginationInternalMethodTest(unittest.TestCase): + """Test TokenPagination internal methods""" + + def setUp(self): + self.version = Mock() + self.version.domain = Mock() + self.version.domain.twilio = Mock() + self.version.domain.absolute_url = Mock( + side_effect=lambda uri: f"https://api.twilio.com{uri}" + ) + + # Mock first page response + self.first_response = Mock() + self.first_response.text = """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": "token_page2", + "previousToken": null + }, + "items": [{"id": 1}, {"id": 2}] + } + """ + self.first_response.status_code = 200 + + # Mock next page response + self.next_response = Mock() + self.next_response.text = """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": null, + "previousToken": "token_prev" + }, + "items": [{"id": 3}, {"id": 4}] + } + """ + self.next_response.status_code = 200 + + self.solution = {"account_sid": "ACxxxx"} + self.page = MockTokenPaginationPage( + self.version, + self.first_response, + "/2010-04-01/Accounts/ACxxxx/Resources.json", + self.solution, + ) + + def test_get_page_with_valid_token(self): + """Test _get_page() with a valid token""" + self.version.domain.twilio.request = Mock(return_value=self.next_response) + + result = self.page._get_page("token_page2") + + self.assertIsNotNone(result) + self.assertIsInstance(result, MockTokenPaginationPage) + self.version.domain.twilio.request.assert_called_once_with( + "GET", + "https://api.twilio.com/2010-04-01/Accounts/ACxxxx/Resources.json?pageToken=token_page2", + ) + + def test_get_page_with_none_token(self): + """Test _get_page() with None token returns None""" + result = self.page._get_page(None) + self.assertIsNone(result) + + def test_get_page_without_uri(self): + """Test _get_page() raises error when URI is missing""" + page = MockTokenPaginationPage( + self.version, self.first_response, "", self.solution + ) + + with self.assertRaises(TwilioException) as context: + page._get_page("some_token") + + self.assertIn("URI must be provided", str(context.exception)) + + def test_repr(self): + """Test __repr__ method returns correct string""" + self.assertEqual(repr(self.page), "") + + +class TokenPaginationAsyncTest(unittest.IsolatedAsyncioTestCase): + """Test TokenPagination async methods""" + + def setUp(self): + self.version = Mock() + self.version.domain = Mock() + self.version.domain.twilio = Mock() + self.version.domain.absolute_url = Mock( + side_effect=lambda uri: f"https://api.twilio.com{uri}" + ) + + # Mock first page response + self.first_response = Mock() + self.first_response.text = """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": "token_page2", + "previousToken": null + }, + "items": [{"id": 1}, {"id": 2}] + } + """ + self.first_response.status_code = 200 + + # Mock next page response + self.next_response = Mock() + self.next_response.text = """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": "token_page3", + "previousToken": "token_prev" + }, + "items": [{"id": 3}, {"id": 4}] + } + """ + self.next_response.status_code = 200 + + # Mock previous page response + self.prev_response = Mock() + self.prev_response.text = """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": "token_page2", + "previousToken": null + }, + "items": [{"id": 1}, {"id": 2}] + } + """ + self.prev_response.status_code = 200 + + self.solution = {"account_sid": "ACxxxx"} + + # Page with next token + self.page = MockTokenPaginationPage( + self.version, + self.first_response, + "/2010-04-01/Accounts/ACxxxx/Resources.json", + self.solution, + ) + + # Page with previous token (page 2) + self.page_with_prev = MockTokenPaginationPage( + self.version, + self.next_response, + "/2010-04-01/Accounts/ACxxxx/Resources.json", + self.solution, + ) + + async def test_get_page_async_with_valid_token(self): + """Test _get_page_async() with a valid token""" + self.version.domain.twilio.request_async = AsyncMock( + return_value=self.next_response + ) + + result = await self.page._get_page_async("token_page2") + + self.assertIsNotNone(result) + self.assertIsInstance(result, MockTokenPaginationPage) + self.version.domain.twilio.request_async.assert_called_once_with( + "GET", + "https://api.twilio.com/2010-04-01/Accounts/ACxxxx/Resources.json?pageToken=token_page2", + ) + + async def test_get_page_async_with_none_token(self): + """Test _get_page_async() with None token returns None""" + result = await self.page._get_page_async(None) + self.assertIsNone(result) + + async def test_get_page_async_without_uri(self): + """Test _get_page_async() raises error when URI is missing""" + page = MockTokenPaginationPage( + self.version, self.first_response, "", self.solution + ) + + with self.assertRaises(TwilioException) as context: + await page._get_page_async("some_token") + + self.assertIn("URI must be provided", str(context.exception)) + + async def test_next_page_async(self): + """Test next_page_async() navigates to next page""" + self.version.domain.twilio.request_async = AsyncMock( + return_value=self.next_response + ) + + next_page = await self.page.next_page_async() + + self.assertIsNotNone(next_page) + self.assertIsInstance(next_page, MockTokenPaginationPage) + self.assertEqual(next_page.next_token, "token_page3") + self.assertEqual(next_page.previous_token, "token_prev") + + async def test_next_page_async_none_when_no_token(self): + """Test next_page_async() returns None when there's no next token""" + # Create page with no next token + no_next_response = Mock() + no_next_response.text = """ + { + "meta": { + "key": "items", + "pageSize": 2, + "nextToken": null, + "previousToken": "token_prev" + }, + "items": [{"id": 5}] + } + """ + no_next_response.status_code = 200 + + page = MockTokenPaginationPage( + self.version, + no_next_response, + "/2010-04-01/Accounts/ACxxxx/Resources.json", + self.solution, + ) + + result = await page.next_page_async() + self.assertIsNone(result) + + async def test_previous_page_async(self): + """Test previous_page_async() navigates to previous page""" + self.version.domain.twilio.request_async = AsyncMock( + return_value=self.prev_response + ) + + prev_page = await self.page_with_prev.previous_page_async() + + self.assertIsNotNone(prev_page) + self.assertIsInstance(prev_page, MockTokenPaginationPage) + self.assertIsNone(prev_page.previous_token) + self.assertEqual(prev_page.next_token, "token_page2") + + async def test_previous_page_async_none_when_no_token(self): + """Test previous_page_async() returns None when there's no previous token""" + # First page has no previous token + result = await self.page.previous_page_async() + self.assertIsNone(result) + + if __name__ == "__main__": unittest.main()