Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.