test(IdP): dynamic loading and use of external IdP plugin

Author: Brooke-white

test/integration/plugin/test_credentials_providers.py

@@ -138,5 +138,5 @@ def testSslAndIam(idp_arg):
 @pytest.mark.parametrize("idp_arg", ALL_IDP, indirect=True)
 def testWrongCredentialsProvider(idp_arg):
     idp_arg["credentials_provider"] = "WrongProvider"
-    with pytest.raises(redshift_connector.InterfaceError, match="Invalid credentials provider" + "WrongProvider"):
+    with pytest.raises(redshift_connector.InterfaceError, match="Invalid credentials provider WrongProvider"):
         redshift_connector.connect(**idp_arg)

test/unit/__init__.py

@@ -0,0 +1 @@
+from .mocks import MockCredentialsProvider

test/unit/mocks/__init__.py

@@ -0,0 +1 @@
+from .mock_external_credential_provider import MockCredentialsProvider

test/unit/mocks/mock_external_credential_provider.py

@@ -0,0 +1,6 @@
+from redshift_connector.plugin import SamlCredentialsProvider
+
+
+class MockCredentialsProvider(SamlCredentialsProvider):
+    def get_saml_assertion(self: "SamlCredentialsProvider"):
+        return "mocked"

test/unit/plugin/test_credentials_providers.py

@@ -20,7 +20,7 @@
 @pytest.mark.parametrize("idp_arg", ALL_IDP, indirect=True)
 def test_credential_provider_dne_should_fail(idp_arg):
     idp_arg["credentials_provider"] = "WrongProvider"
-    with pytest.raises(redshift_connector.InterfaceError, match="Invalid credentials provider" + "WrongProvider"):
+    with pytest.raises(redshift_connector.InterfaceError, match="Invalid credentials provider WrongProvider"):
         redshift_connector.connect(**idp_arg)
 
 

test/unit/test_iam_helper.py

@@ -1,4 +1,5 @@
 import typing
+from test.unit import MockCredentialsProvider
 
 import pytest  # type: ignore
 from pytest_mock import mocker
@@ -321,3 +322,19 @@ def test_set_iam_credentials_via_aws_credentials(mocker):
     assert spy.called is True
     assert spy.call_count == 1
     assert spy.call_args[0][1] == redshift_property
+
+
+def test_dynamically_loading_credential_holder(mocker):
+    external_class_name: str = "test.unit.MockCredentialsProvider"
+    mocker.patch("{}.get_credentials".format(external_class_name))
+    mocker.patch("redshift_connector.iam_helper.set_cluster_credentials", return_value=None)
+    rp: RedshiftProperty = make_redshift_property()
+    rp.credentials_provider = external_class_name
+
+    spy = mocker.spy(MockCredentialsProvider, "add_parameter")
+
+    set_iam_credentials(rp)
+    assert spy.called
+    assert spy.call_count == 1
+    # ensure call to add_Parameter was made on the expected Provider class
+    assert isinstance(spy.call_args[0][0], MockCredentialsProvider) is True