adding test cases

Author: rakshil14-2

awscrt/aws_iot_metrics.py

@@ -68,6 +68,7 @@ def _certificate_source_metrics_value(source):
 
     Mapping: CERTIFICATE_FILES->A, PKCS11->B, WINDOWS_CERT_STORE->C, PKCS12_FILE->E.
     "D" is reserved for Java KeyStore.
+    Returns None if source is None or unrecognized.
     """
     from awscrt.io import _CertificateSource
     mapping = {
@@ -237,7 +238,7 @@ def _get_encoded_feature_list(client_options):
         - D (outbound_topic_alias_behavior): from topic_aliasing_options.outbound_behavior
         - E (inbound_topic_alias_behavior): from topic_aliasing_options.inbound_behavior
         - H (http_proxy_type): HTTP or HTTPS based on proxy TLS settings
-        - I (certificate_source): Detected from TlsContextOptions
+        - I (certificate_source): detected from TlsContextOptions
         - J (tls_cipher_preference): mapped from TlsCipherPref on the TLS context
         - K (minimum_tls_version): mapped from TlsVersion on the TLS context
 
@@ -292,7 +293,7 @@ def _get_encoded_feature_list(client_options):
         val = _http_proxy_type_metrics_value(client_options.http_proxy_options)
         features.append(f"{_MetricsFeatureId.HTTP_PROXY_TYPE.value}/{val}")
 
-    # I: certificate_source - Auto-detected from TlsContextOptions factory method
+    # I: certificate_source - detected from TlsContextOptions factory method
     if client_options.tls_ctx is not None:
         val = _certificate_source_metrics_value(getattr(client_options.tls_ctx, '_certificate_source', None))
         if val:
@@ -325,7 +326,7 @@ def _get_encoded_feature_list_mqtt3(proxy_options, tls_ctx=None):
 
     Conditionally includes:
         - H (http_proxy_type): HTTP or HTTPS based on proxy TLS settings
-        - I (certificate_source): auto-detected from TlsContextOptions factory method
+        - I (certificate_source): detected from TlsContextOptions
         - J (tls_cipher_preference): mapped from TlsCipherPref on the TLS context
         - K (minimum_tls_version): mapped from TlsVersion on the TLS context
 
@@ -344,7 +345,7 @@ def _get_encoded_feature_list_mqtt3(proxy_options, tls_ctx=None):
         val = _http_proxy_type_metrics_value(proxy_options)
         features.append(f"{_MetricsFeatureId.HTTP_PROXY_TYPE.value}/{val}")
 
-    # I: certificate_source - Auto-detected from TlsContextOptions factory method
+    # I: certificate_source - detected from TlsContextOptions factory method
     if tls_ctx is not None:
         val = _certificate_source_metrics_value(getattr(tls_ctx, '_certificate_source', None))
         if val:
@@ -461,7 +462,7 @@ def _create_metrics_mqtt5(client_options):
     """Create the final AWSIoTMetrics object for an MQTT5 client.
 
     Generates the CRT feature list from the full set of MQTT5 ClientOptions,
-    including auto-detected certificate source from the TLS context.
+    including detected certificate source from the TLS context.
 
     Args:
         client_options: MQTT5 ClientOptions dataclass containing all
@@ -477,7 +478,7 @@ def _create_metrics_mqtt3(user_metrics=None, proxy_options=None, tls_ctx=None):
     """Create the final AWSIoTMetrics object for an MQTT3 connection.
 
     Generates the CRT feature list from the MQTT3 connection parameters,
-    including auto-detected certificate source from the TLS context.
+    including detected certificate source from the TLS context.
 
     Args:
         user_metrics: Optional AWSIoTMetrics configuration provided by the IoT SDK.

awscrt/io.py

@@ -19,13 +19,13 @@ class _CertificateSource(IntEnum):
     """Certificate source types for mTLS authentication.
 
     Used by TlsContextOptions to track which factory method created the
-    TLS configuration. The CRT detects this and encodes it as
+    TLS configuration. The CRT reads this and encodes it as
     feature ID "I" in the metrics string.
     """
-    CERTIFICATE_FILES = 0
-    PKCS11 = 1
-    WINDOWS_CERT_STORE = 2
-    PKCS12_FILE = 3
+    CERTIFICATE_FILES = 0 # PEM cert + key files
+    PKCS11 = 1 # Hardware security module via PKCS11
+    WINDOWS_CERT_STORE = 2 # windows certificate source
+    PKCS12_FILE = 4 # PKCS#12 (.p12/.pfx) files
 
 
 class LogLevel(IntEnum):

test/test_aws_iot_metrics.py

@@ -11,6 +11,7 @@
     _protocol_version_metrics_value,
     _socket_implementation_metrics_value,
     _http_proxy_type_metrics_value,
+    _certificate_source_metrics_value,
     IOT_SDK_METRICS_FEATURE_VERSION,
     _get_encoded_feature_list,
     _get_encoded_feature_list_mqtt3,
@@ -26,7 +27,7 @@
     InboundTopicAliasBehaviorType,
     TopicAliasingOptions,
 )
-from awscrt.io import ClientTlsContext, TlsContextOptions, TlsConnectionOptions, TlsVersion, TlsCipherPref
+from awscrt.io import ClientTlsContext, TlsContextOptions, TlsConnectionOptions, TlsVersion, TlsCipherPref, _CertificateSource
 from awscrt.http import HttpProxyOptions
 
 
@@ -419,5 +420,30 @@ def test_custom_library_name(self):
         self.assertEqual("MyCustomSDK/1.0", result.library_name)
 
 
+class TestCertificateSourceMetrics(NativeResourceTest):
+    """Test certificate source metrics encoding."""
+
+    def test_certificate_source_mapping_values(self):
+        """Verify each _CertificateSource maps to the correct metrics value."""
+        self.assertEqual("A", _certificate_source_metrics_value(_CertificateSource.CERTIFICATE_FILES))
+        self.assertEqual("B", _certificate_source_metrics_value(_CertificateSource.PKCS11))
+        self.assertEqual("C", _certificate_source_metrics_value(_CertificateSource.WINDOWS_CERT_STORE))
+        self.assertEqual("E", _certificate_source_metrics_value(_CertificateSource.PKCS12_FILE))
+        self.assertIsNone(_certificate_source_metrics_value(None))
+
+    def test_certificate_source_in_feature_list(self):
+        """TLS context with certificate source should include feature I in metrics."""
+        tls_options = TlsContextOptions()
+        tls_options._certificate_source = _CertificateSource.CERTIFICATE_FILES
+        tls_ctx = ClientTlsContext(tls_options)
+
+        options = ClientOptions(host_name="localhost", port=8883, tls_ctx=tls_ctx)
+        result = _get_encoded_feature_list(options)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/A", result)
+
+        result_mqtt3 = _get_encoded_feature_list_mqtt3(proxy_options=None, tls_ctx=tls_ctx)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/A", result_mqtt3)
+
+
 if __name__ == '__main__':
     unittest.main()