Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use select_resource_config method in localstack

Best Python code snippet using localstack_python

EC2_INSTANCE_EBS_VOLUME_TAGS_MATCH_test.py

Source:EC2_INSTANCE_EBS_VOLUME_TAGS_MATCH_test.py

1# Copyright 2017-2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You may4# not use this file except in compliance with the License. A copy of the License is located at5#6#        http://aws.amazon.com/apache2.0/7#8# or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS,9# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for10# the specific language governing permissions and limitations under the License.11import unittest12from unittest.mock import MagicMock, patch13from rdklib import ComplianceType, Evaluation14from rdklibtest import assert_successful_evaluation15##############16# Parameters #17##############18# Define the default resource to report to Config Rules19RESOURCE_TYPE = "AWS::EC2::Volume"20#############21# Main Code #22#############23MODULE = __import__("EC2_INSTANCE_EBS_VOLUME_TAGS_MATCH")24RULE = MODULE.EC2_INSTANCE_EBS_VOLUME_TAGS_MATCH()25CLIENT_FACTORY = MagicMock()26EC2_CLIENT_MOCK = MagicMock()27CONFIG_CLIENT_MOCK = MagicMock()28def mock_get_client(client_name, *args, **kwargs):29    if client_name == "config":30        return CONFIG_CLIENT_MOCK31    raise Exception("Attempting to create an unknown client")32def mock_evaluator_handle(event, context):33    return f"Event: {event} - Context: {context}"34@patch.object(CLIENT_FACTORY, "build_client", MagicMock(side_effect=mock_get_client))35class ComplianceTest(unittest.TestCase):36    ci_wrong_resource_type = {"resourceType": "AWS::EC2::WRONG"}37    # AWS::EC2::Volume - Configuration Items38    ci_ebs_volume_not_attached = {"resourceType": "AWS::EC2::Volume", "configuration": {"attachments": []}}39    ci_ebs_volume_attached_with_tags1 = {40        "resourceType": "AWS::EC2::Volume",41        "resourceId": "vol-123abc",42        "configuration": {43            "attachments": [{"instanceId": "i-123abc", "volumeId": "vol-123abc"}],44            "tags": [45                {"key": "tag1", "value": "item1"},46                {"key": "tag2", "value": "item2"},47                {"key": "aws:test1", "value": "awsskip"},48                {"key": "tag3", "value": "exclusive2volume"},49            ],50        },51    }52    ci_ebs_volume_attached_with_tags2 = {53        "resourceType": "AWS::EC2::Volume",54        "resourceId": "vol-123abc",55        "configuration": {56            "attachments": [{"instanceId": "i-123abc", "volumeId": "vol-123abc"}],57            "tags": [58                {"key": "tag1", "value": "item55"},59                {"key": "tag2", "value": "item77"},60                {"key": "aws:test2", "value": "awsskip"},61                {"key": "tag3", "value": "exclusive2volume"},62            ],63        },64    }65    # AWS::EC2::Instance - Configuration Items66    ci_ec2_instance_no_ebs_volumes_attached = {67        "resourceType": "AWS::EC2::Instance",68        "resourceId": "i-123abc",69        "configuration": {"blockDeviceMappings": []},70    }71    ci_ec2_instance_with_tags1 = {72        "resourceType": "AWS::EC2::Instance",73        "resourceId": "i-123abc",74        "configuration": {75            "blockDeviceMappings": [{"ebs": {"volumeId": "vol-123abc"}}],76            "tags": [77                {"key": "tag1", "value": "item1"},78                {"key": "tag2", "value": "item2"},79                {"key": "aws:test1", "value": "awsskip"},80            ],81        },82    }83    ci_ec2_instance_with_tags2 = {84        "resourceType": "AWS::EC2::Instance",85        "resourceId": "i-123abc",86        "configuration": {87            "blockDeviceMappings": [{"ebs": {"volumeId": "vol-123abc"}}],88            "tags": [89                {"key": "tag1", "value": "item55"},90                {"key": "tag2", "value": "item77"},91                {"key": "aws:test2", "value": "awsskip"},92            ],93        },94    }95    ebs_volume_results_tags2 = {96        "Results": [97            '{"tags":[{"key":"tag1","value":"item55"},{"key":"tag2","value":"item77"},{"key":"aws:test2","value":"awsskip"},{"key":"tag3","value":"exclusive2volume"}]}'98        ]99    }100    ebs_volume_results_tags2_with_token = {101        "Results": [102            '{"tags":[{"key":"tag1","value":"item55"},{"key":"tag2","value":"item77"},{"key":"aws:test2","value":"awsskip"},{"key":"tag3","value":"exclusive2volume"}]}'103        ],104        "NextToken": False,105    }106    ec2_instance_results_tags2 = {107        "Results": ['{"tags":[{"key":"tag1","value":"item55"},{"key":"tag2","value":"item77"},{"key":"aws:test2","value":"awsskip"}]}']108    }109    ec2_instance_results_tags2_with_token = {110        "Results": ['{"tags":[{"key":"tag1","value":"item55"},{"key":"tag2","value":"item77"},{"key":"aws:test2","value":"awsskip"}]}'],111        "NextToken": False,112    }113    def setUp(self):114        CONFIG_CLIENT_MOCK.reset_mock()115    # Scenario - Returns Empty if Resource Type is received116    def test_scenario_evaluatechange_wrong_resource_type_returnsempty(self):117        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_wrong_resource_type, {})118        response_expected = []119        self.assertEqual(response, response_expected)120    # Scenario 1: EC2 Resource Type - EC2 Instance has no EBS volumes attached.121    def test_scenario1_evaluatechange_ec2_instance_no_ebs_volumes_attached_returnsnotapplicable(self):122        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_ec2_instance_no_ebs_volumes_attached, {})123        response_expected = [Evaluation(ComplianceType.NOT_APPLICABLE)]124        assert_successful_evaluation(self, response, response_expected)125    # Scenario 2: Volume Resource Type - EBS Volume not attached to an EC2 instance.126    def test_scenario2_evaluatechange_ebs_volume_not_attached_returnsnotapplicable(self):127        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_ebs_volume_not_attached, {})128        response_expected = [Evaluation(ComplianceType.NOT_APPLICABLE)]129        assert_successful_evaluation(self, response, response_expected)130    # Scenario 3: EC2 Resource Type - EBS Volumes attached, includes same tags as EC2 Instance.131    def test_scenario3a_evaluatechange_ec2_instance_volumes_has_tags_from_ec2_instance_returnscompliant(self):132        CONFIG_CLIENT_MOCK.select_resource_config = MagicMock(return_value=self.ebs_volume_results_tags2)133        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_ec2_instance_with_tags2, {})134        response_expected = [Evaluation(ComplianceType.COMPLIANT, "vol-123abc", RESOURCE_TYPE)]135        assert_successful_evaluation(self, response, response_expected)136    def test_scenario3b_evaluatechange_ec2_instance_volumes_has_tags_from_ec2_instance_with_token_returnscompliant(self):137        CONFIG_CLIENT_MOCK.select_resource_config = MagicMock(return_value=self.ebs_volume_results_tags2_with_token)138        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_ec2_instance_with_tags2, {})139        response_expected = [Evaluation(ComplianceType.COMPLIANT, "vol-123abc", RESOURCE_TYPE)]140        assert_successful_evaluation(self, response, response_expected)141    # Scenario 4: EC2 Resource Type - EBS Volumes attached, does not include same tags as EC2 Instance.142    def test_scenario4_evaluatechange_ec2_instance_volumes_missing_tags_from_ec2_instance_returnsnoncompliant(self):143        CONFIG_CLIENT_MOCK.select_resource_config = MagicMock(return_value=self.ebs_volume_results_tags2)144        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_ec2_instance_with_tags1, {})145        response_expected = [Evaluation(ComplianceType.NON_COMPLIANT, "vol-123abc", RESOURCE_TYPE)]146        assert_successful_evaluation(self, response, response_expected)147    # Scenario 5: Volume Resource Type - EBS Volume Tags includes same tags as EC2 Instance its attached to.148    def test_scenario5a_evaluatechange_ebs_volume_attached_has_tags_from_ec2_instance_returnscompliant(self):149        CONFIG_CLIENT_MOCK.select_resource_config = MagicMock(return_value=self.ec2_instance_results_tags2)150        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_ebs_volume_attached_with_tags2, {})151        response_expected = [Evaluation(ComplianceType.COMPLIANT)]152        assert_successful_evaluation(self, response, response_expected)153    def test_scenario5b_evaluatechange_ebs_volume_attached_has_tags_from_ec2_instance_with_token_returnscompliant(self):154        CONFIG_CLIENT_MOCK.select_resource_config = MagicMock(return_value=self.ec2_instance_results_tags2_with_token)155        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_ebs_volume_attached_with_tags2, {})156        response_expected = [Evaluation(ComplianceType.COMPLIANT)]157        assert_successful_evaluation(self, response, response_expected)158    # Scenario 6: Volume Resource Type - EBS Volume Tags does not include the same tags as EC2 Instance its attached to.159    def test_scenario6_evaluatechange_ebs_volume_attached_missing_tags_from_ec2_instance_returnsnoncompliant(self):160        CONFIG_CLIENT_MOCK.select_resource_config = MagicMock(return_value=self.ec2_instance_results_tags2)161        response = RULE.evaluate_change({}, CLIENT_FACTORY, self.ci_ebs_volume_attached_with_tags1, {})162        response_expected = [Evaluation(ComplianceType.NON_COMPLIANT)]163        assert_successful_evaluation(self, response, response_expected)164    # No scenario lambda handler passed an event and context165    @patch.object(MODULE.Evaluator, "handle", side_effect=mock_evaluator_handle)166    def test_lambda_handler_called_with_event_and_context(self, mock_evaluator):167        response = MODULE.lambda_handler("event", "context")168        response_expected = "Event: event - Context: context"...

