1# Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"). You 4# may not use this file except in compliance with the License. A copy of 5# the License is located at 6# 7# http://aws.amazon.com/apache2.0/ 8# 9# or in the "license" file accompanying this file. This file is 10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF 11# ANY KIND, either express or implied. See the License for the specific 12# language governing permissions and limitations under the License. 13import time 14 15from tests import unittest 16 17import botocore.session 18from botocore import exceptions 19 20 21class TestApigateway(unittest.TestCase): 22 def setUp(self): 23 self.session = botocore.session.get_session() 24 self.client = self.session.create_client('apigateway', 'us-east-1') 25 26 # Create a resource to use with this client. 27 self.api_name = 'mytestapi' 28 self.api_id = self.create_rest_api_or_skip() 29 30 def create_rest_api_or_skip(self): 31 try: 32 api_id = self.client.create_rest_api(name=self.api_name)['id'] 33 except exceptions.ClientError as e: 34 if e.response['Error']['Code'] == 'TooManyRequestsException': 35 raise unittest.SkipTest( 36 "Hit API gateway throttle limit, skipping test.") 37 raise 38 return api_id 39 40 def delete_api(self): 41 retries = 0 42 while retries < 10: 43 try: 44 self.client.delete_rest_api(restApiId=self.api_id) 45 break 46 except exceptions.ClientError as e: 47 if e.response['Error']['Code'] == 'TooManyRequestsException': 48 retries += 1 49 time.sleep(5) 50 else: 51 raise 52 53 def tearDown(self): 54 self.delete_api() 55 56 def test_put_integration(self): 57 # The only resource on a brand new api is the path. So use that ID. 58 path_resource_id = self.client.get_resources( 59 restApiId=self.api_id)['items'][0]['id'] 60 61 # Create a method for the resource. 62 self.client.put_method( 63 restApiId=self.api_id, 64 resourceId=path_resource_id, 65 httpMethod='GET', 66 authorizationType='None' 67 ) 68 69 # Put an integration on the method. 70 response = self.client.put_integration( 71 restApiId=self.api_id, 72 resourceId=path_resource_id, 73 httpMethod='GET', 74 type='HTTP', 75 integrationHttpMethod='GET', 76 uri='https://api.endpoint.com' 77 ) 78 # Assert the response was successful by checking the integration type 79 self.assertEqual(response['type'], 'HTTP') 80