diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/test_base.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/test_base.py index 0d81fcb4f53..8593d0f38c1 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/test_base.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/test_base.py @@ -15,12 +15,18 @@ import logging import unittest from contextlib import contextmanager -from typing import Tuple +from typing import Optional, Tuple, Union from opentelemetry import metrics as metrics_api from opentelemetry import trace as trace_api from opentelemetry.sdk.metrics import MeterProvider -from opentelemetry.sdk.metrics.export import InMemoryMetricReader, MetricReader +from opentelemetry.sdk.metrics._internal.point import Metric +from opentelemetry.sdk.metrics.export import ( + HistogramDataPoint, + InMemoryMetricReader, + MetricReader, + NumberDataPoint, +) from opentelemetry.sdk.trace import TracerProvider, export from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, @@ -85,6 +91,63 @@ def sorted_spans(self, spans): # pylint: disable=R0201 reverse=True, ) + def sorted_metrics(self, metrics): # pylint: disable=R0201 + """ + Sorts metrics by metric name. + """ + return sorted( + metrics, + key=lambda m: m.name, + ) + + def get_sorted_metrics(self): + resource_metrics = ( + self.memory_metrics_reader.get_metrics_data().resource_metrics + ) + + all_metrics = [] + for metrics in resource_metrics: + for scope_metrics in metrics.scope_metrics: + all_metrics.extend(scope_metrics.metrics) + + return self.sorted_metrics(all_metrics) + + def assert_metric_expected( + self, + metric: Metric, + expected_value: Union[int, float], + expected_attributes: dict, + est_delta: Optional[float] = None, + ): + data_point = next(iter(metric.data.data_points)) + + if isinstance(data_point, HistogramDataPoint): + self.assertEqual( + data_point.count, + 1, + ) + if est_delta is None: + self.assertEqual( + data_point.sum, + expected_value, + ) + else: + self.assertAlmostEqual( + data_point.sum, + expected_value, + delta=est_delta, + ) + elif isinstance(data_point, NumberDataPoint): + self.assertEqual( + data_point.value, + expected_value, + ) + + self.assertDictEqual( + expected_attributes, + dict(data_point.attributes), + ) + @staticmethod def create_tracer_provider(**kwargs): """Helper to create a configured tracer provider.