KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _resource_arn method in localstack
Best Python code snippet using localstack_python
aws_stack.py
Source:aws_stack.py 
...501 if ":policy/" in policy_name:502 return policy_name503 account_id = account_id or get_aws_account_id()504 return "arn:aws:iam::{}:policy/{}".format(account_id, policy_name)505def iam_resource_arn(resource, role=None, env=None):506 env = get_environment(env)507 if not role:508 role = get_iam_role(resource, env=env)509 return role_arn(role_name=role, account_id=get_aws_account_id())510def get_iam_role(resource, env=None):511 env = get_environment(env)512 return "role-%s" % resource513def secretsmanager_secret_arn(secret_id, account_id=None, region_name=None, random_suffix=None):514 if ":" in (secret_id or ""):515 return secret_id516 pattern = "arn:aws:secretsmanager:%s:%s:secret:%s"517 arn = _resource_arn(secret_id, pattern, account_id=account_id, region_name=region_name)518 if random_suffix:519 arn += f"-{random_suffix}"520 return arn521def cloudformation_stack_arn(stack_name, stack_id=None, account_id=None, region_name=None):522 stack_id = stack_id or "id-123"523 pattern = "arn:aws:cloudformation:%s:%s:stack/%s/{stack_id}".format(stack_id=stack_id)524 return _resource_arn(stack_name, pattern, account_id=account_id, region_name=region_name)525def cf_change_set_arn(change_set_name, change_set_id=None, account_id=None, region_name=None):526 change_set_id = change_set_id or "id-456"527 pattern = "arn:aws:cloudformation:%s:%s:changeSet/%s/{cs_id}".format(cs_id=change_set_id)528 return _resource_arn(change_set_name, pattern, account_id=account_id, region_name=region_name)529def dynamodb_table_arn(table_name, account_id=None, region_name=None):530 table_name = table_name.split(":table/")[-1]531 pattern = "arn:aws:dynamodb:%s:%s:table/%s"532 return _resource_arn(table_name, pattern, account_id=account_id, region_name=region_name)533def dynamodb_stream_arn(table_name, latest_stream_label, account_id=None):534 account_id = account_id or get_aws_account_id()535 return "arn:aws:dynamodb:%s:%s:table/%s/stream/%s" % (536 get_region(),537 account_id,538 table_name,539 latest_stream_label,540 )541def cloudwatch_alarm_arn(alarm_name, account_id=None, region_name=None):542 pattern = "arn:aws:cloudwatch:%s:%s:alarm:%s"543 return _resource_arn(alarm_name, pattern, account_id=account_id, region_name=region_name)544def log_group_arn(group_name, account_id=None, region_name=None):545 pattern = "arn:aws:logs:%s:%s:log-group:%s"546 return _resource_arn(group_name, pattern, account_id=account_id, region_name=region_name)547def events_rule_arn(rule_name, account_id=None, region_name=None):548 pattern = "arn:aws:events:%s:%s:rule/%s"549 return _resource_arn(rule_name, pattern, account_id=account_id, region_name=region_name)550def lambda_function_arn(function_name, account_id=None, region_name=None):551 return lambda_function_or_layer_arn(552 "function", function_name, account_id=account_id, region_name=region_name553 )554def lambda_layer_arn(layer_name, version=None, account_id=None):555 return lambda_function_or_layer_arn("layer", layer_name, version=None, account_id=account_id)556def lambda_function_or_layer_arn(557 type, entity_name, version=None, account_id=None, region_name=None558):559 pattern = "arn:([a-z-]+):lambda:.*:.*:(function|layer):.*"560 if re.match(pattern, entity_name):561 return entity_name562 if ":" in entity_name:563 client = connect_to_service("lambda")564 entity_name, _, alias = entity_name.rpartition(":")565 try:566 alias_response = client.get_alias(FunctionName=entity_name, Name=alias)567 version = alias_response["FunctionVersion"]568 except Exception as e:569 msg = f"Alias {alias} of {entity_name} not found"570 LOG.info(f"{msg}: {e}")571 raise Exception(msg)572 account_id = account_id or get_aws_account_id()573 region_name = region_name or get_region()574 result = f"arn:aws:lambda:{region_name}:{account_id}:{type}:{entity_name}"575 if version:576 result = f"{result}:{version}"577 return result578def lambda_function_name(name_or_arn):579 if ":" in name_or_arn:580 arn = parse_arn(name_or_arn)581 if arn["service"] != "lambda":582 raise ValueError("arn is not a lambda arn %s" % name_or_arn)583 return parse_arn(name_or_arn)["resource"].split(":")[1]584 else:585 return name_or_arn586def state_machine_arn(name, account_id=None, region_name=None):587 pattern = "arn:aws:states:%s:%s:stateMachine:%s"588 return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)589def stepfunctions_activity_arn(name, account_id=None, region_name=None):590 pattern = "arn:aws:states:%s:%s:activity:%s"591 return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)592def fix_arn(arn):593 """Function that attempts to "canonicalize" the given ARN. This includes converting594 resource names to ARNs, replacing incorrect regions, account IDs, etc."""595 if arn.startswith("arn:aws:lambda"):596 parts = arn.split(":")597 region = parts[3] if parts[3] in get_valid_regions() else get_region()598 return lambda_function_arn(lambda_function_name(arn), region_name=region)599 LOG.warning("Unable to fix/canonicalize ARN: %s", arn)600 return arn601def cognito_user_pool_arn(user_pool_id, account_id=None, region_name=None):602 pattern = "arn:aws:cognito-idp:%s:%s:userpool/%s"603 return _resource_arn(user_pool_id, pattern, account_id=account_id, region_name=region_name)604def kinesis_stream_arn(stream_name, account_id=None, region_name=None):605 pattern = "arn:aws:kinesis:%s:%s:stream/%s"606 return _resource_arn(stream_name, pattern, account_id=account_id, region_name=region_name)607def elasticsearch_domain_arn(domain_name, account_id=None, region_name=None):608 pattern = "arn:aws:es:%s:%s:domain/%s"609 return _resource_arn(domain_name, pattern, account_id=account_id, region_name=region_name)610def firehose_stream_arn(stream_name, account_id=None, region_name=None):611 pattern = "arn:aws:firehose:%s:%s:deliverystream/%s"612 return _resource_arn(stream_name, pattern, account_id=account_id, region_name=region_name)613def es_domain_arn(domain_name, account_id=None, region_name=None):614 pattern = "arn:aws:es:%s:%s:domain/%s"615 return _resource_arn(domain_name, pattern, account_id=account_id, region_name=region_name)616def kms_key_arn(key_id: str, account_id: str = None, region_name: str = None) -> str:617 pattern = "arn:aws:kms:%s:%s:key/%s"618 return _resource_arn(key_id, pattern, account_id=account_id, region_name=region_name)619def kms_alias_arn(alias_name: str, account_id: str = None, region_name: str = None):620 if not alias_name.startswith("alias/"):621 alias_name = "alias/" + alias_name622 pattern = "arn:aws:kms:%s:%s:%s"623 return _resource_arn(alias_name, pattern, account_id=account_id, region_name=region_name)624def code_signing_arn(code_signing_id: str, account_id: str = None, region_name: str = None) -> str:625 pattern = "arn:aws:lambda:%s:%s:code-signing-config:%s"626 return _resource_arn(code_signing_id, pattern, account_id=account_id, region_name=region_name)627def ssm_parameter_arn(param_name: str, account_id: str = None, region_name: str = None) -> str:628 pattern = "arn:aws:ssm:%s:%s:parameter/%s"629 param_name = param_name.lstrip("/")630 return _resource_arn(param_name, pattern, account_id=account_id, region_name=region_name)631def s3_bucket_arn(bucket_name_or_arn: str, account_id=None):632 bucket_name = s3_bucket_name(bucket_name_or_arn)633 return "arn:aws:s3:::%s" % bucket_name634def s3_bucket_name(bucket_name_or_arn: str) -> str:635 return bucket_name_or_arn.split(":::")[-1]636def _resource_arn(name: str, pattern: str, account_id: str = None, region_name: str = None) -> str:637 if ":" in name:638 return name639 account_id = account_id or get_aws_account_id()640 region_name = region_name or get_region()641 if len(pattern.split("%s")) == 3:642 return pattern % (account_id, name)643 return pattern % (region_name, account_id, name)644def get_events_target_attributes(target):645 return pick_attributes(target, EVENT_TARGET_PARAMETERS)646def get_or_create_bucket(bucket_name: str, s3_client=None):647 s3_client = s3_client or connect_to_service("s3")648 try:649 return s3_client.head_bucket(Bucket=bucket_name)650 except Exception:651 return create_s3_bucket(bucket_name, s3_client=s3_client)652def create_s3_bucket(bucket_name: str, s3_client=None):653 """Creates a bucket in the region that is associated with the current request654 context, or with the given boto3 S3 client, if specified."""655 s3_client = s3_client or connect_to_service("s3")656 region = s3_client.meta.region_name657 kwargs = {}658 if region != AWS_REGION_US_EAST_1:659 kwargs = {"CreateBucketConfiguration": {"LocationConstraint": region}}660 return s3_client.create_bucket(Bucket=bucket_name, **kwargs)661def create_sqs_queue(queue_name, env=None):662 env = get_environment(env)663 # queue664 conn = connect_to_service("sqs", env=env)665 return conn.create_queue(QueueName=queue_name)666def sqs_queue_arn(queue_name, account_id=None, region_name=None):667 account_id = account_id or get_aws_account_id()668 region_name = region_name or get_region()669 queue_name = queue_name.split("/")[-1]670 return "arn:aws:sqs:%s:%s:%s" % (region_name, account_id, queue_name)671def apigateway_restapi_arn(api_id, account_id=None, region_name=None):672 account_id = account_id or get_aws_account_id()673 region_name = region_name or get_region()674 return "arn:aws:apigateway:%s:%s:/restapis/%s" % (region_name, account_id, api_id)675def sqs_queue_name(queue_arn):676 if ":" in queue_arn:677 return parse_arn(queue_arn)["resource"]678 else:679 return queue_arn680def sns_topic_arn(topic_name, account_id=None):681 account_id = account_id or get_aws_account_id()682 return "arn:aws:sns:%s:%s:%s" % (get_region(), account_id, topic_name)683def sqs_receive_message(queue_arn):684 region_name = extract_region_from_arn(queue_arn)685 client = connect_to_service("sqs", region_name=region_name)686 queue_url = get_sqs_queue_url(queue_arn)687 response = client.receive_message(QueueUrl=queue_url)688 return response689def firehose_name(firehose_arn):690 return firehose_arn.split("/")[-1]691def kinesis_stream_name(kinesis_arn):692 return kinesis_arn.split(":stream/")[-1]693def mock_aws_request_headers(694 service="dynamodb", region_name=None, access_key=None, internal=False695) -> Dict[str, str]:696 ctype = APPLICATION_AMZ_JSON_1_0697 if service == "kinesis":698 ctype = APPLICATION_AMZ_JSON_1_1699 elif service in ["sns", "sqs", "sts", "cloudformation"]:700 ctype = APPLICATION_X_WWW_FORM_URLENCODED701 # For S3 presigned URLs, we require that the client and server use the same702 # access key ID to sign requests. So try to use the access key ID for the703 # current request if available704 access_key = access_key or get_ctx_aws_access_key_id() or TEST_AWS_ACCESS_KEY_ID705 region_name = region_name or get_region()706 headers = {707 "Content-Type": ctype,708 "Accept-Encoding": "identity",709 "X-Amz-Date": "20160623T103251Z",710 "Authorization": (711 "AWS4-HMAC-SHA256 "712 + f"Credential={access_key}/20160623/{region_name}/{service}/aws4_request, "713 + "SignedHeaders=content-type;host;x-amz-date;x-amz-target, Signature=1234"714 ),715 }716 if internal:717 headers[HEADER_LOCALSTACK_ACCOUNT_ID] = get_aws_account_id()718 return headers719def inject_region_into_auth_headers(region, headers):720 auth_header = headers.get("Authorization")721 if auth_header:722 regex = r"Credential=([^/]+)/([^/]+)/([^/]+)/"723 auth_header = re.sub(regex, r"Credential=\1/\2/%s/" % region, auth_header)724 headers["Authorization"] = auth_header725def dynamodb_get_item_raw(request):726 headers = mock_aws_request_headers()727 headers["X-Amz-Target"] = "DynamoDB_20120810.GetItem"728 new_item = make_http_request(729 url=config.service_url("dynamodb"),730 method="POST",731 data=json.dumps(request),732 headers=headers,733 )734 new_item = new_item.text735 new_item = new_item and json.loads(new_item)736 return new_item737def create_dynamodb_table(738 table_name: str,739 partition_key: str,740 stream_view_type: str = None,741 region_name: str = None,742 client=None,743 wait_for_active: bool = True,744):745 """Utility method to create a DynamoDB table"""746 dynamodb = client or connect_to_service("dynamodb", region_name=region_name)747 stream_spec = {"StreamEnabled": False}748 key_schema = [{"AttributeName": partition_key, "KeyType": "HASH"}]749 attr_defs = [{"AttributeName": partition_key, "AttributeType": "S"}]750 if stream_view_type is not None:751 stream_spec = {"StreamEnabled": True, "StreamViewType": stream_view_type}752 table = None753 try:754 table = dynamodb.create_table(755 TableName=table_name,756 KeySchema=key_schema,757 AttributeDefinitions=attr_defs,758 BillingMode="PAY_PER_REQUEST",759 StreamSpecification=stream_spec,760 )761 except Exception as e:762 if "ResourceInUseException" in str(e):763 # Table already exists -> return table reference764 return connect_to_resource("dynamodb", region_name=region_name).Table(table_name)765 if "AccessDeniedException" in str(e):766 raise767 def _is_active():768 return dynamodb.describe_table(TableName=table_name)["Table"]["TableStatus"] == "ACTIVE"769 if wait_for_active:770 poll_condition(_is_active)771 return table772def get_apigateway_integration(api_id, method, path, env=None):773 apigateway = connect_to_service(service_name="apigateway", client=True, env=env)774 resources = apigateway.get_resources(restApiId=api_id, limit=100)775 resource_id = None776 for r in resources["items"]:777 if r["path"] == path:778 resource_id = r["id"]779 if not resource_id:780 raise Exception('Unable to find apigateway integration for path "%s"' % path)781 integration = apigateway.get_integration(782 restApiId=api_id, resourceId=resource_id, httpMethod=method783 )784 return integration785def get_apigateway_resource_for_path(api_id, path, parent=None, resources=None):786 if resources is None:787 apigateway = connect_to_service(service_name="apigateway")788 resources = apigateway.get_resources(restApiId=api_id, limit=100)789 if not isinstance(path, list):790 path = path.split("/")791 if not path:792 return parent793 for resource in resources:794 if resource["pathPart"] == path[0] and (not parent or parent["id"] == resource["parentId"]):795 return get_apigateway_resource_for_path(796 api_id, path[1:], parent=resource, resources=resources797 )798 return None799def get_apigateway_path_for_resource(800 api_id, resource_id, path_suffix="", resources=None, region_name=None801):802 if resources is None:803 apigateway = connect_to_service(service_name="apigateway", region_name=region_name)804 resources = apigateway.get_resources(restApiId=api_id, limit=100)["items"]805 target_resource = list(filter(lambda res: res["id"] == resource_id, resources))[0]806 path_part = target_resource.get("pathPart", "")807 if path_suffix:808 if path_part:809 path_suffix = "%s/%s" % (path_part, path_suffix)810 else:811 path_suffix = path_part812 parent_id = target_resource.get("parentId")813 if not parent_id:814 return "/%s" % path_suffix815 return get_apigateway_path_for_resource(816 api_id,817 parent_id,818 path_suffix=path_suffix,819 resources=resources,820 region_name=region_name,821 )822def create_api_gateway(823 name,824 description=None,825 resources=None,826 stage_name=None,827 enabled_api_keys=None,828 env=None,829 usage_plan_name=None,830 region_name=None,831 auth_creator_func=None, # function that receives an api_id and returns an authorizer_id832 client=None,833):834 if enabled_api_keys is None:835 enabled_api_keys = []836 if not client:837 client = connect_to_service("apigateway", env=env, region_name=region_name)838 resources = resources or []839 stage_name = stage_name or "testing"840 usage_plan_name = usage_plan_name or "Basic Usage"841 description = description or 'Test description for API "%s"' % name842 LOG.info('Creating API resources under API Gateway "%s".', name)843 api = client.create_rest_api(name=name, description=description)844 api_id = api["id"]845 auth_id = None846 if auth_creator_func:847 auth_id = auth_creator_func(api_id)848 resources_list = client.get_resources(restApiId=api_id)849 root_res_id = resources_list["items"][0]["id"]850 # add API resources and methods851 for path, methods in resources.items():852 # create resources recursively853 parent_id = root_res_id854 for path_part in path.split("/"):855 api_resource = client.create_resource(856 restApiId=api_id, parentId=parent_id, pathPart=path_part857 )858 parent_id = api_resource["id"]859 # add methods to the API resource860 for method in methods:861 kwargs = {"authorizerId": auth_id} if auth_id else {}862 client.put_method(863 restApiId=api_id,864 resourceId=api_resource["id"],865 httpMethod=method["httpMethod"],866 authorizationType=method.get("authorizationType") or "NONE",867 apiKeyRequired=method.get("apiKeyRequired") or False,868 requestParameters=method.get("requestParameters") or {},869 requestModels=method.get("requestModels") or {},870 **kwargs,871 )872 # create integrations for this API resource/method873 integrations = method["integrations"]874 create_api_gateway_integrations(875 api_id,876 api_resource["id"],877 method,878 integrations,879 env=env,880 region_name=region_name,881 client=client,882 )883 # deploy the API gateway884 client.create_deployment(restApiId=api_id, stageName=stage_name)885 return api886def create_api_gateway_integrations(887 api_id, resource_id, method, integrations=None, env=None, region_name=None, client=None888):889 if integrations is None:890 integrations = []891 if not client:892 client = connect_to_service("apigateway", env=env, region_name=region_name)893 for integration in integrations:894 req_templates = integration.get("requestTemplates") or {}895 res_templates = integration.get("responseTemplates") or {}896 success_code = integration.get("successCode") or "200"897 client_error_code = integration.get("clientErrorCode") or "400"898 server_error_code = integration.get("serverErrorCode") or "500"899 request_parameters = integration.get("requestParameters") or {}900 # create integration901 client.put_integration(902 restApiId=api_id,903 resourceId=resource_id,904 httpMethod=method["httpMethod"],905 integrationHttpMethod=method.get("integrationHttpMethod") or method["httpMethod"],906 type=integration["type"],907 uri=integration["uri"],908 requestTemplates=req_templates,909 requestParameters=request_parameters,910 )911 response_configs = [912 {"pattern": "^2.*", "code": success_code, "res_templates": res_templates},913 {"pattern": "^4.*", "code": client_error_code, "res_templates": {}},914 {"pattern": "^5.*", "code": server_error_code, "res_templates": {}},915 ]916 # create response configs917 for response_config in response_configs:918 # create integration response919 client.put_integration_response(920 restApiId=api_id,921 resourceId=resource_id,922 httpMethod=method["httpMethod"],923 statusCode=response_config["code"],924 responseTemplates=response_config["res_templates"],925 selectionPattern=response_config["pattern"],926 )927 # create method response928 client.put_method_response(929 restApiId=api_id,930 resourceId=resource_id,931 httpMethod=method["httpMethod"],932 statusCode=response_config["code"],933 )934def apigateway_invocations_arn(lambda_uri, region_name: str = None):935 return "arn:aws:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations" % (936 region_name or get_region(),937 lambda_uri,938 )939def get_opensearch_endpoint(domain_arn: str) -> str:940 """941 Get an OpenSearch cluster endpoint by describing the cluster associated with the domain_arn942 :param domain_arn: ARN of the cluster.943 :returns: cluster endpoint944 :raises: ValueError if the domain_arn is malformed945 """946 region_name = extract_region_from_arn(domain_arn)947 if region_name is None:948 raise ValueError("unable to parse region from opensearch domain ARN")949 opensearch_client = connect_to_service(service_name="opensearch", region_name=region_name)950 domain_name = domain_arn.rpartition("/")[2]951 info = opensearch_client.describe_domain(DomainName=domain_name)952 base_domain = info["DomainStatus"]["Endpoint"]953 endpoint = base_domain if base_domain.startswith("http") else f"https://{base_domain}"954 return endpoint955def get_search_db_connection(endpoint: str, region_name: str):956 """957 Get a connection to an ElasticSearch or OpenSearch DB958 :param endpoint: cluster endpoint959 :param region_name: cluster region e.g. us-east-1960 """961 from opensearchpy import OpenSearch, RequestsHttpConnection962 from requests_aws4auth import AWS4Auth963 verify_certs = False964 use_ssl = False965 # use ssl?966 if "https://" in endpoint:967 use_ssl = True968 # TODO remove this condition once ssl certs are available for .es.localhost.localstack.cloud domains969 endpoint_netloc = urlparse(endpoint).netloc970 if not re.match(r"^.*(localhost(\.localstack\.cloud)?)(:\d+)?$", endpoint_netloc):971 verify_certs = True972 LOG.debug("Creating ES client with endpoint %s", endpoint)973 if ENV_ACCESS_KEY in os.environ and ENV_SECRET_KEY in os.environ:974 access_key = os.environ.get(ENV_ACCESS_KEY)975 secret_key = os.environ.get(ENV_SECRET_KEY)976 session_token = os.environ.get(ENV_SESSION_TOKEN)977 awsauth = AWS4Auth(access_key, secret_key, region_name, "es", session_token=session_token)978 connection_class = RequestsHttpConnection979 return OpenSearch(980 hosts=[endpoint],981 verify_certs=verify_certs,982 use_ssl=use_ssl,983 connection_class=connection_class,984 http_auth=awsauth,985 )986 return OpenSearch(hosts=[endpoint], verify_certs=verify_certs, use_ssl=use_ssl)987def create_kinesis_stream(stream_name, shards=1, env=None, delete=False):988 env = get_environment(env)989 stream = KinesisStream(id=stream_name, num_shards=shards)990 conn = connect_to_service("kinesis", env=env)991 stream.connect(conn)992 if delete:993 run_safe(lambda: stream.destroy(), print_error=False)994 stream.create()995 # Note: Returning the stream without awaiting its creation (via wait_for()) to avoid API call timeouts/retries.996 return stream997def kinesis_get_latest_records(stream_name, shard_id, count=10, env=None):998 kinesis = connect_to_service("kinesis", env=env)999 result = []1000 response = kinesis.get_shard_iterator(1001 StreamName=stream_name, ShardId=shard_id, ShardIteratorType="TRIM_HORIZON"1002 )1003 shard_iterator = response["ShardIterator"]1004 while shard_iterator:1005 records_response = kinesis.get_records(ShardIterator=shard_iterator)1006 records = records_response["Records"]1007 for record in records:1008 try:1009 record["Data"] = to_str(record["Data"])1010 except Exception:1011 pass1012 result.extend(records)1013 shard_iterator = records_response["NextShardIterator"] if records else False1014 while len(result) > count:1015 result.pop(0)1016 return result1017def get_stack_details(stack_name, region_name=None):1018 cloudformation = connect_to_service("cloudformation", region_name=region_name)1019 stacks = cloudformation.describe_stacks(StackName=stack_name)1020 for stack in stacks["Stacks"]:1021 if stack["StackName"] == stack_name:1022 return stack1023def await_stack_status(stack_name, expected_statuses, retries=20, sleep=2, region_name=None):1024 def check_stack():1025 stack = get_stack_details(stack_name, region_name=region_name)1026 if stack["StackStatus"] not in expected_statuses:1027 raise Exception(1028 'Status "%s" for stack "%s" not in expected list: %s'1029 % (stack["StackStatus"], stack_name, expected_statuses)1030 )1031 return stack1032 expected_statuses = (1033 expected_statuses if isinstance(expected_statuses, list) else [expected_statuses]1034 )1035 return retry(check_stack, retries, sleep)1036def await_stack_completion(stack_name, retries=20, sleep=2, statuses=None, region_name=None):1037 statuses = statuses or ["CREATE_COMPLETE", "UPDATE_COMPLETE", "DELETE_COMPLETE"]1038 return await_stack_status(1039 stack_name, statuses, retries=retries, sleep=sleep, region_name=region_name1040 )1041def get_ecr_repository_arn(name, account_id=None, region_name=None):1042 pattern = "arn:aws:ecr:%s:%s:repository/%s"1043 return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)1044def get_route53_resolver_firewall_rule_group_arn(1045 id: str, account_id: str = None, region_name: str = None1046):1047 pattern = "arn:aws:route53resolver:%s:%s:firewall-rule-group/%s"1048 return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)1049def get_route53_resolver_firewall_domain_list_arn(1050 id: str, account_id: str = None, region_name: str = None1051):1052 pattern = "arn:aws:route53resolver:%s:%s:firewall-domain-list/%s"1053 return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)1054def get_route53_resolver_firewall_rule_group_associations_arn(1055 id: str, account_id: str = None, region_name: str = None1056):1057 pattern = "arn:aws:route53resolver:%s:%s:firewall-rule-group-association/%s"1058 return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)1059def get_trace_id():...
rds_aurora.py
Source:rds_aurora.py
1import boto32class DbClient:3 _service = 'rds-data'4 _region = 'us-east-1'5 _database = 'voxxdb'6 _aws_access_key_id = 'AKIAVI5BSANHXPPLC3NI'7 _aws_secret_access_key = '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'8 _resource_arn = 'arn:aws:rds:us-east-1:362726687567:cluster:voxx-db-1'9 _secret_arn = 'arn:aws:secretsmanager:us-east-1:362726687567:secret:voxx-user-bYZCyV'10 def __init__(self):11 self._client = boto3.client(12 self._service,13 region_name=self._region,14 aws_access_key_id=self._aws_access_key_id,15 aws_secret_access_key=self._aws_secret_access_key,16 )17 def execute_qry(self, qry, params=[]):18 return self._client.execute_statement(19 database=self._database,20 continueAfterTimeout=False,21 includeResultMetadata=True,22 resourceArn=self._resource_arn,23 secretArn=self._secret_arn,24 parameters=params,25 sql=qry,...
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?
