KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_moto_default_account_id method in localstack

Best Python code snippet using localstack_python

test_moto.py

Source:test_moto.py

1from io import BytesIO2from typing import Optional3import pytest4from moto.core import DEFAULT_ACCOUNT_ID5import localstack.aws.accounts6from localstack import config7from localstack.aws.api import ServiceException, handler8from localstack.services import moto9from localstack.services.moto import MotoFallbackDispatcher10from localstack.utils.common import short_uid11def test_call_with_sqs_creates_state_correctly():12    qname = f"queue-{short_uid()}"13    response = moto.call_moto(14        moto.create_aws_request_context("sqs", "CreateQueue", {"QueueName": qname}),15        include_response_metadata=True,16    )17    url = response["QueueUrl"]18    try:19        assert response["ResponseMetadata"]["HTTPStatusCode"] == 20020        assert response["QueueUrl"].endswith(f"/{qname}")21        response = moto.call_moto(moto.create_aws_request_context("sqs", "ListQueues"))22        assert url in response["QueueUrls"]23    finally:24        moto.call_moto(moto.create_aws_request_context("sqs", "DeleteQueue", {"QueueUrl": url}))25    response = moto.call_moto(moto.create_aws_request_context("sqs", "ListQueues"))26    assert url not in response.get("QueueUrls", [])27def test_call_sqs_invalid_call_raises_http_exception():28    with pytest.raises(ServiceException) as e:29        moto.call_moto(30            moto.create_aws_request_context(31                "sqs",32                "DeleteQueue",33                {34                    "QueueUrl": "http://0.0.0.0/nonexistingqueue",35                },36            )37        )38    e.match("The specified queue does not exist")39def test_call_non_implemented_operation():40    with pytest.raises(NotImplementedError):41        # we'll need to keep finding methods that moto doesn't implement ;-)42        moto.call_moto(43            moto.create_aws_request_context("athena", "DeleteDataCatalog", {"Name": "foo"})44        )45def test_call_with_sqs_modifies_state_in_moto_backend():46    """Whitebox test to check that moto backends are populated correctly"""47    from moto.sqs.models import sqs_backends48    qname = f"queue-{short_uid()}"49    response = moto.call_moto(50        moto.create_aws_request_context("sqs", "CreateQueue", {"QueueName": qname})51    )52    url = response["QueueUrl"]53    assert qname in sqs_backends[DEFAULT_ACCOUNT_ID][config.AWS_REGION_US_EAST_1].queues54    moto.call_moto(moto.create_aws_request_context("sqs", "DeleteQueue", {"QueueUrl": url}))55    assert qname not in sqs_backends[DEFAULT_ACCOUNT_ID][config.AWS_REGION_US_EAST_1].queues56@pytest.mark.parametrize(57    "payload", ["foobar", b"foobar", BytesIO(b"foobar")], ids=["str", "bytes", "IO[bytes]"]58)59def test_call_s3_with_streaming_trait(payload, monkeypatch):60    monkeypatch.setenv("MOTO_S3_CUSTOM_ENDPOINTS", "s3.localhost.localstack.cloud:4566")61    # In this test we use low-level interface with Moto and skip the standard setup62    # In the absence of below patch, Moto and LocalStack uses difference AWS Account IDs causing the test to fail63    monkeypatch.setattr(64        localstack.aws.accounts,65        "account_id_resolver",66        localstack.aws.accounts.get_moto_default_account_id,67    )68    bucket_name = f"bucket-{short_uid()}"69    key_name = f"key-{short_uid()}"70    # create the bucket71    moto.call_moto(moto.create_aws_request_context("s3", "CreateBucket", {"Bucket": bucket_name}))72    moto.call_moto(73        moto.create_aws_request_context(74            "s3", "PutObject", {"Bucket": bucket_name, "Key": key_name, "Body": payload}75        )76    )77    # check whether it was created/received correctly78    response = moto.call_moto(79        moto.create_aws_request_context("s3", "GetObject", {"Bucket": bucket_name, "Key": key_name})80    )81    assert hasattr(82        response["Body"], "read"83    ), f"expected Body to be readable, was {type(response['Body'])}"84    assert response["Body"].read() == b"foobar"85    # cleanup86    moto.call_moto(87        moto.create_aws_request_context(88            "s3", "DeleteObject", {"Bucket": bucket_name, "Key": key_name}89        )90    )91    moto.call_moto(moto.create_aws_request_context("s3", "DeleteBucket", {"Bucket": bucket_name}))92def test_call_include_response_metadata():93    ctx = moto.create_aws_request_context("sqs", "ListQueues")94    response = moto.call_moto(ctx)95    assert "ResponseMetadata" not in response96    response = moto.call_moto(ctx, include_response_metadata=True)97    assert "ResponseMetadata" in response98def test_call_with_modified_request():99    from moto.sqs.models import sqs_backends100    qname1 = f"queue-{short_uid()}"101    qname2 = f"queue-{short_uid()}"102    context = moto.create_aws_request_context("sqs", "CreateQueue", {"QueueName": qname1})103    response = moto.call_moto_with_request(context, {"QueueName": qname2})  # overwrite old request104    url = response["QueueUrl"]105    assert qname2 in sqs_backends[DEFAULT_ACCOUNT_ID][config.AWS_REGION_US_EAST_1].queues106    assert qname1 not in sqs_backends[DEFAULT_ACCOUNT_ID][config.AWS_REGION_US_EAST_1].queues107    moto.call_moto(moto.create_aws_request_context("sqs", "DeleteQueue", {"QueueUrl": url}))108def test_call_with_es_creates_state_correctly():109    domain_name = f"domain-{short_uid()}"110    response = moto.call_moto(111        moto.create_aws_request_context(112            "es",113            "CreateElasticsearchDomain",114            {115                "DomainName": domain_name,116                "ElasticsearchVersion": "7.10",117            },118        ),119        include_response_metadata=True,120    )121    try:122        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200123        assert response["DomainStatus"]["DomainName"] == domain_name124        assert response["DomainStatus"]["ElasticsearchVersion"] == "7.10"125    finally:126        response = moto.call_moto(127            moto.create_aws_request_context(128                "es", "DeleteElasticsearchDomain", {"DomainName": domain_name}129            ),130            include_response_metadata=True,131        )132        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200133def test_call_multi_region_backends():134    from moto.sqs.models import sqs_backends135    qname_us = f"queue-us-{short_uid()}"136    qname_eu = f"queue-eu-{short_uid()}"137    moto.call_moto(138        moto.create_aws_request_context(139            "sqs", "CreateQueue", {"QueueName": qname_us}, region="us-east-1"140        )141    )142    moto.call_moto(143        moto.create_aws_request_context(144            "sqs", "CreateQueue", {"QueueName": qname_eu}, region="eu-central-1"145        )146    )147    assert qname_us in sqs_backends[DEFAULT_ACCOUNT_ID]["us-east-1"].queues148    assert qname_eu not in sqs_backends[DEFAULT_ACCOUNT_ID]["us-east-1"].queues149    assert qname_us not in sqs_backends[DEFAULT_ACCOUNT_ID]["eu-central-1"].queues150    assert qname_eu in sqs_backends[DEFAULT_ACCOUNT_ID]["eu-central-1"].queues151    del sqs_backends[DEFAULT_ACCOUNT_ID]["us-east-1"].queues[qname_us]152    del sqs_backends[DEFAULT_ACCOUNT_ID]["eu-central-1"].queues[qname_eu]153def test_call_with_sqs_invalid_call_raises_exception():154    with pytest.raises(ServiceException):155        moto.call_moto(156            moto.create_aws_request_context(157                "sqs",158                "DeleteQueue",159                {160                    "QueueUrl": "http://0.0.0.0/nonexistingqueue",161                },162            )163        )164def test_call_with_sqs_returns_service_response():165    qname = f"queue-{short_uid()}"166    create_queue_response = moto.call_moto(167        moto.create_aws_request_context("sqs", "CreateQueue", {"QueueName": qname})168    )169    assert "QueueUrl" in create_queue_response170    assert create_queue_response["QueueUrl"].endswith(qname)171class FakeSqsApi:172    @handler("ListQueues", expand=False)173    def list_queues(self, context, request):174        raise NotImplementedError175    @handler("CreateQueue", expand=False)176    def create_queue(self, context, request):177        raise NotImplementedError178class FakeSqsProvider(FakeSqsApi):179    def __init__(self) -> None:180        super().__init__()181        self.calls = []182    @handler("ListQueues", expand=False)183    def list_queues(self, context, request):184        self.calls.append(context)185        return moto.call_moto(context)186def test_moto_fallback_dispatcher():187    provider = FakeSqsProvider()188    dispatcher = MotoFallbackDispatcher(provider)189    assert "ListQueues" in dispatcher190    assert "CreateQueue" in dispatcher191    def _dispatch(action, params):192        context = moto.create_aws_request_context("sqs", action, params)193        return dispatcher[action](context, params)194    qname = f"queue-{short_uid()}"195    # when falling through the dispatcher returns the appropriate ServiceResponse (in this case a CreateQueueResult)196    create_queue_response = _dispatch("CreateQueue", {"QueueName": qname})197    assert "QueueUrl" in create_queue_response198    # this returns a ListQueuesResult199    list_queues_response = _dispatch("ListQueues", None)200    assert len(provider.calls) == 1201    assert len([url for url in list_queues_response["QueueUrls"] if qname in url])202class FakeS3Provider:203    class FakeNoSuchBucket(ServiceException):204        code: str = "NoSuchBucket"205        sender_fault: bool = False206        status_code: int = 404207        BucketName: Optional[str]208    def __init__(self) -> None:209        super().__init__()210        self.calls = []211    @handler("GetObject", expand=False)212    def get_object(self, _, __):213        # Test fall-through raises exception214        raise NotImplementedError215    @handler("ListObjects", expand=False)216    def list_objects(self, _, request):217        # Test provider implementation raises exception218        ex = self.FakeNoSuchBucket()219        ex.BucketName = request["Bucket"]220        raise ex221    @handler("ListObjectsV2", expand=False)222    def list_objects_v2(self, context, _):223        # Test call_moto raises exception224        return moto.call_moto(context)225def test_moto_fallback_dispatcher_error_handling(monkeypatch):226    """227    This test checks if the error handling (marshalling / unmarshalling) works correctly on all levels, including228    additional (even non-officially supported) fields on exception (like NoSuchBucket#BucketName).229    """230    monkeypatch.setenv("MOTO_S3_CUSTOM_ENDPOINTS", "s3.localhost.localstack.cloud:4566")231    provider = FakeS3Provider()232    dispatcher = MotoFallbackDispatcher(provider)233    def _dispatch(action, params):234        context = moto.create_aws_request_context("s3", action, params)235        return dispatcher[action](context, params)236    bucket_name = f"bucket-{short_uid()}"237    # Test fallback implementation raises a service exception which has the additional attribute "BucketName"238    with pytest.raises(ServiceException) as e:239        _dispatch("GetObject", {"Bucket": bucket_name, "Key": "key"})240    assert getattr(e.value, "BucketName") == bucket_name241    # Test provider implementation raises a service exception242    with pytest.raises(ServiceException) as e:243        _dispatch("ListObjects", {"Bucket": bucket_name})244    assert getattr(e.value, "BucketName") == bucket_name245    # Test provider uses call_moto, which raises a service exception246    with pytest.raises(ServiceException) as e:247        _dispatch("ListObjectsV2", {"Bucket": bucket_name})248    assert getattr(e.value, "BucketName") == bucket_name249def test_request_with_response_header_location_fields():250    # CreateHostedZoneResponse has a member "Location" that's located in the headers251    zone_name = f"zone-{short_uid()}.com"252    request = moto.create_aws_request_context(253        "route53", "CreateHostedZone", {"Name": zone_name, "CallerReference": "test"}254    )255    response = moto.call_moto(request, include_response_metadata=True)256    # assert response["Location"]  # FIXME: this is required according to the spec, but not returned by moto257    assert response["HostedZone"]["Id"]258    # clean up259    moto.call_moto(260        moto.create_aws_request_context(261            "route53", "DeleteHostedZone", {"Id": response["HostedZone"]["Id"]}262        )...

accounts.py

Source:accounts.py

...21    """Return the AWS account ID for the current context."""22    return account_id_resolver()23def get_default_account_id() -> str:24    return DEFAULT_AWS_ACCOUNT_ID25def get_moto_default_account_id() -> str:26    return moto.core.DEFAULT_ACCOUNT_ID27account_id_resolver = get_default_account_id28#29# Utils30#31def get_account_id_from_access_key_id(access_key_id: str) -> str:32    """Return the Account ID associated the Access Key ID."""33    # This utility ignores IAM mappings.34    # For now, we assume the client sends Account ID in Access Key ID field.35    if re.match(r"\d{12}", access_key_id):36        return access_key_id37    else:...

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.