How to use describe_authentication_profiles method in localstack

Best Python code snippet using localstack_python

test_iam_helper.py

Source:test_iam_helper.py

1import datetime2import typing3from test.unit import MockCredentialsProvider4from unittest import mock5from unittest.mock import MagicMock, call6import pytest  # type: ignore7from dateutil.tz import tzutc8from pytest_mock import mocker9from redshift_connector import InterfaceError, ProgrammingError, RedshiftProperty10from redshift_connector.auth import AWSCredentialsProvider11from redshift_connector.config import ClientProtocolVersion12from redshift_connector.iam_helper import IamHelper13from redshift_connector.plugin import (14    AdfsCredentialsProvider,15    AzureCredentialsProvider,16    BasicJwtCredentialsProvider,17    BrowserAzureCredentialsProvider,18    BrowserSamlCredentialsProvider,19    OktaCredentialsProvider,20    PingCredentialsProvider,21)22from .helpers import make_redshift_property23@pytest.fixture24def mock_set_iam_credentials(mocker):25    mocker.patch("redshift_connector.iam_helper.IamHelper.set_iam_credentials", return_value=None)26@pytest.fixture27def mock_set_cluster_credentials(mocker):28    mocker.patch("redshift_connector.iam_helper.IamHelper.set_cluster_credentials", return_value=None)29@pytest.fixture30def mock_all_provider_get_credentials(mocker):31    for provider in [32        "OktaCredentialsProvider",33        "AzureCredentialsProvider",34        "BrowserAzureCredentialsProvider",35        "PingCredentialsProvider",36        "AdfsCredentialsProvider",37        "BrowserSamlCredentialsProvider",38        "SamlCredentialsProvider",39    ]:40        mocker.patch("redshift_connector.plugin.{}.get_credentials".format(provider), return_value=None)41def make_basic_redshift_property(**kwargs) -> RedshiftProperty:42    rp: RedshiftProperty = RedshiftProperty()43    for k, v in kwargs.items():44        rp.put(k, v)45    rp.put("user_name", "awsuser")46    rp.put("host", "localhost")47    rp.put("db_name", "dev")48    return rp49@pytest.mark.usefixtures("mock_set_iam_credentials")50def test_set_iam_properties_fails_when_non_str_credential_provider():51    keywords: typing.Dict = {52        "credentials_provider": 1,53        "iam": True,54    }55    with pytest.raises(InterfaceError) as excinfo:56        IamHelper.set_iam_properties(make_basic_redshift_property(**keywords))57    assert "Invalid connection property setting" in str(excinfo.value)58ssl_mode_descriptions: typing.List[typing.Tuple[typing.Optional[str], str]] = [59    ("verify-ca", "verify-ca"),60    ("verify-full", "verify-full"),61    ("disable", "verify-ca"),62    ("allow", "verify-ca"),63    ("prefer", "verify-ca"),64    ("require", "verify-ca"),65    ("bogus", "verify-ca"),66    (None, "verify-ca"),67]68@pytest.mark.usefixtures("mock_set_iam_credentials")69@pytest.mark.parametrize("ssl_param", ssl_mode_descriptions)70def test_set_iam_properties_enforce_min_ssl_mode(ssl_param):71    test_input, expected_mode = ssl_param72    keywords: typing.Dict = {"sslmode": test_input, "ssl": True}73    rp: RedshiftProperty = make_basic_redshift_property(**keywords)74    if test_input is None:75        assert rp.sslmode == expected_mode76    else:77        assert rp.sslmode == test_input78    IamHelper.set_iam_properties(rp)79    assert rp.sslmode == expected_mode80client_protocol_version_values: typing.List[int] = ClientProtocolVersion.list()81@pytest.mark.parametrize("_input", client_protocol_version_values)82def test_set_iam_properties_enforce_client_protocol_version(_input):83    keywords: typing.Dict = {"client_protocol_version": _input}84    rp: RedshiftProperty = make_basic_redshift_property(**keywords)85    assert rp.client_protocol_version == _input86    IamHelper.set_iam_properties(rp)87    assert rp.client_protocol_version == _input88multi_req_params: typing.List[typing.Tuple[typing.Dict, str]] = [89    ({"ssl": False, "iam": True}, "Invalid connection property setting. SSL must be enabled when using IAM"),90    (91        {"iam": False, "credentials_provider": "anything"},92        "Invalid connection property setting",93    ),94    (95        {"iam": False, "profile": "default"},96        "Invalid connection property setting",97    ),98    (99        {"iam": False, "access_key_id": "my_key"},100        "Invalid connection property setting",101    ),102    (103        {"iam": False, "secret_access_key": "shh it's a secret"},104        "Invalid connection property setting",105    ),106    (107        {"iam": False, "session_token": "my_session"},108        "Invalid connection property setting",109    ),110    (111        {"iam": True, "ssl": True},112        "Invalid connection property setting",113    ),114    (115        {116            "iam": True,117            "ssl": True,118            "access_key_id": "my_key",119            "credentials_provider": "OktaCredentialsProvider",120            "cluster_identifier": "my_cluster",121        },122        "Invalid connection property setting. It is not valid to provide both Credentials provider and AWS credentials or AWS profile",123    ),124    (125        {126            "iam": True,127            "ssl": True,128            "secret_access_key": "my_secret",129            "credentials_provider": "OktaCredentialsProvider",130            "cluster_identifier": "my_cluster",131        },132        "Invalid connection property setting. It is not valid to provide both Credentials provider and AWS credentials or AWS profile",133    ),134    (135        {136            "iam": True,137            "ssl": True,138            "session_token": "token",139            "credentials_provider": "OktaCredentialsProvider",140            "cluster_identifier": "my_cluster",141        },142        "Invalid connection property setting. It is not valid to provide both Credentials provider and AWS credentials or AWS profile",143    ),144    (145        {146            "iam": True,147            "ssl": True,148            "profile": "default",149            "credentials_provider": "OktaCredentialsProvider",150            "cluster_identifier": "my_cluster",151        },152        "Invalid connection property setting. It is not valid to provide both Credentials provider and AWS credentials or AWS profile",153    ),154    (155        {"iam": True, "ssl": True, "credentials_provider": 1, "cluster_identifier": "my_cluster"},156        "Invalid connection property setting. It is not valid to provide a non-string value to credentials_provider.",157    ),158    (159        {"iam": True, "ssl": True, "profile": "default", "access_key_id": "my_key", "cluster_identifier": "my_cluster"},160        "Invalid connection property setting. It is not valid to provide any of access_key_id, secret_access_key, or session_token when profile is provided",161    ),162    (163        {164            "iam": True,165            "ssl": True,166            "profile": "default",167            "secret_access_key": "my_secret",168            "cluster_identifier": "my_cluster",169        },170        "Invalid connection property setting. It is not valid to provide any of access_key_id, secret_access_key, or session_token when profile is provided",171    ),172    (173        {"iam": True, "ssl": True, "profile": "default", "session_token": "token", "cluster_identifier": "my_cluster"},174        "Invalid connection property setting. It is not valid to provide any of access_key_id, secret_access_key, or session_token when profile is provided",175    ),176    (177        {"iam": True, "ssl": True, "access_key_id": "my_access_key", "cluster_identifier": "my_cluster"},178        "Invalid connection property setting. secret access key must be provided in either secret_access_key or password field",179    ),180    (181        {"iam": True, "ssl": True, "cluster_identifier": "my_cluster", "secret_access_key": "someSecretAccessKey"},182        "Invalid connection property setting. access_key_id is required when secret_access_key is provided",183    ),184    (185        {"iam": True, "ssl": True, "cluster_identifier": "my_cluster", "session_token": "someSessionToken"},186        "Invalid connection property setting. access_key_id and secret_access_key are  required when session_token is provided",187    ),188    (189        {"iam": True, "ssl": True, "secret_access_key": "my_secret"},190        "Invalid connection property setting",191    ),192    (193        {"iam": True, "ssl": True, "session_token": "token"},194        "Invalid connection property setting",195    ),196    (197        {"iam": True, "ssl": True, "access_key_id": "my_key", "password": ""},198        "Invalid connection property setting",199    ),200    (201        {"iam": False, "ssl_insecure": False},202        "Invalid connection property setting",203    ),204    (205        {"iam": False, "client_protocol_version": max(ClientProtocolVersion.list()) + 1},206        "Invalid connection property setting. client_protocol_version must be in",207    ),208    (209        {210            "iam": True,211            "ssl": True,212        },213        "Invalid connection property setting. Credentials provider, AWS credentials, Redshift auth profile or AWS profile must be provided when IAM is enabled",214    ),215    (216        {"iam": True, "ssl": True, "credentials_provider": "SomeProvider"},217        "Invalid connection property setting. cluster_identifier must be provided when IAM is enabled",218    ),219    (220        {"iam": True, "ssl": True, "cluster_identifier": "some-cluster", "auth_profile": "someAuthProfile"},221        "Invalid connection property setting. access_key_id, secret_access_key, and region are required for authentication via Redshift auth_profile",222    ),223]224@pytest.mark.usefixtures("mock_set_iam_credentials")225@pytest.mark.parametrize("joint_params", multi_req_params)226def test_set_iam_properties_enforce_setting_compatibility(mocker, joint_params):227    test_input, expected_exception_msg = joint_params228    with pytest.raises(InterfaceError) as excinfo:229        IamHelper.set_iam_properties(make_basic_redshift_property(**test_input))230    assert expected_exception_msg in str(excinfo.value)231valid_credential_providers: typing.List[typing.Tuple[str, typing.Any]] = [232    ("OktaCredentialsProvider", OktaCredentialsProvider),233    ("AzureCredentialsProvider", AzureCredentialsProvider),234    ("BrowserAzureCredentialsProvider", BrowserAzureCredentialsProvider),235    ("PingCredentialsProvider", PingCredentialsProvider),236    ("BrowserSamlCredentialsProvider", BrowserSamlCredentialsProvider),237    ("AdfsCredentialsProvider", AdfsCredentialsProvider),238    ("BasicJwtCredentialsProvider", BasicJwtCredentialsProvider),239]240def test_set_iam_properties_raises_exception_when_info_is_none():241    with pytest.raises(InterfaceError) as excinfo:242        IamHelper.set_iam_properties(None)243    assert "Invalid connection property setting. info must be specified" in str(excinfo.value)244def mock_add_parameter():245    return mocker.patch246@pytest.mark.usefixtures("mock_all_provider_get_credentials")247@pytest.mark.usefixtures("mock_set_cluster_credentials")248@pytest.mark.parametrize("provider", valid_credential_providers)249def test_set_iam_properties_provider_assigned(mocker, provider):250    test_input, expectedProvider = provider251    mocker.patch("redshift_connector.plugin.{}.get_credentials".format(test_input))252    rp: RedshiftProperty = make_redshift_property()253    rp.credentials_provider = test_input254    spy = mocker.spy(expectedProvider, "add_parameter")255    IamHelper.set_iam_credentials(rp)256    assert spy.called257    assert spy.call_count == 1258    # ensure call to add_Parameter was made on the expected Provider class259    assert isinstance(spy.call_args[0][0], expectedProvider) is True260valid_aws_credential_args: typing.List[typing.Dict[str, str]] = [261    {"user": "", "password": "", "profile": "default"},262    {"user": "", "password": "", "access_key_id": "myAccessKey", "secret_access_key": "mySecret"},263    {"user": "", "secret_access_key": "", "access_key_id": "myAccessKey", "password": "myHiddenSecret"},264    {265        "user": "",266        "password": "",267        "access_key_id": "myAccessKey",268        "secret_access_key": "mySecret",269        "session_token": "mySession",270    },271    {"user": "", "password": "", "credentials_provider": "myCredentialsProvider"},272]273@pytest.mark.parametrize("test_input", valid_aws_credential_args)274def test_set_iam_properties_via_aws_credentials(mocker, test_input):275    # spy = mocker.spy("redshift_connector", "set_iam_credentials")276    rp: RedshiftProperty = make_basic_redshift_property(277        **{**test_input, **{"ssl": True, "iam": True, "cluster_identifier": "blah"}}278    )279    mocker.patch("redshift_connector.iam_helper.IamHelper.set_iam_credentials", return_value=None)280    IamHelper.set_iam_properties(rp)281    for aws_cred_key, aws_cred_val in enumerate(test_input):282        if aws_cred_key == "profile":283            assert rp.profile == aws_cred_val284        if aws_cred_key == "access_key_id":285            assert rp.access_key_id == aws_cred_val286        if aws_cred_key == "secret_access_key":287            assert rp.secret_access_key == aws_cred_val288        if aws_cred_key == "password":289            assert rp.password == aws_cred_val290        if aws_cred_key == "session_token":291            assert rp.session_token == aws_cred_val292def test_set_iam_credentials_via_aws_credentials(mocker):293    redshift_property: RedshiftProperty = RedshiftProperty()294    redshift_property.profile = "profile_val"295    redshift_property.access_key_id = "access_val"296    redshift_property.secret_access_key = "secret_val"297    redshift_property.session_token = "session_val"298    mocker.patch("redshift_connector.iam_helper.IamHelper.set_cluster_credentials", return_value=None)299    spy = mocker.spy(AWSCredentialsProvider, "add_parameter")300    IamHelper.set_iam_credentials(redshift_property)301    assert spy.called is True302    assert spy.call_count == 1303    assert spy.call_args[0][1] == redshift_property304def test_dynamically_loading_credential_holder(mocker):305    external_class_name: str = "test.unit.MockCredentialsProvider"306    mocker.patch("{}.get_credentials".format(external_class_name))307    mocker.patch("redshift_connector.iam_helper.IamHelper.set_cluster_credentials", return_value=None)308    rp: RedshiftProperty = make_redshift_property()309    rp.credentials_provider = external_class_name310    spy = mocker.spy(MockCredentialsProvider, "add_parameter")311    IamHelper.set_iam_credentials(rp)312    assert spy.called313    assert spy.call_count == 1314    # ensure call to add_Parameter was made on the expected Provider class315    assert isinstance(spy.call_args[0][0], MockCredentialsProvider) is True316def test_get_credentials_cache_key():317    rp: RedshiftProperty = RedshiftProperty()318    rp.db_user = "2"319    rp.db_name = "1"320    rp.db_groups = ["4", "3", "5"]321    rp.cluster_identifier = "6"322    rp.auto_create = False323    res_cache_key: str = IamHelper.get_credentials_cache_key(rp)324    assert res_cache_key is not None325    assert res_cache_key == "2;1;3,4,5;6;False"326def test_get_credentials_cache_key_no_db_groups():327    rp: RedshiftProperty = RedshiftProperty()328    rp.db_user = "2"329    rp.db_name = "1"330    rp.cluster_identifier = "6"331    rp.auto_create = False332    res_cache_key: str = IamHelper.get_credentials_cache_key(rp)333    assert res_cache_key is not None334    assert res_cache_key == "2;1;;6;False"335@mock.patch("boto3.client.get_cluster_credentials")336@mock.patch("boto3.client.describe_clusters")337@mock.patch("boto3.client")338def test_set_cluster_credentials_caches_credentials(339    mock_boto_client, mock_describe_clusters, mock_get_cluster_credentials340):341    mock_cred_provider = MagicMock()342    mock_cred_holder = MagicMock()343    mock_cred_provider.get_credentials.return_value = mock_cred_holder344    mock_cred_holder.has_associated_session = False345    rp: RedshiftProperty = make_redshift_property()346    IamHelper.credentials_cache.clear()347    IamHelper.set_cluster_credentials(mock_cred_provider, rp)348    assert len(IamHelper.credentials_cache) == 1349    assert mock_boto_client.called is True350    mock_boto_client.assert_has_calls(351        [352            call().get_cluster_credentials(353                AutoCreate=rp.auto_create,354                ClusterIdentifier=rp.cluster_identifier,355                DbGroups=rp.db_groups,356                DbName=rp.db_name,357                DbUser=rp.db_user,358            )359        ]360    )361@mock.patch("boto3.client.get_cluster_credentials")362@mock.patch("boto3.client.describe_clusters")363@mock.patch("boto3.client")364def test_set_cluster_credentials_honors_iam_disable_cache(365    mock_boto_client, mock_describe_clusters, mock_get_cluster_credentials366):367    mock_cred_provider = MagicMock()368    mock_cred_holder = MagicMock()369    mock_cred_provider.get_credentials.return_value = mock_cred_holder370    mock_cred_holder.has_associated_session = False371    rp: RedshiftProperty = make_redshift_property()372    rp.iam_disable_cache = True373    IamHelper.credentials_cache.clear()374    IamHelper.set_cluster_credentials(mock_cred_provider, rp)375    assert len(IamHelper.credentials_cache) == 0376    assert mock_boto_client.called is True377    mock_boto_client.assert_has_calls(378        [379            call().get_cluster_credentials(380                AutoCreate=rp.auto_create,381                ClusterIdentifier=rp.cluster_identifier,382                DbGroups=rp.db_groups,383                DbName=rp.db_name,384                DbUser=rp.db_user,385            )386        ]387    )388@mock.patch("boto3.client.get_cluster_credentials")389@mock.patch("boto3.client.describe_clusters")390@mock.patch("boto3.client")391def test_set_cluster_credentials_ignores_cache_when_disabled(392    mock_boto_client, mock_describe_clusters, mock_get_cluster_credentials393):394    mock_cred_provider = MagicMock()395    mock_cred_holder = MagicMock()396    mock_cred_provider.get_credentials.return_value = mock_cred_holder397    mock_cred_holder.has_associated_session = False398    rp: RedshiftProperty = make_redshift_property()399    rp.iam_disable_cache = True400    # mock out the boto3 response temporary credentials stored from prior auth401    mock_cred_obj: typing.Dict[str, typing.Union[str, datetime.datetime]] = {402        "DbUser": "xyz",403        "DbPassword": "turtle",404        "Expiration": datetime.datetime(9999, 1, 1, tzinfo=tzutc()),405    }406    # populate the cache407    IamHelper.credentials_cache.clear()408    IamHelper.credentials_cache[IamHelper.get_credentials_cache_key(rp)] = mock_cred_obj409    IamHelper.set_cluster_credentials(mock_cred_provider, rp)410    assert len(IamHelper.credentials_cache) == 1411    assert IamHelper.credentials_cache[IamHelper.get_credentials_cache_key(rp)] is mock_cred_obj412    assert mock_boto_client.called is True413    # we should not have retrieved user/password from the cache414    assert rp.user_name != mock_cred_obj["DbUser"]415    assert rp.password != mock_cred_obj["DbPassword"]416    assert (417        call().get_cluster_credentials(418            AutoCreate=rp.auto_create,419            ClusterIdentifier=rp.cluster_identifier,420            DbGroups=rp.db_groups,421            DbName=rp.db_name,422            DbUser=rp.db_user,423        )424        in mock_boto_client.mock_calls425    )426@mock.patch("boto3.client.get_cluster_credentials")427@mock.patch("boto3.client.describe_clusters")428@mock.patch("boto3.client")429def test_set_cluster_credentials_uses_cache_if_possible(430    mock_boto_client, mock_describe_clusters, mock_get_cluster_credentials431):432    mock_cred_provider = MagicMock()433    mock_cred_holder = MagicMock()434    mock_cred_provider.get_credentials.return_value = mock_cred_holder435    mock_cred_holder.has_associated_session = False436    rp: RedshiftProperty = make_redshift_property()437    # mock out the boto3 response temporary credentials stored from prior auth438    mock_cred_obj: typing.Dict[str, typing.Union[str, datetime.datetime]] = {439        "DbUser": "xyz",440        "DbPassword": "turtle",441        "Expiration": datetime.datetime(9999, 1, 1, tzinfo=tzutc()),442    }443    # populate the cache444    IamHelper.credentials_cache.clear()445    IamHelper.credentials_cache[IamHelper.get_credentials_cache_key(rp)] = mock_cred_obj446    IamHelper.set_cluster_credentials(mock_cred_provider, rp)447    assert len(IamHelper.credentials_cache) == 1448    assert IamHelper.credentials_cache[IamHelper.get_credentials_cache_key(rp)] is mock_cred_obj449    assert mock_boto_client.called is True450    assert rp.user_name == mock_cred_obj["DbUser"]451    assert rp.password == mock_cred_obj["DbPassword"]452    assert (453        call().get_cluster_credentials(454            AutoCreate=rp.auto_create,455            ClusterIdentifier=rp.cluster_identifier,456            DbGroups=rp.db_groups,457            DbName=rp.db_name,458            DbUser=rp.db_user,459        )460        not in mock_boto_client.mock_calls461    )462@mock.patch("boto3.client.get_cluster_credentials")463@mock.patch("boto3.client.describe_clusters")464@mock.patch("boto3.client")465def test_set_cluster_credentials_refreshes_stale_credentials(466    mock_boto_client, mock_describe_clusters, mock_get_cluster_credentials467):468    mock_cred_provider = MagicMock()469    mock_cred_holder = MagicMock()470    mock_cred_provider.get_credentials.return_value = mock_cred_holder471    mock_cred_holder.has_associated_session = False472    rp: RedshiftProperty = make_redshift_property()473    # mock out the boto3 response temporary credentials stored from prior auth (now stale)474    mock_cred_obj: typing.Dict[str, typing.Union[str, datetime.datetime]] = {475        "DbUser": "xyz",476        "DbPassword": "turtle",477        "Expiration": datetime.datetime(1, 1, 1, tzinfo=tzutc()),478    }479    # populate the cache480    IamHelper.credentials_cache.clear()481    IamHelper.credentials_cache[IamHelper.get_credentials_cache_key(rp)] = mock_cred_obj482    IamHelper.set_cluster_credentials(mock_cred_provider, rp)483    assert len(IamHelper.credentials_cache) == 1484    # ensure new temporary credentials have been replaced in cache485    assert IamHelper.get_credentials_cache_key(rp) in IamHelper.credentials_cache486    assert IamHelper.credentials_cache[IamHelper.get_credentials_cache_key(rp)] is not mock_cred_obj487    assert mock_boto_client.called is True488    mock_boto_client.assert_has_calls(489        [490            call().get_cluster_credentials(491                AutoCreate=rp.auto_create,492                ClusterIdentifier=rp.cluster_identifier,493                DbGroups=rp.db_groups,494                DbName=rp.db_name,495                DbUser=rp.db_user,496            )497        ]498    )499@pytest.mark.parametrize(500    "boto3_version",501    (502        "1.17.110",503        "1.17.100",504    ),505)506def test_set_iam_properties_raises_exception_when_insufficient_boto3_version(mocker, boto3_version):507    mock_boto3_dist_obj = MagicMock()508    mock_boto3_dist_obj.version = boto3_version509    mocker.patch("pkg_resources.get_distribution", return_value=mock_boto3_dist_obj)510    import pkg_resources511    with pytest.raises(pkg_resources.VersionConflict) as excinfo:512        IamHelper.set_iam_properties(513            make_basic_redshift_property(**{"iam": True, "ssl": True, "auth_profile": "SomeTestProfile"})514        )515    assert "boto3 >= 1.17.111 required for authentication via Amazon Redshift authentication profile." in str(516        excinfo.value517    )518def test_set_iam_properties_use_redshift_auth_profile_calls_read_auth_profile(mocker):519    mocker.patch(520        "redshift_connector.iam_helper.IamHelper.read_auth_profile", return_value=RedshiftProperty(kwargs={"": ""})521    )522    mocker.patch("redshift_connector.iam_helper.IamHelper.set_iam_credentials", return_value=None)523    spy = mocker.spy(IamHelper, "read_auth_profile")524    # anticipate read_auth_profile being called with the following parameters525    exp_call_arg: typing.Dict[str, str] = {526        "auth_profile": "someTestProfile",527        "access_key_id": "someAccessKeyIdValue",528        "session_token": "someSessionTokenValue",529        "secret_access_key": "someSecretAccessValue",530        "region": "someRegion",531        "endpoint_url": "someEndpointUrl",532    }533    rp: RedshiftProperty = make_basic_redshift_property(534        **{**{"iam": True, "ssl": True, "cluster_identifier": "someCluster"}, **exp_call_arg}535    )536    IamHelper.set_iam_properties(rp)537    assert spy.called is True538    assert spy.call_count == 1539    assert "auth_profile" in spy.call_args[1]540    assert spy.call_args[1]["auth_profile"] == exp_call_arg["auth_profile"]541    assert "iam_access_key_id" in spy.call_args[1]542    assert spy.call_args[1]["iam_access_key_id"] == exp_call_arg["access_key_id"]543    assert "iam_session_token" in spy.call_args[1]544    assert spy.call_args[1]["iam_session_token"] == exp_call_arg["session_token"]545    assert "iam_secret_key" in spy.call_args[1]546    assert spy.call_args[1]["iam_secret_key"] == exp_call_arg["secret_access_key"]547    assert "info" in spy.call_args[1]548    assert spy.call_args[1]["info"].region == exp_call_arg["region"]549    assert spy.call_args[1]["info"].endpoint_url == exp_call_arg["endpoint_url"]550def test_set_iam_properties_redshift_auth_profile_does_override(mocker):551    mock_contents: typing.Dict[str, str] = {552        "password": "overridePassword",553    }554    mock_auth_profile_contents: RedshiftProperty = RedshiftProperty(**mock_contents)555    mocker.patch("redshift_connector.iam_helper.IamHelper.read_auth_profile", return_value=mock_auth_profile_contents)556    mocker.patch("redshift_connector.iam_helper.IamHelper.set_iam_credentials", return_value=None)557    redshift_auth_profile_spy = mocker.spy(IamHelper, "read_auth_profile")558    set_iam_crednetials_spy = mocker.spy(IamHelper, "set_iam_credentials")559    exp_call_arg: typing.Dict[str, str] = {560        "auth_profile": "someTestProfile",561        "access_key_id": "someAccessKeyIdValue",562        "session_token": "someSessionTokenValue",563        "secret_access_key": "someSecretAccessValue",564        "region": "someRegion",565        "endpoint_url": "someEndpointUrl",566    }567    rp: RedshiftProperty = make_basic_redshift_property(568        **{569            **{"iam": True, "ssl": True, "cluster_identifier": "someCluster"},570            **exp_call_arg,571        }572    )573    res = IamHelper.set_iam_properties(rp)574    assert rp.password == mock_auth_profile_contents.password575    assert redshift_auth_profile_spy.called is True576    assert redshift_auth_profile_spy.call_count == 1577    assert res.password == mock_auth_profile_contents.password578    assert set_iam_crednetials_spy.called is True579    assert set_iam_crednetials_spy.call_count == 1580    assert set_iam_crednetials_spy.call_args[0][0].password == mock_auth_profile_contents.password581def test_read_auth_profile_raises_exception_if_profile_dne(mocker):582    from botocore import exceptions583    req_params: typing.Dict = {584        "auth_profile": "testProfile",585        "iam_access_key_id": "testAccessKey",586        "iam_secret_key": "someSecretKey",587        "iam_session_token": "someToken",588        "info": RedshiftProperty(),589    }590    req_params["info"].put("region", "us-east-1")591    mock_redshift_client: MagicMock = MagicMock()592    mock_redshift_client.describe_authentication_profiles.side_effect = exceptions.ClientError(593        operation_name="ErrorOp", error_response=MagicMock()594    )595    mocker.patch("boto3.client", return_value=mock_redshift_client)596    with pytest.raises(597        InterfaceError, match="Unable to retrieve contents of Redshift authentication profile from server"598    ):599        IamHelper.read_auth_profile(**req_params)600def test_read_auth_profile_loads_json_payload(mocker):601    import json602    req_params: typing.Dict = {603        "auth_profile": "testProfile",604        "iam_access_key_id": "testAccessKey",605        "iam_secret_key": "someSecretKey",606        "iam_session_token": "someToken",607        "info": RedshiftProperty(),608    }609    req_params["info"].put("region", "us-east-1")610    mock_payload: typing.Dict[str, str] = {"region": "someTestRegion", "cluster_identifier": "someCluster"}611    mock_redshift_client: MagicMock = MagicMock()612    mock_redshift_client.describe_authentication_profiles.return_value = {613        "AuthenticationProfiles": [{"AuthenticationProfileContent": json.dumps(mock_payload)}]614    }615    mocker.patch("boto3.client", return_value=mock_redshift_client)616    result = IamHelper.read_auth_profile(**req_params)617    assert result.region == mock_payload["region"]618    assert result.cluster_identifier == mock_payload["cluster_identifier"]619def test_read_auth_profile_invalid_json_payload_raises_exception(mocker):620    import json621    req_params: typing.Dict = {622        "auth_profile": "testProfile",623        "iam_access_key_id": "testAccessKey",624        "iam_secret_key": "someSecretKey",625        "iam_session_token": "someToken",626        "info": RedshiftProperty(),627    }628    req_params["info"].put("region", "us-east-1")629    mock_redshift_client: MagicMock = MagicMock()630    mock_redshift_client.describe_authentication_profiles.return_value = {631        "AuthenticationProfiles": [{"AuthenticationProfileContent": "{{{{"}]632    }633    mocker.patch("boto3.client", return_value=mock_redshift_client)634    with pytest.raises(635        ProgrammingError, match="Unable to decode the JSON content of the Redshift authentication profile"636    ):...

idp_auth_helper.py

Source:idp_auth_helper.py

...157            client = boto3.client(service_name="redshift", **creds)158            _logger.debug("Requesting authentication profiles")159            # 2nd phase - request Amazon Redshift authentication profiles and record contents for retrieving160            # temporary credentials for the Amazon Redshift cluster specified by end user161            response = client.describe_authentication_profiles(AuthenticationProfileName=auth_profile)162        except ClientError as e:163            raise InterfaceError(e)164        _logger.debug("Received {} authentication profiles".format(len(response["AuthenticationProfiles"])))165        # the first matching authentication profile will be used166        profile_content: typing.Union[str] = response["AuthenticationProfiles"][0]["AuthenticationProfileContent"]167        try:168            profile_content_dict: typing.Dict = json.loads(profile_content)169            return RedshiftProperty(**profile_content_dict)170        except ValueError:171            raise ProgrammingError(172                "Unable to decode the JSON content of the Redshift authentication profile: {}".format(auth_profile)173            )174    @staticmethod175    def load_credentials_provider(info: RedshiftProperty) -> IPlugin:...

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.