test_inventory_reader.py

Source:test_inventory_reader.py

1#!/usr/bin/env python2# AWS DISCLAMER3# ---4# The following files are provided by AWS Professional Services describe the process to create a IAM Policy with description.5# These are non-production ready and are to be used for testing purposes.6# These files is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES7# OR CONDITIONS OF ANY KIND, either express or implied. See the License8# for the specific language governing permissions and limitations under the License.9# (c) 2019 Amazon Web Services, Inc. or its affiliates. All Rights Reserved.10# This AWS Content is provided subject to the terms of the AWS Customer Agreement available at11# http://aws.amazon.com/agreement or other written agreement between Customer and Amazon Web Services, Inc.â12from botocore.exceptions import ClientError13from callee import String, Contains14import json15import os16from unittest.mock import MagicMock, Mock, patch, ANY17import pytest18from inventory.mappers import DataMapper19import inventory.readers20from inventory.readers import AwsConfigInventoryReader21def setup_function():22    os.environ["ACCOUNT_LIST"] = '[ { "name": "foo", "id": "210987654321"} ]'23    os.environ["CROSS_ACCOUNT_ROLE_NAME"] = "foobar"24def test_given_valid_arn_then_aws_partition_determined():25    mock_lambda_context = Mock()26    mock_lambda_context.invoked_function_arn = "arn:aws:lambda:us-east-1:123456789012:function:testing"27    reader = AwsConfigInventoryReader(lambda_context=mock_lambda_context)28    aws_partition = reader._get_aws_partition()29    assert aws_partition == "aws", "Partition of ARN in context is aws"30@patch("inventory.readers._logger", autospec=True)31def test_given_unsupported_resource_type_then_warning_is_logged(mock_logger):32    mock_mapper = Mock(spec=DataMapper)33    mock_mapper.can_map.return_value = False34    mock_config_client_factory = Mock()35    mock_config_client_factory.return_value \36                              .select_resource_config \37                              .return_value = { "NextToken": None,38                                                "Results": [ json.dumps({ "resourceType": "foobar" }) ] }39    reader = AwsConfigInventoryReader(lambda_context=MagicMock(), sts_client=Mock(), mappers=[mock_mapper])40    reader._get_config_client = mock_config_client_factory41    all_inventory = reader.get_resources_from_all_accounts()42    assert len(all_inventory) == 0, "no inventory should be returned since there was nothing to map"43    mock_logger.warning.assert_called_with(String() & Contains("skipping mapping"))44@patch("inventory.readers._logger", autospec=True)45def test_given_error_from_boto_then_account_is_skipped_but_others_still_processed(mock_logger):46    os.environ["ACCOUNT_LIST"] = '[ { "name": "foo", "id": "210987654321" }, { "name": "bar", "id": "123456789012" } ]'47    mock_mapper = Mock(spec=DataMapper)48    mock_mapper.can_map.return_value = True49    mock_mapper.map.return_value = [ { "test": True }]50    mock_select_resource_config = Mock(side_effect=[ ClientError(error_response={'Error': {'Code': 'ResourceInUseException'}}, operation_name="select_resource_config"),51                                                    { "NextToken": None,52                                                      "Results": [ json.dumps({ "resourceType": "foobar" }) ] }])53    mock_config_client_factory = Mock()54    mock_config_client_factory.return_value \55                              .select_resource_config = mock_select_resource_config56    reader = AwsConfigInventoryReader(lambda_context=MagicMock(), sts_client=Mock(), mappers=[mock_mapper])57    reader._get_config_client = mock_config_client_factory58    59    all_inventory = reader.get_resources_from_all_accounts()60    assert len(all_inventory) == 1, "inventory from the successful call should be returned"61    assert len(mock_select_resource_config.mock_calls) == 2, "boto should have been called twice to page through results"62    mock_logger.error.assert_called_with(String() & Contains("moving onto next account"), ANY, ANY, exc_info=True)63def test_given_multiple_resource_pages_from_boto_then_reader_loops_through_all_pages():64    mock_mapper = Mock(spec=DataMapper)65    mock_mapper.can_map.return_value = False66    mock_select_resource_config = Mock(side_effect=[{ "NextToken": "nextpage",67                                                      "Results": [ json.dumps({ "resourceType": "foobar" }) ] },68                                                    { "NextToken": None,69                                                      "Results": [ json.dumps({ "resourceType": "foobar" }) ] }])70    mock_config_client_factory = Mock()71    mock_config_client_factory.return_value \72                              .select_resource_config = mock_select_resource_config73    readerx = AwsConfigInventoryReader(lambda_context=MagicMock(), sts_client=Mock(), mappers=[mock_mapper])74    readerx._get_config_client = mock_config_client_factory75    all_inventory = readerx.get_resources_from_all_accounts()76    assert len(all_inventory) == 0, "no inventory should be returned since there was nothing to map"77    assert len(mock_select_resource_config.mock_calls) == 2, "boto should have been called twice to page through results"...

