Support explicit offset in PricingRead.get

ci/run_offline.py

@@ -18,5 +18,5 @@ def deny(event,args):
 sys.path.insert(0,str(ROOT))
 suite=unittest.defaultTestLoader.discover(str(ROOT/'tests_offline'))
 r=unittest.TextTestRunner(verbosity=2).run(suite)
-assert r.testsRun==208, 'unexpected test count'
+assert r.testsRun==216, 'unexpected test count'
 sys.exit(not r.wasSuccessful())

src/cloudsigma/account_reads.py

@@ -13,11 +13,22 @@ class PricingRead(ResourceBase):
     def __init__(self, *args, **kwargs):
         self.c = _RawReadClient(*args, **kwargs)
 
-    def get(self, limit=None):
+    def get(self, limit=None, offset=None):
+        """Read one pricing page with explicit limit and zero-based offset.
+
+        Omitted values preserve service defaults; limit=0 requests all objects.
+        Existing positional limit calls remain unchanged. No traversal occurs.
+        """
         if limit is not None and (type(limit) is not int or limit < 0):
             raise ValueError('limit must be a nonnegative integer')
-        query = None if limit is None else {'limit': limit}
-        return self.c.get(self._get_url(), query_params=query)
+        if offset is not None and (type(offset) is not int or offset < 0):
+            raise ValueError('offset must be a nonnegative integer')
+        query = {}
+        if limit is not None:
+            query['limit'] = limit
+        if offset is not None:
+            query['offset'] = offset
+        return self.c.get(self._get_url(), query_params=query or None)
 
 
 class LedgerRead(ResourceBase):

tests_offline/test_pricing_offset.py

@@ -0,0 +1,129 @@
+"""Explicit pricing pages preserve the existing complete/Decimal read contract."""
+import copy
+import unittest
+from decimal import Decimal
+from urllib.parse import urlencode
+
+import requests
+from cloudsigma.account_reads import PricingRead
+from cloudsigma import errors
+
+BASE = 'https://fixture.invalid/api/2.0/'
+
+
+class Adapter(requests.adapters.BaseAdapter):
+    def __init__(self, indices, status=200, failure=None):
+        self.indices, self.status, self.failure = indices, status, failure
+        self.seen = []
+
+    def send(self, request, **kwargs):
+        self.seen.append((request, kwargs))
+        if self.failure:
+            raise self.failure
+        response = requests.Response()
+        response.status_code = self.status
+        objects = ','.join('{"id":"%s","name":"synthetic price",'
+                           '"numeric":9007199254740993.00001,"nullable":null}' % i
+                           for i in self.indices)
+        response._content = ('{"meta":{"total_count":5},"objects":[' +
+                             objects + '],"extension":null}').encode('ascii')
+        response.headers['X-REQUEST-ID'] = 'synthetic-pricing-page'
+        response.request = request
+        return response
+
+    def close(self):
+        pass
+
+
+class PricingOffsetTests(unittest.TestCase):
+    def invoke(self, params, indices, omitted=False, status=200,
+               failure=None, error=None):
+        original = copy.deepcopy(params)
+        adapter = Adapter(indices, status, failure)
+        with requests.Session() as session:
+            session.mount('https://', adapter)
+            resource = PricingRead(api_endpoint=BASE, username='fixture',
+                                  password='synthetic', session=session)
+            kwargs = {} if omitted else (params or {})
+            if error:
+                with self.assertRaises(error) as caught:
+                    resource.get(**kwargs)
+                if failure:
+                    self.assertIs(caught.exception, failure)
+                else:
+                    self.assertEqual(caught.exception.status_code, status)
+                    self.assertEqual(caught.exception.request_id, 'synthetic-pricing-page')
+            else:
+                result = resource.get(**kwargs)
+                self.assertEqual(result, {
+                    'meta': {'total_count': 5}, 'extension': None,
+                    'objects': [{'id': str(i), 'name': 'synthetic price',
+                                 'numeric': Decimal('9007199254740993.00001'),
+                                 'nullable': None} for i in indices]})
+                for obj in result['objects']:
+                    self.assertIs(type(obj['numeric']), Decimal)
+            self.assertEqual(params, original)
+            self.assertEqual(len(adapter.seen), 1)
+            request, options = adapter.seen[0]
+            query_values = {k: v for k, v in (params or {}).items() if v is not None}
+            query = '?' + urlencode(query_values) if query_values else ''
+            self.assertEqual(request.url, BASE + 'pricing/' + query)
+            self.assertEqual(request.method, 'GET')
+            self.assertIn(request.body, (None, b'', ''))
+            self.assertEqual(request.headers['Authorization'], 'Basic Zml4dHVyZTpzeW50aGV0aWM=')
+            self.assertEqual(options['timeout'], 15)
+
+    def test_explicit_pages(self):
+        # Literal expected members: fixture ordering is not a server guarantee.
+        for params, indices in [({'limit': 2, 'offset': 0}, [0, 1]),
+                                ({'limit': 2, 'offset': 2}, [2, 3]),
+                                ({'limit': 2, 'offset': 4}, [4]),
+                                ({'limit': 2, 'offset': 5}, [])]:
+            with self.subTest(params=params):
+                self.invoke(params, indices)
+
+    def test_unlimited_and_offset_only(self):
+        self.invoke({'limit': 0}, [0, 1, 2, 3, 4])
+        self.invoke({'offset': 4}, [4])
+
+    def test_legacy_default(self):
+        self.invoke(None, [0], omitted=True)
+
+    def test_none_and_empty(self):
+        self.invoke({'limit': None, 'offset': None}, [])
+        self.invoke({}, [])
+
+    def test_errors_without_traversal(self):
+        for status, error in ((403, errors.PermissionError),
+                              (404, errors.ClientError), (500, errors.ServerError),
+                              (302, errors.ApiClientError)):
+            with self.subTest(status=status):
+                self.invoke({'limit': 2, 'offset': 4}, [], status=status, error=error)
+
+    def test_positional_limit_compatibility(self):
+        adapter = Adapter([0])
+        with requests.Session() as session:
+            session.mount('https://', adapter)
+            resource = PricingRead(api_endpoint=BASE, username='fixture',
+                                   password='synthetic', session=session)
+            resource.get(5)
+            resource.get(5, 2)
+            self.assertEqual([r.url for r, _ in adapter.seen],
+                             [BASE + 'pricing/?limit=5',
+                              BASE + 'pricing/?limit=5&offset=2'])
+
+    def test_invalid_offset_zero_dispatch(self):
+        adapter = Adapter([])
+        with requests.Session() as session:
+            session.mount('https://', adapter)
+            resource = PricingRead(api_endpoint=BASE, username='fixture',
+                                   password='synthetic', session=session)
+            for offset in (-1, True, False, 1.5, '2', [], {}):
+                with self.subTest(offset=offset), self.assertRaises(ValueError):
+                    resource.get(limit=2, offset=offset)
+            self.assertEqual(adapter.seen, [])
+
+    def test_transport_error_identity(self):
+        failure = requests.exceptions.Timeout('synthetic price page failure')
+        self.invoke({'limit': 2}, [], failure=failure,
+                    error=requests.exceptions.Timeout)