KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use prepare_event method in localstack
Best Python code snippet using localstack_python
events.py
Source:events.py 
...69 self.type = event["eventType"]70 self.event = event71 self.event_history = event_history72 self.payload = None73 self.prepare_event()74 def prepare_event(self, event):75 raise NotImplementedError()76 def deserialize(self, data):77 return compression.decompress_b64_json(data)78 def serialize(self, data):79 return compression.compress_b64_json(data)80class WorkflowExecutionStarted(DeciderEvent):81 """WorkflowExecutionStarted Event82 {83 u'eventId': 1,84 u'eventTimestamp': datetime.datetime(2016, 3, 26, 22, 20, 7, 17000, tzinfo=tzlocal()),85 u'eventType': u'WorkflowExecutionStarted',86 u'workflowExecutionStartedEventAttributes': {87 u'childPolicy': u'TERMINATE',88 u'executionStartToCloseTimeout': u'60',89 u'input': u'H4sIADdu91YC/6tWyk0tSVSyUqiu1VFQSkmEsJVApFKSkg6IsFIwrK0FALiLFCcoAAAA',90 u'parentInitiatedEventId': 0,91 u'taskList': {u'name': u'flowbee-test-tasks'},92 u'taskPriority': u'0',93 u'taskStartToCloseTimeout': u'10',94 u'workflowType': {95 u'name': u'MyWorkflow.MyActivities',96 u'version': u'0.0.1'}97 }98 }99 """100 def prepare_event(self):101 try:102 attributes = self.event["workflowExecutionStartedEventAttributes"]103 except KeyError as e:104 message = "Unable to lookup '{0}' in {1}".format(e.message, self.event)105 log.error(message)106 raise exceptions.EventException(message=message)107 data = attributes.get("input", None)108 try:109 self.workflow_name = attributes["workflowType"]["name"]110 self.workflow_version = attributes["workflowType"]["version"]111 except KeyError as e:112 message = "Unable to lookup '{0}' in {1}".format(e.message, attributes)113 log.error(message)114 raise exceptions.EventException(message=message)115 if data is not None:116 self.payload = self.deserialize(data)117 else:118 self.payload = None119class ActivityAbstractFailure(DeciderEvent):120 def retry(self):121 log.info(122 "Retrying task '%s@%s'. Retry attempt: %s",123 self.task_name, self.task_version, self.num_retries124 )125 utils.schedule_activity(126 client=self.client,127 tasklist=self.tasklist,128 activity_id=self.activity_id,129 task_token=self.meta.task_token,130 name=self.task_name,131 version=self.task_version,132 payload=self.payload,133 attempt=self.num_retries134 )135 def process_history(self, attributes):136 try:137 scheduled_event_id = attributes["scheduledEventId"]138 except KeyError as e:139 message = "Unable to lookup '{0}' in {1}".format(e.message, self.event)140 log.error(message)141 raise exceptions.EventException(message=message)142 try:143 scheduled_activity_event = [evt for evt in self.event_history if evt["eventId"] == scheduled_event_id][0]144 except IndexError:145 message = "Unable to find event id '{0}' in event_history".format(scheduled_event_id)146 log.error(message)147 raise exceptions.EventException(message=message)148 try:149 activity = scheduled_activity_event["activityTaskScheduledEventAttributes"]150 except KeyError as e:151 message = "Unable to lookup '{0}' in {1}".format(e.message, scheduled_activity_event)152 log.error(message)153 raise exceptions.EventException(message=message)154 try:155 self.activity_id = activity["activityId"].rsplit("-", 1)[0]156 self.tasklist = activity["taskList"]["name"]157 self.task_name = activity["activityType"]["name"]158 self.task_version = activity["activityType"]["version"]159 self.payload = activity["input"]160 except KeyError as e:161 message = "Unable to find key '{0}' in 'activityTaskScheduledEventAttributes'".format(e.message)162 log.error(message)163 raise exceptions.EventException(message=message)164 self.num_retries = sum([165 1 for evt in self.event_history166 if evt["eventType"] == "ActivityTaskScheduled" and167 evt["activityTaskScheduledEventAttributes"]["activityId"].startswith(self.activity_id)168 ])169class ActivityTaskScheduled(DeciderEvent):170 """ActivityTaskScheduled Event171 {172 u'activityTaskScheduledEventAttributes': {173 u'activityId': u'com.flowbee-test.MyWorkflow.MyWorkflow.MyActivities-eb4d44a2c088452a8de053caf50209f7.23gHXuoeTXnzl8Xts+14bNNscjpxZaCJmit8tr2y2Ofzs=.stage1@0.0.1-0',174 u'activityType': {175 u'name': u'stage1',176 u'version': u'0.0.1'},177 u'decisionTaskCompletedEventId': 9,178 u'heartbeatTimeout': u'NONE',179 u'input': u'H4sIADxu91YC/6tWSixKL1ayUohWyilNrlSK1VFQyi6HilUrpeXng+lEIKmUpKQDIqwUDGtrawHg8m1aOQAAAA==',180 u'scheduleToCloseTimeout': u'10',181 u'scheduleToStartTimeout': u'10',182 u'startToCloseTimeout': u'NONE',183 u'taskList': {u'name': u'flowbee-test-tasks'},184 u'taskPriority': u'0'},185 u'eventId': 10,186 u'eventTimestamp': datetime.datetime(2016, 3, 26, 22, 20, 12, 560000, tzinfo=tzlocal()),187 u'eventType': u'ActivityTaskScheduled'188 }189 """190 def prepare_event(self):191 try:192 attributes = self.event["activityTaskScheduledEventAttributes"]193 except KeyError:194 message = "Unable to lookup '{0}' in {1}".format(e.message, self.event)195 log.error(message)196 raise exceptions.EventException(message=message)197 self.tasklist = attributes["taskList"]["name"]198 self.priority = attributes["taskPriority"]199 self.name = attributes["activityType"]["name"]200 self.version = attributes["activityType"]["version"]201 self.activity_id = attributes["activityId"]202 data = attributes.get("input", None)203 if data is not None:204 self.payload = self.deserialize(data)205 else:206 self.payload = None207class ActivityTaskStarted(DeciderEvent):208 """ActivityTaskStarted209 {210 u'activityTaskStartedEventAttributes': {211 u'identity': u'MyWorkflow',212 u'scheduledEventId': 10},213 u'eventId': 11,214 u'eventTimestamp': datetime.datetime(2016, 3, 26, 22, 20, 12, 599000, tzinfo=tzlocal()),215 u'eventType': u'ActivityTaskStarted'216 }217 """218 def prepare_event(self):219 return220class ActivityTaskCompleted(DeciderEvent):221 """ActivityTaskCompleted Event222 {223 u'eventId': 15,224 u'eventType': u'ActivityTaskCompleted',225 u'activityTaskCompletedEventAttributes': {226 u'startedEventId': 14,227 u'scheduledEventId': 13,228 u'result': u'H4sIABZt91YC/1MqLilKLE9KLSrKTC1WAgBhRJKGDgAAAA=='},229 u'eventTimestamp': datetime.datetime(2016, 3, 26, 22, 15, 17, 771000, tzinfo=tzlocal())230 }231 """232 def prepare_event(self):233 data = self.event \234 .get("activityTaskCompletedEventAttributes", {}) \235 .get("result", None)236 if data is not None:237 self.payload = self.deserialize(data)238 else:239 self.payload = None240class ActivityTaskTimedOut(ActivityAbstractFailure):241 """ActivityTaskTimedOut Event242 {243 u'activityTaskTimedOutEventAttributes': {244 u'scheduledEventId': 16,245 u'startedEventId': 17,246 u'timeoutType': u'SCHEDULE_TO_CLOSE'},247 u'eventId': 18,248 u'eventTimestamp': datetime.datetime(2016, 3, 26, 22, 29, 57, 609000, tzinfo=tzlocal()),249 u'eventType': u'ActivityTaskTimedOut'250 }251 """252 def prepare_event(self):253 self.num_retries = 0254 attributes = self.event.get("activityTaskTimedOutEventAttributes")255 self.process_history(attributes)256class ActivityTaskFailed(ActivityAbstractFailure):257 def prepare_event(self):258 self.num_retries = 0259 attributes = self.event.get("activityTaskFailedEventAttributes")260 self.process_history(attributes)261class ScheduleActivityTaskFailed(DeciderEvent):262 def prepare_event(self):263 attributes = self.event["scheduleActivityTaskFailed"]264 activity_id = attributes.get("activityId", "unknown activity id")265 activity_name = attributes.get("activityType", {}).get("name", "unknown name")266 activity_version = attributes.get("activityType", {}).get("version", "unknown version")267 cause = attributes.get("cause", "unknown")268 message = "Failed to schedule activity[%s@%s]: %s - %s" \269 .format(cause, activity_name, activity_version, activity_id)270 log.error(message)271 raise exceptions.EventException(message=message)272class TimerStarted(DeciderEvent):273 """TimerStarted Event274 {275 u'eventId': 5,276 u'eventTimestamp': datetime.datetime(2016, 3, 26, 22, 20, 7, 363000, tzinfo=tzlocal()),277 u'eventType': u'TimerStarted',278 u'timerStartedEventAttributes': {279 u'decisionTaskCompletedEventId': 4,280 u'startToFireTimeout': u'5',281 u'timerId': u'com.flowbee-test.MyWorkflow.MyWorkflow.MyActivities-eb4d44a2c088452a8de053caf50209f7.23gHXuoeTXnzl8Xts+14bNNscjpxZaCJmit8tr2y2Ofzs='}282 }283 """284 def prepare_event(self):285 try:286 attributes = self.event["timerStartedEventAttributes"]287 except KeyError:288 message = "Unable to locate 'timerStartedEventAttributes' on {0}".format(self.event)289 log.error(message)290 raise exceptions.EventException(message=message)291 self.timer_id = attributes["timerId"]292 self.seconds = int(attributes["startToFireTimeout"])293 try:294 data = attributes["control"]295 except KeyError:296 data = None297 if data is None:298 self.payload = None299 else:300 self.payload = self.deserialize(data)301class TimerFired(DeciderEvent):302 """TimerFired Event303 {304 u'eventId': 6,305 u'eventTimestamp': datetime.datetime(2016, 3, 26, 22, 20, 12, 367000, tzinfo=tzlocal()),306 u'eventType': u'TimerFired',307 u'timerFiredEventAttributes': {308 u'startedEventId': 5,309 u'timerId': u'com.flowbee-test.MyWorkflow.MyWorkflow.MyActivities-eb4d44a2c088452a8de053caf50209f7.23gHXuoeTXnzl8Xts+14bNNscjpxZaCJmit8tr2y2Ofzs='}310 }311 """312 def prepare_event(self):313 timer_id = self.event.get("timerFiredEventAttributes", {}).get("timerId")314 self.timer_id = timer_id315 if timer_id is None:316 message = "Unable to locate 'timerId' on 'timerFiredEventAttributes'"317 log.error(message)318 raise exceptions.EventException(message=message)319 try:320 timer_started_event = [321 x for x in self.event_history322 if x["eventType"] == "TimerStarted" and323 x["timerStartedEventAttributes"]["timerId"] == timer_id][0]324 except KeyError as e:325 message = "Failed to find key in event_history '{0}'".format(e.message)326 log.error(message)...
test_pyseco.py
Source:test_pyseco.py
1import abc2import pytest3from collections import namedtuple4from unittest.mock import Mock, call5from random import randint6from src.api.tm_types import Status, ChallengeInfo7from src.errors import NotAnEvent, EventDiscarded8from src.pyseco import Listener, Pyseco9DummyEventData = namedtuple('DummyEventData', ['name', 'data'])10DummyEvent = namedtuple('DummyEvent', ['name'])11CallAssert = namedtuple('CallAssert', ['call', 'params'])12TM_FOREVER = 113DummyConfig = namedtuple('Dummyconfig', ['prefix', 'color', 'tm_login', 'rcp_login', 'rcp_password', 'rcp_ip',14 'rcp_port', 'db_hostname', 'db_user', 'db_password', 'db_name', 'db_charset'])15DUMMY_CONFIG = DummyConfig("T", "$00f", "server_login", "login", "password", "11.22.33.44", 5002, "localhost", "root",16 "passwd", "aseco", "utf8")17DUMMY_PATH_TO_CONFIG = '/path/to/config.yaml'18class Events:19 EVENT1 = DummyEvent('EVENT1')20 EVENT2 = DummyEvent('EVENT2')21 EVENT3 = DummyEvent('EVENT3')22 EVENT4 = DummyEvent('EVENT4')23 EVENT5 = DummyEvent('EVENT5')24class DummyListener(Listener):25 def __init__(self, mocker):26 super().__init__('dummy_listener', Mock())27 self.on_dummy_event1 = mocker.stub(name='on_dummy_event1_stub')28 self.on_dummy_event2 = mocker.stub(name='on_dummy_event2_stub')29 self.on_dummy_event3 = mocker.stub(name='on_dummy_event3_stub')30 self.on_dummy_event4 = mocker.stub(name='on_dummy_event4_stub')31 self.on_dummy_event5 = mocker.stub(name='on_dummy_event5_stub')32# .assert_called_with(Any(str, int))33def Any(*cls):34 class AnyComparator(metaclass=abc.ABCMeta):35 def __eq__(self, other):36 return isinstance(other, cls)37 for c in cls:38 AnyComparator.register(c)39 return AnyComparator()40def random_data():41 return randint(0, 500), randint(0, 500)42def make_event(ev):43 return DummyEventData(name=ev.name, data=random_data())44def make_empty_event(ev):45 return DummyEventData(name=ev.name, data=None)46@pytest.fixture47def pyseco(mocker):48 mocker.patch('src.pyseco.is_bound').return_value = True49 mocker.patch('src.pyseco.Pyseco.start_listening')50 pyseco = Pyseco(DUMMY_PATH_TO_CONFIG)51 return pyseco52@pytest.fixture(autouse=True)53def transport(mocker):54 return mocker.patch('src.pyseco.Transport')55@pytest.fixture(autouse=True)56def config(mocker):57 config_mock = mocker.patch('src.pyseco.Config')58 config_mock.return_value = DUMMY_CONFIG59 return config_mock60@pytest.fixture(autouse=True)61def server(mocker):62 config_mock = mocker.patch('src.pyseco.ServerCtx')63 return config_mock64@pytest.fixture(autouse=True)65def mysql(mocker):66 return mocker.patch('src.pyseco.MySqlWrapper')67@pytest.fixture(autouse=True)68def rpc(mocker):69 rpc = mocker.patch('src.pyseco.XmlRpc')70 rpc.return_value.get_status.return_value = Status({'id': 4, 'name': 'status'})71 rpc.return_value.get_current_challenge_info.return_value = ChallengeInfo()72 return rpc73def test_should_create_object(transport, mysql, config, rpc, server):74 pyseco = Pyseco(DUMMY_PATH_TO_CONFIG)75 transport.assert_called_once_with(DUMMY_CONFIG.rcp_ip, DUMMY_CONFIG.rcp_port, pyseco.events_queue)76 config.assert_called_once_with(DUMMY_PATH_TO_CONFIG)77 mysql.assert_called_once_with(DUMMY_CONFIG)78 rpc.assert_called_once_with(pyseco.transport)79 server.assert_called_once_with(pyseco.rpc, pyseco.config)80def test_should_disconnect_on_exit(transport):81 with Pyseco(DUMMY_PATH_TO_CONFIG):82 transport.return_value.disconnect.assert_not_called()83 transport.return_value.disconnect.assert_called_once()84def test_events_map_should_be_empty_when_no_listeners_registered():85 pyseco = Pyseco(DUMMY_PATH_TO_CONFIG)86 assert len(pyseco.events_matrix) == 087def test_should_sync_data_on_run(transport, pyseco, rpc, server):88 rpc.return_value.get_player_list.return_value = []89 pyseco.run()90 transport.return_value.connect.assert_called_once()91 rpc.return_value.authenticate.assert_called_once_with(DUMMY_CONFIG.rcp_login, DUMMY_CONFIG.rcp_password)92 rpc.return_value.enable_callbacks.assert_called_once_with(True)93 server.return_value.synchronize.assert_called_once()94 rpc.return_value.get_player_list.assert_called_once()95def test_an_exception_other_than_keyboardinterrupt_should_be_passed_further(rpc, pyseco):96 rpc.return_value.authenticate.return_value = True97 rpc.return_value.chat_send_server_message.side_effect = Exception98 with pytest.raises(Exception):99 pyseco.run()100def test_should_disconnect_on_keyboardinterrupt(rpc, pyseco, transport):101 rpc.return_value.authenticate.return_value = True102 rpc.return_value.chat_send_server_message.side_effect = KeyboardInterrupt103 pyseco.run()104 transport.return_value.disconnect.assert_called_once()105def test_should_add_registered_listener_to_events_map(mocker, pyseco):106 listener = DummyListener(mocker)107 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)108 assert listener.on_dummy_event1 in pyseco.events_matrix[Events.EVENT1.name]109def test_only_registered_listener_method_should_be_called_on_event(mocker, pyseco):110 prepare_event = mocker.patch.object(pyseco, '_prepare_event')111 listener = DummyListener(mocker)112 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)113 event = make_event(Events.EVENT1)114 prepare_event.return_value = event115 pyseco.handle_event(None)116 listener.on_dummy_event1.assert_called_with(event.data)117 listener.on_dummy_event2.assert_not_called()118 listener.on_dummy_event3.assert_not_called()119 listener.on_dummy_event4.assert_not_called()120 listener.on_dummy_event5.assert_not_called()121def test_registered_method_should_be_without_params_on_empty_event(mocker, pyseco):122 prepare_event = mocker.patch.object(pyseco, '_prepare_event')123 listener = DummyListener(mocker)124 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)125 event = make_empty_event(Events.EVENT1)126 prepare_event.return_value = event127 pyseco.handle_event(None)128 listener.on_dummy_event1.assert_called_with()129def test_method_registered_multiple_times_should_be_called_once(mocker, pyseco):130 prepare_event = mocker.patch.object(pyseco, '_prepare_event')131 listener = DummyListener(mocker)132 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)133 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)134 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)135 event = make_event(Events.EVENT1)136 prepare_event.return_value = event137 pyseco.handle_event(None)138 listener.on_dummy_event1.assert_called_once()139 listener.on_dummy_event1.assert_called_with(event.data)140def test_listener_should_be_called_multiple_times_on_multiple_events(mocker, pyseco):141 prepare_event = mocker.patch.object(pyseco, '_prepare_event')142 listener = DummyListener(mocker)143 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)144 events_queue = [make_event(Events.EVENT1), make_event(Events.EVENT1), make_event(Events.EVENT1)]145 prepare_event.side_effect = events_queue146 [pyseco.handle_event(None) for _ in events_queue]147 listener.on_dummy_event1.assert_has_calls([call(event.data) for event in events_queue])148def test_multiple_listeners_registered_for_one_event(mocker, pyseco):149 prepare_event = mocker.patch.object(pyseco, '_prepare_event')150 listener1 = DummyListener(mocker)151 listener2 = DummyListener(mocker)152 listener3 = DummyListener(mocker)153 pyseco.register(Events.EVENT1.name, listener1.on_dummy_event1)154 pyseco.register(Events.EVENT1.name, listener2.on_dummy_event1)155 pyseco.register(Events.EVENT1.name, listener3.on_dummy_event1)156 event = make_event(Events.EVENT1)157 prepare_event.return_value = event158 pyseco.handle_event(None)159 listener1.on_dummy_event1.assert_called_with(event.data)160 listener2.on_dummy_event1.assert_called_with(event.data)161 listener3.on_dummy_event1.assert_called_with(event.data)162def test_multiple_methods_registered_for_one_event(mocker, pyseco):163 prepare_event = mocker.patch.object(pyseco, '_prepare_event')164 listener = DummyListener(mocker)165 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)166 pyseco.register(Events.EVENT2.name, listener.on_dummy_event2)167 pyseco.register(Events.EVENT3.name, listener.on_dummy_event3)168 events_queue = [make_event(Events.EVENT1), make_event(Events.EVENT2), make_event(Events.EVENT3)]169 prepare_event.side_effect = events_queue170 [pyseco.handle_event(None) for _ in events_queue]171 listener.on_dummy_event1.assert_called_with(events_queue[0].data)172 listener.on_dummy_event2.assert_called_with(events_queue[1].data)173 listener.on_dummy_event3.assert_called_with(events_queue[2].data)174def test_should_not_call_method_on_exception_raised(mocker, pyseco):175 prepare_event = mocker.patch.object(pyseco, '_prepare_event')176 listener = DummyListener(mocker)177 pyseco.register(Events.EVENT1.name, listener.on_dummy_event1)178 pyseco.register(Events.EVENT2.name, listener.on_dummy_event2)179 pyseco.register(Events.EVENT3.name, listener.on_dummy_event3)180 pyseco.register(Events.EVENT4.name, listener.on_dummy_event4)181 pyseco.register(Events.EVENT5.name, listener.on_dummy_event5)182 events_queue = [make_event(Events.EVENT1),183 NotAnEvent,184 EventDiscarded,185 make_empty_event(Events.EVENT4),186 make_event(Events.EVENT5)]187 prepare_event.side_effect = events_queue188 [pyseco.handle_event(None) for _ in events_queue]189 listener.on_dummy_event1.assert_called_with(events_queue[0].data)190 listener.on_dummy_event2.assert_not_called()191 listener.on_dummy_event3.assert_not_called()192 listener.on_dummy_event4.assert_called_with()...
app.py
Source:app.py
...19 return float(o)20 else:21 return int(o)22 return super(DecimalEncoder, self).default(o)23def prepare_event(event):24 params = {}25 if event.get('queryStringParameters'):26 params = {**event.get('queryStringParameters'), **params}27 if event.get('body'):28 body = json.loads(event.get('body'))29 params = {**body, **params}30 return params31def post_list(event, context):32 """ List Post33 """34 dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')35 table = dynamodb.Table('Blog')36 response = table.scan()37 return {38 "statusCode": 200,39 'headers': {40 "Access-Control-Allow-Origin": "*",41 "Access-Control-Allow-Headers": "Content-Type",42 "Access-Control-Allow-Methods": "OPTIONS,POST,GET"43 },44 "body": json.dumps(response, indent=4, cls=DecimalEncoder),45 }46def post_create(event, context):47 event = prepare_event(event)48 user_name = event['username']49 post_content = event['content']50 image = "null"51 if event.get('image'):52 image = event['image']53 now_timestamp = utcnow()54 dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')55 table = dynamodb.Table('Blog')56 table.put_item(57 Item={58 "id": f"{now_timestamp}-{user_name}",59 "content": post_content,60 "username": user_name,61 "comments": [],62 "image": image,63 "timestamp": now_timestamp64 }65 )66 response = table.get_item(67 Key={68 'id': f"{now_timestamp}-{user_name}"69 }70 )71 item = response['Item']72 return {73 "statusCode": 200,74 'headers': {75 "Access-Control-Allow-Origin": "*",76 "Access-Control-Allow-Headers": "Content-Type",77 "Access-Control-Allow-Methods": "OPTIONS,POST,GET"78 },79 "body": json.dumps(item, indent=4, cls=DecimalEncoder),80 }81def post_delete(event, context):82 event = prepare_event(event)83 post_id = event['id']84 dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')85 table = dynamodb.Table('Blog')86 table.delete_item(87 Key={88 'id': post_id89 }90 )91 return {92 "statusCode": 200,93 'headers': {94 "Access-Control-Allow-Origin": "*",95 "Access-Control-Allow-Headers": "Content-Type",96 "Access-Control-Allow-Methods": "OPTIONS,POST,GET"97 },98 "body": json.dumps({'id': post_id }, indent=4, cls=DecimalEncoder),99 }100def comment_create(event, context):101 event = prepare_event(event)102 post_id = event['postid']103 user_name = event['username']104 comment = event['comment']105 now_timestamp = utcnow()106 comment_item = dict()107 comment_item["comment"] = comment108 comment_item["liked"] = "false"109 comment_item["timestamp"] = now_timestamp110 comment_item["username"] = user_name111 dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')112 table = dynamodb.Table('Blog')113 response = table.get_item(114 Key={115 'id': post_id116 }117 )118 item = response['Item']119 item['comments'].append(comment_item)120 table.update_item(121 Key={122 'id': post_id123 },124 UpdateExpression='SET comments = :val1',125 ExpressionAttributeValues={126 ':val1': item['comments']127 }128 )129 return {130 "statusCode": 200,131 'headers': {132 "Access-Control-Allow-Origin": "*",133 "Access-Control-Allow-Headers": "Content-Type",134 "Access-Control-Allow-Methods": "OPTIONS,POST,GET"135 },136 "body": json.dumps(item, indent=4, cls=DecimalEncoder),137 }138def comment_delete(event, context):139 event = prepare_event(event)140 post_id = event['id']141 user_name = event['username']142 timestamp = event['timestamp']143 dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')144 table = dynamodb.Table('Blog')145 response = table.get_item(146 Key={147 'id': post_id148 }149 )150 item = response['Item']151 an_iterator = filter(lambda comment: comment['timestamp'] != Decimal(timestamp) or comment['username']!=user_name, item['comments'])152 item['comments'] = list(an_iterator)153 table.update_item(...
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?
