Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.