Best Python code snippet using localstack_python
stackdeploy.py
Source:stackdeploy.py
...17 self.ec2 = boto3.client('ec2',18 aws_session_token=self.aws_token,19 region_name=self.aws_region,20 )21 self.ec2.describe_account_attributes()22 except botocore.exceptions.ClientError as err:23 print('[*] {}'.format(err))24 sys.exit(1)25 try:26 self.rds = boto3.client('rds',27 aws_session_token=self.aws_token,28 region_name=self.aws_region,29 )30 self.rds.describe_account_attributes()31 except botocore.exceptions.ClientError as err:32 print('[*] {}'.format(err))33 sys.exit(1)34 elif auth_type == 'AWS Key':35 try:36 self.ec2 = boto3.client('ec2',37 aws_access_key_id=self.aws_acces,38 aws_secret_access_key=self.aws_secret,39 region_name=self.aws_region,40 )41 self.ec2.describe_account_attributes()42 except botocore.exceptions.ClientError as err:43 print('[*] {}'.format(err))44 sys.exit(1)45 try:46 self.rds = boto3.client('rds',47 aws_access_key_id=self.aws_acces,48 aws_secret_access_key=self.aws_secret,49 region_name=self.aws_region,50 )51 self.rds.describe_account_attributes()52 except botocore.exceptions.ClientError as err:53 print('[*] {}'.format(err))54 sys.exit(1)55 def get_vpcs(self):56 v = []57 vpcs = self.ec2.describe_vpcs()58 if vpcs['Vpcs']:59 for vpc in vpcs['Vpcs']:60 if 'Tags' in vpc:61 for tag in vpc['Tags']:62 if tag['Key'] == 'Name':63 name = tag['Value']64 v.append({'name': 'Vpc Name {}, Vpc Id {}'.format(name, vpc['VpcId'])})65 else:66 v.append({'name': 'Vpc Name {}, Vpc Id {}'.format('', vpc['VpcId'])})67 else:68 v.append({'name': 'Vpc Name {}, Vpc Id {}'.format('', vpc['VpcId'])})69 return v70 def get_sg_from_vpc(self, vpc):71 s = []72 sgs = self.ec2.describe_security_groups(73 Filters=[74 {75 'Name': 'vpc-id',76 'Values': [77 vpc78 ]79 }80 ]81 )82 for sg in sgs['SecurityGroups']:83 s.append({'name': 'SG Name {}, SG Id {}'.format(sg['GroupName'], sg['GroupId'])})84 return s85 def get_db_subnet_group(self, vpc):86 db_subnet = self.rds.describe_db_subnet_groups()87 dbsg = []88 for r in db_subnet['DBSubnetGroups']:89 if r['VpcId'] == vpc:90 dbsg.append({'name': r['DBSubnetGroupName']})91 return dbsg92 def create_db_subnet_group(self, vpc, name):93 vpc_subnets = self.ec2.describe_subnets()94 subnetids = []95 for subnet in vpc_subnets['Subnets']:96 if subnet['VpcId'] == vpc:97 subnetids.append(subnet['SubnetId'])98 if subnetids:99 response = self.rds.create_db_subnet_group(100 DBSubnetGroupName='dbsg-{}'.format(name),101 DBSubnetGroupDescription='DB Subnet for {}'.format(vpc),102 SubnetIds=subnetids103 )104 return [{'name': response['DBSubnetGroup']['DBSubnetGroupName']}]105 else:106 print('[*] Please review your subnet configuration for Vpc-Id {}'.format(vpc))107 sys.exit(1)108 def get_az_from_db_subnet_group(self, dbsname):109 response = self.rds.describe_db_subnet_groups(110 DBSubnetGroupName=dbsname111 )112 availabilityzone = []113 for res in response['DBSubnetGroups'][0]['Subnets']:114 availabilityzone.append(res['SubnetAvailabilityZone']['Name'])115 return random.choice(availabilityzone)116 def get_rds_engines(self):117 versions = []118 if self.auth_type == 'AWS Token':119 try:120 client = boto3.client('rds',121 aws_session_token=self.aws_token,122 )123 client.describe_account_attributes()124 except botocore.exceptions.ClientError as err:125 print('[*] {}'.format(err))126 sys.exit(1)127 elif self.auth_type == 'AWS Key':128 try:129 client = boto3.client('rds',130 aws_access_key_id=self.aws_acces,131 aws_secret_access_key=self.aws_secret,132 )133 client.describe_account_attributes()134 except botocore.exceptions.ClientError as err:135 print('[*] {}'.format(err))136 sys.exit(1)137 paginator = client.get_paginator('describe_db_engine_versions')138 pages = paginator.paginate()139 for page in pages:140 for obj in page['DBEngineVersions']:141 e = {'Engine': obj['Engine'],142 'Version': obj['EngineVersion'],143 'ParameterGroupFamily': obj['DBParameterGroupFamily']}144 versions.append(e)145 return versions146 def get_rds_major_engines(self, engines):147 result = []148 engine_vendors = []149 for engine in engines:150 engine_vendors.append(engine['Engine'])151 engines = set(engine_vendors)152 for engine in sorted(engines):153 # TODO unlock all engines in rds154 if engine.startswith('oracle'):155 result.append({'name': engine})156 return result157 def get_rds_minor_engine_version(self, engine, versions):158 minor_versions = []159 for version in versions:160 if version['Engine'] == engine:161 minor_versions.append({'name': version['Version']})162 return minor_versions163 def get_rds_parameter_group_family(self, engines, engine, version):164 for e in engines:165 if e['Engine'] == engine['engine_type'] and e['Version'] == version['engine_version']:166 return e['ParameterGroupFamily']167 def rds_instance_is_available(self):168 waiter = self.rds.get_waiter('db_instance_available')169 try:170 waiter.wait()171 print("[*] Instance created")172 except botocore.exceptions.WaiterError as e:173 if "Max attempts exceded" in e.message:174 print("[*] Error Max attempts exceded")175 sys.exit(1)176 else:177 print(e)178 def create_rds_parameter_group(self, name, instance, family):179 new_parameter_group = self.rds.create_db_parameter_group(180 DBParameterGroupName=name,181 Description='Parameter group for instance {}'.format(instance),182 DBParameterGroupFamily=family183 )184 if new_parameter_group['ResponseMetadata']['HTTPStatusCode'] == 200:185 print("[*] Creating parameter group")186 def get_rds_parameter_groups(self):187 pgs = []188 rds_parameters = self.rds.describe_db_parameter_groups()189 for pg in rds_parameters['DBParameterGroups']:190 pgs.append(pg['DBParameterGroupName'])191 return pgs192 def get_rds_instance_name(self):193 ins_names = []194 instances = self.rds.describe_db_instances()195 for instance in instances['DBInstances']:196 ins_names.append(instance['DBInstanceIdentifier'])197 return ins_names198 def modify_parameter(self, parameter_group, data):199 for key in data.keys():200 modify_parameter = self.rds.modify_db_parameter_group(201 DBParameterGroupName=parameter_group,202 Parameters=[203 {204 'ParameterName': key,205 'ParameterValue': data[key],206 'ApplyMethod': 'immediate',207 },208 ]209 )210 if modify_parameter['ResponseMetadata']['HTTPStatusCode'] == 200:211 print("[*] Modifing parameter group")212 def create_rds_instance(self, data):213 new_instance = self.rds.create_db_instance(214 DBName=data['DBName'],215 DBInstanceIdentifier=data['DBInstanceIdentifier'],216 AllocatedStorage=data['AllocatedStorage'],217 DBInstanceClass=data['DBInstanceClass'],218 Engine=data['Engine'],219 MasterUsername=data['MasterUsername'],220 MasterUserPassword=data['MasterUserPassword'],221 DBSubnetGroupName=data['DBSubnetGroupName'],222 VpcSecurityGroupIds=[223 data['VpcSecurityGroupIds'],224 ],225 AvailabilityZone=data['AvailabilityZone'],226 # PreferredMaintenanceWindow='string',227 DBParameterGroupName=data['DBParameterGroupName'],228 BackupRetentionPeriod=7,229 # PreferredBackupWindow='string',230 Port=data['Port'],231 MultiAZ=data['MultiAZ'],232 EngineVersion=data['EngineVersion'],233 AutoMinorVersionUpgrade=False,234 LicenseModel=data['LicenseModel'],235 # Iops=123,236 # OptionGroupName='default:oracle-se1-11-2',237 CharacterSetName=data['CharacterSetName'],238 PubliclyAccessible=data['PubliclyAccessible'],239 # Tags=[240 # {241 # 'Key': 'Environment',242 # 'Value': 'PRO'243 # },244 # ],245 # DBClusterIdentifier='string',246 StorageType=data['StorageType'],247 # TdeCredentialArn='string',248 # TdeCredentialPassword='string',249 StorageEncrypted=data['StorageEncrypted'],250 # KmsKeyId='string',251 # Domain='string',252 # CopyTagsToSnapshot=True,253 MonitoringInterval=0,254 # MonitoringRoleArn='string',255 # DomainIAMRoleName='string',256 # PromotionTier=123,257 # Timezone='string',258 EnableIAMDatabaseAuthentication=False,259 EnablePerformanceInsights=True,260 # PerformanceInsightsKMSKeyId='string',261 PerformanceInsightsRetentionPeriod=7,262 # EnableCloudwatchLogsExports=[263 # 'string',264 # ],265 # ProcessorFeatures=[266 # {267 # 'Name': 'string',268 # 'Value': 'string'269 # },270 # ],271 DeletionProtection=False,272 # MaxAllocatedStorage=20273 )274 if new_instance['ResponseMetadata']['HTTPStatusCode'] == 200:275 print("[*] Creating instance")276 def get_instance_types(self):277 allowed_family = ['db.m5.', 'db.m4.', 'db.m3.', 'db.m1.', 'db.z1d.', 'db.x1e.', 'db.x1.', 'db.r5.', 'db.r4.',278 'db.r3.', 'db.m2.', 'db.t3.', 'db.t2.']279 paginator = self.ec2.get_paginator('describe_instance_types')280 pages = paginator.paginate()281 instances = []282 for page in pages:283 for instance in page['InstanceTypes']:284 for family in allowed_family:285 if 'db.{}'.format(instance['InstanceType']).startswith(family):286 instances.append('db.{}'.format(instance['InstanceType']))287 res = []288 for ins in sorted(instances):289 res.append({'name': ins})290 return res291 def get_regions(self):292 region_list = []293 if self.auth_type == 'AWS Token':294 try:295 client = boto3.client('ec2',296 aws_session_token=self.aws_token,297 )298 client.describe_account_attributes()299 except botocore.exceptions.ClientError as err:300 print('[*] {}'.format(err))301 sys.exit(1)302 elif self.auth_type == 'AWS Key':303 try:304 client = boto3.client('ec2',305 aws_access_key_id=self.aws_acces,306 aws_secret_access_key=self.aws_secret,307 )308 client.describe_account_attributes()309 except botocore.exceptions.ClientError as err:310 print('[*] {}'.format(err))311 sys.exit(1)312 regions = client.describe_regions()313 for region in regions['Regions']:314 region_list.append({'name': region['RegionName']})315 return region_list316 # # Not implemented317 # def create_db_security_group(self, name):318 # new_sg = self.rds.create_db_security_group(319 # DBSecurityGroupName='sg-{}'.format(name),320 # DBSecurityGroupDescription='SG for instance {}.'.format(name)321 # )322 # if new_sg['ResponseMetadata']['HTTPStatusCode'] == 200:...
not_used_operations.py
Source:not_used_operations.py
1not_used_operations = {2 'ec2': [3 'describe_load_balancer_policies',4 'describe_vpc_endpoint_services',5 'describe_aggregate_id_format',6 'describe_load_balancer_policy_types',7 'describe_endpoints',8 'describe_availability_zones',9 'describe_account_attributes',10 'describe_dhcp_options',11 'describe_fpga_images',12 'describe_host_reservation_offerings',13 'describe_id_format',14 'describe_instance_event_notification_attributes',15 'describe_instance_type_offerings',16 'describe_instance_types',17 'describe_managed_prefix_lists',18 'describe_prefix_lists',19 'describe_regions',20 'describe_launch_template_versions',21 'describe_principal_id_format',22 'describe_spot_price_history',23 'describe_flow_logs',24 ### The blow 2 require extra parameters.25 'describe_images',26 'describe_snapshots',27 ],28 'ecr': [29 'get_authorization_token',30 ],31 'ecs': [32 'describe_capacity_providers',33 ],34 'efs': [35 'describe_account_preferences',36 ],37 'acm': [38 'get_account_configuration',39 ],40 'athena': [41 'list_data_catalogs',42 ],43 'apigateway': [44 'get_account',45 'get_sdk_types',46 ],47 'xray': [48 'get_encryption_config',49 'get_sample_rules',50 ],51 'waf': [52 'get_change_token',53 ],54 'waf-regional': [55 'get_change_token',56 ],57 'sts': [58 'get_caller_identity',59 'get_session_token',60 ],61 'ssm': [62 'describe_available_patches',63 'describe_patch_baselines',64 'get_inventory_schema',65 ],66 'ses': [67 'get_account_sending_enabled',68 'get_send_quota',69 ],70 'sesv2': [71 'get_account',72 'get_deliverability_dashboard_options',73 ],74 'signer': [75 'list_signing_platforms',76 ],77 'shield': [78 'describe_attack_statistics',79 'get_subscription_state',80 ],81 'lex-models': [82 'get_builtin_intents',83 'get_builtin_slot_types',84 ],85 'dynamodb': [86 'describe_limits',87 ],88 'lightsail': [89 'get_blueprints',90 'get_container_service_powers',91 'get_container_api_metadata',92 'get_bundles',93 'get_relational_database_blueprints',94 'get_relational_database_bundles',95 'get_regions',96 'get_bucket_bundles',97 ],98 'inspector': [99 'describe_cross_account_access_role',100 'list_rules_packages',101 ],102 'lambda': [103 'get_account_settings',104 ],105 'kinesis': [106 'describe_limits',107 ],108 'kafka': [109 'list_kafka_versions',110 'get_compatible_kafka_versions',111 ],112 'iot': [113 'describe_account_audit_configuration',114 'describe_event_configurations',115 'get_indexing_configuration',116 'get_registration_code',117 'list_domain_configurations'118 ],119 'iotsitewise': [120 'describe_logging_options',121 'describe_storage_configuration'122 ],123 'guardduty': [124 'get_invitations_count',125 ],126 'gamelift': [127 'describe_ec2_instance_limits',128 ],129 'elb': [130 'describe_account_limits',131 'describe_load_balancer_policies',132 ],133 'elbv2': [134 'describe_account_limits',135 'describe_ssl_policies',136 ],137 'elasticache': [138 'describe_cache_parameter_groups',139 'describe_service_updates',140 'list_allowed_node_type_modifications',141 'describe_cache_security_groups',142 'describe_users',143 ],144 'elasticbeanstalk': [145 'describe_account_attributes',146 'list_available_solution_stacks'147 ],148 'clouddirectory': [149 'list_managed_schema_arns',150 ],151 'codebuild': [152 'list_builds',153 'list_curated_environment_images'154 ],155 'cloudtrail': [156 'list_public_keys',157 'get_event_selectors',158 ],159 'cloudwatch': [160 'describe_alarm_history',161 'list_metrics',162 ],163 'snowball': [164 'get_snowball_usage',165 'list_compatible_images'166 ],167 'sms': [168 'get_servers'169 ],170 'sagemaker': [171 'get_sagemaker_servicecatalog_portfolio_status',172 ],173 'transfer': [174 'list_security_policies'175 ],176 'schemas': [177 'list_registries',178 ],179 'securityhub': [180 'describe_standards',181 ],182 'secretsmanager': [183 'get_random_password',184 ],185 'service-quotes': [186 'list_services',187 ],188 'wellarchitected': [189 'list_lenses',190 ],191 'route53': [192 'get_checker_ip_ranges',193 'get_geo_location',194 'get_traffic_policy_instance_count',195 'get_hosted_zone_count',196 'get_health_check_count',197 'list_geo_locations',198 ],199 'resourcegroupstaggingapi': [200 'get_resources',201 'get_tag_keys',202 ],203 'events': [204 'describe_event_bus',205 'list_event_buses',206 ],207 'fis': [208 'list_actions',209 ],210 'frauddetector': [211 'get_kms_encryption_key',212 ],213 'glacier': [214 'get_data_retrieval_policy',215 ],216 'glue': [217 'get_catalog_import_status',218 ],219 'apprunner': [220 'list_auto_scaling_configurations',221 ],222 'rds': [223 'describe_account_attributes',224 'describe_certificates',225 'describe_source_regions',226 ],227 'ram': [228 'list_permissions',229 ],230 'redshift': [231 'describe_account_attributes',232 'describe_cluster_tracks',233 'describe_orderable_cluster_options',234 'describe_storage',235 ],236 'polly': [237 'describe_voices',238 ],239 'opsworks': [240 'describe_my_user_profile',241 'describe_operating_systems',242 'describe_user_profiles',243 ],244 'opsworkscm': [245 'describe_account_attributes',246 ],247 'emr': [248 'get_block_public_access_configuration',249 'list_release_labels',250 ],251 'elastictranscoder': [252 'list_presets',253 ],254 'dms': [255 'describe_account_attributes',256 'describe_applicable_individual_assessments',257 'describe_orderable_replication_instances',258 ],259 'docdb': [260 'describe_certificates',261 'describe_db_subnet_groups',262 ],263 'ds': [264 'get_directory_limits',265 ],266 'compute-optimizer': [267 'get_enrollment_status',268 ],269 'config': [270 'get_compliance_summary_by_config_rule',271 'get_compliance_summary_by_resource_type',272 'get_discovered_resource_counts',273 ],274 'dax': [275 'describe_parameter_groups'276 ],277 'codedeploy': [278 'list_deployment_configs',279 ],280 'backup': [281 'list_backup_plan_templates',282 ],283 'autoscaling': [284 'describe_account_limits',285 ],286 'auditmanger': [287 'get_account_status',288 ],289 'cloudformation': [290 'describe_account_limits',291 ],292 'nimble': [293 'list_eulas',294 ],295 'cloudfront': [296 'list_cache_polocies',297 ]...
aws_opsworkscm_info.py
Source:aws_opsworkscm_info.py
...73 if client.can_paginate('describe_account_attributes'):74 paginator = client.get_paginator('describe_account_attributes')75 return paginator.paginate(), True76 else:77 return client.describe_account_attributes(), False78 elif module.params['describe_backups']:79 if client.can_paginate('describe_backups'):80 paginator = client.get_paginator('describe_backups')81 return paginator.paginate(), True82 else:83 return client.describe_backups(), False84 elif module.params['describe_servers']:85 if client.can_paginate('describe_servers'):86 paginator = client.get_paginator('describe_servers')87 return paginator.paginate(), True88 else:89 return client.describe_servers(), False90 else:91 return None, False...
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!!