KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use set_response method in localstack
Best Python code snippet using localstack_python
player_test.py
Source:player_test.py 
...33 logger = Mock()34 logger.log = Mock()35 player_service = Player(api_wrapper, event_manager, logger)36 pgo = api_wrapper.get_api()37 pgo.set_response('get_player', self._create_generic_player_response())38 pgo.set_response('get_inventory', self._create_generic_inventory_response())39 player = player_service.get_player()40 assert isinstance(player, PlayerData)41 assert player.username == 'test_account'42 assert pgo.call_stack_size() == 043 def test_get_inventory(self):44 config = create_core_test_config()45 api_wrapper = create_mock_api_wrapper(config)46 event_manager = Mock()47 logger = Mock()48 logger.log = Mock()49 player_service = Player(api_wrapper, event_manager, logger)50 pgo = api_wrapper.get_api()51 pgo.set_response('get_player', self._create_generic_player_response())52 pgo.set_response('get_inventory', self._create_generic_inventory_response())53 inventory = player_service.get_inventory()54 assert inventory['count'] == 3655 assert pgo.call_stack_size() == 056 def test_get_pokemon(self):57 config = create_core_test_config()58 api_wrapper = create_mock_api_wrapper(config)59 event_manager = Mock()60 logger = Mock()61 logger.log = Mock()62 player_service = Player(api_wrapper, event_manager, logger)63 pgo = api_wrapper.get_api()64 pgo.set_response('get_player', self._create_generic_player_response())65 pgo.set_response('get_inventory', self._create_generic_inventory_response())66 pokemon = player_service.get_pokemon()67 assert len(pokemon) == 268 assert pgo.call_stack_size() == 069 def test_get_candies(self):70 config = create_core_test_config()71 api_wrapper = create_mock_api_wrapper(config)72 event_manager = Mock()73 logger = Mock()74 logger.log = Mock()75 player_service = Player(api_wrapper, event_manager, logger)76 pgo = api_wrapper.get_api()77 pgo.set_response('get_player', self._create_generic_player_response())78 pgo.set_response('get_inventory', self._create_generic_inventory_response())79 candies = player_service.get_candies()80 assert len(candies) == 181 assert candies[1] == 10082 assert pgo.call_stack_size() == 083 def test_get_candy(self):84 config = create_core_test_config()85 api_wrapper = create_mock_api_wrapper(config)86 event_manager = Mock()87 logger = Mock()88 logger.log = Mock()89 player_service = Player(api_wrapper, event_manager, logger)90 pgo = api_wrapper.get_api()91 pgo.set_response('get_player', self._create_generic_player_response())92 pgo.set_response('get_inventory', self._create_generic_inventory_response())93 candies = player_service.get_candy(1)94 assert candies == 10095 assert pgo.call_stack_size() == 096 def test_get_candy_key_error(self):97 config = create_core_test_config()98 api_wrapper = create_mock_api_wrapper(config)99 event_manager = Mock()100 logger = Mock()101 logger.log = Mock()102 player_service = Player(api_wrapper, event_manager, logger)103 pgo = api_wrapper.get_api()104 pgo.set_response('get_player', self._create_generic_player_response())105 pgo.set_response('get_inventory', self._create_generic_inventory_response())106 candies = player_service.get_candy(100)107 assert candies == 0108 assert pgo.call_stack_size() == 0109 def test_add_candy(self):110 config = create_core_test_config()111 api_wrapper = create_mock_api_wrapper(config)112 event_manager = Mock()113 logger = Mock()114 logger.log = Mock()115 player_service = Player(api_wrapper, event_manager, logger)116 pgo = api_wrapper.get_api()117 pgo.set_response('get_player', self._create_generic_player_response())118 pgo.set_response('get_inventory', self._create_generic_inventory_response())119 before_candies = player_service.get_candy(1)120 assert before_candies == 100121 player_service.add_candy(1, 3)122 after_candies = player_service.get_candy(1)123 assert after_candies == 103124 assert pgo.call_stack_size() == 0125 def test_add_candy_new(self):126 config = create_core_test_config()127 api_wrapper = create_mock_api_wrapper(config)128 event_manager = Mock()129 logger = Mock()130 logger.log = Mock()131 player_service = Player(api_wrapper, event_manager, logger)132 pgo = api_wrapper.get_api()133 pgo.set_response('get_player', self._create_generic_player_response())134 pgo.set_response('get_inventory', self._create_generic_inventory_response())135 before_candies = player_service.get_candy(1)136 assert before_candies == 100137 player_service.add_candy(10, 3)138 after_candies = player_service.get_candy(10)139 assert after_candies == 3140 assert pgo.call_stack_size() == 0141 def test_get_pokeballs(self):142 config = create_core_test_config()143 api_wrapper = create_mock_api_wrapper(config)144 event_manager = Mock()145 logger = Mock()146 logger.log = Mock()147 player_service = Player(api_wrapper, event_manager, logger)148 pgo = api_wrapper.get_api()149 pgo.set_response('get_player', self._create_generic_player_response())150 pgo.set_response('get_inventory', self._create_generic_inventory_response())151 pokeballs = player_service.get_pokeballs()152 assert Item.ITEM_POKE_BALL.value in pokeballs153 assert pokeballs[Item.ITEM_POKE_BALL.value] == 11154 assert Item.ITEM_GREAT_BALL.value in pokeballs155 assert pokeballs[Item.ITEM_GREAT_BALL.value] == 12156 assert Item.ITEM_ULTRA_BALL.value in pokeballs157 assert pokeballs[Item.ITEM_ULTRA_BALL.value] == 13158 assert Item.ITEM_MASTER_BALL.value in pokeballs159 assert pgo.call_stack_size() == 0160 def test_print_stats(self):161 config = create_core_test_config()162 api_wrapper = create_mock_api_wrapper(config)163 event_manager = Mock()164 logger = Mock()165 logger.log = Mock()166 player_service = Player(api_wrapper, event_manager, logger)167 pgo = api_wrapper.get_api()168 pgo.set_response('get_player', self._create_generic_player_response())169 pgo.set_response('get_inventory', self._create_generic_inventory_response())170 logger.log.return_value = None171 player_service.print_stats()172 assert logger.log.call_count == 15173 self._assert_log_call(logger.log, 'Username: test_account')174 self._assert_log_call(logger.log, 'Bag storage: 36/350')175 self._assert_log_call(logger.log, 'Pokemon storage: 2/250')176 self._assert_log_call(logger.log, 'Stardust: 20,000')177 self._assert_log_call(logger.log, 'Pokecoins: 10')178 self._assert_log_call(logger.log, 'Poke Balls: 11')179 self._assert_log_call(logger.log, 'Great Balls: 12')180 self._assert_log_call(logger.log, 'Ultra Balls: 13')181 self._assert_log_call(logger.log, '-- Level: 14')182 self._assert_log_call(logger.log, '-- Experience: 15')183 self._assert_log_call(logger.log, '-- Experience until next level: 985')184 self._assert_log_call(logger.log, '-- Pokemon captured: 17')185 self._assert_log_call(logger.log, '-- Pokestops visited: 18')186 assert pgo.call_stack_size() == 0187 def test_print_stats_no_update(self):188 config = create_core_test_config()189 api_wrapper = create_mock_api_wrapper(config)190 event_manager = Mock()191 logger = Mock()192 logger.log = Mock()193 player_service = Player(api_wrapper, event_manager, logger)194 api_wrapper.call = Mock(return_value=None)195 logger.log.return_value = None196 player_service.print_stats()197 self._assert_log_call(logger.log, 'Failed to retrieve player and inventory stats', color='red')198 def test_heartbeat(self):199 config = create_core_test_config()200 api_wrapper = create_mock_api_wrapper(config)201 event_manager = Mock()202 logger = Mock()203 logger.log = Mock()204 player_service = Player(api_wrapper, event_manager, logger)205 pgo = api_wrapper.get_api()206 pgo.set_response('get_player', self._create_generic_player_response())207 pgo.set_response('get_inventory', self._create_generic_inventory_response())208 player_service.heartbeat()209 assert pgo.call_stack_size() == 0210 @staticmethod211 def _create_generic_player_response():212 return {213 'player_data': {214 'username': 'test_account',215 'max_pokemon_storage': 250,216 'max_item_storage': 350,217 'creation_timestamp_ms': 1470009600000,218 'currencies': [219 {220 'name': 'stardust',221 'amount': 20000,...
UserModel.py
Source:UserModel.py
...23 def _login(self,username=None,password=None):24 t_token=self.db.init_transaction() 25 resp=self.tk.check_user_session(username=username)26 if(resp):27 self.dto.set_response(token=resp,message='already_logged_in',message_connection='login')28 return29 statement=self.transactionGenerator.get_password(username=username)30 resp=self.db.make_transaction_commit(data=statement,token=t_token)31 if(len(resp['Result']['rows'])==0):32 self.dto.set_response(message='wrong_input',message_connection='login')33 return34 else:35 stored_hash=resp['Result']['rows'][0][0]36 if(self.us.check_password(password,stored_hash)):37 token=str(uuid.uuid1())38 resp=self.tk.set_token(username=username,token=token)39 if(resp>0):40 self.dto.set_response(token=token)41 else:42 self.dto.set_response(message='system_error',message_connection='login')43 else:44 self.dto.set_response(message='wrong_input',message_connection='login')45 def _get_session(self,token=None):46 resp=self.tk.retrieve_user(token=token)47 return resp48 def _logout(self,token=None):49 uname=self._get_session(token)50 if(uname):51 resp=self.tk.delete_token(token)52 if(resp>0):53 self.dto.set_response() 54 else:55 self.dto.set_response(message='system_error',message_connection='logout')56 else:57 self.dto.set_response(message='already_logged_out',message_connection='logout')58 def _get_details(self,token=None):59 uname=self._get_session(token)60 if(uname):61 statement=self.transactionGenerator.get_user_details(username=uname)62 db_token=self.db.init_transaction()63 resp=self.db.make_transaction_commit(data=statement,token=db_token)64 if(len(resp['Error'])==0):65 id_num=int(resp['Result']['rows'][0][0])66 firstname=resp['Result']['rows'][0][1]67 lastname=resp['Result']['rows'][0][2]68 mail=resp['Result']['rows'][0][3]69 self.dto.set_response(first_name=firstname,70 last_name=lastname,mail=mail,username=uname,id_number=id_num)71 else:72 self.dto.set_response(message='system_error',message_connection='user_information')73 return True74 return result75 def _set_details(self,token=None,mail=None,firstname=None,lastname=None,):76 uname=self._get_session(token)77 if(uname):78 passwordHash=self.us.hash_password(password)79 statement=statement=self.transactionGenerator.set_details(username=uname,password=passwordHash,mail=mail,80 firstname=firstname,lastname=lastname)81 db_token=self.db.init_transaction()82 resp=self.db.make_transaction_commit(data=statement,token=db_token)83 if(len(resp['Error']) == 0):84 self.dto.set_response()85 else:86 self.dto.set_response(message='system_error',message_connection='user')87 else:88 self.dto.set_response(message='session_expired',message_connection='session')89 def _register(self,username=None, password=None, mail=None,login=False,firstname=None,lastname=None):90 t_token=self.db.init_transaction()91 statement=self.transactionGenerator.get_user(username=username)92 resp=self.db.make_transaction(data=statement,token=t_token)93 if(len(resp['Error'])==0):94 if(int(resp['Result']['rows'][0][0])==0):95 passwordHash=self.us.hash_password(password)96 statement=self.transactionGenerator.register(username=username,97 password=passwordHash,mail=mail,firstname=firstname,lastname=lastname)98 resp=self.db.make_transaction_commit(data=statement,token=t_token)99 if(len(resp['Error'])==0):100 if(login):101 self._login(username=username,password=password)102 else:103 self.dto.set_response()104 else:105 self.dto.set_response(message='System Error')106 else:107 self.dto.set_response(message='User Taken') 108 else:109 self.dto.set_response(message='System Error')110 111 def get_response(self):...
handler.py
Source:handler.py
...7import constants8def start_stop(event, context):9 reserved_instances = ec2_util.describe_ec2_instances()10 if not reserved_instances:11 return set_response(400, {'message': 'no ec2 instance found'})12 start_instances, stop_instances = ec2_util.get_all_instances_to_start_stop(reserved_instances)13 ec2_util.start_stop_instances('start', start_instances)14 ec2_util.start_stop_instances('stop', stop_instances)15 body = {16 "message": "Ec2 Scheduler run successfully!",17 "instances": {'start': start_instances, 'stop': stop_instances}18 }19 response = {20 "statusCode": 200,21 "body": json.dumps(body)22 }23 return response24def handle_multi_ec2(event, context):25 body = json.loads(event['body'])26 if 'instance_ids' not in body or 'action' not in body:27 return set_response(400, {'message': 'instance_ids and action are required parameters'})28 if not body['action'] in constants.action_supported:29 return set_response(400, {'message': body['action'] + ' action not supported'})30 instance_ids = body['instance_ids']31 if type(instance_ids) != list:32 return set_response(400, {'message': 'type of instance_ids must be list'})33 ec2_reserved = ec2_util.describe_ec2_instances(34 ec2_filter=[{35 'Name': 'instance-id',36 'Values': instance_ids37 }])38 if not ec2_reserved:39 return set_response(400, {40 "instance_ids": instance_ids,41 'message': 'instance ids not found'42 })43 instances = ec2_util.get_all_instances(ec2_reserved)44 ec2_util.start_stop_instances(body['action'], instances)45 if body['action'] == 'start':46 ec2_util.create_tag(instances, [{'Key': 'ScheduledShutdown','Value':'false'}])47 else:48 ec2_util.create_tag(instances, [{'Key': 'ScheduledShutdown','Value':'true'}])49 response_body = {50 "instance_ids": instance_ids,51 "message": body['action'] + " executed successfully",52 }53 return set_response(200, response_body)54def handle_group_ec2(event, context):55 body = json.loads(event['body'])56 if 'group_ids' not in body or 'action' not in body:57 return set_response(400, {'message': 'group_ids and action are required parameters'})58 if not body['action'] in constants.action_supported:59 return set_response(400, {'message': body['action'] + ' action not supported'})60 group_ids = body['group_ids']61 if type(group_ids) != list:62 return set_response(400, {'message': 'type of group_ids must be list'})63 group_ec2 = ec2_util.describe_ec2_instances(ec2_filter=[64 {65 'Name': 'tag:Domain',66 'Values': group_ids67 }68 ])69 group_instances = ec2_util.get_all_instances(group_ec2)70 ec2_util.start_stop_instances(body['action'], group_instances)71 if body['action'] == 'start':72 ec2_util.create_tag(group_instances, [{'Key': 'ScheduledShutdown','Value':'false'}])73 else:74 ec2_util.create_tag(group_instances, [{'Key': 'ScheduledShutdown','Value':'true'}])75 response_body = {76 "instance_ids": group_instances,77 "message": body['action'] + " executed successfully",78 }79 return set_response(200, response_body)80def set_response(status_code, message):81 return {82 "statusCode": status_code,83 "body": json.dumps(message)...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
