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

How to use create_instance_profile method in localstack

Best Python code snippet using localstack_python

test_create_instance_profile.py

Source:test_create_instance_profile.py

1import os2import sys3import logging4import unittest5import ConfigParser6import boto37import mock8DOC_DIR = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))9REPOROOT = os.path.dirname(DOC_DIR)10# Import shared testing code11sys.path.append(12    os.path.join(13        REPOROOT,14        'Testing'15    )16)17sys.path.append(os.path.join(18    DOC_DIR, "Documents/Lambdas"19))20sys.path.append(21    os.path.abspath(os.path.join(22        os.path.dirname(os.path.realpath(__file__)),23        "lib/"24    ))25)26import create_instance_profile27import ssm_testing  # noqa pylint: disable=import-error,wrong-import-position28import managedinstanceutil as util29CONFIG = ConfigParser.ConfigParser()30CONFIG.readfp(open(os.path.join(REPOROOT, 'Testing', 'defaults.cfg')))31CONFIG.read([os.path.join(REPOROOT, 'Testing', 'local.cfg')])32REGION = CONFIG.get('general', 'region')33PREFIX = CONFIG.get('general', 'resource_prefix')34SERVICE_ROLE_NAME = CONFIG.get('general', 'automation_service_role_name')35WINDOWS_AMI_ID = CONFIG.get('windows', 'windows2016.{}'.format(REGION))36LINUX_AMI_ID = CONFIG.get('linux', 'ami')37INSTANCE_TYPE = CONFIG.get('windows', 'instance_type')38SSM_DOC_NAME = PREFIX + 'automation-asg'39CFN_STACK_NAME = PREFIX + 'automation-asg'40logging.basicConfig(level=CONFIG.get('general', 'log_level').upper())41LOGGER = logging.getLogger(__name__)42logging.getLogger('botocore').setLevel(level=logging.WARNING)43boto3.setup_default_session(region_name=REGION)44orig_client = boto3.client45ec2_client = boto3.client('ec2')46as_client = boto3.client('autoscaling')47iam_client = boto3.client('iam')48def create_send_mock(result):49    return util.create_send_mock(result)50def cleanup(name):51    util.cleanup_instance_profile(iam_client, name)52def mock_boto_client(client):53    if client == "cloudformation":54        class TestCFClass:55            def __init__(self):56                pass57            def describe_stacks(self, *args, **kwargs):58                return {"Stacks": [59                    {"StackStatus": "ROLLBACK_IN_PROGRESS"}60                ]}61        return TestCFClass()62    return orig_client(client)63class CreateInstanceTest(unittest.TestCase):64    def test_create_new_profile(self):65        result = {}66        name = "{}SomeReallyRandomRoleNameThatShouldNotExist".format(PREFIX)67        try:68            with mock.patch("create_instance_profile.cfnresponse.send", side_effect=create_send_mock(result)):69                with mock.patch("create_instance_profile.boto3.client", side_effect=mock_boto_client):70                    cleanup(name)71                    event = {72                        "RequestType": "Create",73                        "StackId": "FakeID",74                        "ResourceProperties": {75                            "InstanceProfileName": name76                        }77                    }78                    context = {}79                    create_instance_profile.handler(event, context)80                    print result["args"]81                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]82                    self.assertEqual(responseStatus, "SUCCESS")83                    self.assertTrue(physicalResourceId.startswith("created:"))84                    iam_client.get_role(RoleName=name)85                    # verify instance profile was created86                    instance_profile = iam_client.get_instance_profile(InstanceProfileName=name)87                    # verify policy was added88                    attached = iam_client.list_attached_role_policies(RoleName=name)89                    arns = set([])90                    for policy in attached["AttachedPolicies"]:91                        arns.add(policy["PolicyArn"])92                    # verify role was added to profile93                    is_role_found = False94                    for role in instance_profile["InstanceProfile"]["Roles"]:95                        if role["RoleName"] == name:96                            is_role_found = True97                            continue98                    self.assertEquals(arns, set(create_instance_profile.POLICY_ARNS))99                    self.assertTrue(is_role_found, "Role was not added to the profile correctly")100        finally:101            cleanup(name)102    def test_create_existing_profile(self):103        result = {}104        name = "{}SomeReallyRandomRoleNameThatShouldNotExist".format(PREFIX)105        try:106            with mock.patch("create_instance_profile.cfnresponse.send", side_effect=create_send_mock(result)):107                with mock.patch("create_instance_profile.boto3.client", side_effect=mock_boto_client):108                    cleanup(name)109                    event = {110                        "RequestType": "Create",111                        "StackId": "FakeID",112                        "ResourceProperties": {113                            "InstanceProfileName": name114                        }115                    }116                    context = {}117                    create_instance_profile.handler(event, context)118                    # make sure role and instance profile exists119                    iam_client.get_role(RoleName=name)120                    iam_client.get_instance_profile(InstanceProfileName=name)121                    create_instance_profile.handler(event, context)122                    print result["args"]123                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]124                    self.assertEqual(responseStatus, "SUCCESS")125                    self.assertTrue(physicalResourceId.startswith("existing:"))126        finally:127            cleanup(name)128    def test_delete_created_instance_profile(self):129        result = {}130        name = "{}SomeReallyRandomRoleNameThatShouldNotExist".format(PREFIX)131        try:132            with mock.patch("create_instance_profile.cfnresponse.send", side_effect=create_send_mock(result)):133                with mock.patch("create_instance_profile.boto3.client", side_effect=mock_boto_client):134                    cleanup(name)135                    event = {136                        "RequestType": "Create",137                        "StackId": "FakeID",138                        "ResourceProperties": {139                            "InstanceProfileName": name140                        }141                    }142                    context = {}143                    create_instance_profile.handler(event, context)144                    # make sure role and instance profile exists145                    iam_client.get_role(RoleName=name)146                    iam_client.get_instance_profile(InstanceProfileName=name)147                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]148                    event = {149                        "RequestType": "Delete",150                        "StackId": "FakeID",151                        "PhysicalResourceId": physicalResourceId152                    }153                    create_instance_profile.handler(event, context)154                    print result["args"]155                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]156                    self.assertEqual(responseStatus, "SUCCESS")157                    try:158                        iam_client.get_role(RoleName=name)159                        self.assertTrue(False, "Role still exists in account")160                    except Exception as e:161                        pass162                    try:163                        iam_client.get_instance_profile(RoleName=name)164                        self.assertTrue(False, "Instance still exists in account")165                    except Exception as e:166                        pass167        finally:168            cleanup(name)169    def test_delete_existing_instance_profile(self):170        result = {}171        name = "{}SomeReallyRandomRoleNameThatShouldNotExist".format(PREFIX)172        try:173            with mock.patch("create_instance_profile.cfnresponse.send", side_effect=create_send_mock(result)):174                with mock.patch("create_instance_profile.boto3.client", side_effect=mock_boto_client):175                    cleanup(name)176                    event = {177                        "RequestType": "Create",178                        "StackId": "FakeID",179                        "ResourceProperties": {180                            "InstanceProfileName": name181                        }182                    }183                    context = {}184                    create_instance_profile.handler(event, context)185                    create_instance_profile.handler(event, context)186                    # make sure role and instance profile exists187                    iam_client.get_role(RoleName=name)188                    iam_client.get_instance_profile(InstanceProfileName=name)189                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]190                    event = {191                        "RequestType": "Delete",192                        "StackId": "FakeID",193                        "PhysicalResourceId": physicalResourceId194                    }195                    create_instance_profile.handler(event, context)196                    print result["args"]197                    (event, context, responseStatus, responseData, physicalResourceId) = result["args"]198                    self.assertEqual(responseStatus, "SUCCESS")199                    self.assertTrue(physicalResourceId.startswith("existing:"))200                    # make sure role and instance profile exists201                    iam_client.get_role(RoleName=name)202                    iam_client.get_instance_profile(InstanceProfileName=name)203        finally:...

