KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use aws_provider method in localstack
Best Python code snippet using localstack_python
providers.py
Source:providers.py 
1from localstack import config2from localstack.aws.proxy import AwsApiListener3from localstack.services.moto import MotoFallbackDispatcher4from localstack.services.plugins import Service, aws_provider5@aws_provider()6def acm():7 from localstack.services.acm.provider import AcmProvider8 from localstack.services.moto import MotoFallbackDispatcher9 provider = AcmProvider()10 return Service("acm", listener=AwsApiListener("acm", MotoFallbackDispatcher(provider)))11@aws_provider(api="apigateway", name="legacy")12def apigateway_legacy():13 from localstack.services.apigateway.provider import ApigatewayApiListener, ApigatewayProvider14 provider = ApigatewayProvider()15 listener = ApigatewayApiListener("apigateway", MotoFallbackDispatcher(provider))16 return Service("apigateway", listener=listener, lifecycle_hook=provider)17@aws_provider(api="apigateway", name="default")18def apigateway_asf():19 from localstack.services.apigateway.provider_asf import AsfApigatewayProvider20 provider = AsfApigatewayProvider()21 listener = AwsApiListener("apigateway", MotoFallbackDispatcher(provider))22 return Service("apigateway", listener=listener, lifecycle_hook=provider)23@aws_provider()24def cloudformation():25 from localstack.services.cloudformation.provider import CloudformationProvider26 provider = CloudformationProvider()27 return Service("cloudformation", listener=AwsApiListener("cloudformation", provider))28@aws_provider(api="config")29def awsconfig():30 from localstack.services.configservice.provider import ConfigProvider31 from localstack.services.moto import MotoFallbackDispatcher32 provider = ConfigProvider()33 return Service("config", listener=AwsApiListener("config", MotoFallbackDispatcher(provider)))34@aws_provider()35def cloudwatch():36 from localstack.services.cloudwatch.provider import CloudwatchProvider37 from localstack.services.moto import MotoFallbackDispatcher38 provider = CloudwatchProvider()39 listener = AwsApiListener("cloudwatch", MotoFallbackDispatcher(provider))40 return Service(41 "cloudwatch",42 listener=listener,43 lifecycle_hook=provider,44 )45@aws_provider()46def dynamodb():47 from localstack.services.dynamodb.provider import DynamoDBApiListener48 listener = DynamoDBApiListener()49 return Service(50 "dynamodb",51 listener=listener,52 lifecycle_hook=listener.provider,53 )54@aws_provider()55def dynamodbstreams():56 from localstack.aws.proxy import AwsApiListener57 from localstack.services.dynamodbstreams.provider import DynamoDBStreamsProvider58 provider = DynamoDBStreamsProvider()59 return Service(60 "dynamodbstreams",61 listener=AwsApiListener("dynamodbstreams", provider),62 lifecycle_hook=provider,63 )64@aws_provider()65def ec2():66 from localstack.services.ec2.provider import Ec2Provider67 from localstack.services.moto import MotoFallbackDispatcher68 provider = Ec2Provider()69 return Service(70 "ec2",71 listener=AwsApiListener("ec2", MotoFallbackDispatcher(provider)),72 )73@aws_provider()74def es():75 from localstack.aws.proxy import AwsApiListener76 from localstack.services.es.provider import EsProvider77 provider = EsProvider()78 return Service("es", listener=AwsApiListener("es", provider))79@aws_provider()80def firehose():81 from localstack.aws.proxy import AwsApiListener82 from localstack.services.firehose.provider import FirehoseProvider83 provider = FirehoseProvider()84 return Service("firehose", listener=AwsApiListener("firehose", provider))85@aws_provider()86def iam():87 from localstack.services.iam.provider import IamProvider88 from localstack.services.moto import MotoFallbackDispatcher89 provider = IamProvider()90 return Service(91 "iam",92 listener=AwsApiListener("iam", MotoFallbackDispatcher(provider)),93 )94@aws_provider()95def sts():96 from localstack.services.sts.provider import StsAwsApiListener97 listener = StsAwsApiListener()98 return Service("sts", listener=listener)99@aws_provider()100def kinesis():101 from localstack.services.kinesis import kinesis_listener, kinesis_starter102 return Service(103 "kinesis",104 listener=kinesis_listener.UPDATE_KINESIS,105 start=kinesis_starter.start_kinesis,106 check=kinesis_starter.check_kinesis,107 )108@aws_provider()109def kms():110 if config.KMS_PROVIDER == "local-kms":111 from localstack.services.kms import kms_starter112 return Service("kms", start=kms_starter.start_kms_local)113 # fall back to default provider114 from localstack.services.kms.provider import KmsProvider115 provider = KmsProvider()116 return Service("kms", listener=AwsApiListener("kms", MotoFallbackDispatcher(provider)))117@aws_provider(api="lambda")118def awslambda():119 from localstack.services.awslambda import lambda_starter120 return Service(121 "lambda",122 start=lambda_starter.start_lambda,123 stop=lambda_starter.stop_lambda,124 check=lambda_starter.check_lambda,125 )126@aws_provider(api="lambda", name="asf")127def awslambda_asf():128 from localstack.aws.proxy import AwsApiListener129 from localstack.services.awslambda.provider import LambdaProvider130 provider = LambdaProvider()131 return Service("lambda", listener=AwsApiListener("lambda", provider), lifecycle_hook=provider)132@aws_provider()133def logs():134 from localstack.services.logs.provider import LogsAwsApiListener135 listener = LogsAwsApiListener()136 return Service("logs", listener=listener)137@aws_provider()138def opensearch():139 from localstack.aws.proxy import AwsApiListener140 from localstack.services.opensearch.provider import OpensearchProvider141 provider = OpensearchProvider()142 return Service("opensearch", listener=AwsApiListener("opensearch", provider))143@aws_provider()144def redshift():145 from localstack.services.redshift.provider import RedshiftProvider146 provider = RedshiftProvider()147 listener = AwsApiListener("redshift", MotoFallbackDispatcher(provider))148 return Service("redshift", listener=listener)149@aws_provider()150def route53():151 from localstack.services.route53.provider import Route53Provider152 provider = Route53Provider()153 return Service("route53", listener=AwsApiListener("route53", MotoFallbackDispatcher(provider)))154@aws_provider()155def route53resolver():156 from localstack.services.route53resolver.provider import Route53ResolverProvider157 provider = Route53ResolverProvider()158 return Service(159 "route53resolver",160 listener=AwsApiListener("route53resolver", MotoFallbackDispatcher(provider)),161 )162@aws_provider()163def s3():164 from localstack.services.s3 import s3_listener, s3_starter165 return Service(166 "s3", listener=s3_listener.UPDATE_S3, start=s3_starter.start_s3, check=s3_starter.check_s3167 )168@aws_provider()169def s3control():170 from localstack.services.moto import MotoFallbackDispatcher171 from localstack.services.s3control.provider import S3ControlProvider172 provider = S3ControlProvider()173 return Service(174 "s3control", listener=AwsApiListener("s3control", MotoFallbackDispatcher(provider))175 )176@aws_provider()177def secretsmanager():178 from localstack.services.moto import MotoFallbackDispatcher179 from localstack.services.secretsmanager.provider import SecretsmanagerProvider180 provider = SecretsmanagerProvider()181 return Service(182 "secretsmanager",183 listener=AwsApiListener("secretsmanager", MotoFallbackDispatcher(provider)),184 )185@aws_provider()186def ses():187 from localstack.services.moto import MotoFallbackDispatcher188 from localstack.services.ses.provider import SesProvider189 provider = SesProvider()190 return Service(191 "ses",192 listener=AwsApiListener("ses", MotoFallbackDispatcher(provider)),193 lifecycle_hook=provider,194 )195@aws_provider()196def sns():197 from localstack.aws.proxy import AwsApiListener198 from localstack.services.sns.provider import SnsProvider199 provider = SnsProvider()200 return Service("sns", listener=AwsApiListener("sns", provider), lifecycle_hook=provider)201@aws_provider()202def sqs():203 from localstack.aws.proxy import AwsApiListener204 from localstack.services import edge205 from localstack.services.sqs import query_api206 from localstack.services.sqs.provider import SqsProvider207 query_api.register(edge.ROUTER)208 provider = SqsProvider()209 return Service("sqs", listener=AwsApiListener("sqs", provider), lifecycle_hook=provider)210@aws_provider(api="sqs", name="legacy")211def sqs_legacy():212 from localstack.services.sqs.legacy import sqs_listener, sqs_starter213 return Service(214 "sqs",215 listener=sqs_listener.UPDATE_SQS,216 start=sqs_starter.start_sqs,217 check=sqs_starter.check_sqs,218 )219@aws_provider()220def ssm():221 from localstack.services.moto import MotoFallbackDispatcher222 from localstack.services.ssm.provider import SsmProvider223 provider = SsmProvider()224 return Service(225 "ssm",226 listener=AwsApiListener("ssm", MotoFallbackDispatcher(provider)),227 )228@aws_provider()229def events():230 from localstack.services.events.provider import EventsProvider231 from localstack.services.moto import MotoFallbackDispatcher232 provider = EventsProvider()233 return Service(234 "events",235 listener=AwsApiListener("events", MotoFallbackDispatcher(provider)),236 )237@aws_provider()238def stepfunctions():239 from localstack.services.stepfunctions.provider import StepFunctionsApiListener240 listener = StepFunctionsApiListener()241 return Service(242 "stepfunctions",243 listener=listener,244 lifecycle_hook=listener.provider,245 )246@aws_provider()247def swf():248 from localstack.services.moto import MotoFallbackDispatcher249 from localstack.services.swf.provider import SWFProvider250 provider = SWFProvider()251 return Service(252 "swf",253 listener=AwsApiListener("swf", MotoFallbackDispatcher(provider)),254 )255@aws_provider()256def resourcegroupstaggingapi():257 from localstack.services.moto import MotoFallbackDispatcher258 from localstack.services.resourcegroupstaggingapi.provider import (259 ResourcegroupstaggingapiProvider,260 )261 provider = ResourcegroupstaggingapiProvider()262 return Service(263 "resourcegroupstaggingapi",264 listener=AwsApiListener("resourcegroupstaggingapi", MotoFallbackDispatcher(provider)),265 )266@aws_provider(api="resource-groups")267def resource_groups():268 from localstack.services.moto import MotoFallbackDispatcher269 from localstack.services.resourcegroups.provider import ResourceGroupsProvider270 provider = ResourceGroupsProvider()271 return Service(272 "resource-groups",273 listener=AwsApiListener("resource-groups", MotoFallbackDispatcher(provider)),274 )275@aws_provider()276def support():277 from localstack.services.moto import MotoFallbackDispatcher278 from localstack.services.support.provider import SupportProvider279 provider = SupportProvider()280 return Service(281 "support",282 listener=AwsApiListener("support", MotoFallbackDispatcher(provider)),...
test_aws_provider.py
Source:test_aws_provider.py
1import unittest2from unittest import mock3import pytest4from ViniAudit.providers import get_provider5from ViniAudit.providers.aws.authentication_strategy import AWSCredentials6from ViniAudit.providers.base.authentication_strategy import AuthenticationException7from ViniAudit.providers.base.authentication_strategy_factory import get_authentication_strategy8class Object(object):9 pass10# Test methods for AWS Provider11class TestAWSProviderClass(unittest.TestCase):12 @mock.patch("ViniAudit.providers.aws.authentication_strategy.boto3")13 @mock.patch("ViniAudit.providers.aws.authentication_strategy.get_caller_identity")14 def test_authenticate(self, mock_get_caller_identity, mock_boto3):15 aws_authentication_strategy = get_authentication_strategy("aws")16 boto3_session = Object()17 boto3_session._session = Object()18 mock_boto3.Session.return_value = boto3_session19 test_cases = [20 # no params21 {22 "profile": None,23 "aws_access_key_id": None,24 "aws_secret_access_key": None,25 "aws_session_token": None,26 "call_dict": {},27 },28 # profile29 {30 "profile": "123",31 "aws_access_key_id": None,32 "aws_secret_access_key": None,33 "aws_session_token": None,34 "call_dict": {"profile_name": "123"},35 },36 # access and secret key37 {38 "profile": None,39 "aws_access_key_id": "456",40 "aws_secret_access_key": "789",41 "aws_session_token": None,42 "call_dict": {43 "aws_access_key_id": "456",44 "aws_secret_access_key": "789",45 },46 },47 # access, secret key and token48 {49 "profile": None,50 "aws_access_key_id": "456",51 "aws_secret_access_key": "789",52 "aws_session_token": "101112",53 "call_dict": {54 "aws_access_key_id": "456",55 "aws_secret_access_key": "789",56 "aws_session_token": "101112",57 },58 },59 ]60 for test_case in test_cases:61 result = aws_authentication_strategy.authenticate(62 test_case["profile"],63 test_case["aws_access_key_id"],64 test_case["aws_secret_access_key"],65 test_case["aws_session_token"],66 )67 mock_boto3.Session.assert_called_with(**test_case["call_dict"])68 mock_get_caller_identity.assert_called_with(boto3_session)69 assert isinstance(result, AWSCredentials)70 assert result.session == boto3_session71 # exception test72 mock_boto3.Session.side_effect = Exception("an exception")73 with pytest.raises(AuthenticationException):74 result = aws_authentication_strategy.authenticate(None, None, None, None)75 # mock two separate places from which get_aws_account_id is called76 @mock.patch("ViniAudit.providers.aws.facade.base.get_aws_account_id")77 @mock.patch("ViniAudit.providers.aws.provider.get_aws_account_id")78 @mock.patch("ViniAudit.providers.aws.provider.get_partition_name")79 def test_get_report_name(80 self,81 mock_get_partiton_name,82 mock_get_aws_account_id,83 mock_facade_aws_account_id,84 ):85 # no account_id, no profile86 mock_get_aws_account_id.return_value = None87 mock_get_partiton_name.return_value = None88 aws_provider = get_provider(89 provider="aws", credentials=mock.MagicMock(session="123"),90 )91 assert aws_provider.get_report_name() == "aws"92 # profile and account_id93 mock_get_aws_account_id.return_value = "12345"94 aws_provider = get_provider(95 provider="aws", profile="9999", credentials=mock.MagicMock(session="123"),96 )97 assert aws_provider.get_report_name() == "aws-9999"98 # account_id99 aws_provider = get_provider(100 provider="aws", credentials=mock.MagicMock(session="123"),101 )...
aws.py
Source:aws.py
1from sqlalchemy.orm import Session2from sqlalchemy import func3import datetime4from security.vault import vault_encrypt, vault_decrypt5import db.models as models6import schemas.schemas as schemas7@vault_encrypt8def encrypt(secreto):9 try:10 return secreto11 except Exception as err:12 raise err13@vault_decrypt14def decrypt(secreto):15 try:16 return secreto17 except Exception as err:18 raise err19def create_aws_profile(db: Session, aws: schemas.AwsAsumeProfile):20 encrypt_access_key_id = encrypt(aws.access_key_id)21 encrypt_secret_access_key = encrypt(aws.secret_access_key)22 db_aws = models.Aws_provider(access_key_id=encrypt_access_key_id,23 secret_access_key=encrypt_secret_access_key,24 environment=aws.environment,25 default_region=aws.default_region,26 profile_name=aws.profile_name,27 role_arn=aws.role_arn,28 source_profile=aws.source_profile,29 created_at=datetime.datetime.now(),30 squad=aws.squad)31 check_None = [None, "string"]32 if db_aws.role_arn in check_None:33 db_aws.role_arn = ""34 if db_aws.profile_name in check_None:35 db_aws.profile_name = ""36 if db_aws.source_profile in check_None:37 db_aws.source_profile = ""38 try:39 db.add(db_aws)40 db.commit()41 db.refresh(db_aws)42 return db_aws43 except Exception as err:44 raise err45def get_credentials_aws_profile(db: Session, environment: str, squad: str):46 get_access_key = db.query(models.Aws_provider.access_key_id).filter(47 models.Aws_provider.environment == environment).filter(models.Aws_provider.squad == squad).first()48 get_secret_access_key = db.query(models.Aws_provider.secret_access_key).filter(49 models.Aws_provider.environment == environment).filter(models.Aws_provider.squad == squad).first()50 default_region = db.query(models.Aws_provider.default_region).filter(51 models.Aws_provider.environment == environment).filter(models.Aws_provider.squad == squad).first()52 profile_name = db.query(models.Aws_provider.profile_name).filter(53 models.Aws_provider.environment == environment).filter(models.Aws_provider.squad == squad).first()54 role_arn = db.query(models.Aws_provider.role_arn).filter(55 models.Aws_provider.environment == environment).filter(models.Aws_provider.squad == squad).first()56 source_profile = db.query(models.Aws_provider.source_profile).filter(57 models.Aws_provider.environment == environment).filter(models.Aws_provider.squad == squad).first()58 try:59 return {60 "access_key": decrypt(get_access_key[0]),61 "secret_access_key": decrypt(get_secret_access_key[0]),62 "default_region": default_region[0],63 "profile_name": profile_name[0],64 "role_arn": role_arn[0],65 "source_profile": source_profile[0]66 }67 except Exception as err:68 raise err69def get_squad_aws_profile(db: Session, squad: str, environment: str):70 try:71 if environment != None:72 return db.query(models.Aws_provider).filter(models.Aws_provider.squad == squad).filter(73 models.Aws_provider.environment == environment).first()74 result = []75 for i in squad:76 result.extend(db.query(models.Aws_provider).filter(models.Aws_provider.squad == i).all())77 return set(result)78 except Exception as err:79 raise err80def get_all_aws_profile(db: Session):81 try:82 return db.query(models.Aws_provider).all()83 except Exception as err:84 raise err85def delete_aws_profile_by_id(db: Session, aws_profile_id: int):86 db.query(models.Aws_provider).filter(87 models.Aws_provider.id == aws_profile_id).delete()88 try:89 db.commit()90 return {aws_profile_id: "deleted", "aws_profile_id": aws_profile_id}91 except Exception as err:...
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?
