KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use describe_import_snapshot_tasks method in localstack

Best Python code snippet using localstack_python

vhd-to-ami.py

Source:vhd-to-ami.py

...52        aws_import_snapshot_task_status = aws_ec2_client.import_snapshot(DiskContainer=disk_container)53        aws_import_snapshot_task_complete_or_failed = False54        last_aws_import_snapshot_task_detail = False55        while not aws_import_snapshot_task_complete_or_failed:56            aws_import_snapshot_task_detail = aws_ec2_client.describe_import_snapshot_tasks(ImportTaskIds=[aws_import_snapshot_task_status['ImportTaskId']])['ImportSnapshotTasks'][0]['SnapshotTaskDetail']57            if ((last_aws_import_snapshot_task_detail and ('Progress' in aws_import_snapshot_task_detail) and ((last_aws_import_snapshot_task_detail['Status'] != aws_import_snapshot_task_detail['Status']) or (last_aws_import_snapshot_task_detail['StatusMessage'] != aws_import_snapshot_task_detail['StatusMessage']) or (last_aws_import_snapshot_task_detail['Progress'] < aws_import_snapshot_task_detail['Progress']))) or (last_aws_import_snapshot_task_detail == False)):58                print('  snapshot import task in progress with id: {}, progress: {}%, status: {}; {}'.format(aws_import_snapshot_task_status['ImportTaskId'], aws_import_snapshot_task_detail['Progress'], aws_import_snapshot_task_detail['Status'], aws_import_snapshot_task_detail['StatusMessage']))59            last_aws_import_snapshot_task_detail = aws_import_snapshot_task_detail60            aws_import_snapshot_task_complete_or_failed = aws_import_snapshot_task_detail['Status'] in ['completed', 'deleted'] or aws_import_snapshot_task_detail['StatusMessage'].startswith('ServerError') or aws_import_snapshot_task_detail['StatusMessage'].startswith('ClientError')61            sleep(1)62        aws_import_snapshot_task_detail = aws_ec2_client.describe_import_snapshot_tasks(ImportTaskIds=[aws_import_snapshot_task_status['ImportTaskId']])['ImportSnapshotTasks'][0]['SnapshotTaskDetail']63        if aws_import_snapshot_task_detail['Status'] != 'completed':64            print('  snapshot import failed. status: {}; {}'.format(aws_import_snapshot_task_detail['Status'], aws_import_snapshot_task_detail['StatusMessage']))65        else:66            print('  snapshot import complete. snapshot id: {}, status: {}'.format(aws_import_snapshot_task_detail['SnapshotId'], aws_import_snapshot_task_detail['Status']))67            aws_snapshot = aws_ec2_resource.Snapshot(aws_import_snapshot_task_detail['SnapshotId'])68            while aws_snapshot.state != 'completed':69                print('  waiting for snapshot {} availability. current state: {}'.format(aws_snapshot.snapshot_id, aws_snapshot.state))70                sleep(1)71                aws_snapshot = aws_ec2_resource.Snapshot(aws_import_snapshot_task_detail['SnapshotId'])72            print('  snapshot id: {}, state: {}, progress: {}, size: {}gb'.format(aws_snapshot.snapshot_id, aws_snapshot.state, aws_snapshot.progress, aws_snapshot.volume_size))73            print('    https://{}.console.aws.amazon.com/ec2/v2/home?region={}#Snapshots:visibility=owned-by-me;snapshotId={}'.format(aws_region_name, aws_region_name, aws_snapshot.snapshot_id))74            # tag snapshot75            aws_ec2_client.create_tags(Resources=[aws_snapshot.snapshot_id], Tags=aws_resource_tags)76            # create and tag volume...

import_rhcos.py

Source:import_rhcos.py

...148        )['ImportTaskId']149        time_elapsed = 0150        while True:151            logger.info('Checking status of snapshot import task {}'.format(import_task_id))152            snapshot_task = ec2.describe_import_snapshot_tasks(153                ImportTaskIds=[import_task_id],154            )155            if snapshot_task['ImportSnapshotTasks'][0]['SnapshotTaskDetail']['Status'] == 'completed':156                snapshot_id = snapshot_task['ImportSnapshotTasks'][0]['SnapshotTaskDetail']['SnapshotId']157                logger.info('Snapshot {} created'.format(snapshot_id))158                logger.info('Tagging snapshot {} with rhcos_version={}'.format(snapshot_id, self.rhcos_version))159                ec2.create_tags(160                    Resources=[snapshot_id],161                    Tags=[162                        {163                            'Key': 'rhcos_version',164                            'Value': self.rhcos_version,165                        }166                    ],...

aws_snapshot.py

Source:aws_snapshot.py

...89            'completed' status."""90            try:91                LOGGER.trace("Querying the status of import-task [%s].", import_task_id)92                response = \93                    self.ec2_client.describe_import_snapshot_tasks(94                        ImportTaskIds=[import_task_id])95                if not response:96                    raise RuntimeError("describe_import_snapshot_tasks() returned none response!")97                LOGGER.trace("Response from describe_import_snapshot_tasks => '%s'",98                             response)99                task_status = response['ImportSnapshotTasks'][0]['SnapshotTaskDetail']['Status']100                if task_status == 'error':101                    # Print the response before raising an exception.102                    LOGGER.debug("describe_import_snapshot_tasks() response for [%s] => [%s]",103                                 import_task_id, response)104                    raise RuntimeError("import-snapshot task [{}] in unrecoverable 'error' state.".105                                       format(import_task_id))106                return task_status == 'completed'107            except ClientError as client_error:108                LOGGER.exception(client_error)109                raise RuntimeError("describe_import_snapshot_tasks() failed for [{}]!".110                                   format(import_task_id)) from client_error111        retrier = Retrier(_is_snapshot_ready)112        retrier.tries = int(get_config_value('AWS_IMPORT_SNAPSHOT_TASK_RETRY_COUNT'))113        retrier.delay = int(get_config_value('AWS_IMPORT_SNAPSHOT_TASK_RETRY_DELAY'))114        LOGGER.info("Waiting for the import snapshot task [%s] to complete.", import_task_id)115        try:116            if retrier.execute():117                LOGGER.info("import_snapshot_task [%s] is completed.", import_task_id)118                # Call it one last time to get the snapshot_id.119                response = \120                self.ec2_client.describe_import_snapshot_tasks(121                    ImportTaskIds=[import_task_id])122                self.snapshot_id = \123                    response['ImportSnapshotTasks'][0]['SnapshotTaskDetail']['SnapshotId']124                LOGGER.info("SnapshotID = [%s].", self.snapshot_id)125                return True126            LOGGER.warning("import_snapshot_task [%s] didn't complete after checking [%d] times!",127                           import_task_id, retrier.tries)128            return False129        except RuntimeError as runtime_exception:130            LOGGER.exception(runtime_exception)131            raise132    def get_snapshot_tag_metadata(self):133        """Returns associated snapshot metadata tags."""134        metadata_tags = CloudImageTags(self.metadata)...

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.