Convert list[object] to List[object] to support python 3.8

Author: thpierce

contract-tests/images/mock-collector/mock_collector_client.py

@@ -3,7 +3,7 @@
 from datetime import datetime, timedelta
 from logging import Logger, getLogger
 from time import sleep
-from typing import Callable, TypeVar
+from typing import Callable, TypeVar, List
 
 from google.protobuf.internal.containers import RepeatedScalarFieldContainer
 from grpc import Channel, insecure_channel
@@ -62,46 +62,46 @@ def clear_signals(self) -> None:
         """Clear all the signals in the backend collector"""
         self.client.clear(ClearRequest())
 
-    def get_traces(self) -> list[ResourceScopeSpan]:
+    def get_traces(self) -> List[ResourceScopeSpan]:
         """Get all traces that are currently stored in the collector
 
         Returns:
             List of `ResourceScopeSpan` which is essentially a flat list containing all the spans and their related
             scope and resources.
         """
 
-        def get_export() -> list[ExportTraceServiceRequest]:
+        def get_export() -> List[ExportTraceServiceRequest]:
             response: GetTracesResponse = self.client.get_traces(GetTracesRequest())
             serialized_traces: RepeatedScalarFieldContainer[bytes] = response.traces
             return list(map(ExportTraceServiceRequest.FromString, serialized_traces))
 
-        def wait_condition(exported: list[ExportTraceServiceRequest], current: list[ExportTraceServiceRequest]) -> bool:
+        def wait_condition(exported: List[ExportTraceServiceRequest], current: List[ExportTraceServiceRequest]) -> bool:
             return 0 < len(exported) == len(current)
 
-        exported_traces: list[ExportTraceServiceRequest] = _wait_for_content(get_export, wait_condition)
-        spans: list[ResourceScopeSpan] = []
+        exported_traces: List[ExportTraceServiceRequest] = _wait_for_content(get_export, wait_condition)
+        spans: List[ResourceScopeSpan] = []
         for exported_trace in exported_traces:
             for resource_span in exported_trace.resource_spans:
                 for scope_span in resource_span.scope_spans:
                     for span in scope_span.spans:
                         spans.append(ResourceScopeSpan(resource_span, scope_span, span))
         return spans
 
-    def get_metrics(self, present_metrics: set[str]) -> list[ResourceScopeMetric]:
+    def get_metrics(self, present_metrics: set[str]) -> List[ResourceScopeMetric]:
         """Get all metrics that are currently stored in the mock collector.
 
         Returns:
              List of `ResourceScopeMetric` which is a flat list containing all metrics and their related scope and
              resources.
         """
 
-        def get_export() -> list[ExportMetricsServiceRequest]:
+        def get_export() -> List[ExportMetricsServiceRequest]:
             response: GetMetricsResponse = self.client.get_metrics(GetMetricsRequest())
             serialized_metrics: RepeatedScalarFieldContainer[bytes] = response.metrics
             return list(map(ExportMetricsServiceRequest.FromString, serialized_metrics))
 
         def wait_condition(
-            exported: list[ExportMetricsServiceRequest], current: list[ExportMetricsServiceRequest]
+            exported: List[ExportMetricsServiceRequest], current: List[ExportMetricsServiceRequest]
         ) -> bool:
             received_metrics: set[str] = set()
             for exported_metric in current:
