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

How to use create_security_group method in localstack

Best Python code snippet using localstack_python

test_create_security_group.py

Source:test_create_security_group.py

1import os2import sys3import logging4import unittest5import ConfigParser6import boto37import json8import mock9DOC_DIR = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))10REPOROOT = os.path.dirname(DOC_DIR)11# Import shared testing code12sys.path.append(13    os.path.join(14        REPOROOT,15        'Testing'16    )17)18sys.path.append(os.path.join(19    DOC_DIR, "Documents/Lambdas"20))21sys.path.append(22    os.path.abspath(os.path.join(23        os.path.dirname(os.path.realpath(__file__)),24        "lib/"25    ))26)27import create_security_group28import ssm_testing  # noqa pylint: disable=import-error,wrong-import-position29import managedinstanceutil as util30CONFIG = ConfigParser.ConfigParser()31CONFIG.readfp(open(os.path.join(REPOROOT, 'Testing', 'defaults.cfg')))32CONFIG.read([os.path.join(REPOROOT, 'Testing', 'local.cfg')])33REGION = CONFIG.get('general', 'region')34PREFIX = CONFIG.get('general', 'resource_prefix')35SERVICE_ROLE_NAME = CONFIG.get('general', 'automation_service_role_name')36WINDOWS_AMI_ID = CONFIG.get('windows', 'windows2016.{}'.format(REGION))37LINUX_AMI_ID = CONFIG.get('linux', 'ami')38INSTANCE_TYPE = CONFIG.get('windows', 'instance_type')39SSM_DOC_NAME = PREFIX + 'automation-asg'40CFN_STACK_NAME = PREFIX + 'automation-asg'41logging.basicConfig(level=CONFIG.get('general', 'log_level').upper())42LOGGER = logging.getLogger(__name__)43logging.getLogger('botocore').setLevel(level=logging.WARNING)44boto3.setup_default_session(region_name=REGION)45orig_client = boto3.client46ec2_client = boto3.client('ec2')47as_client = boto3.client('autoscaling')48iam_client = boto3.client('iam')49def create_send_mock(result):50    return util.create_send_mock(result)51def find_default_vpc():52    return util.find_default_vpc(ec2_client)53def cleanup(vpc_id, group_name):54    util.cleanup_security_groups(ec2_client, vpc_id, group_name)55def mock_boto_client(client):56    if client == "cloudformation":57        class TestCFClass:58            def __init__(self):59                pass60            def describe_stacks(self, *args, **kwargs):61                return {"Stacks": [62                    {"StackStatus": "ROLLBACK_IN_PROGRESS"}63                ]}64        return TestCFClass()65    return orig_client(client)66class CreateSecurityGroupTests(unittest.TestCase):67    def test_create_new_security_group_linux_no_cidr(self):68        result = {"args": None}69        name = "{}SomeReallyRandomGroupName".format(PREFIX)70        default_vpc = None71        try:72            with mock.patch("create_security_group.cfnresponse.send", side_effect=create_send_mock(result)):73                with mock.patch("create_security_group.boto3.client", side_effect=mock_boto_client):74                    default_vpc = find_default_vpc()75                    cleanup(default_vpc["VpcId"], name)76                    event = {77                        "RequestType": "Create",78                        "StackId": "FakeID",79                        "ResourceProperties": {80                            "GroupName": name,81                            "VpcId": default_vpc["VpcId"],82                            "Platform": "linux",83                            "RemoteAccessCidr": "10.0.0.0/24"84                        }85                    }86                    context = {}87                    create_security_group.handler(event, context)88                    print result["args"]89                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]90                    self.assertEqual(responseStatus, "SUCCESS")91                    self.assertTrue(physicalResourceId.startswith("created:"))92                    security_group = \93                        ec2_client.describe_security_groups(GroupIds=[responseData["SecurityGroupId"]])[94                            "SecurityGroups"][0]95                    self.assertEqual(security_group["GroupName"], name)96                    self.assertEqual(len(security_group["IpPermissions"]), 0)97        finally:98            if default_vpc is not None:99                cleanup(default_vpc["VpcId"], name)100    def test_create_new_security_group_linux_with_cidr(self):101        result = {"args": None}102        name = "{}SomeReallyRandomGroupName".format(PREFIX)103        default_vpc = None104        try:105            with mock.patch("create_security_group.cfnresponse.send", side_effect=create_send_mock(result)):106                with mock.patch("create_security_group.boto3.client", side_effect=mock_boto_client):107                    default_vpc = find_default_vpc()108                    cleanup(default_vpc["VpcId"], name)109                    event = {110                        "RequestType": "Create",111                        "StackId": "FakeID",112                        "ResourceProperties": {113                            "GroupName": name,114                            "VpcId": default_vpc["VpcId"],115                            "Platform": "linux",116                            'AccessCidr': "10.0.0.0/24"117                        }118                    }119                    context = {}120                    create_security_group.handler(event, context)121                    print result["args"]122                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]123                    self.assertEqual(responseStatus, "SUCCESS")124                    self.assertTrue(physicalResourceId.startswith("created:"))125                    print [responseData["SecurityGroupId"]]126                    security_group = \127                        ec2_client.describe_security_groups(GroupIds=[responseData["SecurityGroupId"]])[128                            "SecurityGroups"][0]129                    self.assertEqual(security_group["GroupName"], name)130                    self.assertEqual(len(security_group["IpPermissions"]), 1)131                    self.assertDictContainsSubset({132                        "FromPort": 22,133                        "IpRanges": [{"CidrIp": "10.0.0.0/24"}]134                    }, security_group["IpPermissions"][0])135        finally:136            if default_vpc is not None:137                cleanup(default_vpc["VpcId"], name)138    def test_create_new_security_group_windows_with_cidr(self):139        result = {"args": None}140        name = "{}SomeReallyRandomGroupName".format(PREFIX)141        default_vpc = None142        try:143            with mock.patch("create_security_group.cfnresponse.send", side_effect=create_send_mock(result)):144                with mock.patch("create_security_group.boto3.client", side_effect=mock_boto_client):145                    default_vpc = find_default_vpc()146                    cleanup(default_vpc["VpcId"], name)147                    event = {148                        "RequestType": "Create",149                        "StackId": "FakeID",150                        "ResourceProperties": {151                            "GroupName": name,152                            "VpcId": default_vpc["VpcId"],153                            "Platform": "windows",154                            'AccessCidr': "10.0.0.0/24"155                        }156                    }157                    context = {}158                    create_security_group.handler(event, context)159                    print result["args"]160                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]161                    self.assertEqual(responseStatus, "SUCCESS")162                    self.assertTrue(physicalResourceId.startswith("created:"))163                    print [responseData["SecurityGroupId"]]164                    security_group = \165                        ec2_client.describe_security_groups(GroupIds=[responseData["SecurityGroupId"]])[166                            "SecurityGroups"][0]167                    self.assertEqual(security_group["GroupName"], name)168                    self.assertEqual(len(security_group["IpPermissions"]), 1)169                    self.assertDictContainsSubset({170                        "FromPort": 3389,171                        "IpRanges": [{"CidrIp": "10.0.0.0/24"}]172                    }, security_group["IpPermissions"][0])173        finally:174            if default_vpc is not None:175                cleanup(default_vpc["VpcId"], name)176    def test_existing_new_security_group_windows_with_cidr(self):177        result = {"args": None}178        name = "{}SomeReallyRandomGroupName".format(PREFIX)179        default_vpc = None180        try:181            with mock.patch("create_security_group.cfnresponse.send", side_effect=create_send_mock(result)):182                with mock.patch("create_security_group.boto3.client", side_effect=mock_boto_client):183                    default_vpc = find_default_vpc()184                    cleanup(default_vpc["VpcId"], name)185                    event = {186                        "RequestType": "Create",187                        "StackId": "FakeID",188                        "ResourceProperties": {189                            "GroupName": name,190                            "VpcId": default_vpc["VpcId"],191                            "Platform": "windows",192                            'AccessCidr': "10.0.0.0/24"193                        }194                    }195                    context = {}196                    create_security_group.handler(event, context)197                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]198                    _ = \199                        ec2_client.describe_security_groups(GroupIds=[responseData["SecurityGroupId"]])[200                            "SecurityGroups"][0]201                    create_security_group.handler(event, context)202                    print result["args"]203                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]204                    self.assertEqual(responseStatus, "SUCCESS")205                    self.assertTrue(physicalResourceId.startswith("existing:"))206        finally:207            if default_vpc is not None:208                cleanup(default_vpc["VpcId"], name)209    def test_created_delete(self):210        result = {"args": None}211        name = "{}SomeReallyRandomGroupName".format(PREFIX)212        default_vpc = None213        try:214            with mock.patch("create_security_group.cfnresponse.send", side_effect=create_send_mock(result)):215                with mock.patch("create_security_group.boto3.client", side_effect=mock_boto_client):216                    default_vpc = find_default_vpc()217                    cleanup(default_vpc["VpcId"], name)218                    event = {219                        "RequestType": "Create",220                        "StackId": "FakeID",221                        "ResourceProperties": {222                            "GroupName": name,223                            "VpcId": default_vpc["VpcId"],224                            "Platform": "windows",225                            'AccessCidr': "10.0.0.0/24"226                        }227                    }228                    context = {}229                    create_security_group.handler(event, context)230                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]231                    security_group_id = responseData["SecurityGroupId"]232                    _ = ec2_client.describe_security_groups(GroupIds=[security_group_id])["SecurityGroups"][0]233                    event = {234                        "StackId": "FakeID",235                        "RequestType": "Delete",236                        "PhysicalResourceId": physicalResourceId237                    }238                    create_security_group.handler(event, context)239                    print result["args"]240                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]241                    self.assertEqual(responseStatus, "SUCCESS")242                    try:243                        ec2_client.describe_security_groups(GroupIds=[security_group_id])244                        self.assertTrue(False, "SecurityGroup still exists")245                    except Exception:246                        pass247        finally:248            if default_vpc is not None:249                cleanup(default_vpc["VpcId"], name)250    def test_existing_delete(self):251        result = {"args": None}252        name = "{}SomeReallyRandomGroupName".format(PREFIX)253        default_vpc = None254        try:255            with mock.patch("create_security_group.cfnresponse.send", side_effect=create_send_mock(result)):256                with mock.patch("create_security_group.boto3.client", side_effect=mock_boto_client):257                    default_vpc = find_default_vpc()258                    cleanup(default_vpc["VpcId"], name)259                    event = {260                        "RequestType": "Create",261                        "StackId": "FakeID",262                        "ResourceProperties": {263                            "GroupName": name,264                            "VpcId": default_vpc["VpcId"],265                            "Platform": "windows",266                            'AccessCidr': "10.0.0.0/24"267                        }268                    }269                    context = {}270                    create_security_group.handler(event, context)271                    create_security_group.handler(event, context)272                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]273                    security_group_id = responseData["SecurityGroupId"]274                    _ = ec2_client.describe_security_groups(GroupIds=[security_group_id])["SecurityGroups"][0]275                    event = {276                        "StackId": "FakeID",277                        "RequestType": "Delete",278                        "PhysicalResourceId": physicalResourceId279                    }280                    create_security_group.handler(event, context)281                    print result["args"]282                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]283                    self.assertEqual(responseStatus, "SUCCESS")284                    groups = ec2_client.describe_security_groups(GroupIds=[security_group_id])["SecurityGroups"]285                    self.assertEqual(len(groups), 1)286        finally:287            if default_vpc is not None:...

