Better ApmConfig behaviour including try-except

solarwinds_apm/apm_config.py

@@ -393,9 +393,18 @@ def _calculate_service_name_apm_proto(
                 "unknown_service"
             ):
                 # When agent_enabled, assume service_key exists and is formatted correctly.
-                service_name = self.__config.get("service_key", ":").split(
-                    ":"
-                )[1]
+                # This is a precaution. Validate defensively instead of relying on
+                # exception handling so unexpected non-string values cannot crash
+                # the instrumented application.
+                service_key = self.__config.get("service_key", ":")
+                if isinstance(service_key, str):
+                    service_key_parts = service_key.split(":", 1)
+                    if len(service_key_parts) > 1:
+                        service_name = service_key_parts[1]
+                    else:
+                        service_name = ""
+                else:
+                    service_name = ""
             else:
                 service_name = otel_service_name
         return service_name
@@ -443,14 +452,18 @@ def _update_service_key_name(
         if agent_enabled and service_key and service_name:
             # Only update if service_name and service_key exist and non-empty,
             # and service_key in correct format.
-            key_parts = service_key.split(":")
+            try:
+                key_parts = service_key.split(":")
+            except AttributeError:
+                logger.debug("Service key is not a valid string. Skipping.")
+                return service_key
             if len(key_parts) < 2:
                 logger.debug(
                     "Service key is not in the correct format to update its own service name. Skipping."
                 )
                 return service_key
 
-            return ":".join([service_key.split(":")[0], service_name])
+            return ":".join([key_parts[0], service_name])
 
         # Else no need to update service_key when not reporting
         return service_key
@@ -465,12 +478,15 @@ def mask_service_key(self) -> str:
         service_key = self.__config.get("service_key")
         if not service_key:
             return ""
-        if service_key.strip() == "":
-            return service_key
 
-        key_parts = service_key.split(":")
+        try:
+            if service_key.strip() == "":
+                return service_key
+            key_parts = service_key.split(":")
+        except AttributeError:
+            return ""
         if len(key_parts) < 2:
-            bad_format_key = key_parts[0]
+            bad_format_key = key_parts[0] if key_parts else ""
             if len(bad_format_key) < 5:
                 return self._KEY_MASK_BAD_FORMAT_SHORT.format(bad_format_key)
             return self._KEY_MASK_BAD_FORMAT.format(
@@ -855,11 +871,10 @@ def to_configuration(cls, apm_config) -> Configuration:
         Returns:
         Configuration: Configuration object for sampler initialization.
         """
-        token = (
-            apm_config.get("service_key").split(":")[0]
-            if len(apm_config.get("service_key").split(":")) > 0
-            else ""
-        )
+        try:
+            token = apm_config.get("service_key").split(":")[0]
+        except (AttributeError, IndexError):
+            token = ""
         filters = apm_config.get("transaction_filters")
         transaction_settings = []
         for transaction_filter in filters:

tests/unit/test_apm_config/test_apm_config.py

@@ -326,6 +326,12 @@ def test_config_mask_service_key(
         self._mock_service_key(mocker, "valid-and-long:key")
         assert apm_config.SolarWindsApmConfig()._config_mask_service_key().get("service_key") == "vali...long:key"
 
+    def test_mask_service_key_attribute_error_non_string_key(self):
+        test_config = apm_config.SolarWindsApmConfig()
+        test_config._SolarWindsApmConfig__config["service_key"] = 123
+        result = test_config.mask_service_key()
+        assert result == ""
+
     def test_str(
         self,
         mocker,
@@ -664,3 +670,15 @@ def test_to_configuration_with_enabled_agent(apm):
     apm.agent_enabled = True
     config = apm_config.SolarWindsApmConfig.to_configuration(apm_config=apm)
     assert config.enabled is True
+
+def test_to_configuration_attribute_error_non_string_service_key(
+    mocker,
+):
+    mocker.patch.dict(os.environ, {
+        "SW_APM_SERVICE_KEY": "token:service",
+    })
+    test_apm_config = apm_config.SolarWindsApmConfig()
+    test_apm_config._SolarWindsApmConfig__config["service_key"] = None
+    config = apm_config.SolarWindsApmConfig.to_configuration(test_apm_config)
+    # Should default to empty string for token in Authorization header
+    assert config.headers["Authorization"] == "Bearer "

tests/unit/test_apm_config/test_apm_config_service_name.py

@@ -120,6 +120,32 @@ def test__calculate_service_name_apm_proto_use_otel_service_name(
         )
         assert result == "foobar"
 
+    def test__calculate_service_name_apm_proto_malformed_service_key_only_token(
+        self,
+        mocker,
+    ):
+        mocker.patch.dict(os.environ, {
+            "SW_APM_SERVICE_KEY": "token:",
+        })
+        test_config = apm_config.SolarWindsApmConfig()
+        result = test_config._calculate_service_name_apm_proto(
+            True,
+            Resource.create()  # default is unknown_service
+        )
+        assert result == ""
+
+    def test__calculate_service_name_apm_proto_non_string_service_key(
+        self,
+        mocker,
+    ):
+        test_config = apm_config.SolarWindsApmConfig()
+        test_config._SolarWindsApmConfig__config["service_key"] = 123
+        result = test_config._calculate_service_name_apm_proto(
+            True,
+            Resource.create()  # default is unknown_service
+        )
+        assert result == ""
+
 class TestSolarWindsApmConfigServiceNameLambda:
     def test__calculate_service_name_lambda_no_otel_name(
         self,