@@ -111,8 +111,8 @@ def wait_condition(
                             received_metrics.add(metric.name)
             return 0 < len(exported) == len(current) and present_metrics.issubset(received_metrics)
 
-        exported_metrics: list[ExportMetricsServiceRequest] = _wait_for_content(get_export, wait_condition)
-        metrics: list[ResourceScopeMetric] = []
+        exported_metrics: List[ExportMetricsServiceRequest] = _wait_for_content(get_export, wait_condition)
+        metrics: List[ResourceScopeMetric] = []
         for exported_metric in exported_metrics:
             for resource_metric in exported_metric.resource_metrics:
                 for scope_metric in resource_metric.scope_metrics:
@@ -121,14 +121,14 @@ def wait_condition(
         return metrics
 
 
-def _wait_for_content(get_export: Callable[[], list[T]], wait_condition: Callable[[list[T], list[T]], bool]) -> list[T]:
+def _wait_for_content(get_export: Callable[[], List[T]], wait_condition: Callable[[list[T], List[T]], bool]) -> List[T]:
     # Verify that there is no more data to be received
     deadline: datetime = datetime.now() + _TIMEOUT_DELAY
-    exported: list[T] = []
+    exported: List[T] = []
 
     while deadline > datetime.now():
         try:
-            current_exported: list[T] = get_export()
+            current_exported: List[T] = get_export()
             if wait_condition(exported, current_exported):
                 return current_exported
             exported = current_exported

contract-tests/images/mock-collector/mock_collector_metrics_service.py

@@ -1,6 +1,7 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 from queue import Queue
+from typing import List
 
 from grpc import ServicerContext
 from typing_extensions import override
@@ -15,7 +16,7 @@
 class MockCollectorMetricsService(MetricsServiceServicer):
     _export_requests: Queue = Queue(maxsize=-1)
 
-    def get_requests(self) -> list[ExportMetricsServiceRequest]:
+    def get_requests(self) -> List[ExportMetricsServiceRequest]:
         with self._export_requests.mutex:
             return list(self._export_requests.queue)
 

contract-tests/images/mock-collector/mock_collector_service.py

@@ -1,5 +1,7 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
+from typing import List
+
 from grpc import ServicerContext
 from mock_collector_metrics_service import MockCollectorMetricsService
 from mock_collector_service_pb2 import (
@@ -37,14 +39,14 @@ def clear(self, request: ClearRequest, context: ServicerContext) -> ClearRespons
 
     @override
     def get_traces(self, request: GetTracesRequest, context: ServicerContext) -> GetTracesResponse:
-        trace_requests: list[ExportTraceServiceRequest] = self.trace_collector.get_requests()
-        traces: list[bytes] = list(map(ExportTraceServiceRequest.SerializeToString, trace_requests))
+        trace_requests: List[ExportTraceServiceRequest] = self.trace_collector.get_requests()
+        traces: List[bytes] = list(map(ExportTraceServiceRequest.SerializeToString, trace_requests))
         response: GetTracesResponse = GetTracesResponse(traces=traces)
         return response
 
     @override
     def get_metrics(self, request: GetMetricsRequest, context: ServicerContext) -> GetMetricsResponse:
-        metric_requests: list[ExportMetricsServiceRequest] = self.metrics_collector.get_requests()
-        metrics: list[bytes] = list(map(ExportTraceServiceRequest.SerializeToString, metric_requests))
+        metric_requests: List[ExportMetricsServiceRequest] = self.metrics_collector.get_requests()
+        metrics: List[bytes] = list(map(ExportTraceServiceRequest.SerializeToString, metric_requests))
         response: GetMetricsResponse = GetMetricsResponse(metrics=metrics)
         return response

contract-tests/images/mock-collector/mock_collector_trace_service.py

@@ -1,6 +1,7 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 from queue import Queue
+from typing import List
 
 from grpc import ServicerContext
 from typing_extensions import override
@@ -15,7 +16,7 @@
 class MockCollectorTraceService(TraceServiceServicer):
     _export_requests: Queue = Queue(maxsize=-1)
 
-    def get_requests(self) -> list[ExportTraceServiceRequest]:
+    def get_requests(self) -> List[ExportTraceServiceRequest]:
         with self._export_requests.mutex:
             return list(self._export_requests.queue)
 

contract-tests/tests/test/amazon/base/contract_test_base.py

@@ -1,5 +1,6 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
+from typing import List
 from unittest import TestCase
 
 from docker import DockerClient
@@ -86,7 +87,7 @@ def get_application_port(self) -> int:
     def get_application_extra_environment_variables(self) -> dict[str, str]:
         return {}
 
-    def get_application_network_aliases(self) -> list[str]:
+    def get_application_network_aliases(self) -> List[str]:
         return []
 
     def get_application_image_name(self) -> str:

contract-tests/tests/test/amazon/requests/requests_test.py

@@ -1,5 +1,7 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
+from typing import List
+
 from mock_collector_client import ResourceScopeMetric, ResourceScopeSpan
 from requests import Response, request
 from typing_extensions import override
@@ -27,7 +29,7 @@ def get_application_image_name(self) -> str:
         return "aws-appsignals-tests-requests-app"
 
     @override
-    def get_application_network_aliases(self) -> list[str]:
+    def get_application_network_aliases(self) -> List[str]:
         """
         This will be the target hostname of the clients making http requests in the application image, so that they
         don't use localhost.
@@ -70,21 +72,21 @@ def do_test_requests(
 
         self.assertEqual(status_code, response.status_code)
 
-        resource_scope_spans: list[ResourceScopeSpan] = self._mock_collector_client.get_traces()
+        resource_scope_spans: List[ResourceScopeSpan] = self._mock_collector_client.get_traces()
         self._assert_aws_span_attributes(resource_scope_spans, method, path)
         self._assert_semantic_conventions_span_attributes(resource_scope_spans, method, path, status_code)
 
-        metrics: list[ResourceScopeMetric] = self._mock_collector_client.get_metrics(
+        metrics: List[ResourceScopeMetric] = self._mock_collector_client.get_metrics(
             {LATENCY_METRIC, ERROR_METRIC, FAULT_METRIC}
         )
         self._assert_metric_attributes(metrics, method, path, LATENCY_METRIC, 5000)
         self._assert_metric_attributes(metrics, method, path, ERROR_METRIC, expected_error)
         self._assert_metric_attributes(metrics, method, path, FAULT_METRIC, expected_fault)
 
     def _assert_aws_span_attributes(
-        self, resource_scope_spans: list[ResourceScopeSpan], method: str, path: str
+        self, resource_scope_spans: List[ResourceScopeSpan], method: str, path: str
     ) -> None:
-        target_spans: list[Span] = []
+        target_spans: List[Span] = []
         for resource_scope_span in resource_scope_spans:
             # pylint: disable=no-member
             if resource_scope_span.span.kind == Span.SPAN_KIND_CLIENT:
@@ -93,7 +95,7 @@ def _assert_aws_span_attributes(
         self.assertEqual(len(target_spans), 1)
         self._assert_aws_attributes(target_spans[0].attributes, method, path)
 
-    def _assert_aws_attributes(self, attributes_list: list[KeyValue], method: str, endpoint: str) -> None:
+    def _assert_aws_attributes(self, attributes_list: List[KeyValue], method: str, endpoint: str) -> None:
         attributes_dict: dict[str, AnyValue] = self._get_attributes_dict(attributes_list)
         self._assert_str_attribute(attributes_dict, AWS_LOCAL_SERVICE, self.get_application_otel_service_name())
         # InternalOperation as OTEL does not instrument the basic server we are using, so the client span is a local
@@ -106,7 +108,7 @@ def _assert_aws_attributes(self, attributes_list: list[KeyValue], method: str, e
         # See comment above AWS_LOCAL_OPERATION
         self._assert_str_attribute(attributes_dict, AWS_SPAN_KIND, "LOCAL_ROOT")
 
-    def _get_attributes_dict(self, attributes_list: list[KeyValue]) -> dict[str, AnyValue]:
+    def _get_attributes_dict(self, attributes_list: List[KeyValue]) -> dict[str, AnyValue]:
         attributes_dict: dict[str, AnyValue] = {}
         for attribute in attributes_list:
             key: str = attribute.key
@@ -129,9 +131,9 @@ def _assert_int_attribute(self, attributes_dict: dict[str, AnyValue], key: str,
         self.assertEqual(expected_value, actual_value.int_value)
 
     def _assert_semantic_conventions_span_attributes(
-        self, resource_scope_spans: list[ResourceScopeSpan], method: str, path: str, status_code: int
+        self, resource_scope_spans: List[ResourceScopeSpan], method: str, path: str, status_code: int
     ) -> None:
-        target_spans: list[Span] = []
+        target_spans: List[Span] = []
         for resource_scope_span in resource_scope_spans:
             # pylint: disable=no-member
             if resource_scope_span.span.kind == Span.SPAN_KIND_CLIENT:
@@ -142,7 +144,7 @@ def _assert_semantic_conventions_span_attributes(
         self._assert_semantic_conventions_attributes(target_spans[0].attributes, method, path, status_code)
 
     def _assert_semantic_conventions_attributes(
-        self, attributes_list: list[KeyValue], method: str, endpoint: str, status_code: int
+        self, attributes_list: List[KeyValue], method: str, endpoint: str, status_code: int
     ) -> None:
         attributes_dict: dict[str, AnyValue] = self._get_attributes_dict(attributes_list)
         # TODO: requests instrumentation is not populating net peer attributes
@@ -156,20 +158,20 @@ def _assert_semantic_conventions_attributes(
 
     def _assert_metric_attributes(
         self,
-        resource_scope_metrics: list[ResourceScopeMetric],
+        resource_scope_metrics: List[ResourceScopeMetric],
         method: str,
         path: str,
         metric_name: str,
         expected_sum: float,
     ) -> None:
-        target_metrics: list[Metric] = []
+        target_metrics: List[Metric] = []
         for resource_scope_metric in resource_scope_metrics:
             if resource_scope_metric.metric.name == metric_name:
                 target_metrics.append(resource_scope_metric.metric)
 
         self.assertEqual(len(target_metrics), 1)
         target_metric: Metric = target_metrics[0]
-        dp_list: list[ExponentialHistogramDataPoint] = target_metric.exponential_histogram.data_points
+        dp_list: List[ExponentialHistogramDataPoint] = target_metric.exponential_histogram.data_points
 
         self.assertEqual(len(dp_list), 2)
         dp: ExponentialHistogramDataPoint = dp_list[0]