# Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the 'License'). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the 'license' file accompanying this file. This file is # distributed on an 'AS IS' BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import pytest from boto3.resources.base import ResourceMeta, ServiceResource from boto3.resources.factory import ResourceFactory from boto3.resources.model import Parameter, ResponseResource from boto3.resources.response import ( RawHandler, ResourceHandler, build_empty_response, build_identifiers, ) from boto3.utils import ServiceContext from tests import BaseTestCase, mock class TestBuildIdentifiers(BaseTestCase): def test_build_identifier_from_res_path_scalar(self): identifiers = [ Parameter(target='Id', source='response', path='Container.Frob.Id') ] parent = mock.Mock() params = {} response = {'Container': {'Frob': {'Id': 'response-path'}}} values = build_identifiers(identifiers, parent, params, response) # Verify identifier loaded from responsePath scalar set assert values[0][1] == 'response-path' def test_build_identifier_from_res_path_list(self): identifiers = [ Parameter( target='Id', source='response', path='Container.Frobs[].Id' ) ] parent = mock.Mock() params = {} response = {'Container': {'Frobs': [{'Id': 'response-path'}]}} values = build_identifiers(identifiers, parent, params, response) # Verify identifier loaded from responsePath scalar set assert values[0][1] == ['response-path'] def test_build_identifier_from_parent_identifier(self): identifiers = [Parameter(target='Id', source='identifier', name='Id')] parent = mock.Mock() parent.id = 'identifier' params = {} response = {'Container': {'Frobs': []}} values = build_identifiers(identifiers, parent, params, response) # Verify identifier loaded from responsePath scalar set assert values[0][1] == 'identifier' def test_build_identifier_from_parent_data_member(self): identifiers = [Parameter(target='Id', source='data', path='Member')] parent = mock.Mock() parent.meta = ResourceMeta('test', data={'Member': 'data-member'}) params = {} response = {'Container': {'Frobs': []}} values = build_identifiers(identifiers, parent, params, response) # Verify identifier loaded from responsePath scalar set assert values[0][1] == 'data-member' def test_build_identifier_from_req_param(self): identifiers = [ Parameter(target='Id', source='requestParameter', path='Param') ] parent = mock.Mock() params = {'Param': 'request-param'} response = {'Container': {'Frobs': []}} values = build_identifiers(identifiers, parent, params, response) # Verify identifier loaded from responsePath scalar set assert values[0][1] == 'request-param' def test_build_identifier_from_invalid_source_type(self): identifiers = [Parameter(target='Id', source='invalid')] parent = mock.Mock() params = {} response = {'Container': {'Frobs': []}} with pytest.raises(NotImplementedError): build_identifiers(identifiers, parent, params, response) class TestBuildEmptyResponse(BaseTestCase): def setUp(self): super().setUp() self.search_path = '' self.operation_name = 'GetFrobs' self.output_shape = mock.Mock() operation_model = mock.Mock() operation_model.output_shape = self.output_shape self.service_model = mock.Mock() self.service_model.operation_model.return_value = operation_model def get_response(self): return build_empty_response( self.search_path, self.operation_name, self.service_model ) def test_empty_structure(self): self.output_shape.type_name = 'structure' response = self.get_response() # Structure should default to empty dictionary assert isinstance(response, dict) assert response == {} def test_empty_list(self): self.output_shape.type_name = 'list' response = self.get_response() assert isinstance(response, list) assert len(response) == 0 def test_empty_map(self): self.output_shape.type_name = 'map' response = self.get_response() assert isinstance(response, dict) assert response == {} def test_empty_string(self): self.output_shape.type_name = "string" response = self.get_response() assert response is None def test_empty_integer(self): self.output_shape.type_name = "integer" response = self.get_response() assert response is None def test_empty_unknown_returns_none(self): self.output_shape.type_name = "invalid" response = self.get_response() assert response is None def test_path_structure(self): self.search_path = 'Container.Frob' frob = mock.Mock() frob.type_name = 'integer' container = mock.Mock() container.type_name = 'structure' container.members = {'Frob': frob} self.output_shape.type_name = 'structure' self.output_shape.members = {'Container': container} response = self.get_response() assert response is None def test_path_list(self): self.search_path = 'Container[1].Frob' frob = mock.Mock() frob.type_name = 'integer' container = mock.Mock() container.type_name = 'list' container.member = frob self.output_shape.type_name = 'structure' self.output_shape.members = {'Container': container} response = self.get_response() assert response is None def test_path_invalid(self): self.search_path = 'Container.Invalid' container = mock.Mock() container.type_name = 'invalid' self.output_shape.type_name = 'structure' self.output_shape.members = {'Container': container} with pytest.raises(NotImplementedError): self.get_response() class TestRawHandler(BaseTestCase): def test_raw_handler_response(self): parent = mock.Mock() params = {} response = {'Id': 'foo'} handler = RawHandler(search_path=None) parsed_response = handler(parent, params, response) # verify response is unmodified assert parsed_response == response def test_raw_handler_response_path(self): parent = mock.Mock() params = {} frob = {'Id': 'foo'} response = {'Container': {'Frob': frob}} handler = RawHandler(search_path='Container.Frob') parsed_response = handler(parent, params, response) assert parsed_response == frob class TestResourceHandler(BaseTestCase): def setUp(self): super().setUp() self.identifier_path = '' self.factory = ResourceFactory(mock.Mock()) self.resource_defs = { 'Frob': {'shape': 'Frob', 'identifiers': [{'name': 'Id'}]} } self.service_model = mock.Mock() shape = mock.Mock() shape.members = {} self.service_model.shape_for.return_value = shape frobs = mock.Mock() frobs.type_name = 'list' container = mock.Mock() container.type_name = 'structure' container.members = {'Frobs': frobs} self.output_shape = mock.Mock() self.output_shape.type_name = 'structure' self.output_shape.members = {'Container': container} operation_model = mock.Mock() operation_model.output_shape = self.output_shape self.service_model.operation_model.return_value = operation_model self.parent = mock.Mock() self.parent.meta = ResourceMeta('test', client=mock.Mock()) self.params = {} def get_resource(self, search_path, response): request_resource_def = { 'type': 'Frob', 'identifiers': [ { 'target': 'Id', 'source': 'response', 'path': self.identifier_path, }, ], } resource_model = ResponseResource( request_resource_def, self.resource_defs ) handler = ResourceHandler( search_path=search_path, factory=self.factory, resource_model=resource_model, service_context=ServiceContext( service_name='myservice', resource_json_definitions=self.resource_defs, service_model=self.service_model, service_waiter_model=None, ), operation_name='GetFrobs', ) return handler(self.parent, self.params, response) def test_create_resource_scalar(self): self.identifier_path = 'Container.Id' search_path = 'Container' response = { 'Container': { 'Id': 'a-frob', 'OtherValue': 'other', } } resource = self.get_resource(search_path, response) assert isinstance(resource, ServiceResource) @mock.patch('boto3.resources.response.build_empty_response') def test_missing_data_scalar_builds_empty_response(self, build_mock): self.identifier_path = 'Container.Id' search_path = 'Container' response = {'something': 'irrelevant'} resources = self.get_resource(search_path, response) assert build_mock.called assert resources == build_mock.return_value def test_create_resource_list(self): self.identifier_path = 'Container.Frobs[].Id' search_path = 'Container.Frobs[]' response = { 'Container': { 'Frobs': [ { 'Id': 'a-frob', 'OtherValue': 'other', }, { 'Id': 'another-frob', 'OtherValue': 'foo', }, ] } } resources = self.get_resource(search_path, response) assert isinstance(resources, list) assert len(resources) == 2 assert isinstance(resources[0], ServiceResource) def test_create_resource_list_no_search_path(self): self.identifier_path = '[].Id' search_path = '' response = [{'Id': 'a-frob', 'OtherValue': 'other'}] resources = self.get_resource(search_path, response) assert isinstance(resources, list) assert len(resources) == 1 assert isinstance(resources[0], ServiceResource) @mock.patch('boto3.resources.response.build_empty_response') def test_missing_data_list_builds_empty_response(self, build_mock): self.identifier_path = 'Container.Frobs[].Id' search_path = 'Container.Frobs[]' response = {'something': 'irrelevant'} resources = self.get_resource(search_path, response) assert build_mock.called, 'build_empty_response was never called' assert resources == build_mock.return_value