diff --git a/ci/run_offline.py b/ci/run_offline.py index 64dc44f..1c29b00 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==216, 'unexpected test count' +assert r.testsRun==222, 'unexpected test count' sys.exit(not r.wasSuccessful()) diff --git a/src/cloudsigma/vmware_read.py b/src/cloudsigma/vmware_read.py index 52f46c0..06debb4 100644 --- a/src/cloudsigma/vmware_read.py +++ b/src/cloudsigma/vmware_read.py @@ -12,5 +12,10 @@ class VMwareRead: def __init__(self, *args, **kwargs): self.c = _RawReadClient(*args, **kwargs) - def list(self): - return self.c.get('/vmware_servers/') + def list(self, query_params=None): + """Read one page with optional explicit limit/offset values. + + Omitted parameters preserve the service default; limit=0 requests all + objects. No automatic traversal or new filter support is implied. + """ + return self.c.get('/vmware_servers/', query_params=query_params) diff --git a/tests_offline/test_vmware_pagination.py b/tests_offline/test_vmware_pagination.py new file mode 100644 index 0000000..9f3c43d --- /dev/null +++ b/tests_offline/test_vmware_pagination.py @@ -0,0 +1,105 @@ +"""Explicit VMware 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.vmware_read import VMwareRead +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 VMware server",' + '"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-vmware-page' + response.request = request + return response + + def close(self): + pass + + +class VMwarePaginationTests(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 = VMwareRead(api_endpoint=BASE, username='fixture', + password='synthetic', session=session) + kwargs = {} if omitted else {'query_params': params} + if error: + with self.assertRaises(error) as caught: + resource.list(**kwargs) + if failure: + self.assertIs(caught.exception, failure) + else: + self.assertEqual(caught.exception.status_code, status) + self.assertEqual(caught.exception.request_id, 'synthetic-vmware-page') + else: + result = resource.list(**kwargs) + self.assertEqual(result, { + 'meta': {'total_count': 5}, 'extension': None, + 'objects': [{'id': str(i), 'name': 'synthetic VMware server', + '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 = '?' + urlencode(params) if params else '' + self.assertEqual(request.url, BASE + 'vmware_servers/' + 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(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_transport_error_identity(self): + failure = requests.exceptions.Timeout('synthetic VMware server page failure') + self.invoke({'limit': 2}, [], failure=failure, + error=requests.exceptions.Timeout)