test(connection, startup): support new startup message properties

Author: Brooke-white

test/integration/plugin/test_credentials_providers.py

@@ -8,6 +8,7 @@
 import pytest  # type: ignore
 
 import redshift_connector
+from redshift_connector import DriverInfo
 
 conf = configparser.ConfigParser()
 root_path = os.path.dirname(os.path.dirname(os.path.abspath(os.path.join(__file__, os.pardir))))
@@ -160,3 +161,24 @@ def use_cached_temporary_credentials(idp_arg):
     # holds these
     assert len(redshift_connector.IamHelper.credentials_cache) == 1
     assert first_cred_cache_entry == redshift_connector.IamHelper.credentials_cache.popitem()
+
+
+@pytest.mark.skip
+# TODO: https://docs.aws.amazon.com/redshift/latest/dg/r_STL_CONNECTION_LOG.html plugin_name column character limit
+# TODO: causes field value cut-off
+@pytest.mark.parametrize("idp_arg", NON_BROWSER_IDP, indirect=True)
+def test_stl_connection_log_contains_plugin_name(idp_arg, db_kwargs):
+    idp_arg["auto_create"] = True
+    with redshift_connector.connect(**idp_arg) as conn:
+        pass
+    with redshift_connector.connect(**db_kwargs) as conn:
+        with conn.cursor() as cursor:
+            # verify stl_connection_log contains driver version as expected
+            cursor.execute(
+                "select top 1 1 from stl_connection_log where driver_version = '{}' and plugin_name = 'redshift_connector.plugin.{}'".format(
+                    DriverInfo.driver_full_name(), idp_arg["credentials_provider"]
+                )
+            )
+            res = cursor.fetchone()
+            assert res is not None
+            assert res[0] == 1

test/integration/test_connection.py

@@ -1,12 +1,15 @@
 import configparser
 import os
+import random
 import socket
+import string
 import struct
 import sys
 
 import pytest  # type: ignore
 
 import redshift_connector
+from redshift_connector import DriverInfo
 from redshift_connector.config import ClientProtocolVersion
 
 conf = configparser.ConfigParser()
@@ -198,3 +201,49 @@ def test_client_protocol_version_invalid_warns_user(db_kwargs):
     del db_kwargs["cluster_identifier"]
     with pytest.warns(UserWarning):
         redshift_connector.Connection(**db_kwargs)
+
+
+def test_stl_connection_log_contains_driver_version(db_kwargs):
+    with redshift_connector.connect(**db_kwargs) as conn:
+        with conn.cursor() as cursor:
+            # verify stl_connection_log contains driver version as expected
+            cursor.execute(
+                "select top 1 1 from stl_connection_log where driver_version = '{}'".format(
+                    DriverInfo.driver_full_name()
+                )
+            )
+            res = cursor.fetchone()
+            assert res is not None
+            assert res[0] == 1
+
+
+def test_stl_connection_log_contains_os_version(db_kwargs):
+    with redshift_connector.connect(**db_kwargs) as conn:
+        with conn.cursor() as cursor:
+            # verify stl_connection_log contains driver version as expected
+            cursor.execute(
+                "select top 1 1 from stl_connection_log where driver_version = '{}' and os_version = '{}'".format(
+                    DriverInfo.driver_full_name(), conn.client_os_version
+                )
+            )
+            res = cursor.fetchone()
+            assert res is not None
+            assert res[0] == 1
+
+
+def test_stl_connection_log_contains_application_name(db_kwargs):
+    # make some connection so this unique application name is logged
+    mock_application_name: str = "".join(random.choice(string.ascii_letters) for x in range(10))
+    db_kwargs["application_name"] = mock_application_name
+
+    with redshift_connector.connect(**db_kwargs) as conn:
+        with conn.cursor() as cursor:
+            # verify stl_connection_log contains driver version as expected
+            cursor.execute(
+                "select top 1 1 from stl_connection_log where driver_version = '{}' and application_name = '{}'".format(
+                    DriverInfo.driver_full_name(), mock_application_name
+                )
+            )
+            res = cursor.fetchone()
+            assert res is not None
+            assert res[0] == 1

test/unit/test_connection.py

@@ -1,6 +1,7 @@
 import typing
 from collections import deque
 from decimal import Decimal
+from unittest.mock import patch
 
 import pytest  # type: ignore
 
@@ -310,3 +311,16 @@ def test_is_single_database_metadata(_input):
         mock_connection.parameter_statuses.append((b"datashare_enabled", param_status.encode()))
 
     assert mock_connection.is_single_database_metadata == exp_val
+
+
+def test_client_os_version_is_present():
+    mock_connection = Connection.__new__(Connection)
+    assert mock_connection.client_os_version is not None
+    assert isinstance(mock_connection.client_os_version, str)
+
+
+def test_client_os_version_is_not_present():
+    mock_connection = Connection.__new__(Connection)
+
+    with patch("platform.platform", side_effect=Exception("not for you")):
+        assert mock_connection.client_os_version == "unknown"

test/unit/test_driver_info.py

@@ -0,0 +1,38 @@
+import re
+
+import pytest
+
+from redshift_connector.utils import DriverInfo
+
+
+def test_version_is_not_none():
+    assert DriverInfo.version() is not None
+
+
+def test_version_is_str():
+    assert isinstance(DriverInfo.version(), str)
+
+
+def test_version_proper_format():
+    version_regex = re.compile(r"^\d+(\.\d+){2,3}$")
+    assert version_regex.match(DriverInfo.version())
+
+
+def test_driver_name_is_not_none():
+    assert DriverInfo.driver_name() is not None
+
+
+def test_driver_short_name_is_not_none():
+    assert DriverInfo.driver_short_name() is not None
+
+
+def test_driver_full_name_is_not_none():
+    assert DriverInfo.driver_full_name() is not None
+
+
+def test_driver_full_name_contains_name():
+    assert DriverInfo.driver_name() in DriverInfo.driver_full_name()
+
+
+def test_driver_full_name_contains_version():
+    assert DriverInfo.version() in DriverInfo.driver_full_name()

test/unit/test_iam_helper.py

@@ -113,6 +113,14 @@ def test_set_iam_properties_fails_when_info_is_none(missing_param):
     assert "Invalid connection property setting" in str(excinfo.value)
 
 
+@pytest.mark.usefixtures("mock_set_iam_credentials")
+def test_set_iam_properties_fails_when_non_str_credential_provider():
+    keywords: typing.Dict = {"credentials_provider": 1}
+    with pytest.raises(InterfaceError) as excinfo:
+        IamHelper.set_iam_properties(**get_set_iam_properties_args(**keywords))
+    assert "Invalid connection property setting" in str(excinfo.value)
+
+
 ssl_mode_descriptions: typing.List[typing.Tuple[typing.Optional[str], str]] = [
     ("verify-ca", "verify-ca"),
     ("verify-full", "verify-full"),