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

How to use get_caller_identity method in localstack

Best Python code snippet using localstack_python

test_iam.py

Source:test_iam.py

1#2# (c) 2020 Red Hat Inc.3#4# This file is part of Ansible5# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)6from __future__ import (absolute_import, division, print_function)7__metaclass__ = type8import pytest9import botocore10from ansible_collections.amazon.aws.tests.unit.compat.mock import MagicMock11from ansible_collections.amazon.aws.tests.unit.compat import unittest12import ansible_collections.amazon.aws.plugins.module_utils.iam as utils_iam13from ansible_collections.amazon.aws.plugins.module_utils.ec2 import HAS_BOTO314if not HAS_BOTO3:15    pytestmark = pytest.mark.skip("test_iam.py requires the python modules 'boto3' and 'botocore'")16class IamUtilsTestSuite(unittest.TestCase):17    def _make_denied_exception(self, partition):18        return botocore.exceptions.ClientError(19            {20                "Error": {21                    "Code": "AccessDenied",22                    "Message": "User: arn:" + partition + ":iam::123456789012:user/ExampleUser "23                               + "is not authorized to perform: iam:GetUser on resource: user ExampleUser"24                },25                "ResponseMetadata": {26                    "RequestId": "01234567-89ab-cdef-0123-456789abcdef"27                }28            }, 'getUser')29    def _make_unexpected_exception(self):30        return botocore.exceptions.ClientError(31            {32                "Error": {33                    "Code": "SomeThingWentWrong",34                    "Message": "Boom!"35                },36                "ResponseMetadata": {37                    "RequestId": "01234567-89ab-cdef-0123-456789abcdef"38                }39            }, 'someCall')40    def _make_encoded_exception(self):41        return botocore.exceptions.ClientError(42            {43                "Error": {44                    "Code": "AccessDenied",45                    "Message": "You are not authorized to perform this operation. Encoded authorization failure message: " +46                               "fEwXX6llx3cClm9J4pURgz1XPnJPrYexEbrJcLhFkwygMdOgx_-aEsj0LqRM6Kxt2HVI6prUhDwbJqBo9U2V7iRKZ" +47                               "T6ZdJvHH02cXmD0Jwl5vrTsf0PhBcWYlH5wl2qME7xTfdolEUr4CzumCiti7ETiO-RDdHqWlasBOW5bWsZ4GSpPdU" +48                               "06YAX0TfwVBs48uU5RpCHfz1uhSzez-3elbtp9CmTOHLt5pzJodiovccO55BQKYLPtmJcs6S9YLEEogmpI4Cb1D26" +49                               "fYahDh51jEmaohPnW5pb1nQe2yPEtuIhtRzNjhFCOOMwY5DBzNsymK-Gj6eJLm7FSGHee4AHLU_XmZMe_6bcLAiOx" +50                               "6Zdl65Kdd0hLcpwVxyZMi27HnYjAdqRlV3wuCW2PkhAW14qZQLfiuHZDEwnPe2PBGSlFcCmkQvJvX-YLoA7Uyc2wf" +51                               "NX5RJm38STwfiJSkQaNDhHKTWKiLOsgY4Gze6uZoG7zOcFXFRyaA4cbMmI76uyBO7j-9uQUCtBYqYto8x_9CUJcxI" +52                               "VC5SPG_C1mk-WoDMew01f0qy-bNaCgmJ9TOQGd08FyuT1SaMpCC0gX6mHuOnEgkFw3veBIowMpp9XcM-yc42fmIOp" +53                               "FOdvQO6uE9p55Qc-uXvsDTTvT3A7EeFU8a_YoAIt9UgNYM6VTvoprLz7dBI_P6C-bdPPZCY2amm-dJNVZelT6TbJB" +54                               "H_Vxh0fzeiSUBersy_QzB0moc-vPWgnB-IkgnYLV-4L3K0L2"55                },56                "ResponseMetadata": {57                    "RequestId": "01234567-89ab-cdef-0123-456789abcdef"58                }59            }, 'someCall')60    def _make_botocore_exception(self):61        return botocore.exceptions.EndpointConnectionError(endpoint_url='junk.endpoint')62    def setUp(self):63        self.sts_client = MagicMock()64        self.iam_client = MagicMock()65        self.module = MagicMock()66        clients = {'sts': self.sts_client, 'iam': self.iam_client}67        def get_client(*args, **kwargs):68            return clients[args[0]]69        self.module.client.side_effect = get_client70        self.module.fail_json_aws.side_effect = SystemExit(1)71        self.module.fail_json.side_effect = SystemExit(2)72    # ========== get_aws_account_id ============73    # This is just a minimal (compatability) wrapper around get_aws_account_info74    # Perform some basic testing and call it a day.75    # Test the simplest case - We're permitted to call GetCallerIdentity76    def test_get_aws_account_id__caller_success(self):77        # Prepare78        self.sts_client.get_caller_identity.side_effect = [{'UserId': 'AIDA1234567890ABCDEFG',79                                                            'Account': '123456789012',80                                                            'Arn': 'arn:aws:iam::123456789012:user/ExampleUser'}]81        # Run module82        return_value = utils_iam.get_aws_account_id(self.module)83        # Check we only saw the calls we mocked out84        self.module.client.assert_called_once()85        self.sts_client.get_caller_identity.assert_called_once()86        # Check we got the values back we expected.87        self.assertEqual(return_value, '123456789012')88    # Test the simplest case - We're permitted to call GetCallerIdentity89    # (China partition)90    def test_get_aws_account_id__caller_success_cn(self):91        # Prepare92        self.sts_client.get_caller_identity.side_effect = [{'UserId': 'AIDA1234567890ABCDEFG',93                                                            'Account': '123456789012',94                                                            'Arn': 'arn:aws-cn:iam::123456789012:user/ExampleUser'}]95        # Run module96        return_value = utils_iam.get_aws_account_id(self.module)97        # Check we only saw the calls we mocked out98        self.module.client.assert_called_once()99        self.sts_client.get_caller_identity.assert_called_once()100        # Check we got the values back we expected.101        self.assertEqual(return_value, '123456789012')102    # ========== get_aws_account_info ============103    # Test the simplest case - We're permitted to call GetCallerIdentity104    def test_get_aws_account_info__caller_success(self):105        # Prepare106        self.sts_client.get_caller_identity.side_effect = [{'UserId': 'AIDA1234567890ABCDEFG',107                                                            'Account': '123456789012',108                                                            'Arn': 'arn:aws:iam::123456789012:user/ExampleUser'}]109        # Run module110        return_value = utils_iam.get_aws_account_info(self.module)111        # Check we only saw the calls we mocked out112        self.module.client.assert_called_once()113        self.sts_client.get_caller_identity.assert_called_once()114        # Check we got the values back we expected.115        self.assertEqual(return_value, ('123456789012', 'aws',))116    # (China partition)117    def test_get_aws_account_info__caller_success_cn(self):118        # Prepare119        self.sts_client.get_caller_identity.side_effect = [{'UserId': 'AIDA1234567890ABCDEFG',120                                                            'Account': '123456789012',121                                                            'Arn': 'arn:aws-cn:iam::123456789012:user/ExampleUser'}]122        # Run module123        return_value = utils_iam.get_aws_account_info(self.module)124        # Check we only saw the calls we mocked out125        self.module.client.assert_called_once()126        self.sts_client.get_caller_identity.assert_called_once()127        # Check we got the values back we expected.128        self.assertEqual(return_value, ('123456789012', 'aws-cn',))129    # (US-Gov partition)130    def test_get_aws_account_info__caller_success_gov(self):131        # Prepare132        self.sts_client.get_caller_identity.side_effect = [{'UserId': 'AIDA1234567890ABCDEFG',133                                                            'Account': '123456789012',134                                                            'Arn': 'arn:aws-us-gov:iam::123456789012:user/ExampleUser'}]135        # Run module136        return_value = utils_iam.get_aws_account_info(self.module)137        # Check we only saw the calls we mocked out138        self.module.client.assert_called_once()139        self.sts_client.get_caller_identity.assert_called_once()140        # Check we got the values back we expected.141        self.assertEqual(return_value, ('123456789012', 'aws-us-gov',))142    # If sts:get_caller_identity fails (most likely something wierd on the143    # client side), then try a few extra options.144    # Test response if STS fails and we need to fall back to GetUser145    def test_get_aws_account_info__user_success(self):146        # Prepare147        self.sts_client.get_caller_identity.side_effect = [self._make_botocore_exception()]148        self.iam_client.get_user.side_effect = [{"User": {"Path": "/", "UserName": "ExampleUser", "UserId": "AIDA1234567890ABCDEFG",149                                                          "Arn": "arn:aws:iam::123456789012:user/ExampleUser", "CreateDate": "2020-09-08T14:04:32Z"}}]150        # Run module151        return_value = utils_iam.get_aws_account_info(self.module)152        # Check we only saw the calls we mocked out153        self.assertEqual(self.module.client.call_count, 2)154        self.sts_client.get_caller_identity.assert_called_once()155        self.iam_client.get_user.assert_called_once()156        # Check we got the values back we expected.157        self.assertEqual(return_value, ('123456789012', 'aws',))158    # (China partition)159    def test_get_aws_account_info__user_success_cn(self):160        # Prepare161        self.sts_client.get_caller_identity.side_effect = [self._make_botocore_exception()]162        self.iam_client.get_user.side_effect = [{"User": {"Path": "/", "UserName": "ExampleUser", "UserId": "AIDA1234567890ABCDEFG",163                                                          "Arn": "arn:aws-cn:iam::123456789012:user/ExampleUser", "CreateDate": "2020-09-08T14:04:32Z"}}]164        # Run module165        return_value = utils_iam.get_aws_account_info(self.module)166        # Check we only saw the calls we mocked out167        self.assertEqual(self.module.client.call_count, 2)168        self.sts_client.get_caller_identity.assert_called_once()169        self.iam_client.get_user.assert_called_once()170        # Check we got the values back we expected.171        self.assertEqual(return_value, ('123456789012', 'aws-cn',))172    # (US-Gov partition)173    def test_get_aws_account_info__user_success_gov(self):174        # Prepare175        self.sts_client.get_caller_identity.side_effect = [self._make_botocore_exception()]176        self.iam_client.get_user.side_effect = [{"User": {"Path": "/", "UserName": "ExampleUser", "UserId": "AIDA1234567890ABCDEFG",177                                                          "Arn": "arn:aws-us-gov:iam::123456789012:user/ExampleUser", "CreateDate": "2020-09-08T14:04:32Z"}}]178        # Run module179        return_value = utils_iam.get_aws_account_info(self.module)180        # Check we only saw the calls we mocked out181        self.assertEqual(self.module.client.call_count, 2)182        self.sts_client.get_caller_identity.assert_called_once()183        self.iam_client.get_user.assert_called_once()184        # Check we got the values back we expected.185        self.assertEqual(return_value, ('123456789012', 'aws-us-gov',))186    # Test response if STS and IAM fails and we need to fall back to the denial message187    def test_get_aws_account_info__user_denied(self):188        # Prepare189        self.sts_client.get_caller_identity.side_effect = [self._make_botocore_exception()]190        self.iam_client.get_user.side_effect = [self._make_denied_exception('aws')]191        # Run module192        return_value = utils_iam.get_aws_account_info(self.module)193        # Check we only saw the calls we mocked out194        self.assertEqual(self.module.client.call_count, 2)195        self.sts_client.get_caller_identity.assert_called_once()196        self.iam_client.get_user.assert_called_once()197        # Check we got the values back we expected.198        self.assertEqual(return_value, ('123456789012', 'aws',))199    # (China partition)200    def test_get_aws_account_info__user_denied_cn(self):201        # Prepare202        self.sts_client.get_caller_identity.side_effect = [self._make_botocore_exception()]203        self.iam_client.get_user.side_effect = [self._make_denied_exception('aws-cn')]204        # Run module205        return_value = utils_iam.get_aws_account_info(self.module)206        # Check we only saw the calls we mocked out207        self.assertEqual(self.module.client.call_count, 2)208        self.sts_client.get_caller_identity.assert_called_once()209        self.iam_client.get_user.assert_called_once()210        # Check we got the values back we expected.211        self.assertEqual(return_value, ('123456789012', 'aws-cn',))212    # (US-Gov partition)213    def test_get_aws_account_info__user_denied_gov(self):214        # Prepare215        self.sts_client.get_caller_identity.side_effect = [self._make_botocore_exception()]216        self.iam_client.get_user.side_effect = [self._make_denied_exception('aws-us-gov')]217        # Run module218        return_value = utils_iam.get_aws_account_info(self.module)219        # Check we only saw the calls we mocked out220        self.assertEqual(self.module.client.call_count, 2)221        self.sts_client.get_caller_identity.assert_called_once()222        self.iam_client.get_user.assert_called_once()223        # Check we got the values back we expected.224        self.assertEqual(return_value, ('123456789012', 'aws-us-gov',))225    # Test that we fail gracefully if Boto throws exceptions at us...226    def test_get_aws_account_info__boto_failures(self):227        # Prepare228        self.sts_client.get_caller_identity.side_effect = [self._make_botocore_exception()]229        self.iam_client.get_user.side_effect = [self._make_botocore_exception()]230        # Run module231        with pytest.raises(SystemExit) as e:232            utils_iam.get_aws_account_info(self.module)233        # Check we only saw the calls we mocked out234        self.assertEqual(self.module.client.call_count, 2)235        self.sts_client.get_caller_identity.assert_called_once()236        self.iam_client.get_user.assert_called_once()237        # Check we got the values back we expected.238        assert e.type == SystemExit239        assert e.value.code == 1  # 1 == fail_json_aws240    def test_get_aws_account_info__client_failures(self):241        # Prepare242        self.sts_client.get_caller_identity.side_effect = [self._make_unexpected_exception()]243        self.iam_client.get_user.side_effect = [self._make_unexpected_exception()]244        # Run module245        with pytest.raises(SystemExit) as e:246            utils_iam.get_aws_account_info(self.module)247        # Check we only saw the calls we mocked out248        self.assertEqual(self.module.client.call_count, 2)249        self.sts_client.get_caller_identity.assert_called_once()250        self.iam_client.get_user.assert_called_once()251        # Check we got the values back we expected.252        assert e.type == SystemExit253        assert e.value.code == 1  # 1 == fail_json_aws254    def test_get_aws_account_info__encoded_failures(self):255        # Prepare256        self.sts_client.get_caller_identity.side_effect = [self._make_encoded_exception()]257        self.iam_client.get_user.side_effect = [self._make_encoded_exception()]258        # Run module259        with pytest.raises(SystemExit) as e:260            utils_iam.get_aws_account_info(self.module)261        # Check we only saw the calls we mocked out262        self.assertEqual(self.module.client.call_count, 2)263        self.sts_client.get_caller_identity.assert_called_once()264        self.iam_client.get_user.assert_called_once()265        # Check we got the values back we expected.266        assert e.type == SystemExit...

