Best Python code snippet using localstack_python
test_apigw_rest_api.py
Source:test_apigw_rest_api.py
1#!/usr/bin/python2# TODO: License goes here3import library.apigw_rest_api as apigw_rest_api4from library.apigw_rest_api import ApiGwRestApi5import mock6from mock import patch7from mock import create_autospec8from mock import ANY9import unittest10import boto11from botocore.exceptions import BotoCoreError12class TestApiGwRestApi(unittest.TestCase):13 def setUp(self):14 self.module = mock.MagicMock()15 self.module.check_mode = False16 self.module.exit_json = mock.MagicMock()17 self.module.fail_json = mock.MagicMock()18 self.restapi = ApiGwRestApi(self.module)19 self.restapi.client = mock.MagicMock()20 reload(apigw_rest_api)21 def test_boto_module_not_found(self):22 # Setup Mock Import Function23 import __builtin__ as builtins24 real_import = builtins.__import__25 def mock_import(name, *args):26 if name == 'boto': raise ImportError27 return real_import(name, *args)28 with mock.patch('__builtin__.__import__', side_effect=mock_import):29 reload(apigw_rest_api)30 ApiGwRestApi(self.module)31 self.module.fail_json.assert_called_with(msg='boto and boto3 are required for this module')32 def test_boto3_module_not_found(self):33 # Setup Mock Import Function34 import __builtin__ as builtins35 real_import = builtins.__import__36 def mock_import(name, *args):37 if name == 'boto3': raise ImportError38 return real_import(name, *args)39 with mock.patch('__builtin__.__import__', side_effect=mock_import):40 reload(apigw_rest_api)41 ApiGwRestApi(self.module)42 self.module.fail_json.assert_called_with(msg='boto and boto3 are required for this module')43 @patch.object(apigw_rest_api, 'boto3')44 def test_boto3_client_properly_instantiated(self, mock_boto):45 ApiGwRestApi(self.module)46 mock_boto.client.assert_called_once_with('apigateway')47 def test_process_request_calls_boto3_get_rest_apis(self):48 self.restapi.module.params = { 'name': 'whatever' }49 self.restapi.process_request()50 self.restapi.client.get_rest_apis.assert_called_once_with()51 def test_process_request_fails_when_get_rest_apis_returns_error(self):52 self.restapi.module.params = { 'name': 'whatever' }53 self.restapi.client.get_rest_apis = mock.MagicMock(side_effect=BotoCoreError())54 self.restapi.process_request()55 self.restapi.module.fail_json.assert_called_once_with(56 msg='Encountered fatal error calling boto3 get_rest_apis function: An unspecified error occurred')57 def test_process_request_exits_with_no_change_when_removing_non_existent_api(self):58 self.restapi.module.params = { 'name': 'whatever', 'state': 'absent' }59 self.restapi.client.get_rest_apis = mock.MagicMock(return_value={'items': []})60 self.restapi.process_request()61 self.restapi.module.exit_json.assert_called_once_with(changed=False, api=None)62 def test_process_request_exits_with_no_change_when_adding_existing_and_unchanged_api(self):63 get_response = {64 'items': [{65 'id': 12345,66 'name': 'whatever',67 'description': 'very awesome'68 }]69 }70 self.restapi.module.params = { 'name': 'whatever', 'state': 'present', 'description': 'very awesome' }71 self.restapi.client.get_rest_apis = mock.MagicMock(return_value=get_response)72 self.restapi.process_request()73 self.restapi.module.exit_json.assert_called_once_with(changed=False, api=get_response['items'][0])74 def test_process_request_creates_api_when_missing(self):75 create_response = {76 'id': 'aws-whatever',77 'name': 'a-name',78 'description': 'a-desc',79 'createdDate': 'some-date'80 }81 self.restapi.module.params = { 'name': 'whatever', 'state': 'present', 'description': 'very awesome' }82 self.restapi.client.get_rest_apis = mock.MagicMock(return_value={'items': []})83 self.restapi.client.create_rest_api = mock.MagicMock(return_value=create_response)84 self.restapi.process_request()85 self.restapi.client.create_rest_api.assert_called_once_with(name='whatever', description='very awesome')86 self.restapi.module.exit_json.assert_called_once_with(changed=True, api=create_response)87 def test_process_request_fails_when_create_rest_api_throws_error(self):88 self.restapi.module.params = { 'name': 'whatever', 'state': 'present' }89 self.restapi.client.get_rest_apis = mock.MagicMock(return_value={'items': []})90 self.restapi.client.create_rest_api = mock.MagicMock(side_effect=BotoCoreError())91 self.restapi.process_request()92 self.restapi.client.create_rest_api.assert_called_once_with(name='whatever')93 self.restapi.module.fail_json.assert_called_once_with(94 msg='Encountered fatal error calling boto3 create_rest_api function: An unspecified error occurred')95 def test_process_request_updates_api_when_params_do_not_match(self):96 get_response = {97 'items': [{98 'id': 12345,99 'name': 'whatever',100 'description': 'very awesome'101 }]102 }103 self.restapi.module.params = { 'name': 'whatever', 'state': 'present', 'description': 'awesomer' }104 self.restapi.client.get_rest_apis = mock.MagicMock(return_value=get_response)105 self.restapi.client.update_rest_api = mock.MagicMock(return_value='TotallyUpdated')106 self.restapi.process_request()107 self.restapi.client.update_rest_api.assert_called_once_with(restApiId=12345, patchOperations=[108 {'op': 'replace', 'path': '/name', 'value': 'whatever'},109 {'op': 'replace', 'path': '/description', 'value': 'awesomer'},110 ])111 self.restapi.module.exit_json.assert_called_once_with(changed=True, api='TotallyUpdated')112 def test_process_request_sets_description_to_empty_string_when_missing_during_update(self):113 get_response = {114 'items': [{115 'id': 12345,116 'name': 'whatever',117 'description': 'very awesome'118 }]119 }120 self.restapi.module.params = { 'name': 'whatever', 'state': 'present' }121 self.restapi.client.get_rest_apis = mock.MagicMock(return_value=get_response)122 self.restapi.client.update_rest_api = mock.MagicMock(return_value='TotallyUpdated')123 self.restapi.process_request()124 self.restapi.client.update_rest_api.assert_called_once_with(restApiId=12345, patchOperations=[125 {'op': 'replace', 'path': '/name', 'value': 'whatever'},126 {'op': 'replace', 'path': '/description', 'value': ''}127 ])128 self.restapi.module.exit_json.assert_called_once_with(changed=True, api='TotallyUpdated')129 def test_process_request_fails_when_update_rest_api_throws_exception(self):130 get_response = {131 'items': [{132 'id': 12345,133 'name': 'whatever',134 'description': 'very awesome'135 }]136 }137 self.restapi.module.params = { 'name': 'whatever', 'state': 'present', 'description': 'awesomer' }138 self.restapi.client.get_rest_apis = mock.MagicMock(return_value=get_response)139 self.restapi.client.update_rest_api = mock.MagicMock(side_effect=BotoCoreError())140 self.restapi.process_request()141 self.restapi.client.update_rest_api.assert_called_once_with(restApiId=12345, patchOperations=[142 {'op': 'replace', 'path': '/name', 'value': 'whatever'},143 {'op': 'replace', 'path': '/description', 'value': 'awesomer'},144 ])145 self.restapi.module.fail_json.assert_called_once_with(146 msg='Encountered fatal error calling boto3 update_rest_api function: An unspecified error occurred')147 def test_process_request_deletes_api_when_api_is_present(self):148 get_response = {149 'items': [{150 'id': 12345,151 'name': 'whatever',152 'description': 'very awesome'153 }]154 }155 self.restapi.module.params = { 'name': 'whatever', 'state': 'absent' }156 self.restapi.client.get_rest_apis = mock.MagicMock(return_value=get_response)157 self.restapi.client.delete_rest_api = mock.MagicMock(return_value=None)158 self.restapi.process_request()159 self.restapi.client.delete_rest_api.assert_called_once_with(restApiId=12345)160 self.restapi.module.exit_json.assert_called_once_with(changed=True, api=None)161 def test_process_request_fails_when_delete_rest_api_throws_exception(self):162 get_response = {163 'items': [{164 'id': 12345,165 'name': 'whatever',166 'description': 'very awesome'167 }]168 }169 self.restapi.module.params = { 'name': 'whatever', 'state': 'absent' }170 self.restapi.client.get_rest_apis = mock.MagicMock(return_value=get_response)171 self.restapi.client.delete_rest_api = mock.MagicMock(side_effect=BotoCoreError())172 self.restapi.process_request()173 self.restapi.client.delete_rest_api.assert_called_once_with(restApiId=12345)174 self.restapi.module.fail_json.assert_called_once_with(175 msg='Encountered fatal error calling boto3 delete_rest_api function: An unspecified error occurred')176 def test_define_argument_spec(self):177 result = ApiGwRestApi._define_module_argument_spec()178 self.assertIsInstance(result, dict)179 self.assertEqual(result, dict(180 name=dict(required=True),181 description=dict(required=False),182 state=dict(default='present', choices=['present', 'absent'])183 ))184 def test_process_requests_reports_change_and_does_not_call_create_when_check_mode(self):185 self.restapi.module.params = { 'name': 'whatever', 'state': 'present', 'description': 'very awesome' }186 self.restapi.client.get_rest_apis = mock.MagicMock(return_value={'items': []})187 self.restapi.module.check_mode = True188 self.restapi.process_request()189 assert self.restapi.client.create_rest_api.call_count == 0190 self.restapi.module.exit_json.assert_called_once_with(changed=True, api=ANY)191 def test_process_requests_reports_change_and_does_not_call_update_when_check_mode(self):192 get_response = {193 'items': [{194 'id': 12345,195 'name': 'whatever',196 'description': 'very awesome'197 }]198 }199 self.restapi.module.check_mode = True200 self.restapi.module.params = { 'name': 'whatever', 'state': 'present', 'description': 'awesomer' }201 self.restapi.client.get_rest_apis = mock.MagicMock(return_value=get_response)202 self.restapi.process_request()203 assert self.restapi.client.update_rest_api.call_count == 0204 self.restapi.module.exit_json.assert_called_once_with(changed=True, api=ANY)205 def test_process_requests_reports_change_and_does_not_call_delete_when_check_mode(self):206 get_response = {207 'items': [{208 'id': 12345,209 'name': 'whatever',210 'description': 'very awesome'211 }]212 }213 self.restapi.module.check_mode = True214 self.restapi.module.params = { 'name': 'whatever', 'state': 'absent' }215 self.restapi.client.get_rest_apis = mock.MagicMock(return_value=get_response)216 self.restapi.process_request()217 assert self.restapi.client.delete_rest_api.call_count == 0218 self.restapi.module.exit_json.assert_called_with(changed=True, api=ANY)219 @patch.object(apigw_rest_api, 'AnsibleModule')220 @patch.object(apigw_rest_api, 'ApiGwRestApi')221 def test_main(self, mock_ApiGwRestApi, mock_AnsibleModule):222 mock_ApiGwRestApi_instance = mock.MagicMock()223 mock_AnsibleModule_instance = mock.MagicMock()224 mock_ApiGwRestApi.return_value = mock_ApiGwRestApi_instance225 mock_AnsibleModule.return_value = mock_AnsibleModule_instance226 apigw_rest_api.main()227 mock_ApiGwRestApi.assert_called_once_with(mock_AnsibleModule_instance)228 assert mock_ApiGwRestApi_instance.process_request.call_count == 1229if __name__ == '__main__':...
test_gateway_normalizer.py
Source:test_gateway_normalizer.py
1from deployment import gateway_normalizer, chalice_config_reader2import os3import boto34import pytest5from botocore.stub import Stubber6import configparser7TEST_ENVIRONMENT = 'some_environment'8PRIVATE_ENVIRONMENT = "private_env"9VPCE_ID = "vpce-12345"10REST_API_ID = 'some_rest_api_id'11API_GATEWAY_STAGE = 'some_api_stage'12MOCK_ACCOUNT_NUMBER = 4213MOCK_DEPLOYED_GATEWAY = {"rest_api_id": REST_API_ID}14MOCK_CHALICE_CONFIG = {15 "stages": {16 TEST_ENVIRONMENT: {17 "api_gateway_stage": API_GATEWAY_STAGE18 },19 PRIVATE_ENVIRONMENT: {20 "api_gateway_stage": API_GATEWAY_STAGE,21 "api_gateway_endpoint_type": "PRIVATE",22 "api_gateway_endpoint_vpce": VPCE_ID,23 }24 }25}26@pytest.fixture27def api_gateway_client():28 os.environ['AWS_DEFAULT_REGION'] = 'us-east-1'29 api_gateway_client = boto3.client('apigateway')30 return api_gateway_client31@pytest.fixture32def api_gateway_stubber(api_gateway_client):33 return Stubber(api_gateway_client)34@pytest.fixture35def mock_chalice_config_reader(mocker):36 mocker.patch.object(chalice_config_reader, 'find_deployed_config')37 chalice_config_reader.find_deployed_config.return_value = MOCK_DEPLOYED_GATEWAY38 mocker.patch.object(chalice_config_reader, 'chalice_config')39 chalice_config_reader.chalice_config.return_value = MOCK_CHALICE_CONFIG40 return chalice_config_reader41@pytest.fixture42def partially_mocked_gateway_normalizer(mocker, api_gateway_client):43 mocker.patch.object(gateway_normalizer, 'api_gateway_client')44 gateway_normalizer.api_gateway_client.return_value = api_gateway_client45 mocker.patch.object(gateway_normalizer, 'get_region')46 gateway_normalizer.get_region.return_value = 'us-east-1'47 mocker.patch.object(gateway_normalizer, 'get_account_number')48 gateway_normalizer.get_account_number.return_value = MOCK_ACCOUNT_NUMBER49 return gateway_normalizer50def test_normalize_gateway_sets_name_and_stage_logging_and_adds_tags(mock_chalice_config_reader, partially_mocked_gateway_normalizer, api_gateway_stubber):51 environment = 'some_environment'52 api_gateway_stubber.add_response('update_rest_api', {}, {'restApiId': REST_API_ID,53 'patchOperations': [{'op': 'replace', 'path': '/name', 'value': f'{environment}-webportal'},54 {'op': 'replace', 'path': '/description', 'value': f'{environment}-webportal'}]})55 api_gateway_stubber.add_response('update_stage', {}, {'restApiId': REST_API_ID, 'stageName': API_GATEWAY_STAGE,56 'patchOperations': [{'op': 'replace', 'path': '/accessLogSettings/destinationArn',57 'value': f'arn:aws:logs:us-east-1:{MOCK_ACCOUNT_NUMBER}:log-group:/aws/apigateway/{REST_API_ID}/{API_GATEWAY_STAGE}'},58 {'op': 'replace', 'path': '/accessLogSettings/format', 'value': gateway_normalizer.LOG_FORMAT}]})59 api_gateway_stubber.add_response('tag_resource', {}, {'resourceArn': f'arn:aws:apigateway:us-east-1::/restapis/{REST_API_ID}',60 "tags": {"Environment": environment, "Project": "SDC-Platform", "Team": "sdc-platform"}})61 with api_gateway_stubber:62 partially_mocked_gateway_normalizer.normalize_gateway(environment)63 api_gateway_stubber.assert_no_pending_responses()64 chalice_config_reader.find_deployed_config.assert_called_with('rest_api', environment)65 chalice_config_reader.chalice_config.assert_called_with()66def test_normalize_sets_vpce(mock_chalice_config_reader, partially_mocked_gateway_normalizer, api_gateway_stubber):67 environment = PRIVATE_ENVIRONMENT68 api_gateway_stubber.add_response('update_rest_api', {}, {'restApiId': REST_API_ID,69 'patchOperations': [{'op': 'replace', 'path': '/name', 'value': f'{environment}-webportal'},70 {'op': 'replace', 'path': '/description', 'value': f'{environment}-webportal'}]})71 # Additional private API call72 # Stubs are FIFO so order is important!!!73 api_gateway_stubber.add_response('update_rest_api', 74 {}, 75 {'restApiId': REST_API_ID,76 'patchOperations': [{'op': 'add', 'path': '/endpointConfiguration/vpcEndpointIds', 'value': VPCE_ID}]77 })78 api_gateway_stubber.add_response('update_stage', {}, {'restApiId': REST_API_ID, 'stageName': API_GATEWAY_STAGE,79 'patchOperations': [{'op': 'replace', 'path': '/accessLogSettings/destinationArn',80 'value': f'arn:aws:logs:us-east-1:{MOCK_ACCOUNT_NUMBER}:log-group:/aws/apigateway/{REST_API_ID}/{API_GATEWAY_STAGE}'},81 {'op': 'replace', 'path': '/accessLogSettings/format', 'value': gateway_normalizer.LOG_FORMAT}]})82 api_gateway_stubber.add_response('tag_resource', {}, {'resourceArn': f'arn:aws:apigateway:us-east-1::/restapis/{REST_API_ID}',83 "tags": {"Environment": environment, "Project": "SDC-Platform", "Team": "sdc-platform"}})84 85 with api_gateway_stubber:86 partially_mocked_gateway_normalizer.normalize_gateway(environment)87 api_gateway_stubber.assert_no_pending_responses()88 chalice_config_reader.find_deployed_config.assert_called_with('rest_api', environment)89 chalice_config_reader.chalice_config.assert_called_with()90def test_get_session_with_fips_enabled_returns_session_with_default_profile(partially_mocked_gateway_normalizer):91 result = partially_mocked_gateway_normalizer.get_session('tests/fixtures/some_config.json')92 assert result.profile_name == 'default'93def test_get_region_returns_region():94 home = os.path.expanduser("~")95 os.makedirs(f'{home}/.aws', exist_ok=True)96 con_parser = configparser.RawConfigParser()97 config_file = f'{home}/.aws/config'98 con_parser.read(config_file)99 if not con_parser.has_section('profile sdc'):100 con_parser.add_section('profile sdc')101 con_parser.set('profile sdc', 'output', 'json')102 con_parser.set('profile sdc', 'region', 'us-east-1')103 with open(config_file, 'w+') as configfile:104 con_parser.write(configfile)...
apigateway.py
Source:apigateway.py
...17 binary_enabled = ("binaryMediaTypes" in rest_api_info and18 "*/*" in rest_api_info["binaryMediaTypes"])19 if request_type in ("Create", "Update"):20 if not binary_enabled:21 apigw.update_rest_api(restApiId=rest_api_id, patchOperations=[22 {"op": "add", "path": "/binaryMediaTypes/*~1*"}23 ])24 elif request_type == "Delete":25 if binary_enabled:26 apigw.update_rest_api(restApiId=rest_api_id, patchOperations=[27 {"op": "remove", "path": "/binaryMediaTypes/*~1*"}28 ])...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!