diff --git a/awsiot/iot_metrics.py b/awsiot/iot_metrics.py new file mode 100644 index 00000000..82cdba35 --- /dev/null +++ b/awsiot/iot_metrics.py @@ -0,0 +1,140 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0. + +""" +IoT SDK Metrics V2 - SDK layer feature tracking. + +This module implements the SDK-side of the IoT Metrics. +It collects SDK-level feature usage information (such as the certificate source +used for authentication) and packages it into an ``AWSIoTMetrics`` +object. The CRT layer then merges these SDK-level features with its own CRT-level +features and embeds the combined metrics string in the MQTT CONNECT packet's +username field. + +Metrics Flow: + 1. A connection builder determines which certificate source is in use. + 2. :func:`build_sdk_metrics` is called with the appropriate + :class:`CertificateSource` value (or ``None`` for connections that + don't use client certificates, e.g. websocket or custom auth). + 3. The returned ``AWSIoTMetrics`` object is + passed to the CRT connection/client, which handles final encoding into + the CONNECT username. + +Feature Encoding Format: + SDK features are encoded as comma-separated ``ID/Value`` pairs. + For example, certificate source PKCS11 is encoded as ``I/B``. +""" + +from enum import Enum +from typing import Optional + +from awscrt.aws_iot_metrics import AWSIoTMetrics, IoTMetricsMetadata, IOT_SDK_METRICS_FEATURE_VERSION + +SDK_LIBRARY_NAME = "IoTDeviceSDK/Python" + +class FeatureId(str, Enum): + """SDK-layer feature identifiers. + + Each member maps a feature name to the single-character ID + used in the encoded metrics string. + + Attributes: + CERTIFICATE_SOURCE: Tracks which certificate/authentication method is + used for the connection. Encoded values come from :class:`CertificateSource`. + """ + CERTIFICATE_SOURCE = "I" + +class CertificateSource(str, Enum): + """Certificate source identifiers for metrics feature ``I``. + + Each value corresponds to a specific authentication method used by the + MQTT connection. The single-character value is what gets encoded into the + metrics string sent in the CONNECT packet. + + Note: + Value ``"D"`` (Java KeyStore) is reserved for the Java SDK and is not + applicable to the Python SDK. It is intentionally skipped here. + + Attributes: + CERTIFICATE_FILES: Client certificate and private key provided as file paths. + PKCS11: Private key stored in a PKCS#11-compatible hardware security module. + WINDOWS_CERT_STORE: Certificate retrieved from the Windows system certificate store. + PKCS12_FILE: Certificate and private key bundled in a PKCS#12 (.p12/.pfx) file. + """ + CERTIFICATE_FILES = "A" + PKCS11 = "B" + WINDOWS_CERT_STORE = "C" + # "D" is Java KeyStore — not applicable to the Python SDK. + PKCS12_FILE = "E" + + + +def _get_sdk_version(): + """Return the installed ``awsiotsdk`` package version string. + + Falls back to ``"dev"`` if the package metadata is unavailable (e.g. when + running from a source checkout without installing). + + Returns: + str: A version string like ``"1.21.0"`` or ``"dev"``. + """ + try: + import importlib.metadata + return importlib.metadata.version("awsiotsdk") + except Exception: + return "dev" + + +def _encode_feature_list(certificate_source: Optional[CertificateSource] = None) -> str: + """Encode SDK features into the ``ID/Value,...`` wire format. + + Each feature is represented as its :class:`FeatureId` character followed by + a slash and the feature-specific value character. Multiple features would be + separated by commas (currently only one feature is tracked). + + Args: + certificate_source: The certificate method in use, or ``None`` if no + client certificate is involved. + + Returns: + str: Encoded feature string (e.g. ``"I/A"``), or an empty string if no + features apply. + """ + if certificate_source is not None: + return f"{FeatureId.CERTIFICATE_SOURCE.value}/{certificate_source.value}" + return "" + + +def build_sdk_metrics(certificate_source: Optional[CertificateSource] = None) -> AWSIoTMetrics: + """Build an ``AWSIoTMetrics`` instance for the CRT layer. + + This is the main entry point for SDK metrics. Connection builders call this + function to produce the metrics object that the CRT will merge with its own + metrics and embed in the MQTT CONNECT username. + + The returned object always includes: + - ``IoTSDKVersion``: The installed SDK version string. + + When a *certificate_source* is provided, it additionally includes: + - ``IoTSDKFeature``: Encoded feature string (e.g. ``"I/A"``). + - ``IoTSDKMetricsVersion``: The metrics protocol version supported. + + Args: + certificate_source: The certificate/authentication method used by this + connection. Pass ``None`` for connections that don't use client + certificates (e.g. websocket with SigV4, custom authorizers). + + Returns: + AWSIoTMetrics: A metrics object ready to be passed to the CRT + connection or client builder. + """ + metadata = [ + IoTMetricsMetadata(key="IoTSDKVersion", value=_get_sdk_version()), + ] + + feature_list = _encode_feature_list(certificate_source) + if feature_list: + metadata.append(IoTMetricsMetadata(key="IoTSDKFeature", value=feature_list)) + metadata.append(IoTMetricsMetadata(key="IoTSDKMetricsVersion", value=str(IOT_SDK_METRICS_FEATURE_VERSION))) + + return AWSIoTMetrics(library_name=SDK_LIBRARY_NAME, metadata_entries=metadata) diff --git a/awsiot/mqtt5_client_builder.py b/awsiot/mqtt5_client_builder.py index ae9b9750..90c61dbd 100644 --- a/awsiot/mqtt5_client_builder.py +++ b/awsiot/mqtt5_client_builder.py @@ -170,8 +170,8 @@ **cipher_pref** (:class:`awscrt.io.TlsCipherPref`): Cipher preference to use for TLS connection. Default is `TlsCipherPref.DEFAULT`. - **enable_metrics_collection** (`bool`): Whether to send the SDK version number in the CONNECT packet. - Default is True. + **enable_metrics_collection** (`bool`): Controls whether SDK metrics are included in the CONNECT packet. + Defaults to True (metrics enabled). """ @@ -184,6 +184,8 @@ import awscrt.mqtt5 import urllib.parse +from awsiot.iot_metrics import CertificateSource, build_sdk_metrics + DEFAULT_WEBSOCKET_MQTT_PORT = 443 DEFAULT_DIRECT_MQTT_PORT = 8883 @@ -210,35 +212,6 @@ def _get(kwargs, name, default=None): return val -_metrics_str = None - - -def _get_metrics_str(current_username=""): - global _metrics_str - - username_has_query = False - if current_username.find("?") != -1: - username_has_query = True - - if _metrics_str is None: - try: - import importlib.metadata - try: - version = importlib.metadata.version("awsiotsdk") - _metrics_str = "SDK=PythonV2&Version={}".format(version) - except importlib.metadata.PackageNotFoundError: - _metrics_str = "SDK=PythonV2&Version=dev" - except BaseException: - _metrics_str = "" - - if not _metrics_str == "": - if username_has_query: - return "&" + _metrics_str - else: - return "?" + _metrics_str - else: - return "" - def _builder( tls_ctx_options, @@ -246,13 +219,12 @@ def _builder( websocket_handshake_transform=None, use_custom_authorizer=False, cipher_pref=awscrt.io.TlsCipherPref.DEFAULT, + certificate_source=None, **kwargs): assert isinstance(cipher_pref, awscrt.io.TlsCipherPref) username = _get(kwargs, 'username', '') - if _get(kwargs, 'enable_metrics_collection', True): - username += _get_metrics_str(username) client_options = _get(kwargs, 'client_options') if client_options is None: @@ -364,6 +336,11 @@ def _builder( tls_ctx = awscrt.io.ClientTlsContext(tls_ctx_options) client_options.tls_ctx = tls_ctx + + # Set SDK metrics for the CRT layer to embed in the CONNECT packet username + if _get(kwargs, 'enable_metrics_collection', True): + client_options.metrics = build_sdk_metrics(certificate_source) + client = awscrt.mqtt5.Client(client_options=client_options) return client @@ -384,7 +361,7 @@ def mtls_from_path(cert_filepath, pri_key_filepath, **kwargs) -> awscrt.mqtt5.Cl """ _check_required_kwargs(**kwargs) tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls_from_path(cert_filepath, pri_key_filepath) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.CERTIFICATE_FILES, **kwargs) def mtls_from_bytes(cert_bytes, pri_key_bytes, **kwargs) -> awscrt.mqtt5.Client: @@ -402,7 +379,7 @@ def mtls_from_bytes(cert_bytes, pri_key_bytes, **kwargs) -> awscrt.mqtt5.Client: """ _check_required_kwargs(**kwargs) tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls(cert_bytes, pri_key_bytes) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.CERTIFICATE_FILES, **kwargs) def mtls_with_pkcs11(*, @@ -458,7 +435,7 @@ def mtls_with_pkcs11(*, private_key_label=private_key_label, cert_file_path=cert_filepath, cert_file_contents=cert_bytes) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.PKCS11, **kwargs) def mtls_with_pkcs12(*, @@ -484,7 +461,7 @@ def mtls_with_pkcs12(*, tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls_pkcs12( pkcs12_filepath=pkcs12_filepath, pkcs12_password=pkcs12_password) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.PKCS12_FILE, **kwargs) def mtls_with_windows_cert_store_path(*, @@ -507,7 +484,7 @@ def mtls_with_windows_cert_store_path(*, _check_required_kwargs(**kwargs) tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls_windows_cert_store_path(cert_store_path) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.WINDOWS_CERT_STORE, **kwargs) def websockets_with_default_aws_signing( diff --git a/awsiot/mqtt_connection_builder.py b/awsiot/mqtt_connection_builder.py index 75144563..75bfb4ab 100644 --- a/awsiot/mqtt_connection_builder.py +++ b/awsiot/mqtt_connection_builder.py @@ -113,8 +113,8 @@ **cipher_pref** (:class:`awscrt.io.TlsCipherPref`): Cipher preference to use for TLS connection. Default is `TlsCipherPref.DEFAULT`. - **enable_metrics_collection** (`bool`): Whether to send the SDK version number in the CONNECT packet. - Default is True. + **enable_metrics_collection** (`bool`): Controls whether SDK metrics are included in the CONNECT packet. + Default is True (metrics enabled). **http_proxy_options** (:class: 'awscrt.http.HttpProxyOptions'): HTTP proxy options to use """ @@ -127,6 +127,8 @@ import awscrt.mqtt import urllib.parse +from awsiot.iot_metrics import CertificateSource, build_sdk_metrics + def _check_required_kwargs(**kwargs): for required in ['endpoint', 'client_id']: @@ -148,35 +150,6 @@ def _get(kwargs, name, default=None): return val -_metrics_str = None - - -def _get_metrics_str(current_username=""): - global _metrics_str - - username_has_query = False - if current_username.find("?") != -1: - username_has_query = True - - if _metrics_str is None: - try: - import importlib.metadata - try: - version = importlib.metadata.version("awsiotsdk") - _metrics_str = "SDK=PythonV2&Version={}".format(version) - except importlib.metadata.PackageNotFoundError: - _metrics_str = "SDK=PythonV2&Version=dev" - except BaseException: - _metrics_str = "" - - if not _metrics_str == "": - if username_has_query: - return "&" + _metrics_str - else: - return "?" + _metrics_str - else: - return "" - def _builder( tls_ctx_options, @@ -184,6 +157,7 @@ def _builder( websocket_handshake_transform=None, use_custom_authorizer=False, cipher_pref=awscrt.io.TlsCipherPref.DEFAULT, + certificate_source=None, **kwargs): assert isinstance(cipher_pref, awscrt.io.TlsCipherPref) @@ -225,12 +199,15 @@ def _builder( _get(kwargs, 'tcp_keep_alive_max_probes', _get(kwargs, 'tcp_keepalive_max_probes', 0)) username = _get(kwargs, 'username', '') - if _get(kwargs, 'enable_metrics_collection', True): - username += _get_metrics_str(username) if username == "": username = None + # Set SDK metrics for the CRT layer to embed in the CONNECT packet username + metrics = None + if _get(kwargs, 'enable_metrics_collection', True): + metrics = build_sdk_metrics(certificate_source) + client_bootstrap = _get(kwargs, 'client_bootstrap') if client_bootstrap is None: client_bootstrap = awscrt.io.ClientBootstrap.get_or_create_static_default() @@ -262,6 +239,7 @@ def _builder( on_connection_success=_get(kwargs, 'on_connection_success'), on_connection_failure=_get(kwargs, 'on_connection_failure'), on_connection_closed=_get(kwargs, 'on_connection_closed'), + metrics=metrics, ) @@ -280,7 +258,7 @@ def mtls_from_path(cert_filepath, pri_key_filepath, **kwargs) -> awscrt.mqtt.Con """ _check_required_kwargs(**kwargs) tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls_from_path(cert_filepath, pri_key_filepath) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.CERTIFICATE_FILES, **kwargs) def mtls_from_bytes(cert_bytes, pri_key_bytes, **kwargs) -> awscrt.mqtt.Connection: @@ -298,7 +276,7 @@ def mtls_from_bytes(cert_bytes, pri_key_bytes, **kwargs) -> awscrt.mqtt.Connecti """ _check_required_kwargs(**kwargs) tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls(cert_bytes, pri_key_bytes) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.CERTIFICATE_FILES, **kwargs) def mtls_with_pkcs11(*, @@ -355,7 +333,7 @@ def mtls_with_pkcs11(*, cert_file_path=cert_filepath, cert_file_contents=cert_bytes) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.PKCS11, **kwargs) def mtls_with_pkcs12(*, @@ -381,7 +359,7 @@ def mtls_with_pkcs12(*, tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls_pkcs12( pkcs12_filepath=pkcs12_filepath, pkcs12_password=pkcs12_password) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.PKCS12_FILE, **kwargs) def mtls_with_windows_cert_store_path(*, @@ -405,7 +383,7 @@ def mtls_with_windows_cert_store_path(*, tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls_windows_cert_store_path(cert_store_path) - return _builder(tls_ctx_options, **kwargs) + return _builder(tls_ctx_options, certificate_source=CertificateSource.WINDOWS_CERT_STORE, **kwargs) def websockets_with_default_aws_signing( diff --git a/docsrc/awsiot/iot_metrics.rst b/docsrc/awsiot/iot_metrics.rst new file mode 100644 index 00000000..917aa63a --- /dev/null +++ b/docsrc/awsiot/iot_metrics.rst @@ -0,0 +1,4 @@ +awsiot.iot_metrics +===================== + +.. automodule:: awsiot.iot_metrics diff --git a/docsrc/index.rst b/docsrc/index.rst index 68cb82e1..034e4e7d 100644 --- a/docsrc/index.rst +++ b/docsrc/index.rst @@ -28,6 +28,7 @@ API Reference awsiot/iotidentity awsiot/iotjobs awsiot/iotshadow + awsiot/iot_metrics diff --git a/test/test_get_metrics.py b/test/test_get_metrics.py index 0c0d7d8b..1ea323e6 100644 --- a/test/test_get_metrics.py +++ b/test/test_get_metrics.py @@ -2,126 +2,103 @@ # SPDX-License-Identifier: Apache-2.0. import unittest -from unittest.mock import patch +from awsiot.iot_metrics import ( + CertificateSource, + FeatureId, + SDK_LIBRARY_NAME, + _encode_feature_list, + _get_sdk_version, + build_sdk_metrics, +) -class TestImportlibMetadata(unittest.TestCase): - """Test that importlib.metadata is used instead of pkg_resources""" - def setUp(self): - """Reset the metrics string cache before each test""" - # Reset the cached metrics string in both modules - import awsiot.mqtt5_client_builder - import awsiot.mqtt_connection_builder +class TestFeatureEncoding(unittest.TestCase): - # Reset the global _metrics_str variable - awsiot.mqtt_connection_builder._metrics_str = None - awsiot.mqtt5_client_builder._metrics_str = None + def test_certificate_files(self): + self.assertEqual(_encode_feature_list(CertificateSource.CERTIFICATE_FILES), "I/A") - def test_metrics_string_generation_mqtt_connection_builder(self): - """Test that mqtt_connection_builder uses importlib.metadata for version detection""" - from awsiot import mqtt_connection_builder + def test_pkcs11(self): + self.assertEqual(_encode_feature_list(CertificateSource.PKCS11), "I/B") - # Mock importlib.metadata.version to return a known version - with patch("importlib.metadata.version") as mock_version: - mock_version.return_value = "1.2.3" + def test_windows_cert_store(self): + self.assertEqual(_encode_feature_list(CertificateSource.WINDOWS_CERT_STORE), "I/C") - # Call the function that uses version detection - # We need to access the private function for testing - result = mqtt_connection_builder._get_metrics_str("test_username") + def test_pkcs12(self): + self.assertEqual(_encode_feature_list(CertificateSource.PKCS12_FILE), "I/E") - # Verify that importlib.metadata.version was called - mock_version.assert_called_once_with("awsiotsdk") + def test_none_returns_empty(self): + self.assertEqual(_encode_feature_list(None), "") - # Verify the result contains the expected format - self.assertIn("SDK=PythonV2&Version=1.2.3", result) - def test_metrics_string_generation_mqtt5_client_builder(self): - """Test that mqtt5_client_builder uses importlib.metadata for version detection""" - from awsiot import mqtt5_client_builder +class TestGetSdkVersion(unittest.TestCase): - # Mock importlib.metadata.version to return a known version - with patch("importlib.metadata.version") as mock_version: - mock_version.return_value = "1.2.3" + def test_returns_string(self): + version = _get_sdk_version() + self.assertIsInstance(version, str) + self.assertTrue(len(version) > 0) - # Call the function that uses version detection - # We need to access the private function for testing - result = mqtt5_client_builder._get_metrics_str("test_username") - # Verify that importlib.metadata.version was called - mock_version.assert_called_once_with("awsiotsdk") +class TestBuildSdkMetrics(unittest.TestCase): - # Verify the result contains the expected format - self.assertIn("SDK=PythonV2&Version=1.2.3", result) + def test_with_certificate_source(self): + metrics = build_sdk_metrics(CertificateSource.CERTIFICATE_FILES) - def test_package_not_found_handling_mqtt_connection_builder(self): - """Test that PackageNotFoundError is handled correctly in mqtt_connection_builder""" - import importlib.metadata + self.assertEqual(metrics.library_name, SDK_LIBRARY_NAME) + entries = {e.key: e.value for e in metrics.metadata_entries} + self.assertIn("IoTSDKVersion", entries) + self.assertEqual(entries["IoTSDKFeature"], "I/A") + self.assertIn("IoTSDKMetricsVersion", entries) - from awsiot import mqtt_connection_builder + def test_without_certificate_source(self): + metrics = build_sdk_metrics(None) - # Mock importlib.metadata.version to raise PackageNotFoundError - with patch("importlib.metadata.version") as mock_version: - mock_version.side_effect = importlib.metadata.PackageNotFoundError("Package not found") + self.assertEqual(metrics.library_name, SDK_LIBRARY_NAME) + entries = {e.key: e.value for e in metrics.metadata_entries} + self.assertIn("IoTSDKVersion", entries) + self.assertNotIn("IoTSDKFeature", entries) + self.assertNotIn("IoTSDKMetricsVersion", entries) - # Call the function that uses version detection - result = mqtt_connection_builder._get_metrics_str("test_username") + def test_pkcs11_feature(self): + metrics = build_sdk_metrics(CertificateSource.PKCS11) + entries = {e.key: e.value for e in metrics.metadata_entries} + self.assertEqual(entries["IoTSDKFeature"], "I/B") - # Verify that the fallback version is used - self.assertIn("SDK=PythonV2&Version=dev", result) + def test_pkcs12_feature(self): + metrics = build_sdk_metrics(CertificateSource.PKCS12_FILE) + entries = {e.key: e.value for e in metrics.metadata_entries} + self.assertEqual(entries["IoTSDKFeature"], "I/E") - def test_package_not_found_handling_mqtt5_client_builder(self): - """Test that PackageNotFoundError is handled correctly in mqtt5_client_builder""" - import importlib.metadata + def test_windows_cert_store_feature(self): + metrics = build_sdk_metrics(CertificateSource.WINDOWS_CERT_STORE) + entries = {e.key: e.value for e in metrics.metadata_entries} + self.assertEqual(entries["IoTSDKFeature"], "I/C") - from awsiot import mqtt5_client_builder + def test_library_name(self): + metrics = build_sdk_metrics(CertificateSource.CERTIFICATE_FILES) + self.assertEqual(metrics.library_name, "IoTDeviceSDK/Python") - # Mock importlib.metadata.version to raise PackageNotFoundError - with patch("importlib.metadata.version") as mock_version: - mock_version.side_effect = importlib.metadata.PackageNotFoundError("Package not found") + def test_metrics_version_only_set_with_features(self): + metrics_with = build_sdk_metrics(CertificateSource.CERTIFICATE_FILES) + metrics_without = build_sdk_metrics(None) - # Call the function that uses version detection - result = mqtt5_client_builder._get_metrics_str("test_username") + entries_with = {e.key for e in metrics_with.metadata_entries} + entries_without = {e.key for e in metrics_without.metadata_entries} - # Verify that the fallback version is used - self.assertIn("SDK=PythonV2&Version=dev", result) + self.assertIn("IoTSDKMetricsVersion", entries_with) + self.assertNotIn("IoTSDKMetricsVersion", entries_without) - def test_general_exception_handling_mqtt_connection_builder(self): - """Test that general exceptions are handled correctly in mqtt_connection_builder""" - from awsiot import mqtt_connection_builder - # Mock importlib.metadata.version to raise a general exception - with patch("importlib.metadata.version") as mock_version: - mock_version.side_effect = Exception("Some other error") +class TestEnumValues(unittest.TestCase): - # Call the function that uses version detection - result = mqtt_connection_builder._get_metrics_str("test_username") + def test_feature_id(self): + self.assertEqual(FeatureId.CERTIFICATE_SOURCE.value, "I") - # Verify that empty string is returned on general exception - self.assertEqual(result, "") - - def test_general_exception_handling_mqtt5_client_builder(self): - """Test that general exceptions are handled correctly in mqtt5_client_builder""" - from awsiot import mqtt5_client_builder - - # Mock importlib.metadata.version to raise a general exception - with patch("importlib.metadata.version") as mock_version: - mock_version.side_effect = Exception("Some other error") - - # Call the function that uses version detection - result = mqtt5_client_builder._get_metrics_str("test_username") - - # Verify that empty string is returned on general exception - self.assertEqual(result, "") - - def test_no_pkg_resources_import(self): - """Test that pkg_resources is not imported in the modified files""" - import awsiot.mqtt5_client_builder - import awsiot.mqtt_connection_builder - - # Check that pkg_resources is not in the module's globals - self.assertNotIn("pkg_resources", awsiot.mqtt_connection_builder.__dict__) - self.assertNotIn("pkg_resources", awsiot.mqtt5_client_builder.__dict__) + def test_certificate_source_values(self): + self.assertEqual(CertificateSource.CERTIFICATE_FILES.value, "A") + self.assertEqual(CertificateSource.PKCS11.value, "B") + self.assertEqual(CertificateSource.WINDOWS_CERT_STORE.value, "C") + self.assertEqual(CertificateSource.PKCS12_FILE.value, "E") if __name__ == "__main__":