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!!