test_sts.py

Source:test_sts.py

1from moto import mock_sts2from hamcrest import assert_that, is_, has_entries, ends_with3from aws_easy_use import sts4def test_sts_get_caller_identity(aws_account):5    with mock_sts():6        result = sts.get_caller_identity()7        assert_that(result, is_(dict))8        assert_that(result, has_entries(UserId=aws_account["user_id"], Account=aws_account["account_id"], Arn=aws_account["arn"]))9def test_sts_get_caller_identity_when_assuming_role(aws_account, sts_assumed_role_simple):10    result = sts.get_caller_identity()11    assert_that(result, is_(dict))12    assert_that(result, 13        has_entries(14            UserId=sts_assumed_role_simple["role_id"], 15            Account=aws_account["account_id"], 16            Arn=sts_assumed_role_simple["arn"]17        )18    )19def test_sts_assume_role(aws_account, iam_role_simple):20    role_session_name = "sts-assume-role-1"21    region =            "ap-northeast-1"22    @sts.assume_role(23        assume_role_arn=iam_role_simple["arn"], 24        assume_role_session_name=role_session_name,25        assume_role_region=region,26        external_id="4be10b85-4b3c-4b59-a71a-d0c1808ad5ac"27    )28    def _test_func():29        result = sts.get_caller_identity()30        assert_that(result, is_(dict))31        assert_that(result, 32            has_entries(33                UserId=ends_with(f":{role_session_name}"), 34                Account=aws_account["account_id"], 35                Arn=f"arn:aws:sts::{aws_account['account_id']}:assumed-role{iam_role_simple['path']}{iam_role_simple['name']}/{role_session_name}"36            )37        )38    with mock_sts():39        _test_func()40        result = sts.get_caller_identity()41        assert_that(result, is_(dict))...

sts.py

Source:sts.py

...3    def _decorator(func):4        @functools.wraps(func)5        def _wrapper(*args, **kw):6            print(f"Assume role `{assume_role_arn}`")7            print(f"Before assuming role, the identity is `{get_caller_identity()}`")8            assumed_role = boto3.client('sts').assume_role(9                RoleArn =         assume_role_arn,10                RoleSessionName = assume_role_session_name,11                ExternalId=external_id12            )13            boto3.setup_default_session(14                aws_access_key_id     = assumed_role['Credentials']['AccessKeyId'],15                aws_secret_access_key = assumed_role['Credentials']['SecretAccessKey'],16                aws_session_token     = assumed_role['Credentials']['SessionToken'],17                region_name           = assume_role_region18            )19            print(f"After assuming role, the identity is `{get_caller_identity()}`")20            result = func(*args, **kw)21            boto3.setup_default_session()22            return result23        return _wrapper24    return _decorator25def get_caller_identity() -> dict:26    """27    Get caller identity28    :param str reference_arn: reference ARN29    :return: is reference existed?30    :rtype: bool31    """...

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.