Best Python code snippet using localstack_python
Worker.py
Source:Worker.py
...163 logger.debug('t2Unlimited(): Found T2 Flag in DynamoDB: %s' % self.modifiedInstanceFlag)164 try:165 self.checkT2Unlimited()166 except Exception as e:167 self.snsInit.sendSns("Worker::describe_instance_credit_specifications() has encountered an exception" ,str(e))168 if self.current_t2_value == "standard":169 try:170 logger.info('t2Unlimited(): Trying to modify EC2 instance credit specification')171 result = retry(self.ec2_client.modify_instance_credit_specification,attempts=5, sleeptime=0,jitter=0,kwargs= {"InstanceCreditSpecifications":[{'InstanceId': self.instance.id,'CpuCredits': 'unlimited'}]} )172 logger.info('t2Unlimited(): EC2 instance credit specification modified')173 except Exception as e:174 msg = 'Worker::instance.modify_attribute().t2Unlimited Exception for EC2 instance %s, error --> %s' % (self.instance, str(e))175 logger.warning(msg)176 self.snsInit.sendSns("Worker::checkT2Unlimited::instance.modify_attribute().t2Unlimited has encountered an exception",str(e))177 else:178 try:179 self.checkT2Unlimited()180 except Exception as e:181 self.snsInit.sendSns("Worker::checkT2Unlimited::describe_instance_credit_specifications() Exception has encountered an exception",str(e))182 if self.current_t2_value == "unlimited":183 logger.debug('t2Unlimited(): Current T2 value: %s' % self.current_t2_value)184 try:185 logger.info('t2Unlimited(): Trying to modify EC2 instance credit specification')186 result = retry(self.ec2_client.modify_instance_credit_specification, attempts=5, sleeptime=0, jitter=0,187 kwargs= {"InstanceCreditSpecifications":[{'InstanceId': self.instance.id,'CpuCredits': 'standard'}]} )188 logger.info('t2Unlimited(): EC2 instance credit specification modified')189 except Exception as e:190 msg = 'Worker::instance.modify_attribute().t2standard Exception in progress for EC2 instance %s, error --> %s' % (self.instance, str(e))191 logger.warning(msg)192 self.snsInit.sendSns("Worker::instance.modify_attribute().t2standard Exception has encountered an exception",str(e))193 @retriable(attempts=5,sleeptime=0, jitter=0)194 def checkT2Unlimited(self):195 try:196 logger.info('check_t2_unlimited(): Checking current T2 value')197 self.current_t2_value = self.ec2_client.describe_instance_credit_specifications(InstanceIds=[self.instance.id,])['InstanceCreditSpecifications'][0]['CpuCredits']198 except Exception as e:199 msg = 'Worker::describe_instance_credit_specifications() exception for EC2 instance %s, error --> %s' % (self.instance, str(e))200 logger.warning(msg)201 raise e202 return self.current_t2_value203 def start(self):204 self.startInstance()205class StopWorker(Worker):206 def __init__(self, ddbRegion, workloadRegion, instance, dryRunFlag, ec2_client, snsInit):207 super(StopWorker, self).__init__(workloadRegion, instance, ec2_client, dryRunFlag, snsInit)208 self.ddbRegion = ddbRegion209 # MUST convert string False to boolean False210 self.waitFlag = strtobool('False')211 self.overrideFlag = strtobool('False')212 self.snsInit = snsInit213 def stopInstance(self):...
main.py
Source:main.py
...113 # If InstanceType starts with t114 match_instance_type = re.match(r'^t\d\..*$', instance_type)115 # Determine whether resource is standard or unlimited and execute correct alarm creation function116 if match_instance_type:117 credit_specifications = ec2.describe_instance_credit_specifications(InstanceIds=[instance_id])118 credit_type = credit_specifications['InstanceCreditSpecifications'][0]['CpuCredits']119 if credit_type == 'standard':120 create_standard_alarm(instance_id)121 if credit_type == 'unlimited':122 create_unlimited_alarm(instance_id)123 124 # instance Stopped/Terminated event125 if (event['detail']['state'] == 'stopped') or (event['detail']['state'] == 'terminated'):126 # Pull instance from resource by regex group i-.*127 for resource in event['resources']:128 match_instance = re.match(r'.*(i-.*)$', resource)129 instance_id = match_instance.group(1)130 if match_instance:131 # Get instance information and record type132 instance = ec2.describe_instances(InstanceIds=[instance_id])133 instance_type = instance['Reservations'][0]['Instances'][0]['InstanceType']134 # If InstanceType starts with t135 match_instance_type = re.match(r'^t\d\..*$', instance_type)136 # Attempt to delete CloudWatch Alarm137 if match_instance_type:138 credit_specifications = ec2.describe_instance_credit_specifications(InstanceIds=[instance_id])139 credit_type = credit_specifications['InstanceCreditSpecifications'][0]['CpuCredits']140 if credit_type == 'standard':141 delete_standard_alarm(instance_id)142 if credit_type == 'unlimited':...
lambda_function.py
Source:lambda_function.py
...9logger = logging.getLogger()10logger.setLevel(logging.INFO)11def lambda_handler(event, context):12 client = boto3.client('ec2', region_name = AWS_REGION)13 response = client.describe_instance_credit_specifications(InstanceIds=[INSTANCE_ID])14 credit_type = response['InstanceCreditSpecifications'][0]['CpuCredits']15 if credit_type == 'standard':16 print "Unlimited disabled"17 cw = boto3.client('cloudwatch', region_name=AWS_REGION)18 response = cw.get_metric_statistics(19 Namespace='AWS/EC2',20 MetricName='CPUCreditBalance',21 Dimensions=[22 {23 "Name": "InstanceId",24 "Value": INSTANCE_ID25 }26 ],27 StartTime=datetime.utcnow() - timedelta(seconds=CLOUDWATCH_PERIOD),...
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!!