db_in.py

Source:db_in.py

...10    password="interpark",11    host="10.222.10.189",12    db="all_arn",13    charset='utf8')14def select_resource_config():15    config = boto3.client('config')16    arn = []17    result = []18    config_arn = {}19    NextToken = ''20    # cursor = conn.cursor(pymysql.cursors.DictCursor)21    # cursor.execute("DROP TABLE IF EXISTS all_arn")22    # cursor.execute("CREATE TABLE all_arn(arn LONGTEXT, doc JSON)")23    while True:24        response = config.select_resource_config(25            Expression="SELECT * ", Limit=100, NextToken=NextToken26        )27        if 'NextToken' not in response:28            break29        NextToken = response['NextToken']30        if NextToken == '':31            break32        for r in response['Results']:33            js = json.loads(r)  # json(string) -> dict ë³í34            if 'arn' in js:35                config_arn['arn'] = js['arn']36                config_arn['doc'] = js37                result.append(dict(config_arn))38    for i in result:39        sql = "insert into all_arn values('" + \40            i['arn'] + "' , '" + json.dumps(i) + "')"41        # cursor.execute(sql)42    # conn.commit()43    # conn.close()44def extract_tags():45    restag = boto3.client('resourcegroupstaggingapi')46    response = restag.get_resources()47    cursors = conn.cursor(pymysql.cursors.DictCursor)48    cursors.execute("DROP TABLE IF EXISTS tagged_arn")49    cursors.execute("CREATE TABLE tagged_arn(arn LONGTEXT, doc JSON)")50    while 'PaginationToken' in response and response['PaginationToken']:51        token = response['PaginationToken']52        response = restag.get_resources(53            ResourcesPerPage=100, PaginationToken=token)54        for res in response['ResourceTagMappingList']:55            sql = "insert into tagged_arn values('" + \56                res['ResourceARN'] + "', '" + json.dumps(res) + "')"57            print(sql)58            cursors.execute(sql)59            conn.commit()60    conn.close()61def main():62    # select_resource_config()63    extract_tags()64if __name__ == '__main__':65    main()66    67    ...

Automation Testing Tutorials

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.