test_providers_aws_interface.py

Source:test_providers_aws_interface.py

...103        self.boto.get_service_config = tmp_get_service_config104        self.boto.create_security_group = tmp_create_security_group105        self.boto.configure_security_group = tmp_configure_security_group106    def test_create_security_group_when_sg_does_not_exists(self):107        self.assertFalse(self.boto.create_security_group('yegbe'))108        self.boto.client.create_security_group.assert_called_once()109    def test_create_security_group_when_sg_exists(self):110        self.boto.client.create_security_group.side_effect = self.interface.botocore.exceptions.ClientError111        self.assertTrue(self.boto.create_security_group('yegbe'))112        self.boto.client.create_security_group.assert_called_once()113        self.boto.logger.log.warning.assert_called_once()114    def test_configure_security_group_without_exceptions(self):115        configMock.ports = [44]116        self.boto.configure_security_group('gbengbeleku')117        self.boto.client.authorize_security_group_ingress.assert_called_once_with(GroupName='gbengbeleku',118                                                                                  IpProtocol="tcp",119                                                                                  CidrIp="0.0.0.0/0",120                                                                                  FromPort=int(44),121                                                                                  ToPort=int(44))122    def test_configure_security_group_with_exceptions(self):123        configMock.ports = [44]124        self.boto.client.authorize_security_group_ingress.side_effect = self.interface.botocore.exceptions.ClientError125        self.boto.configure_security_group('gbengbeleku')...

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.