How to use _wait_until_running method in avocado

Best Python code snippet using avocado_python

test_deployment.py

Source:test_deployment.py

...107            pass108        else:109            self.fail('TypeError was not thrown')110    def test_wait_until_running_running_instantly(self):111        node2, ips = self.driver._wait_until_running(node=self.node, wait_period=1,112                                                     timeout=10)113        self.assertEqual(self.node.uuid, node2.uuid)114        self.assertEqual(['67.23.21.33'], ips)115    def test_wait_until_running_running_after_1_second(self):116        RackspaceMockHttp.type = '1_SECOND_DELAY'117        node2, ips = self.driver._wait_until_running(node=self.node, wait_period=1,118                                                     timeout=10)119        self.assertEqual(self.node.uuid, node2.uuid)120        self.assertEqual(['67.23.21.33'], ips)121    def test_wait_until_running_running_after_1_second_private_ips(self):122        RackspaceMockHttp.type = '1_SECOND_DELAY'123        node2, ips = self.driver._wait_until_running(node=self.node, wait_period=1,124                                                     timeout=10, ssh_interface='private_ips')125        self.assertEqual(self.node.uuid, node2.uuid)126        self.assertEqual(['10.176.168.218'], ips)127    def test_wait_until_running_invalid_ssh_interface_argument(self):128        try:129            self.driver._wait_until_running(node=self.node, wait_period=1,130                                            ssh_interface='invalid')131        except ValueError:132            pass133        else:134            self.fail('Exception was not thrown')135    def test_wait_until_running_timeout(self):136        RackspaceMockHttp.type = 'TIMEOUT'137        try:138            self.driver._wait_until_running(node=self.node, wait_period=0.5,139                                            timeout=1)140        except LibcloudError:141            e = sys.exc_info()[1]142            self.assertTrue(e.value.find('Timed out') != -1)143        else:144            self.fail('Exception was not thrown')145    def test_wait_until_running_running_node_missing_from_list_nodes(self):146        RackspaceMockHttp.type = 'MISSING'147        try:148            self.driver._wait_until_running(node=self.node, wait_period=0.5,149                                            timeout=1)150        except LibcloudError:151            e = sys.exc_info()[1]152            self.assertTrue(e.value.find('Timed out after 1 second') != -1)153        else:154            self.fail('Exception was not thrown')155    def test_wait_until_running_running_multiple_nodes_have_same_uuid(self):156        RackspaceMockHttp.type = 'SAME_UUID'157        try:158            self.driver._wait_until_running(node=self.node, wait_period=0.5,159                                            timeout=1)160        except LibcloudError:161            e = sys.exc_info()[1]162            self.assertTrue(e.value.find('multiple nodes have same UUID') != -1)163        else:164            self.fail('Exception was not thrown')165    def test_ssh_client_connect_success(self):166        mock_ssh_client = Mock()167        mock_ssh_client.return_value = None168        ssh_client = self.driver._ssh_client_connect(ssh_client=mock_ssh_client,169                                                     timeout=10)170        self.assertEqual(mock_ssh_client, ssh_client)171    def test_ssh_client_connect_timeout(self):172        mock_ssh_client = Mock()...

spot_instance.py

Source:spot_instance.py

...8        raise RuntimeError(request.status_message)9    _wait_until_completed(client, request_id)10    request = _describe_request(client, request_id)11    tag(client, request.instance_id, config)12    _wait_until_running(client, request.instance_id)13    instance = get_by_instance_id(client, request.instance_id)14    if instance.has_security_groups:15        open_port(client, instance, 8888)16    return instance17def _perform_request(client, config):18    random_id = str(random.random() * 1000)19    if config.security_group_id is None:20        security_group_ids = []21    else:22        security_group_ids = [config.security_group_id]23    response = client.request_spot_instances(24        SpotPrice=config.max_bid,25        ClientToken=random_id,26        InstanceCount=1,27        Type='one-time',28        LaunchSpecification={29            'ImageId': config.ami,30            'KeyName': config.key_name,31            'InstanceType': config.type,32            'Placement': {33                'AvailabilityZone': config.az,34            },35            'SecurityGroupIds': security_group_ids,36	    'SubnetId':config.subnet_id37        }38    )39    return response.get('SpotInstanceRequests')[0].get('SpotInstanceRequestId')40def _wait_until_completed(client, request_id):41    waiter = client.get_waiter('spot_instance_request_fulfilled')42    return waiter.wait(SpotInstanceRequestIds=[request_id])43def _describe_request(client, request_id):44    response = client.describe_spot_instance_requests(45        SpotInstanceRequestIds=[request_id],46    )47    return SpotInstanceRequest(response.get('SpotInstanceRequests')[0])48def _wait_until_running(client, instance_id):49    waiter = client.get_waiter('instance_running')50    return waiter.wait(InstanceIds=[instance_id])51class SpotInstanceRequest():52    def __init__(this, response):53        this.instance_id = response.get('InstanceId')54        this.status_code = response.get('Status').get('Code')...

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.