KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use get_event_source_arn method in localstack
Best Python code snippet using localstack_python
test_trigger.py
Source:test_trigger.py 
...21 with open(test_file, "r") as event:22 event = json.load(event)23 ctx = get_mock_context()24 event_source = parse_event_source(event)25 event_source_arn = get_event_source_arn(event_source, event, ctx)26 self.assertEqual(event_source.to_string(), event_sample_source)27 self.assertEqual(28 event_source_arn,29 "arn:aws:apigateway:us-west-1::/restapis/1234567890/stages/prod",30 )31 def test_event_source_api_gateway_non_proxy(self):32 event_sample_source = "api-gateway-non-proxy"33 test_file = event_samples + event_sample_source + ".json"34 with open(test_file, "r") as event:35 event = json.load(event)36 ctx = get_mock_context()37 event_source = parse_event_source(event)38 event_source_arn = get_event_source_arn(event_source, event, ctx)39 self.assertEqual(event_source.to_string(), "api-gateway")40 self.assertEqual(event_source.subtype, EventSubtypes.API_GATEWAY)41 self.assertEqual(42 event_source_arn,43 "arn:aws:apigateway:us-west-1::/restapis/lgxbo6a518/stages/dev",44 )45 def test_event_source_api_gateway_websocket_connect(self):46 event_sample_source = "api-gateway-websocket-connect"47 test_file = event_samples + event_sample_source + ".json"48 with open(test_file, "r") as event:49 event = json.load(event)50 ctx = get_mock_context()51 event_source = parse_event_source(event)52 event_source_arn = get_event_source_arn(event_source, event, ctx)53 self.assertEqual(event_source.to_string(), "api-gateway")54 self.assertEqual(event_source.subtype, EventSubtypes.WEBSOCKET)55 self.assertEqual(56 event_source_arn,57 "arn:aws:apigateway:us-west-1::/restapis/p62c47itsb/stages/dev",58 )59 def test_event_source_api_gateway_websocket_default(self):60 event_sample_source = "api-gateway-websocket-default"61 test_file = event_samples + event_sample_source + ".json"62 with open(test_file, "r") as event:63 event = json.load(event)64 ctx = get_mock_context()65 event_source = parse_event_source(event)66 event_source_arn = get_event_source_arn(event_source, event, ctx)67 self.assertEqual(event_source.to_string(), "api-gateway")68 self.assertEqual(event_source.subtype, EventSubtypes.WEBSOCKET)69 self.assertEqual(70 event_source_arn,71 "arn:aws:apigateway:us-west-1::/restapis/p62c47itsb/stages/dev",72 )73 def test_event_source_api_gateway_websocket_disconnect(self):74 event_sample_source = "api-gateway-websocket-disconnect"75 test_file = event_samples + event_sample_source + ".json"76 with open(test_file, "r") as event:77 event = json.load(event)78 ctx = get_mock_context()79 event_source = parse_event_source(event)80 event_source_arn = get_event_source_arn(event_source, event, ctx)81 self.assertEqual(event_source.to_string(), "api-gateway")82 self.assertEqual(event_source.subtype, EventSubtypes.WEBSOCKET)83 self.assertEqual(84 event_source_arn,85 "arn:aws:apigateway:us-west-1::/restapis/p62c47itsb/stages/dev",86 )87 def test_event_source_api_gateway_http_api(self):88 event_sample_source = "http-api"89 test_file = event_samples + event_sample_source + ".json"90 with open(test_file, "r") as event:91 event = json.load(event)92 ctx = get_mock_context()93 event_source = parse_event_source(event)94 event_source_arn = get_event_source_arn(event_source, event, ctx)95 self.assertEqual(event_source.to_string(), "api-gateway")96 self.assertEqual(event_source.subtype, EventSubtypes.HTTP_API)97 self.assertEqual(98 event_source_arn,99 "arn:aws:apigateway:us-west-1::/restapis/x02yirxc7a/stages/$default",100 )101 def test_event_source_application_load_balancer(self):102 event_sample_source = "application-load-balancer"103 test_file = event_samples + event_sample_source + ".json"104 with open(test_file, "r") as event:105 event = json.load(event)106 ctx = get_mock_context()107 event_source = parse_event_source(event)108 event_source_arn = get_event_source_arn(event_source, event, ctx)109 self.assertEqual(event_source.to_string(), event_sample_source)110 self.assertEqual(111 event_source_arn,112 "arn:aws:elasticloadbalancing:us-east-2:123456789012:targetgroup/lambda-xyz/123abc",113 )114 def test_event_source_cloudfront(self):115 event_sample_source = "cloudfront"116 test_file = event_samples + event_sample_source + ".json"117 with open(test_file, "r") as event:118 event = json.load(event)119 ctx = get_mock_context()120 event_source = parse_event_source(event)121 event_source_arn = get_event_source_arn(event_source, event, ctx)122 self.assertEqual(event_source.to_string(), event_sample_source)123 self.assertEqual(124 event_source_arn, "arn:aws:cloudfront::123457598159:distribution/EXAMPLE"125 )126 def test_event_source_cloudwatch_events(self):127 event_sample_source = "cloudwatch-events"128 test_file = event_samples + event_sample_source + ".json"129 with open(test_file, "r") as event:130 event = json.load(event)131 ctx = get_mock_context()132 event_source = parse_event_source(event)133 event_source_arn = get_event_source_arn(event_source, event, ctx)134 self.assertEqual(event_source.to_string(), event_sample_source)135 self.assertEqual(136 event_source_arn, "arn:aws:events:us-east-1:123456789012:rule/ExampleRule"137 )138 def test_event_source_cloudwatch_logs(self):139 event_sample_source = "cloudwatch-logs"140 test_file = event_samples + event_sample_source + ".json"141 with open(test_file, "r") as event:142 event = json.load(event)143 ctx = get_mock_context()144 event_source = parse_event_source(event)145 event_source_arn = get_event_source_arn(event_source, event, ctx)146 self.assertEqual(event_source.to_string(), event_sample_source)147 self.assertEqual(148 event_source_arn,149 "arn:aws:logs:us-west-1:123457598159:log-group:testLogGroup",150 )151 def test_event_source_dynamodb(self):152 event_sample_source = "dynamodb"153 test_file = event_samples + event_sample_source + ".json"154 with open(test_file, "r") as event:155 event = json.load(event)156 ctx = get_mock_context()157 event_source = parse_event_source(event)158 event_source_arn = get_event_source_arn(event_source, event, ctx)159 self.assertEqual(event_source.to_string(), event_sample_source)160 self.assertEqual(161 event_source_arn,162 "arn:aws:dynamodb:us-east-1:123456789012:table/ExampleTableWithStream/stream/2015-06-27T00:48:05.899",163 )164 def test_event_source_kinesis(self):165 event_sample_source = "kinesis"166 test_file = event_samples + event_sample_source + ".json"167 with open(test_file, "r") as event:168 event = json.load(event)169 ctx = get_mock_context()170 event_source = parse_event_source(event)171 event_source_arn = get_event_source_arn(event_source, event, ctx)172 self.assertEqual(event_source.to_string(), event_sample_source)173 self.assertEqual(174 event_source_arn,175 "arn:aws:kinesis:sa-east-1:601427279990:stream/kinesisStream",176 )177 def test_event_source_s3(self):178 event_sample_source = "s3"179 test_file = event_samples + event_sample_source + ".json"180 with open(test_file, "r") as event:181 event = json.load(event)182 ctx = get_mock_context()183 event_source = parse_event_source(event)184 event_source_arn = get_event_source_arn(event_source, event, ctx)185 self.assertEqual(event_source.to_string(), event_sample_source)186 self.assertEqual(event_source_arn, "arn:aws:s3:::example-bucket")187 def test_event_source_sns(self):188 event_sample_source = "sns"189 event_sample_name = "sns-string-msg-attribute"190 test_file = event_samples + event_sample_name + ".json"191 with open(test_file, "r") as event:192 event = json.load(event)193 ctx = get_mock_context()194 event_source = parse_event_source(event)195 event_source_arn = get_event_source_arn(event_source, event, ctx)196 self.assertEqual(event_source.to_string(), event_sample_source)197 self.assertEqual(198 event_source_arn,199 "arn:aws:sns:sa-east-1:601427279990:serverlessTracingTopicPy",200 )201 def test_event_source_sqs(self):202 event_sample_source = "sqs"203 event_sample_name = "sqs-string-msg-attribute"204 test_file = event_samples + event_sample_name + ".json"205 with open(test_file, "r") as event:206 event = json.load(event)207 ctx = get_mock_context()208 event_source = parse_event_source(event)209 event_source_arn = get_event_source_arn(event_source, event, ctx)210 self.assertEqual(event_source.to_string(), event_sample_source)211 self.assertEqual(212 event_source_arn,213 "arn:aws:sqs:sa-east-1:601427279990:InferredSpansQueueNode",214 )215 def test_event_source_unsupported(self):216 event_sample_source = "custom"217 test_file = event_samples + event_sample_source + ".json"218 with open(test_file, "r") as event:219 event = json.load(event)220 ctx = get_mock_context()221 event_source = parse_event_source(event)222 event_source_arn = get_event_source_arn(event_source, event, ctx)223 self.assertEqual(event_source.to_string(), "unknown")224 self.assertEqual(event_source_arn, None)225class GetTriggerTags(unittest.TestCase):226 def test_extract_trigger_tags_api_gateway(self):227 event_sample_source = "api-gateway"228 test_file = event_samples + event_sample_source + ".json"229 ctx = get_mock_context()230 with open(test_file, "r") as event:231 event = json.load(event)232 tags = extract_trigger_tags(event, ctx)233 self.assertEqual(234 tags,235 {236 "function_trigger.event_source": "api-gateway",...
test_integration.py
Source:test_integration.py
...159 # deploy test lambdas connected to Kinesis streams160 zip_file = testutil.create_lambda_archive(load_file(TEST_LAMBDA_PYTHON), get_content=True,161 libs=TEST_LAMBDA_LIBS, runtime=LAMBDA_RUNTIME_PYTHON27)162 testutil.create_lambda_function(func_name=TEST_CHAIN_LAMBDA1_NAME, zip_file=zip_file,163 event_source_arn=get_event_source_arn(TEST_CHAIN_STREAM1_NAME), runtime=LAMBDA_RUNTIME_PYTHON27)164 testutil.create_lambda_function(func_name=TEST_CHAIN_LAMBDA2_NAME, zip_file=zip_file,165 event_source_arn=get_event_source_arn(TEST_CHAIN_STREAM2_NAME), runtime=LAMBDA_RUNTIME_PYTHON27)166 # publish test record167 test_data = {'test_data': 'forward_chain_data_%s' % short_uid()}168 data = clone(test_data)169 data[lambda_integration.MSG_BODY_MESSAGE_TARGET] = 'kinesis:%s' % TEST_CHAIN_STREAM2_NAME170 kinesis.put_record(Data=to_bytes(json.dumps(data)), PartitionKey='testId', StreamName=TEST_CHAIN_STREAM1_NAME)171 # check results172 time.sleep(5)173 all_objects = testutil.list_all_s3_objects()174 testutil.assert_objects(test_data, all_objects)175# ---------------176# HELPER METHODS177# ---------------178def get_event_source_arn(stream_name):179 kinesis = aws_stack.connect_to_service('kinesis')180 return kinesis.describe_stream(StreamName=stream_name)['StreamDescription']['StreamARN']181def get_lambda_metrics(func_name, metric='Invocations'):182 return cloudwatch_util.get_metric_statistics(183 Namespace='AWS/Lambda',184 MetricName=metric,185 Dimensions=[{'Name': 'FunctionName', 'Value': func_name}],186 Period=60,187 StartTime=datetime.now() - timedelta(minutes=1),188 EndTime=datetime.now(),189 Statistics=['Sum']...
34121_test_integration.py
Source:34121_test_integration.py
...159 # deploy test lambdas connected to Kinesis streams160 zip_file = testutil.create_lambda_archive(load_file(TEST_LAMBDA_PYTHON), get_content=True,161 libs=TEST_LAMBDA_LIBS, runtime=LAMBDA_RUNTIME_PYTHON27)162 testutil.create_lambda_function(func_name=TEST_CHAIN_LAMBDA1_NAME, zip_file=zip_file,163 event_source_arn=get_event_source_arn(TEST_CHAIN_STREAM1_NAME), runtime=LAMBDA_RUNTIME_PYTHON27)164 testutil.create_lambda_function(func_name=TEST_CHAIN_LAMBDA2_NAME, zip_file=zip_file,165 event_source_arn=get_event_source_arn(TEST_CHAIN_STREAM2_NAME), runtime=LAMBDA_RUNTIME_PYTHON27)166 # publish test record167 test_data = {'test_data': 'forward_chain_data_%s' % short_uid()}168 data = clone(test_data)169 data[lambda_integration.MSG_BODY_MESSAGE_TARGET] = 'kinesis:%s' % TEST_CHAIN_STREAM2_NAME170 kinesis.put_record(Data=to_bytes(json.dumps(data)), PartitionKey='testId', StreamName=TEST_CHAIN_STREAM1_NAME)171 # check results172 time.sleep(5)173 all_objects = testutil.list_all_s3_objects()174 testutil.assert_objects(test_data, all_objects)175# ---------------176# HELPER METHODS177# ---------------178def get_event_source_arn(stream_name):179 kinesis = aws_stack.connect_to_service('kinesis')180 return kinesis.describe_stream(StreamName=stream_name)['StreamDescription']['StreamARN']181def get_lambda_metrics(func_name, metric='Invocations'):182 return cloudwatch_util.get_metric_statistics(183 Namespace='AWS/Lambda',184 MetricName=metric,185 Dimensions=[{'Name': 'FunctionName', 'Value': func_name}],186 Period=60,187 StartTime=datetime.now() - timedelta(minutes=1),188 EndTime=datetime.now(),189 Statistics=['Sum']...
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?
