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

How to use run_instances method in localstack

Best Python code snippet using localstack_python

ec2_schd.py

Source:ec2_schd.py

...211    print "finish!!!"212    #conn.create_tags(['r-75625877'], {'Name':'aaaaaaaaa'})213214def request_instances():215    ret = conn.run_instances(216        "ami-a579efa4", 217        min_count = 1,218        max_count = 1,219        #key_name='favbuykey', 220        #security_groups=['sg-5d0b7d5c'], 221        security_group_ids = ['sg-5d0b7d5c'], 222        #instance_profile_name = "aa",223        instance_type="t1.micro", 224        #user_data=get_init_script(*(NUMS.get(itype, (10, 10))),burst=burst)225    )226    return ret227228if __name__ == "__main__":229    log.startLogging(sys.stdout)
...

test_aws.py

Source:test_aws.py

...17                            return True18        return False19    @mock_ec220    def test_create_tags(self):21        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']22        test_tag = {23            'Key': 'test',24            'Value': 'test_value'25        }26        self.wrapper.create_tags(Resources=[instance], Tags=[test_tag])27        response = self.wrapper.ec2.describe_instances(InstanceIds=[instance])28        29        self.assertTrue(self.is_tagged_with_value(instance, response, 'test', 'test_value'))30        31    @mock_ec232    def test_tag_as_low_use(self):33        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']34        response = self.wrapper.ec2.describe_instances(InstanceIds=[instance])35        self.assertFalse(self.is_tagged_with_value(instance, response, 'Low Use', 'true'))36        self.wrapper.tag_as_low_use(instance)37        response_after_tag = self.wrapper.ec2.describe_instances(InstanceIds=[instance])38        self.assertTrue(self.is_tagged_with_value(instance, response_after_tag, 'Low Use', 'true'))39    @mock_ec240    def test_tag_as_whitelisted(self):41        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']42        response = self.wrapper.ec2.describe_instances(InstanceIds=[instance])43        self.assertFalse(self.is_tagged_with_value(instance, response, 'Whitelisted', 'true'))44        self.wrapper.tag_as_whitelisted(instance)45        response_after_tag = self.wrapper.ec2.describe_instances(InstanceIds=[instance])46        self.assertTrue(self.is_tagged_with_value(instance, response_after_tag, 'Whitelisted', 'true'))47    48    @mock_ec249    def test_tag_whitelist_reason(self):50        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']51        response = self.wrapper.ec2.describe_instances(InstanceIds=[instance])52        self.assertFalse(self.is_tagged_with_value(instance, response, 'Reason', 'test_value'))53        self.wrapper.tag_whitelist_reason(instance, 'test_value')54        response_after_tag = self.wrapper.ec2.describe_instances(InstanceIds=[instance])55        self.assertTrue(self.is_tagged_with_value(instance, response_after_tag, 'Reason', 'test_value'))56    57    @mock_ec258    def test_tag_for_deletion(self):59        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']60        response = self.wrapper.ec2.describe_instances(InstanceIds=[instance])61        self.assertFalse(self.is_tagged_with_value(instance, response, 'Scheduled For Deletion', 'true'))62        self.wrapper.tag_for_deletion(instance)63        response_after_tag = self.wrapper.ec2.describe_instances(InstanceIds=[instance])64        self.assertTrue(self.is_tagged_with_value(instance, response_after_tag, 'Scheduled For Deletion', 'true'))65    66    @mock_ec267    def test_tag_instance(self):68        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']69        response = self.wrapper.ec2.describe_instances(InstanceIds=[instance])70        self.assertFalse(self.is_tagged_with_value(instance, response, 'test_key', 'test_value'))71        self.wrapper.tag_instance(instance, 'tag_key','test_value')72        response_after_tag = self.wrapper.ec2.describe_instances(InstanceIds=[instance])73        self.assertTrue(self.is_tagged_with_value(instance, response_after_tag, 'tag_key', 'test_value'))74    75    @mock_ec276    def test_get_creator_for_instance(self):77        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']78        self.wrapper.tag_instance(instance, 'Creator','test_creator')79        self.assertEqual(self.wrapper.get_creator_for_instance(instance), 'test_creator')80    81    @mock_ec282    def test_get_whitelist_reason_for_instance(self):83        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']84        self.wrapper.tag_as_whitelisted(instance)85        self.wrapper.tag_whitelist_reason(instance, 'test_value')86        self.assertEqual(self.wrapper.get_whitelist_reason_for_instance(instance), 'test_value')87    @mock_ec288    def test_get_tag_for_instance(self):89        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']90        self.wrapper.tag_instance(instance, 'test_key', 'test_value')91        self.assertEqual(self.wrapper.get_tag_for_instance(instance, 'test_key'), 'test_value')92    @mock_ec293    def test_get_tags_for_instance(self):94        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']95        self.wrapper.tag_instance(instance, 'test_key', 'test_value')96        expected = [{97            'Key': 'test_key',98            'Value': 'test_value'99        }]100        self.assertEqual(self.wrapper.get_tags_for_instance(instance), expected)101    @mock_ec2102    def test_is_whitelisted(self):103        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']104        self.wrapper.tag_as_whitelisted(instance)105        self.assertTrue(self.wrapper.is_whitelisted(instance))106    @mock_ec2107    def test_is_low_use(self):108        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']109        self.wrapper.tag_as_low_use(instance)110        self.assertTrue(self.wrapper.is_low_use(instance))111    @mock_ec2112    def test_is_scheduled_for_deletion(self):113        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']114        self.wrapper.tag_for_deletion(instance)115        self.assertTrue(self.wrapper.is_scheduled_for_deletion(instance))116    @mock_ec2117    def test_is_tagged(self):118        instance = self.wrapper.ec2.run_instances(MaxCount=1, MinCount=1)['Instances'][0]['InstanceId']119        self.wrapper.tag_instance(instance, 'test_key', 'true')120        self.assertTrue(self.wrapper.is_tagged(instance, 'test_key'))121class TestASGWrapper(unittest.TestCase):122    @mock_autoscaling123    def setUp(self):124        self.session = boto3.Session(region_name='us-west-2')125        self.wrapper = ASGWrapper(self.session)126    @mock_autoscaling127    def test_get_asg_user_tag_by_instance_id(self): 128        pass129class TestTrustedAdvisor(unittest.TestCase):130    pass131class TestDynamoWrapper(unittest.TestCase):132    pass...

