some fixes for linting and unit tests

Author: srprash

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/otlp_udp_exporter.py

@@ -1,7 +1,8 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 import socket
-from typing import Dict, Optional, Sequence
+from logging import Logger, getLogger
+from typing import Dict, Optional, Sequence, Tuple
 
 from typing_extensions import override
 
@@ -15,17 +16,16 @@
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
 
-from logging import getLogger, Logger
-
 DEFAULT_ENDPOINT = "127.0.0.1:2000"
 PROTOCOL_HEADER = '{"format":"json","version":1}\n'
 PROTOCOL_DELIMITER = "\n"
 
 _logger: Logger = getLogger(__name__)
 
+
 class UdpExporter:
     def __init__(self, endpoint: Optional[str] = None):
-        self._endpoint = endpoint or DEFAULT_ENDPOINT # TODO: read from some env var??
+        self._endpoint = endpoint or DEFAULT_ENDPOINT  # TODO: read from some env var??
         self._host, self._port = self._parse_endpoint(self._endpoint)
         self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self._socket.setblocking(False)
@@ -44,7 +44,7 @@ def shutdown(self):
         self._socket.close()
 
     # pylint: disable=no-self-use
-    def _parse_endpoint(self, endpoint: str) -> tuple[str, int]:
+    def _parse_endpoint(self, endpoint: str) -> Tuple[str, int]:
         try:
             vals = endpoint.split(":")
             host = vals[0]
@@ -79,7 +79,9 @@ def export(
         self._udp_exporter.send_data(data=serialized_data, signal_format="OTEL_V1_METRICS")  # TODO: Convert to constant
         return MetricExportResult.SUCCESS  # TODO: send appropriate status back. Need to??
 
+    # pylint: disable=no-self-use
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
+        # TODO: implement force flush
         return True
 
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
@@ -96,8 +98,10 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
         self._udp_exporter.send_data(data=serialized_data, signal_format="OTEL_V1_TRACES")  # TODO: Convert to constant
         return SpanExportResult.SUCCESS  # TODO: send appropriate status back. Need to??
 
+    # pylint: disable=no-self-use
     @override
     def force_flush(self, timeout_millis: int = 30000) -> bool:
+        # TODO: implement force flush
         return True
 
     @override

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/test_aws_opentelementry_configurator.py

@@ -19,7 +19,7 @@
 )
 from amazon.opentelemetry.distro.aws_opentelemetry_distro import AwsOpenTelemetryDistro
 from amazon.opentelemetry.distro.aws_span_metrics_processor import AwsSpanMetricsProcessor
-from amazon.opentelemetry.distro.otlp_udp_exporter import OtlpUdpSpanExporter
+from amazon.opentelemetry.distro.otlp_udp_exporter import OtlpUdpMetricExporter, OtlpUdpSpanExporter
 from amazon.opentelemetry.distro.sampler._aws_xray_sampling_client import _AwsXRaySamplingClient
 from amazon.opentelemetry.distro.sampler.aws_xray_remote_sampler import AwsXRayRemoteSampler
 from opentelemetry.environment_variables import OTEL_LOGS_EXPORTER, OTEL_METRICS_EXPORTER, OTEL_TRACES_EXPORTER
@@ -301,8 +301,9 @@ def test_application_signals_exporter_provider(self):
         # When in Lambda, exporter should be UDP.
         os.environ.setdefault("AWS_LAMBDA_FUNCTION_NAME", "myLambdaFunc")
         exporter: MetricExporter = ApplicationSignalsExporterProvider().create_exporter()
-        self.assertIsInstance(exporter, OtlpUdpSpanExporter)
+        self.assertIsInstance(exporter, OtlpUdpMetricExporter)
         self.assertEqual("127.0.0.1:2000", exporter._udp_exporter._endpoint)
+        os.environ.pop("AWS_LAMBDA_FUNCTION_NAME", None)
 
 
 def validate_distro_environ():

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/test_otlp_udp_exporter.py

@@ -46,7 +46,7 @@ def test_udp_exporter_init_invalid_endpoint(self, mock_socket):
     def test_send_data(self, mock_socket):
         mock_socket_instance = mock_socket.return_value
         exporter = UdpExporter()
-        exporter.send_data('encoded_data', "signal")
+        exporter.send_data("encoded_data", "signal")
         expected_message = PROTOCOL_HEADER + '{"format":"signal","data":encoded_data}'
         mock_socket_instance.sendto.assert_called_once_with(expected_message.encode("utf-8"), ("127.0.0.1", 2000))