ecs_setup.py

Source:ecs_setup.py

...101		print(e)102	else:103		print('%s policy added to %s' %(policy_arn, role_name))104# Create instance profile (Needed to attach role to an instance)105def create_instance_profile(instance_profile_name):106	try:107		iam_client.create_instance_profile(108			InstanceProfileName = instance_profile_name109		)110	except botocore.exceptions.ClientError as e:111		print(e)112	else:113		print('Instance profile %s created'%instance_profile_name)114# Add role to instance profile115def add_role_to_instance_profile(instance_profile_name, role_name):116	try:117		iam_client.add_role_to_instance_profile(118			InstanceProfileName= instance_profile_name,119			RoleName= role_name120		)121	except botocore.exceptions.ClientError as e:122		print(e)123	else:124		print('Role added to instance profile')125def add_to_config(keypairName, sgID, role1ARN, role2ARN, role3ARN, role4ARN):126	data = {'ecs_information': {'keypair_name': str(keypairName), 'security_group_ID': str(sgID), 'ecsInstanceRole_arn' : str(role1ARN), 'ecsTaskExecutionRole_arn' : str(role2ARN), 'ecsS3InputBucketAccess_arn' : str(role3ARN), 'ecsS3OutputBucketAccess_arn' : str(role4ARN)}}127	config_file = open('ecs_config.yml', 'w')128	yaml.dump(data, config_file)129	print('ecs_config file created')130# Create key pair 131key_name = create_keypair('ecs_key')132# Create ECS security group133security_group_id = create_security_group('Security group for ECS Scenario 2', 'ECS group')134# Defining a security group rule - this allows SSH access to the instance 135ipPermissions =[136		{137			'FromPort': 22,138			'IpProtocol': 'tcp',139			'IpRanges': [140				{141					'CidrIp': '0.0.0.0/0',142					'Description': 'SSH access',143				},144			],145			'ToPort': 22,146		}147	]148# Adding rule to the security group 149create_sg_rule(security_group_id, ipPermissions)150# Creating an IAM role for EC2 to access S3 151# Create a trust permission for both EC2 and ECS-Tasks (giving EC2 and ecs-tasks ability to take on the role created)152ecs_task_role_access = {153  "Version": "2012-10-17",154  "Statement": [155	{156	  "Sid": "",157	  "Effect": "Allow",158	  "Principal": {159		"Service": "ecs-tasks.amazonaws.com"160	  },161	  "Action": "sts:AssumeRole"162	}163  ]164}165ec2_role_access = {166  "Version": "2012-10-17",167  "Statement": [168	{169	  "Sid": "",170	  "Effect": "Allow",171	  "Principal": {172		"Service": "ec2.amazonaws.com"173	  },174	  "Action": "sts:AssumeRole"175	}176  ]177}178# Creating the four required roles:179ecsInstanceRole = {180'RoleName':'ecsInstanceRole',181'AssumeRolePolicyDocument' : json.dumps(ec2_role_access),182'Description':'Role to give EC2 access to Amazon EC2 Container Service.',183'MaxSessionDuration' : 43200}184ecsTaskExecutionRole = {185'RoleName':'ecsTaskExecutionRole',186'AssumeRolePolicyDocument' : json.dumps(ecs_task_role_access),187'Description':'Role to provide access to other AWS service resources that are required to run Amazon ECS tasks',188'MaxSessionDuration' : 43200}189ecsS3InputBucketAccess = {190'RoleName':'ecsS3InputBucketAccess',191'AssumeRolePolicyDocument' : json.dumps(ecs_task_role_access),192'Description':'Role to provide access to input bucket to ecs tasks',193'MaxSessionDuration' : 43200}194ecsS3OutputBucketAccess = {195'RoleName':'ecsS3OutputBucketAccess',196'AssumeRolePolicyDocument' : json.dumps(ecs_task_role_access),197'Description':'Role to provide access to output bucket to ecs tasks',198'MaxSessionDuration' : 43200}199# Creating the roles200ecsInstanceRole_arn, ecsInstanceRole_name = create_iam_role(**ecsInstanceRole)201ecsTaskExecutionRole_arn, ecsTaskExecutionRole_name = create_iam_role(**ecsTaskExecutionRole)202ecsS3InputBucketAccess_arn, ecsS3InputBucketAccess_name = create_iam_role(**ecsS3InputBucketAccess)203ecsS3OutputBucketAccess_arn, ecsS3OutputBucketAccess_name = create_iam_role(**ecsS3OutputBucketAccess)204# Adding the aws managed policies to ecsInstanceRole and ecsTaskExecutionRole205add_policy('arn:aws:iam::aws:policy/service-role/AmazonEC2ContainerServiceforEC2Role', ecsInstanceRole_name)206add_policy('arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy', ecsTaskExecutionRole_name)207# Create policies for ecsS3InputBucketAccess and ecsS3OutputBucketAccess roles208input_bucket_access = {209  "Version": "2012-10-17",210  "Statement": [211    {212      "Effect": "Allow",213      "Action": ["s3:ListBucket"],214      "Resource": [input_bucket]215    },216    {217      "Effect": "Allow",218      "Action": [219        "s3:GetObject"220      ],221      "Resource": [input_bucket_files]222    }223  ]224}225output_bucket_access = {226   "Version":"2012-10-17",227   "Statement":[228	  {229		 "Effect":"Allow",230		 "Action":[231			"s3:ListBucket"232		 ],233		 "Resource": [output_bucket]234	  },235	  {236		 "Effect":"Allow",237		 "Action":[238			"s3:PutObject"239		 ],240		 "Resource": [output_bucket_files]241	  }242   ]243}244input_policy = create_policy('ecsS3InputBucketAccess_policy', input_bucket_access)245output_policy = create_policy('ecsS3OutputBucketAccess_policy', output_bucket_access)246# Adding the created policies to ecsS3InputBucketAccess and ecsS3OutputBucketAccess roles247add_policy(input_policy, ecsS3InputBucketAccess_name)248add_policy(output_policy , ecsS3OutputBucketAccess_name)249# Create instance profiles and add roles -> Name of instance profile == same as role name (makes it easier and is how this occurs if done through the console)250# ecsInstanceRole 251create_instance_profile(ecsInstanceRole_name)252add_role_to_instance_profile(ecsInstanceRole_name, ecsInstanceRole_name)253# ecsTaskExecutionRole254create_instance_profile(ecsTaskExecutionRole_name)255add_role_to_instance_profile(ecsTaskExecutionRole_name, ecsTaskExecutionRole_name)256# ecsS3InputBucketAccess257create_instance_profile(ecsS3InputBucketAccess_name)258add_role_to_instance_profile(ecsS3InputBucketAccess_name, ecsS3InputBucketAccess_name)259# ecsS3OutputBucketAccess260create_instance_profile(ecsS3OutputBucketAccess_name)261add_role_to_instance_profile(ecsS3OutputBucketAccess_name, ecsS3OutputBucketAccess_name)262# Load information into config file: ...

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.