diff --git a/ci/run_offline.py b/ci/run_offline.py index dec7258..64dc44f 100644 --- a/ci/run_offline.py +++ b/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()) diff --git a/src/cloudsigma/account_reads.py b/src/cloudsigma/account_reads.py index 86cd9d1..a204278 100644 --- a/src/cloudsigma/account_reads.py +++ b/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): diff --git a/tests_offline/test_pricing_offset.py b/tests_offline/test_pricing_offset.py new file mode 100644 index 0000000..44f5ebc --- /dev/null +++ b/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)