Module tests.aws

Classes

class aws

Expand source code

class aws(cloudprovider):
    ACTION_WAIT_TIMEOUT_SEC = 120
    ec2_client = None

    def __init__(self):
        access_key_id = os.getenv("AWS_ACCESS_KEY_ID")
        if access_key_id is None:
            print("AWS_ACCESS_KEY_ID env variable not defined")
            assert False

        secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")
        if secret_access_key is None:
            print("AWS_SECRET_ACCESS_KEY env variable not defined")
            assert False

        default_region = os.getenv("AWS_DEFAULT_REGION")
        if default_region is None:
            print("AWS_DEFAULT_REGION env variable not defined")
            assert False

        aws.ec2_client = boto3.client(
            'ec2',
            aws_access_key_id=access_key_id,
            aws_secret_access_key=secret_access_key,
            region_name=default_region)

    def instance_id_by_ip(self, instance_ip):
        resource_suffix = os.getenv("RESOURCE_SUFFIX")
        response = aws.ec2_client.describe_instances(
            Filters=[
                {
                    'Name': 'private-ip-address',
                    'Values': [instance_ip]
                },
                {
                    'Name': 'tag:Name',
                    'Values': [f"*{resource_suffix}*"]
                }
            ],
        )

        return response['Reservations'][0]['Instances'][0]['InstanceId']

    def instance_id(self, instance_name):
        response = aws.ec2_client.describe_instances(
            Filters=[
                {
                    'Name': 'tag:Name',
                    'Values': [instance_name]
                },
            ],
        )

        return response['Reservations'][0]['Instances'][0]['InstanceId']

    def instance_status(self, instance_id):
        response = aws.ec2_client.describe_instances(
            InstanceIds=[instance_id]
        )

        return response['Reservations'][0]['Instances'][0]['State']['Name']

    def instance_exec_action(self,
                             instance_id,
                             action,
                             expected_status,
                             desired_status):
        instance_current_status = self.instance_status(instance_id)
        if instance_current_status == expected_status:
            if action == "start":
                aws.ec2_client.start_instances(InstanceIds=[instance_id])
            elif action == "stop":
                aws.ec2_client.stop_instances(InstanceIds=[instance_id])

            action_ok = False
            for i in range(self.ACTION_WAIT_TIMEOUT_SEC):
                instance_status = self.instance_status(instance_id)
                if instance_status == desired_status:
                    action_ok = True
                    break
                elif instance_status == expected_status:
                    time.sleep(1)

            return action_ok

        elif instance_current_status == desired_status:
            return True

    def instance_start(self, instance_id):
        self.instance_exec_action(instance_id=instance_id,
                                  action="start",
                                  expected_status="stopped",
                                  desired_status="running")

    def instance_stop(self, instance_id):
        self.instance_exec_action(instance_id=instance_id,
                                  action="stop",
                                  expected_status="running",
                                  desired_status="stopped")

Ancestors

• cloudprovider.cloudprovider

Class variables

var ACTION_WAIT_TIMEOUT_SEC

The type of the None singleton.

var ec2_client

The type of the None singleton.

Methods

def instance_exec_action(self, instance_id, action, expected_status, desired_status)

Expand source code

def instance_exec_action(self,
                         instance_id,
                         action,
                         expected_status,
                         desired_status):
    instance_current_status = self.instance_status(instance_id)
    if instance_current_status == expected_status:
        if action == "start":
            aws.ec2_client.start_instances(InstanceIds=[instance_id])
        elif action == "stop":
            aws.ec2_client.stop_instances(InstanceIds=[instance_id])

        action_ok = False
        for i in range(self.ACTION_WAIT_TIMEOUT_SEC):
            instance_status = self.instance_status(instance_id)
            if instance_status == desired_status:
                action_ok = True
                break
            elif instance_status == expected_status:
                time.sleep(1)

        return action_ok

    elif instance_current_status == desired_status:
        return True

def instance_id(self, instance_name)

Expand source code

def instance_id(self, instance_name):
    response = aws.ec2_client.describe_instances(
        Filters=[
            {
                'Name': 'tag:Name',
                'Values': [instance_name]
            },
        ],
    )

    return response['Reservations'][0]['Instances'][0]['InstanceId']

def instance_id_by_ip(self, instance_ip)

Expand source code

def instance_id_by_ip(self, instance_ip):
    resource_suffix = os.getenv("RESOURCE_SUFFIX")
    response = aws.ec2_client.describe_instances(
        Filters=[
            {
                'Name': 'private-ip-address',
                'Values': [instance_ip]
            },
            {
                'Name': 'tag:Name',
                'Values': [f"*{resource_suffix}*"]
            }
        ],
    )

    return response['Reservations'][0]['Instances'][0]['InstanceId']

def instance_start(self, instance_id)

Expand source code

def instance_start(self, instance_id):
    self.instance_exec_action(instance_id=instance_id,
                              action="start",
                              expected_status="stopped",
                              desired_status="running")

def instance_status(self, instance_id)

Expand source code

def instance_status(self, instance_id):
    response = aws.ec2_client.describe_instances(
        InstanceIds=[instance_id]
    )

    return response['Reservations'][0]['Instances'][0]['State']['Name']

def instance_stop(self, instance_id)

Expand source code

def instance_stop(self, instance_id):
    self.instance_exec_action(instance_id=instance_id,
                              action="stop",
                              expected_status="running",
                              desired_status="stopped")