Add meter provider and in memory reader to TestBase
import logging
import unittest
from contextlib import contextmanager
+from typing import Tuple
+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.trace import TracerProvider, export
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
-from opentelemetry.test.globals_test import reset_trace_globals
+from opentelemetry.test.globals_test import (
+ reset_metrics_globals,
+ reset_trace_globals,
+)
class TestBase(unittest.TestCase):
@@ -36,6 +43,11 @@ class TestBase(unittest.TestCase):
reset_trace_globals()
trace_api.set_tracer_provider(cls.tracer_provider)
+ result = cls.create_meter_provider()
+ cls.meter_provider, cls.memory_metrics_reader = result
+ reset_metrics_globals()
+ metrics_api.set_meter_provider(cls.meter_provider)
+
@classmethod
def tearDownClass(cls):
# This is done because set_tracer_provider cannot override the
@@ -92,6 +104,21 @@ class TestBase(unittest.TestCase):
return tracer_provider, memory_exporter
+ @staticmethod
+ def create_meter_provider(**kwargs) -> Tuple[MeterProvider, MetricReader]:
+ """Helper to create a configured meter provider
+ Creates a `MeterProvider` and an `InMemoryMetricReader`.
+ Returns:
+ A tuple with the meter provider in the first element and the
+ in-memory metrics exporter in the second
+ """
+ memory_reader = InMemoryMetricReader()
+ metric_readers = kwargs.get("metric_readers", [])
+ metric_readers.append(memory_reader)
+ kwargs["metric_readers"] = metric_readers
+ meter_provider = MeterProvider(**kwargs)
+ return meter_provider, memory_reader
+
@staticmethod
@contextmanager
def disable_logging(highest_level=logging.CRITICAL):
