Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.