From d7acf7434b8e281c1b6094f3143ce9ba9ce8fbb2 Mon Sep 17 00:00:00 2001 From: I335851 Date: Fri, 19 Jun 2026 15:01:25 +0200 Subject: [PATCH 1/2] service: harden Collection FunctionImport handler from PR #298 review - Hoist entity_set lookup above EntityType/Collection branches to eliminate duplication - Guard against missing 'results' key with a descriptive PyODataException - Add tests: empty collection, missing results key, total_count absent (ProgramError) - Fix double blank line in test_function_import_collection_with_pagination_metadata --- pyodata/v2/service.py | 15 ++++++----- tests/test_service_v2.py | 56 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 64 insertions(+), 7 deletions(-) diff --git a/pyodata/v2/service.py b/pyodata/v2/service.py index f1f73e5..85ede9c 100644 --- a/pyodata/v2/service.py +++ b/pyodata/v2/service.py @@ -1707,12 +1707,13 @@ def function_import_handler(fimport, response): response_data = response.json()['d'] - # 1. if return types is "entity type", return instance of appropriate entity proxy - if isinstance(fimport.return_type, model.EntityType): + # 1. if return type is an entity type or collection, resolve the entity set once + if isinstance(fimport.return_type, (model.EntityType, model.Collection)): entity_set = self._service.schema.entity_set(fimport.entity_set_name) + + if isinstance(fimport.return_type, model.EntityType): return EntityProxy(self._service, entity_set, fimport.return_type, response_data) - # 1.b alternatively, if return type is a Collection, return a list of appropriate entity proxy if isinstance(fimport.return_type, model.Collection): total_count = None next_url = None @@ -1720,11 +1721,13 @@ def function_import_handler(fimport, response): total_count = int(response_data['__count']) if '__next' in response_data: next_url = response_data['__next'] + results = response_data.get('results') + if results is None: + raise PyODataException( + f'Function import {fimport.name} returned a Collection response without a "results" key') collection = ListWithTotalCount(total_count, next_url) - - entity_set = self._service.schema.entity_set(fimport.entity_set_name) collection_item_type = fimport.return_type.item_type - for entity in response_data['results']: + for entity in results: collection.append(EntityProxy(self._service, entity_set, collection_item_type, entity)) return collection diff --git a/tests/test_service_v2.py b/tests/test_service_v2.py index 1ca22d6..083c3a4 100644 --- a/tests/test_service_v2.py +++ b/tests/test_service_v2.py @@ -635,7 +635,6 @@ def test_function_import_collection_with_pagination_metadata(service): # pylint: disable=redefined-outer-name - responses.add( responses.GET, f'{service.url}/get_best_measurements?$inlinecount=allpages', @@ -670,6 +669,61 @@ def test_function_import_collection_with_pagination_metadata(service): "FunctionImport Collection should preserve __next like get_entities()" assert result.next_url == f"{service.url}/get_best_measurements?$skiptoken=2" + +@responses.activate +def test_function_import_collection_empty(service): + """Collection FunctionImport with zero results returns an empty ListWithTotalCount""" + + # pylint: disable=redefined-outer-name + + responses.add( + responses.GET, + f'{service.url}/get_best_measurements', + headers={'Content-type': 'application/json'}, + json={'d': {'results': []}}, + status=200) + + result = service.functions.get_best_measurements.execute() + assert isinstance(result, list) + assert len(result) == 0 + + +@responses.activate +def test_function_import_collection_missing_results_key(service): + """Collection FunctionImport response without 'results' raises PyODataException""" + + # pylint: disable=redefined-outer-name + + responses.add( + responses.GET, + f'{service.url}/get_best_measurements', + headers={'Content-type': 'application/json'}, + json={'d': {}}, + status=200) + + with pytest.raises(PyODataException) as ex_info: + service.functions.get_best_measurements.execute() + assert 'results' in str(ex_info.value) + + +@responses.activate +def test_function_import_collection_total_count_absent(service): + """total_count raises ProgramError when __count was not requested""" + + # pylint: disable=redefined-outer-name + + responses.add( + responses.GET, + f'{service.url}/get_best_measurements', + headers={'Content-type': 'application/json'}, + json={'d': {'results': []}}, + status=200) + + result = service.functions.get_best_measurements.execute() + with pytest.raises(ProgramError): + _ = result.total_count + + @responses.activate def test_update_entity(service): """Check updating of entity properties""" From be72b3be9adc44d69fe43281b5d4dd555fb8ccf0 Mon Sep 17 00:00:00 2001 From: I335851 Date: Fri, 19 Jun 2026 15:39:09 +0200 Subject: [PATCH 2/2] service: extract status handling to fix too-many-branches linter error All status-related logic (errors and unexpected-2xx warnings) is consolidated into _handle_response_status, resolving the pylint R0912 violation. --- pyodata/v2/service.py | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/pyodata/v2/service.py b/pyodata/v2/service.py index 85ede9c..b460e0b 100644 --- a/pyodata/v2/service.py +++ b/pyodata/v2/service.py @@ -1659,52 +1659,52 @@ def __getattr__(self, name): fimport = self._service.schema.function_import(name) - def function_import_handler(fimport, response): - """Get function call response from HTTP Response""" - + def _handle_response_status(fimport, response): + # errors — raise on any non-2xx response if 300 <= response.status_code < 400: raise HttpError(f'Function Import {fimport.name} requires Redirection which is not supported', response) - if response.status_code == 401: raise HttpError(f'Not authorized to call Function Import {fimport.name}', response) - if response.status_code == 403: raise HttpError(f'Missing privileges to call Function Import {fimport.name}', response) - if response.status_code == 405: raise HttpError( f'Despite definition Function Import {fimport.name} does not support HTTP {fimport.http_method}', response) - if 400 <= response.status_code < 500: raise HttpError( f'Function Import {fimport.name} call has failed with status code {response.status_code}', response) - if response.status_code >= 500: raise HttpError(f'Server has encountered an error while processing Function Import {fimport.name}', response) + # warnings — unexpected 2xx codes if fimport.return_type is None: if response.status_code != 204: logging.getLogger(LOGGER_NAME).warning( 'The No Return Function Import %s has replied with HTTP Status Code %d instead of 204', fimport.name, response.status_code) + elif response.status_code != 200: + logging.getLogger(LOGGER_NAME).warning( + 'The Function Import %s has replied with HTTP Status Code %d instead of 200', + fimport.name, response.status_code) + + def function_import_handler(fimport, response): + """Get function call response from HTTP Response""" + + _handle_response_status(fimport, response) + if fimport.return_type is None: if response.text: logging.getLogger(LOGGER_NAME).warning( 'The No Return Function Import %s has returned content:\n%s', fimport.name, response.text) return None - if response.status_code != 200: - logging.getLogger(LOGGER_NAME).warning( - 'The Function Import %s has replied with HTTP Status Code %d instead of 200', - fimport.name, response.status_code) - response_data = response.json()['d'] # 1. if return type is an entity type or collection, resolve the entity set once