timeout_test.py

Source:timeout_test.py

1# Copyright 2019 Alibaba Cloud Inc. All rights reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#    http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14from base import SDKTestBase15from mock import MagicMock, patch16from aliyunsdkcore.client import AcsClient17from aliyunsdkcore.http.http_response import HttpResponse18from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest19from aliyunsdkecs.request.v20140526.CreateInstanceRequest import CreateInstanceRequest20from aliyunsdkecs.request.v20140526.DescribeInstanceHistoryEventsRequest import \21    DescribeInstanceHistoryEventsRequest22from aliyunsdkecs.request.v20140526.DescribeDisksRequest import DescribeDisksRequest23from aliyunsdkecs.request.v20140526.RunInstancesRequest import RunInstancesRequest24from aliyunsdkram.request.v20150501.ListUsersRequest import ListUsersRequest25from aliyunsdkcore.acs_exception.exceptions import ClientException26from aliyunsdkcore.acs_exception.exceptions import ServerException27class TimeoutTest(SDKTestBase):28    def setUp(self):29        globals()['_test_patch_client_read_timeout'] = None30        globals()['_test_patch_client_connect_timeout'] = None31    def _patch_client(self, client):32        original_make_http_response = client._make_http_response33        def _make_http_response(endpoint, request, read_timeout, connect_timeout,34                                specific_signer=None):35            globals()["_test_patch_client_read_timeout"] = read_timeout36            globals()["_test_patch_client_connect_timeout"] = connect_timeout37            return original_make_http_response(endpoint, request, read_timeout, connect_timeout,38                                               specific_signer=None)39        client._make_http_response = _make_http_response40    def _test_timeout(self, client, request, expected_read_timeout, expected_connect_timeout):41        request.set_endpoint("somewhere.you.will.never.get")42        with self.assertRaises(ClientException):43            client.do_action_with_exception(request)44        self.assertEqual(expected_read_timeout, _test_patch_client_read_timeout)45        self.assertEqual(expected_connect_timeout, _test_patch_client_connect_timeout)46    def test_request_customized_timeout(self):47        client = AcsClient(self.access_key_id, self.access_key_secret, self.region_id,48                           auto_retry=False)49        describe_instances = DescribeInstancesRequest()50        describe_instances.set_read_timeout(3)51        describe_instances.set_connect_timeout(4)52        list_users = ListUsersRequest()53        list_users.set_read_timeout(6)54        list_users.set_connect_timeout(7)55        run_instances = RunInstancesRequest()56        run_instances.set_read_timeout(21)57        run_instances.set_connect_timeout(15)58        create_instance = CreateInstanceRequest()59        create_instance.set_read_timeout(22)60        create_instance.set_connect_timeout(16)61        self._patch_client(client)62        self._test_timeout(client, describe_instances, 3, 4)63        self._test_timeout(client, list_users, 6, 7)64        self._test_timeout(client, run_instances, 21, 15)65        self._test_timeout(client, create_instance, 22, 16)66    def test_client_customized_timeout(self):67        client = AcsClient(self.access_key_id, self.access_key_secret, self.region_id,68                           timeout=7, connect_timeout=8, auto_retry=False)69        self._patch_client(client)70        self._test_timeout(client, DescribeInstancesRequest(), 7, 8)71        self._test_timeout(client, ListUsersRequest(), 7, 8)72        self._test_timeout(client, RunInstancesRequest(), 7, 8)73        self._test_timeout(client, CreateInstanceRequest(), 7, 8)74    def test_default_request_timeout(self):75        client = AcsClient(self.access_key_id, self.access_key_secret, self.region_id,76                           auto_retry=False)77        self._patch_client(client)78        self._test_timeout(client, CreateInstanceRequest(), 86, 5)79        self._test_timeout(client, DescribeInstancesRequest(), 10, 5)80        self._test_timeout(client, RunInstancesRequest(), 86, 5)81    def test_default_client_timeout(self):82        client = AcsClient(self.access_key_id, self.access_key_secret, self.region_id,83                           auto_retry=False)84        self._patch_client(client)85        self._test_timeout(client, ListUsersRequest(), 10, 5)86    def test_read_timeout_priority(self):87        client = AcsClient(self.access_key_id, self.access_key_secret, self.region_id,88                           timeout=11, auto_retry=False)89        describe_instances = DescribeInstancesRequest()90        describe_instances.set_read_timeout(6)91        list_users = ListUsersRequest()92        list_users.set_read_timeout(7)93        run_instances = RunInstancesRequest()94        run_instances.set_read_timeout(8)95        create_instance = CreateInstanceRequest()96        create_instance.set_read_timeout(9)97        self._patch_client(client)98        self._test_timeout(client, describe_instances, 6, 5)99        self._test_timeout(client, list_users, 7, 5)100        self._test_timeout(client, run_instances, 8, 5)101        self._test_timeout(client, create_instance, 9, 5)102    def test_connect_timeout_priority(self):103        client = AcsClient(self.access_key_id, self.access_key_secret, self.region_id,104                           connect_timeout=12, auto_retry=False)105        describe_instances = DescribeInstancesRequest()106        describe_instances.set_connect_timeout(6)107        list_users = ListUsersRequest()108        list_users.set_connect_timeout(7)109        run_instances = RunInstancesRequest()110        run_instances.set_connect_timeout(8)111        create_instance = CreateInstanceRequest()112        create_instance.set_connect_timeout(9)113        self._patch_client(client)114        self._test_timeout(client, describe_instances, 10, 6)115        self._test_timeout(client, list_users, 10, 7)116        self._test_timeout(client, run_instances, 86, 8)...

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.