# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0

# type: ignore
# pylint: skip-file
from __future__ import annotations

import logging
import logging.config
from collections.abc import Iterable, Sequence
from logging import WARNING, getLogger
from os import environ
from unittest import TestCase, mock
from unittest.mock import Mock, patch

from pytest import raises

from opentelemetry import trace
from opentelemetry.context import Context
from opentelemetry.environment_variables import OTEL_PYTHON_ID_GENERATOR
from opentelemetry.sdk._configuration import (
    _EXPORTER_OTLP,
    _EXPORTER_OTLP_PROTO_GRPC,
    _EXPORTER_OTLP_PROTO_HTTP,
    _get_exporter_names,
    _get_id_generator,
    _get_logger_configurator,
    _get_meter_configurator,
    _get_sampler,
    _get_tracer_configurator,
    _import_config_components,
    _import_exporters,
    _import_id_generator,
    _import_logger_configurator,
    _import_meter_configurator,
    _import_sampler,
    _import_tracer_configurator,
    _init_logging,
    _init_metrics,
    _init_tracing,
    _initialize_components,
    _OTelSDKConfigurator,
)
from opentelemetry.sdk._logs import LoggingHandler, LogRecordProcessor
from opentelemetry.sdk._logs._internal import _RuleBasedLoggerConfigurator
from opentelemetry.sdk._logs._internal.export import LogRecordExporter
from opentelemetry.sdk._logs.export import (
    ConsoleLogRecordExporter,
    SimpleLogRecordProcessor,
)
from opentelemetry.sdk.environment_variables import (
    OTEL_PYTHON_LOGGER_CONFIGURATOR,
    OTEL_PYTHON_METER_CONFIGURATOR,
    OTEL_TRACES_SAMPLER,
    OTEL_TRACES_SAMPLER_ARG,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics._internal import (
    _default_meter_configurator,
    _RuleBasedMeterConfigurator,
)
from opentelemetry.sdk.metrics.export import (
    AggregationTemporality,
    ConsoleMetricExporter,
    Metric,
    MetricExporter,
    MetricReader,
)
from opentelemetry.sdk.metrics.view import Aggregation
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import SpanProcessor, _RuleBasedTracerConfigurator
from opentelemetry.sdk.trace.export import (
    ConsoleSpanExporter,
    SimpleSpanProcessor,
)
from opentelemetry.sdk.trace.id_generator import IdGenerator, RandomIdGenerator
from opentelemetry.sdk.trace.sampling import (
    ALWAYS_ON,
    Decision,
    ParentBased,
    Sampler,
    SamplingResult,
    TraceIdRatioBased,
)
from opentelemetry.test.mock_test_classes import IterEntryPoint
from opentelemetry.trace import Link, SpanKind
from opentelemetry.trace.span import TraceState
from opentelemetry.util.types import Attributes


class Provider:
    def __init__(
        self,
        resource=None,
        sampler=None,
        id_generator=None,
        *,
        _tracer_configurator=None,
    ):
        self.sampler = sampler
        self.id_generator = id_generator
        self.processors = []
        self._tracer_configurator = _tracer_configurator
        self.resource = resource or Resource.create({})

    def add_span_processor(self, processor):
        self.processors.append(processor)


class DummyLoggerProvider:
    def __init__(self, resource=None, *, _logger_configurator=None):
        self.resource = resource
        self.processors = []
        self._logger_configurator = _logger_configurator

    def add_log_record_processor(self, processor):
        self.processors.append(processor)

    def get_logger(self, name, *args, **kwargs):
        return DummyLogger(name, self.resource, self.processors)

    def force_flush(self, *args, **kwargs):
        pass


class DummyMeterProvider(MeterProvider):
    pass


class DummyLogger:
    def __init__(self, name, resource, processors):
        self.name = name
        self.resource = resource
        self.processors = processors

    def emit(
        self,
        record=None,
        *,
        timestamp=None,
        observed_timestamp=None,
        context=None,
        severity_number=None,
        severity_text=None,
        body=None,
        attributes=None,
        event_name=None,
    ):
        for processor in self.processors:
            processor.emit(record)


class DummyLogRecordProcessor:
    def __init__(self, exporter):
        self.exporter = exporter

    def emit(self, record):
        self.exporter.export([record])

    def force_flush(self, time):
        pass

    def shutdown(self):
        pass


class Processor:
    def __init__(self, exporter):
        self.exporter = exporter


class DummyMetricReader(MetricReader):
    def __init__(
        self,
        exporter: MetricExporter,
        preferred_temporality: dict[type, AggregationTemporality] | None = None,
        preferred_aggregation: dict[type, Aggregation] | None = None,
        export_interval_millis: float | None = None,
        export_timeout_millis: float | None = None,
    ) -> None:
        super().__init__(
            preferred_temporality=preferred_temporality,
            preferred_aggregation=preferred_aggregation,
        )
        self.exporter = exporter

    def _receive_metrics(
        self,
        metrics: Iterable[Metric],
        timeout_millis: float = 10_000,
        **kwargs,
    ) -> None:
        self.exporter.export(None)

    def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
        return True


# MetricReader that can be configured as a pull exporter
class DummyMetricReaderPullExporter(MetricReader):
    def _receive_metrics(
        self,
        metrics: Iterable[Metric],
        timeout_millis: float = 10_000,
        **kwargs,
    ) -> None:
        pass

    def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
        return True


class DummyOTLPMetricExporter:
    def __init__(self, compression: str | None = None, *args, **kwargs):
        self.export_called = False
        self.compression = compression

    def export(self, batch):
        self.export_called = True

    def shutdown(self):
        pass


class Exporter:
    def __init__(self):
        tracer_provider = trace.get_tracer_provider()
        self.service_name = (
            tracer_provider.resource.attributes[SERVICE_NAME]
            if getattr(tracer_provider, "resource", None)
            else Resource.create().attributes.get(SERVICE_NAME)
        )

    def shutdown(self):
        pass


class OTLPSpanExporter:
    def __init__(self, compression: str | None = None, *args, **kwargs):
        self.compression = compression


class DummyOTLPLogExporter(LogRecordExporter):
    def __init__(self, compression: str | None = None, *args, **kwargs):
        self.export_called = False
        self.compression = compression

    def export(self, batch):
        self.export_called = True

    def shutdown(self):
        pass

    def force_flush(self, timeout_millis: int = 10_000) -> bool:
        return True


class CustomSampler(Sampler):
    def __init__(self) -> None:
        pass

    def get_description(self) -> str:
        return "CustomSampler"

    def should_sample(
        self,
        parent_context: Context | None,
        trace_id: int,
        name: str,
        kind: SpanKind = None,
        attributes: Attributes = None,
        links: Sequence[Link] = None,
        trace_state: TraceState = None,
    ) -> SamplingResult:
        return SamplingResult(
            Decision.RECORD_AND_SAMPLE,
            None,
            None,
        )


class CustomRatioSampler(TraceIdRatioBased):
    def __init__(self, ratio):
        if not isinstance(ratio, float):
            raise ValueError("CustomRatioSampler ratio argument is not a float.")
        self.ratio = ratio
        super().__init__(ratio)

    def get_description(self) -> str:
        return "CustomSampler"

    def should_sample(
        self,
        parent_context: Context | None,
        trace_id: int,
        name: str,
        kind: SpanKind | None = None,
        attributes: Attributes = None,
        links: Sequence[Link] | None = None,
        trace_state: TraceState | None = None,
    ) -> SamplingResult:
        return SamplingResult(
            Decision.RECORD_AND_SAMPLE,
            None,
            None,
        )


class CustomSamplerFactory:
    @staticmethod
    def get_custom_sampler(unused_sampler_arg):
        return CustomSampler()

    @staticmethod
    def get_custom_ratio_sampler(sampler_arg):
        return CustomRatioSampler(float(sampler_arg))

    @staticmethod
    def empty_get_custom_sampler(sampler_arg):
        return


class CustomIdGenerator(IdGenerator):
    def generate_span_id(self):
        pass

    def generate_trace_id(self):
        pass

    def is_trace_id_random(self):
        return False


class TestTraceInit(TestCase):
    def setUp(self):
        super()
        self.get_provider_patcher = patch("opentelemetry.sdk._configuration.TracerProvider", Provider)
        self.get_processor_patcher = patch("opentelemetry.sdk._configuration.BatchSpanProcessor", Processor)
        self.set_provider_patcher = patch("opentelemetry.sdk._configuration.set_tracer_provider")

        self.get_provider_mock = self.get_provider_patcher.start()
        self.get_processor_mock = self.get_processor_patcher.start()
        self.set_provider_mock = self.set_provider_patcher.start()

    def tearDown(self):
        super()
        self.get_provider_patcher.stop()
        self.get_processor_patcher.stop()
        self.set_provider_patcher.stop()

    # pylint: disable=protected-access
    @patch.dict(environ, {"OTEL_RESOURCE_ATTRIBUTES": "service.name=my-test-service"})
    def test_trace_init_default(self):
        auto_resource = Resource.create(
            {
                "telemetry.auto.version": "test-version",
            }
        )
        _init_tracing(
            {"zipkin": Exporter},
            id_generator=RandomIdGenerator(),
            resource=auto_resource,
        )

        self.assertEqual(self.set_provider_mock.call_count, 1)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider, Provider)
        self.assertIsInstance(provider.id_generator, RandomIdGenerator)
        self.assertEqual(len(provider.processors), 1)
        self.assertIsInstance(provider.processors[0], Processor)
        self.assertIsInstance(provider.processors[0].exporter, Exporter)
        self.assertEqual(provider.processors[0].exporter.service_name, "my-test-service")
        self.assertEqual(
            provider.resource.attributes.get("telemetry.auto.version"),
            "test-version",
        )

    @patch.dict(
        environ,
        {"OTEL_RESOURCE_ATTRIBUTES": "service.name=my-otlp-test-service"},
    )
    def test_trace_init_otlp(self):
        _init_tracing({"otlp": OTLPSpanExporter}, id_generator=RandomIdGenerator())

        self.assertEqual(self.set_provider_mock.call_count, 1)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider, Provider)
        self.assertIsInstance(provider.id_generator, RandomIdGenerator)
        self.assertEqual(len(provider.processors), 1)
        self.assertIsInstance(provider.processors[0], Processor)
        self.assertIsInstance(provider.processors[0].exporter, OTLPSpanExporter)
        self.assertIsInstance(provider.resource, Resource)
        self.assertEqual(
            provider.resource.attributes.get("service.name"),
            "my-otlp-test-service",
        )

    def test_trace_init_exporter_uses_exporter_args_map(self):
        _init_tracing(
            {"otlp": OTLPSpanExporter},
            id_generator=RandomIdGenerator(),
            exporter_args_map={
                OTLPSpanExporter: {"compression": "gzip"},
                DummyMetricReaderPullExporter: {"compression": "no"},
            },
        )

        provider = self.set_provider_mock.call_args[0][0]
        self.assertEqual(len(provider.processors), 1)
        exporter = provider.processors[0].exporter
        self.assertEqual(exporter.compression, "gzip")

    def test_trace_init_custom_span_processors(self):
        span_processor = mock.Mock(spec=SpanProcessor)
        _init_tracing(
            {"otlp": OTLPSpanExporter},
            id_generator=RandomIdGenerator(),
            span_processors=[span_processor],
            export_span_processor=SimpleSpanProcessor,
        )

        provider = self.set_provider_mock.call_args[0][0]
        self.assertEqual(len(provider.processors), 2)
        self.assertEqual(provider.processors[0], span_processor)
        self.assertTrue(isinstance(provider.processors[1], SimpleSpanProcessor))

    @patch.dict(environ, {OTEL_PYTHON_ID_GENERATOR: "custom_id_generator"})
    @patch("opentelemetry.sdk._configuration.IdGenerator", new=IdGenerator)
    @patch("opentelemetry.sdk._configuration.entry_points")
    def test_trace_init_custom_id_generator(self, mock_entry_points):
        mock_entry_points.configure_mock(return_value=[IterEntryPoint("custom_id_generator", CustomIdGenerator)])

        id_generator_name = _get_id_generator()
        id_generator = _import_id_generator(id_generator_name)
        _init_tracing({}, id_generator=id_generator)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider.id_generator, CustomIdGenerator)

    @patch.dict("os.environ", {OTEL_TRACES_SAMPLER: "non_existent_entry_point"})
    def test_trace_init_custom_sampler_with_env_non_existent_entry_point(self):
        sampler_name = _get_sampler()
        with self.assertLogs(level=WARNING):
            sampler = _import_sampler(sampler_name)
        _init_tracing({}, sampler=sampler)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsNone(provider.sampler)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict("os.environ", {OTEL_TRACES_SAMPLER: "custom_sampler_factory"})
    def test_trace_init_custom_sampler_with_env(self, mock_entry_points):
        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_sampler_factory",
                    CustomSamplerFactory.get_custom_sampler,
                )
            ]
        )

        sampler_name = _get_sampler()
        sampler = _import_sampler(sampler_name)
        _init_tracing({}, sampler=sampler)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider.sampler, CustomSampler)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict("os.environ", {OTEL_TRACES_SAMPLER: "custom_sampler_factory"})
    def test_trace_init_custom_sampler_with_env_bad_factory(self, mock_entry_points):
        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_sampler_factory",
                    CustomSamplerFactory.empty_get_custom_sampler,
                )
            ]
        )

        sampler_name = _get_sampler()
        with self.assertLogs(level=WARNING):
            sampler = _import_sampler(sampler_name)
        _init_tracing({}, sampler=sampler)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsNone(provider.sampler)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict(
        "os.environ",
        {
            OTEL_TRACES_SAMPLER: "custom_sampler_factory",
            OTEL_TRACES_SAMPLER_ARG: "0.5",
        },
    )
    def test_trace_init_custom_sampler_with_env_unused_arg(self, mock_entry_points):
        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_sampler_factory",
                    CustomSamplerFactory.get_custom_sampler,
                )
            ]
        )

        sampler_name = _get_sampler()
        sampler = _import_sampler(sampler_name)
        _init_tracing({}, sampler=sampler)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider.sampler, CustomSampler)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict(
        "os.environ",
        {
            OTEL_TRACES_SAMPLER: "custom_ratio_sampler_factory",
            OTEL_TRACES_SAMPLER_ARG: "0.5",
        },
    )
    def test_trace_init_custom_ratio_sampler_with_env(self, mock_entry_points):
        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_ratio_sampler_factory",
                    CustomSamplerFactory.get_custom_ratio_sampler,
                )
            ]
        )

        sampler_name = _get_sampler()
        sampler = _import_sampler(sampler_name)
        _init_tracing({}, sampler=sampler)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider.sampler, CustomRatioSampler)
        self.assertEqual(provider.sampler.ratio, 0.5)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict(
        "os.environ",
        {
            OTEL_TRACES_SAMPLER: "custom_ratio_sampler_factory",
            OTEL_TRACES_SAMPLER_ARG: "foobar",
        },
    )
    def test_trace_init_custom_ratio_sampler_with_env_bad_arg(self, mock_entry_points):
        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_ratio_sampler_factory",
                    CustomSamplerFactory.get_custom_ratio_sampler,
                )
            ]
        )

        sampler_name = _get_sampler()
        with self.assertLogs(level=WARNING):
            sampler = _import_sampler(sampler_name)
        _init_tracing({}, sampler=sampler)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsNone(provider.sampler)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict(
        "os.environ",
        {
            OTEL_TRACES_SAMPLER: "custom_ratio_sampler_factory",
        },
    )
    def test_trace_init_custom_ratio_sampler_with_env_missing_arg(self, mock_entry_points):
        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_ratio_sampler_factory",
                    CustomSamplerFactory.get_custom_ratio_sampler,
                )
            ]
        )

        sampler_name = _get_sampler()
        with self.assertLogs(level=WARNING):
            sampler = _import_sampler(sampler_name)
        _init_tracing({}, sampler=sampler)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsNone(provider.sampler)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict(
        "os.environ",
        {
            OTEL_TRACES_SAMPLER: "custom_sampler_factory",
            OTEL_TRACES_SAMPLER_ARG: "0.5",
        },
    )
    def test_trace_init_custom_ratio_sampler_with_env_multiple_entry_points(self, mock_entry_points):
        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_sampler_factory",
                    CustomSamplerFactory.get_custom_sampler,
                ),
            ]
        )

        sampler_name = _get_sampler()
        sampler = _import_sampler(sampler_name)
        _init_tracing({}, sampler=sampler)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider.sampler, CustomSampler)

    def verify_default_sampler(self, tracer_provider):
        self.assertIsInstance(tracer_provider.sampler, ParentBased)
        # pylint: disable=protected-access
        self.assertEqual(tracer_provider.sampler._root, ALWAYS_ON)

    @patch.dict(
        "os.environ",
        {"OTEL_PYTHON_TRACER_CONFIGURATOR": "non_existent_entry_point"},
    )
    def test_trace_init_custom_tracer_configurator_with_env_non_existent_entry_point(
        self,
    ):
        tracer_configurator_name = _get_tracer_configurator()
        with self.assertLogs(level=WARNING):
            tracer_configurator = _import_tracer_configurator(tracer_configurator_name)
        _init_tracing({}, tracer_configurator=tracer_configurator)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict(
        "os.environ",
        {"OTEL_PYTHON_TRACER_CONFIGURATOR": "custom_tracer_configurator"},
    )
    def test_trace_init_custom_tracer_configurator_with_env(self, mock_entry_points):
        def custom_tracer_configurator(tracer_scope):
            return mock.Mock(spec=_RuleBasedTracerConfigurator)(tracer_scope)

        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_tracer_configurator",
                    custom_tracer_configurator,
                )
            ]
        )

        tracer_configurator_name = _get_tracer_configurator()
        tracer_configurator = _import_tracer_configurator(tracer_configurator_name)
        _init_tracing({}, tracer_configurator=tracer_configurator)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertEqual(provider._tracer_configurator, custom_tracer_configurator)


class TestLoggingInit(TestCase):
    def setUp(self):
        self.processor_patch = patch(
            "opentelemetry.sdk._configuration.BatchLogRecordProcessor",
            DummyLogRecordProcessor,
        )
        self.provider_patch = patch(
            "opentelemetry.sdk._configuration.LoggerProvider",
            DummyLoggerProvider,
        )
        self.set_provider_patch = patch("opentelemetry.sdk._configuration.set_logger_provider")

        self.processor_mock = self.processor_patch.start()
        self.provider_mock = self.provider_patch.start()
        self.set_provider_mock = self.set_provider_patch.start()

    def tearDown(self):
        self.processor_patch.stop()
        self.set_provider_patch.stop()
        self.provider_patch.stop()
        root_logger = getLogger("root")
        root_logger.handlers = [handler for handler in root_logger.handlers if not isinstance(handler, LoggingHandler)]

    def test_logging_init_empty(self):
        with ResetGlobalLoggingState():
            auto_resource = Resource.create(
                {
                    "telemetry.auto.version": "auto-version",
                }
            )
            _init_logging({}, resource=auto_resource)
            self.assertEqual(self.set_provider_mock.call_count, 1)
            provider = self.set_provider_mock.call_args[0][0]
            self.assertIsInstance(provider, DummyLoggerProvider)
            self.assertIsInstance(provider.resource, Resource)
            self.assertEqual(
                provider.resource.attributes.get("telemetry.auto.version"),
                "auto-version",
            )

    @patch.dict(
        environ,
        {"OTEL_RESOURCE_ATTRIBUTES": "service.name=otlp-service"},
    )
    def test_logging_init_exporter(self):
        with ResetGlobalLoggingState():
            resource = Resource.create({})
            _init_logging({"otlp": DummyOTLPLogExporter}, resource=resource)
            self.assertEqual(self.set_provider_mock.call_count, 1)
            provider = self.set_provider_mock.call_args[0][0]
            self.assertIsInstance(provider, DummyLoggerProvider)
            self.assertIsInstance(provider.resource, Resource)
            self.assertEqual(
                provider.resource.attributes.get("service.name"),
                "otlp-service",
            )
            self.assertEqual(len(provider.processors), 1)
            self.assertIsInstance(provider.processors[0], DummyLogRecordProcessor)
            self.assertIsInstance(provider.processors[0].exporter, DummyOTLPLogExporter)
            getLogger(__name__).error("hello")
            self.assertEqual(len(provider.processors), 1)
            self.assertTrue(provider.processors[0].exporter.export_called)

    def test_logging_init_exporter_uses_exporter_args_map(self):
        with ResetGlobalLoggingState():
            resource = Resource.create({})
            _init_logging(
                {"otlp": DummyOTLPLogExporter},
                resource=resource,
                exporter_args_map={
                    DummyOTLPLogExporter: {"compression": "gzip"},
                    DummyOTLPMetricExporter: {"compression": "no"},
                },
            )
            self.assertEqual(self.set_provider_mock.call_count, 1)
            provider = self.set_provider_mock.call_args[0][0]
            self.assertEqual(len(provider.processors), 1)
            self.assertEqual(provider.processors[0].exporter.compression, "gzip")

    def test_logging_init_custom_log_record_processors(self):
        log_record_processor = mock.Mock(spec=LogRecordProcessor)
        with ResetGlobalLoggingState():
            resource = Resource.create({})
            _init_logging(
                {"otlp": DummyOTLPLogExporter},
                resource=resource,
                log_record_processors=[log_record_processor],
                export_log_record_processor=SimpleLogRecordProcessor,
            )
            provider = self.set_provider_mock.call_args[0][0]
            self.assertEqual(len(provider.processors), 2)
            self.assertEqual(provider.processors[0], log_record_processor)
            self.assertIsInstance(provider.processors[1], SimpleLogRecordProcessor)

    @patch.dict(
        environ,
        {"OTEL_RESOURCE_ATTRIBUTES": "service.name=otlp-service"},
    )
    def test_logging_init_exporter_without_handler_setup(self):
        resource = Resource.create({})
        _init_logging(
            {"otlp": DummyOTLPLogExporter},
            resource=resource,
            setup_logging_handler=False,
        )
        self.assertEqual(self.set_provider_mock.call_count, 1)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider, DummyLoggerProvider)
        self.assertIsInstance(provider.resource, Resource)
        self.assertEqual(
            provider.resource.attributes.get("service.name"),
            "otlp-service",
        )
        self.assertEqual(len(provider.processors), 1)
        self.assertIsInstance(provider.processors[0], DummyLogRecordProcessor)
        self.assertIsInstance(provider.processors[0].exporter, DummyOTLPLogExporter)
        getLogger(__name__).error("hello")
        self.assertFalse(provider.processors[0].exporter.export_called)

    def test_logging_init_with_setup_logging_handler_to_true_warns(self):
        resource = Resource.create({})
        with self.assertWarnsRegex(
            DeprecationWarning,
            "and the `LoggingHandler` in `opentelemetry-sdk` that it controls are deprecated",
        ):
            with ResetGlobalLoggingState():
                _init_logging(
                    {"otlp": DummyOTLPLogExporter},
                    resource=resource,
                    setup_logging_handler=True,
                )

    @patch.dict(
        environ,
        {"OTEL_RESOURCE_ATTRIBUTES": "service.name=otlp-service"},
    )
    @patch("opentelemetry.sdk._configuration._init_tracing")
    @patch("opentelemetry.sdk._configuration._init_logging")
    def test_logging_init_disable_default(self, logging_mock, tracing_mock):
        _initialize_components(auto_instrumentation_version="auto-version")
        self.assertEqual(tracing_mock.call_count, 1)
        logging_mock.assert_called_once_with(
            mock.ANY,
            mock.ANY,
            False,
            exporter_args_map=None,
            log_record_processors=None,
            export_log_record_processor=None,
            logger_configurator=None,
        )

    @patch.dict(
        environ,
        {
            "OTEL_RESOURCE_ATTRIBUTES": "service.name=otlp-service",
            "OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED": "True",
        },
    )
    @patch("opentelemetry.sdk._configuration._init_tracing")
    @patch("opentelemetry.sdk._configuration._init_logging")
    def test_logging_init_enable_env(self, logging_mock, tracing_mock):
        with self.assertLogs(level=WARNING):
            _initialize_components(auto_instrumentation_version="auto-version")
        logging_mock.assert_called_once_with(
            mock.ANY,
            mock.ANY,
            True,
            exporter_args_map=None,
            log_record_processors=None,
            export_log_record_processor=None,
            logger_configurator=None,
        )
        self.assertEqual(tracing_mock.call_count, 1)

    @patch.dict(
        environ,
        {
            "OTEL_RESOURCE_ATTRIBUTES": "service.name=otlp-service",
            "OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED": "True",
        },
    )
    @patch("opentelemetry.sdk._configuration._init_tracing")
    @patch("opentelemetry.sdk._configuration._init_logging")
    @patch("opentelemetry.sdk._configuration._init_metrics")
    def test_initialize_components_resource(self, metrics_mock, logging_mock, tracing_mock):
        _initialize_components(auto_instrumentation_version="auto-version")
        self.assertEqual(logging_mock.call_count, 1)
        self.assertEqual(tracing_mock.call_count, 1)
        self.assertEqual(metrics_mock.call_count, 1)

        _, args, _ = logging_mock.mock_calls[0]
        logging_resource = args[1]
        _, _, kwargs = tracing_mock.mock_calls[0]
        tracing_resource = kwargs["resource"]
        _, args, _ = metrics_mock.mock_calls[0]
        metrics_resource = kwargs["resource"]
        self.assertEqual(logging_resource, tracing_resource)
        self.assertEqual(logging_resource, metrics_resource)
        self.assertEqual(tracing_resource, metrics_resource)

    @patch.dict(
        environ,
        {
            "OTEL_TRACES_EXPORTER": _EXPORTER_OTLP,
            "OTEL_METRICS_EXPORTER": _EXPORTER_OTLP_PROTO_GRPC,
            "OTEL_LOGS_EXPORTER": _EXPORTER_OTLP_PROTO_HTTP,
        },
    )
    @patch.dict(
        environ,
        {
            "OTEL_RESOURCE_ATTRIBUTES": "service.name=otlp-service, custom.key.1=env-value",
            "OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED": "False",
        },
    )
    @patch("opentelemetry.sdk._configuration.Resource")
    @patch("opentelemetry.sdk._configuration._import_exporters")
    @patch("opentelemetry.sdk._configuration._get_exporter_names")
    @patch("opentelemetry.sdk._configuration._init_tracing")
    @patch("opentelemetry.sdk._configuration._init_logging")
    @patch("opentelemetry.sdk._configuration._init_metrics")
    def test_initialize_components_kwargs(
        self,
        metrics_mock,
        logging_mock,
        tracing_mock,
        exporter_names_mock,
        import_exporters_mock,
        resource_mock,
    ):
        exporter_names_mock.return_value = [
            "env_var_exporter_1",
            "env_var_exporter_2",
        ]
        import_exporters_mock.return_value = (
            "TEST_SPAN_EXPORTERS_DICT",
            "TEST_METRICS_EXPORTERS_DICT",
            "TEST_LOG_EXPORTERS_DICT",
        )
        resource_mock.create.return_value = "TEST_RESOURCE"
        kwargs = {
            "auto_instrumentation_version": "auto-version",
            "trace_exporter_names": ["custom_span_exporter"],
            "metric_exporter_names": ["custom_metric_exporter"],
            "log_exporter_names": ["custom_log_exporter"],
            "sampler": "TEST_SAMPLER",
            "resource_attributes": {
                "custom.key.1": "pass-in-value-1",
                "custom.key.2": "pass-in-value-2",
            },
            "id_generator": "TEST_GENERATOR",
            "setup_logging_handler": True,
            "exporter_args_map": {1: {"compression": "gzip"}},
            "export_log_record_processor": SimpleLogRecordProcessor,
            "export_span_processor": SimpleSpanProcessor,
            "log_record_processors": [],
            "span_processors": [],
            "tracer_configurator": "tracer_configurator_test",
            "meter_configurator": "meter_configurator_test",
        }
        _initialize_components(**kwargs)

        import_exporters_mock.assert_called_once_with(
            [
                "custom_span_exporter",
                "env_var_exporter_1",
                "env_var_exporter_2",
            ],
            [
                "custom_metric_exporter",
                "env_var_exporter_1",
                "env_var_exporter_2",
            ],
            [
                "custom_log_exporter",
                "env_var_exporter_1",
                "env_var_exporter_2",
            ],
        )
        resource_mock.create.assert_called_once_with(
            {
                "telemetry.auto.version": "auto-version",
                "custom.key.1": "pass-in-value-1",
                "custom.key.2": "pass-in-value-2",
            }
        )
        # Resource is checked separates
        tracing_mock.assert_called_once_with(
            exporters="TEST_SPAN_EXPORTERS_DICT",
            id_generator="TEST_GENERATOR",
            sampler="TEST_SAMPLER",
            resource="TEST_RESOURCE",
            exporter_args_map={1: {"compression": "gzip"}},
            span_processors=[],
            export_span_processor=SimpleSpanProcessor,
            tracer_configurator="tracer_configurator_test",
        )
        metrics_mock.assert_called_once_with(
            exporters_or_readers="TEST_METRICS_EXPORTERS_DICT",
            resource="TEST_RESOURCE",
            exporter_args_map={1: {"compression": "gzip"}},
            meter_configurator="meter_configurator_test",
        )
        logging_mock.assert_called_once_with(
            "TEST_LOG_EXPORTERS_DICT",
            "TEST_RESOURCE",
            True,
            exporter_args_map={1: {"compression": "gzip"}},
            log_record_processors=[],
            export_log_record_processor=SimpleLogRecordProcessor,
            logger_configurator=None,
        )

    def test_basicConfig_works_with_otel_handler(self):
        with ResetGlobalLoggingState():
            _init_logging(
                {"otlp": DummyOTLPLogExporter},
                Resource.create({}),
                setup_logging_handler=True,
            )

            logging.basicConfig(level=logging.INFO)

            root_logger = logging.getLogger()
            stream_handlers = [h for h in root_logger.handlers if isinstance(h, logging.StreamHandler)]
            self.assertEqual(
                len(stream_handlers),
                1,
                "basicConfig should add a StreamHandler even when OTel handler exists",
            )

    def test_basicConfig_preserves_otel_handler(self):
        with ResetGlobalLoggingState():
            _init_logging(
                {"otlp": DummyOTLPLogExporter},
                Resource.create({}),
                setup_logging_handler=True,
            )

            root_logger = logging.getLogger()
            self.assertEqual(
                len(root_logger.handlers),
                1,
                "Should be exactly one OpenTelemetry LoggingHandler",
            )
            handler = root_logger.handlers[0]
            self.assertIsInstance(handler, LoggingHandler)
            logging.basicConfig()

            self.assertGreater(len(root_logger.handlers), 1)

            logging_handlers = [h for h in root_logger.handlers if isinstance(h, LoggingHandler)]
            self.assertEqual(
                len(logging_handlers),
                1,
                "Should still have exactly one OpenTelemetry LoggingHandler",
            )

    def test_dictConfig_preserves_otel_handler(self):
        with ResetGlobalLoggingState():
            _init_logging(
                {"otlp": DummyOTLPLogExporter},
                Resource.create({}),
                setup_logging_handler=True,
            )

            root = logging.getLogger()
            self.assertEqual(
                len(root.handlers),
                1,
                "Should be exactly one OpenTelemetry LoggingHandler",
            )
            logging.config.dictConfig(
                {
                    "version": 1,
                    "disable_existing_loggers": False,  # If this is True all loggers are disabled. Many unit tests assert loggers emit logs.
                    "handlers": {
                        "console": {
                            "class": "logging.StreamHandler",
                            "level": "DEBUG",
                            "stream": "ext://sys.stdout",
                        },
                    },
                    "loggers": {
                        "": {  # root logger
                            "handlers": ["console"],
                        },
                    },
                }
            )
            self.assertEqual(len(root.handlers), 2)

            logging_handlers = [h for h in root.handlers if isinstance(h, LoggingHandler)]
            self.assertEqual(
                len(logging_handlers),
                1,
                "Should still have exactly one OpenTelemetry LoggingHandler",
            )

    def test_logging_init_logger_configurator_none_by_default(self):
        with ResetGlobalLoggingState():
            _init_logging({})
            provider = self.set_provider_mock.call_args[0][0]
            self.assertIsInstance(provider, DummyLoggerProvider)
            self.assertIsNone(provider._logger_configurator)

    def test_logging_init_logger_configurator_passed_directly(self):
        mock_configurator = Mock()
        with ResetGlobalLoggingState():
            _init_logging({}, logger_configurator=mock_configurator)
            provider = self.set_provider_mock.call_args[0][0]
            self.assertIsInstance(provider, DummyLoggerProvider)
            self.assertEqual(provider._logger_configurator, mock_configurator)

    @patch.dict(
        "os.environ",
        {OTEL_PYTHON_LOGGER_CONFIGURATOR: "non_existent_entry_point"},
    )
    def test_logging_init_custom_logger_configurator_with_env_non_existent_entry_point(
        self,
    ):
        logger_configurator_name = _get_logger_configurator()
        with self.assertLogs(level=WARNING):
            logger_configurator = _import_logger_configurator(logger_configurator_name)
        with ResetGlobalLoggingState():
            _init_logging({}, logger_configurator=logger_configurator)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict(
        "os.environ",
        {OTEL_PYTHON_LOGGER_CONFIGURATOR: "custom_logger_configurator"},
    )
    def test_logging_init_custom_logger_configurator_with_env(self, mock_entry_points):
        def custom_logger_configurator(logger_scope):
            return mock.Mock(spec=_RuleBasedLoggerConfigurator)(logger_scope)

        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_logger_configurator",
                    custom_logger_configurator,
                )
            ]
        )

        logger_configurator_name = _get_logger_configurator()
        logger_configurator = _import_logger_configurator(logger_configurator_name)
        with ResetGlobalLoggingState():
            _init_logging({}, logger_configurator=logger_configurator)
            provider = self.set_provider_mock.call_args[0][0]
            self.assertEqual(provider._logger_configurator, custom_logger_configurator)


class TestMetricsInit(TestCase):
    def setUp(self):
        self.metric_reader_patch = patch(
            "opentelemetry.sdk._configuration.PeriodicExportingMetricReader",
            DummyMetricReader,
        )
        self.provider_patch = patch(
            "opentelemetry.sdk._configuration.MeterProvider",
            DummyMeterProvider,
        )
        self.set_provider_patch = patch("opentelemetry.sdk._configuration.set_meter_provider")

        self.metric_reader_mock = self.metric_reader_patch.start()
        self.provider_mock = self.provider_patch.start()
        self.set_provider_mock = self.set_provider_patch.start()

    def tearDown(self):
        self.metric_reader_patch.stop()
        self.set_provider_patch.stop()
        self.provider_patch.stop()

    def test_metrics_init_empty(self):
        auto_resource = Resource.create(
            {
                "telemetry.auto.version": "auto-version",
            }
        )
        _init_metrics({}, resource=auto_resource)
        self.assertEqual(self.set_provider_mock.call_count, 1)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider, DummyMeterProvider)
        self.assertIsInstance(provider._sdk_config.resource, Resource)
        self.assertEqual(
            provider._sdk_config.resource.attributes.get("telemetry.auto.version"),
            "auto-version",
        )

    @patch.dict(
        environ,
        {"OTEL_RESOURCE_ATTRIBUTES": "service.name=otlp-service"},
    )
    def test_metrics_init_exporter(self):
        resource = Resource.create({})
        _init_metrics({"otlp": DummyOTLPMetricExporter}, resource=resource)
        self.assertEqual(self.set_provider_mock.call_count, 1)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider, DummyMeterProvider)
        self.assertIsInstance(provider._sdk_config.resource, Resource)
        self.assertEqual(
            provider._sdk_config.resource.attributes.get("service.name"),
            "otlp-service",
        )
        reader = provider._metric_readers[0]
        self.assertIsInstance(reader, DummyMetricReader)
        self.assertIsInstance(reader.exporter, DummyOTLPMetricExporter)

    def test_metrics_init_pull_exporter(self):
        resource = Resource.create({})
        _init_metrics(
            {"dummy_metric_reader": DummyMetricReaderPullExporter},
            resource=resource,
        )
        self.assertEqual(self.set_provider_mock.call_count, 1)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider, DummyMeterProvider)
        reader = provider._metric_readers[0]
        self.assertIsInstance(reader, DummyMetricReaderPullExporter)

    def test_metrics_init_exporter_uses_exporter_args_map(self):
        resource = Resource.create({})
        _init_metrics(
            {"otlp": DummyOTLPMetricExporter},
            resource=resource,
            exporter_args_map={
                DummyOTLPMetricExporter: {"compression": "gzip"},
                DummyMetricReaderPullExporter: {"compression": "no"},
            },
        )
        provider = self.set_provider_mock.call_args[0][0]
        reader = provider._metric_readers[0]
        self.assertEqual(reader.exporter.compression, "gzip")

    def test_metrics_init_meter_configurator_none_by_default(self):
        _init_metrics({})
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider, DummyMeterProvider)
        self.assertEqual(provider._meter_configurator, _default_meter_configurator)

    def test_metrics_init_meter_configurator_passed_directly(self):
        mock_configurator = Mock()
        _init_metrics({}, meter_configurator=mock_configurator)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertIsInstance(provider, DummyMeterProvider)
        self.assertEqual(provider._meter_configurator, mock_configurator)

    @patch.dict(
        "os.environ",
        {OTEL_PYTHON_METER_CONFIGURATOR: "non_existent_entry_point"},
    )
    def test_metrics_init_custom_meter_configurator_with_env_non_existent_entry_point(
        self,
    ):
        meter_configurator_name = _get_meter_configurator()
        with self.assertLogs(level=WARNING):
            meter_configurator = _import_meter_configurator(meter_configurator_name)
        _init_metrics({}, meter_configurator=meter_configurator)

    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch.dict(
        "os.environ",
        {OTEL_PYTHON_METER_CONFIGURATOR: "custom_meter_configurator"},
    )
    def test_metrics_init_custom_meter_configurator_with_env(self, mock_entry_points):
        def custom_meter_configurator(meter_scope):
            return mock.Mock(spec=_RuleBasedMeterConfigurator)(meter_scope)

        mock_entry_points.configure_mock(
            return_value=[
                IterEntryPoint(
                    "custom_meter_configurator",
                    custom_meter_configurator,
                )
            ]
        )

        meter_configurator_name = _get_meter_configurator()
        meter_configurator = _import_meter_configurator(meter_configurator_name)
        _init_metrics({}, meter_configurator=meter_configurator)
        provider = self.set_provider_mock.call_args[0][0]
        self.assertEqual(provider._meter_configurator, custom_meter_configurator)


class TestExporterNames(TestCase):
    @patch.dict(
        environ,
        {
            "OTEL_TRACES_EXPORTER": _EXPORTER_OTLP,
            "OTEL_METRICS_EXPORTER": _EXPORTER_OTLP_PROTO_GRPC,
            "OTEL_LOGS_EXPORTER": _EXPORTER_OTLP_PROTO_HTTP,
        },
    )
    def test_otlp_exporter(self):
        self.assertEqual(_get_exporter_names("traces"), [_EXPORTER_OTLP_PROTO_GRPC])
        self.assertEqual(_get_exporter_names("metrics"), [_EXPORTER_OTLP_PROTO_GRPC])
        self.assertEqual(_get_exporter_names("logs"), [_EXPORTER_OTLP_PROTO_HTTP])

    @patch.dict(
        environ,
        {
            "OTEL_TRACES_EXPORTER": _EXPORTER_OTLP,
            "OTEL_METRICS_EXPORTER": _EXPORTER_OTLP,
            "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
            "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "grpc",
        },
    )
    def test_otlp_custom_exporter(self):
        self.assertEqual(_get_exporter_names("traces"), [_EXPORTER_OTLP_PROTO_HTTP])
        self.assertEqual(_get_exporter_names("metrics"), [_EXPORTER_OTLP_PROTO_GRPC])

    @patch.dict(
        environ,
        {
            "OTEL_TRACES_EXPORTER": _EXPORTER_OTLP_PROTO_HTTP,
            "OTEL_METRICS_EXPORTER": _EXPORTER_OTLP_PROTO_GRPC,
            "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc",
            "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "http/protobuf",
        },
    )
    def test_otlp_exporter_conflict(self):
        # Verify that OTEL_*_EXPORTER is used, and a warning is logged
        with self.assertLogs(level="WARNING") as logs_context:
            self.assertEqual(_get_exporter_names("traces"), [_EXPORTER_OTLP_PROTO_HTTP])
        assert len(logs_context.output) == 1

        with self.assertLogs(level="WARNING") as logs_context:
            self.assertEqual(_get_exporter_names("metrics"), [_EXPORTER_OTLP_PROTO_GRPC])
        assert len(logs_context.output) == 1

    @patch.dict(environ, {"OTEL_TRACES_EXPORTER": "zipkin"})
    def test_multiple_exporters(self):
        self.assertEqual(sorted(_get_exporter_names("traces")), ["zipkin"])

    @patch.dict(environ, {"OTEL_TRACES_EXPORTER": "none"})
    def test_none_exporters(self):
        self.assertEqual(sorted(_get_exporter_names("traces")), [])

    def test_no_exporters(self):
        self.assertEqual(sorted(_get_exporter_names("traces")), [])

    @patch.dict(environ, {"OTEL_TRACES_EXPORTER": ""})
    def test_empty_exporters(self):
        self.assertEqual(sorted(_get_exporter_names("traces")), [])


class TestImportExporters(TestCase):
    def test_console_exporters(self):
        trace_exporters, metric_exporterts, logs_exporters = _import_exporters(["console"], ["console"], ["console"])
        self.assertEqual(trace_exporters["console"].__class__, ConsoleSpanExporter.__class__)
        self.assertEqual(
            logs_exporters["console"].__class__,
            ConsoleLogRecordExporter.__class__,
        )
        self.assertEqual(
            metric_exporterts["console"].__class__,
            ConsoleMetricExporter.__class__,
        )

    @patch(
        "opentelemetry.sdk._configuration.entry_points",
    )
    def test_metric_pull_exporter(self, mock_entry_points: Mock):
        def mock_entry_points_impl(group, name):
            if name == "dummy_pull_exporter":
                return [IterEntryPoint(name=name, class_type=DummyMetricReaderPullExporter)]
            return []

        mock_entry_points.side_effect = mock_entry_points_impl
        _, metric_exporters, _ = _import_exporters([], ["dummy_pull_exporter"], [])
        self.assertIs(
            metric_exporters["dummy_pull_exporter"],
            DummyMetricReaderPullExporter,
        )


class TestImportConfigComponents(TestCase):
    @patch(
        "opentelemetry.sdk._configuration.entry_points",
        **{"side_effect": KeyError},
    )
    def test__import_config_components_missing_entry_point(self, mock_entry_points):
        with raises(RuntimeError) as error:
            _import_config_components(["a", "b", "c"], "name")
        self.assertEqual(str(error.value), "Requested entry point 'name' not found")

    @patch(
        "opentelemetry.sdk._configuration.entry_points",
        **{"side_effect": StopIteration},
    )
    def test__import_config_components_missing_component(self, mock_entry_points):
        with raises(RuntimeError) as error:
            _import_config_components(["a", "b", "c"], "name")
        self.assertEqual(
            str(error.value),
            "Requested component 'a' not found in entry point 'name'",
        )


class TestConfigurator(TestCase):
    class CustomConfigurator(_OTelSDKConfigurator):
        def _configure(self, **kwargs):
            kwargs["sampler"] = "TEST_SAMPLER"
            super()._configure(**kwargs)

    @patch("opentelemetry.sdk._configuration._initialize_components")
    def test_custom_configurator(self, mock_init_comp):
        custom_configurator = TestConfigurator.CustomConfigurator()
        custom_configurator._configure(auto_instrumentation_version="TEST_VERSION2")
        kwargs = {
            "auto_instrumentation_version": "TEST_VERSION2",
            "sampler": "TEST_SAMPLER",
        }
        mock_init_comp.assert_called_once_with(**kwargs)


# Any test that calls _init_logging with setup_logging_handler=True
# should call _init_logging within this context manager, to
# ensure the global logging state is reset after the test.
class ResetGlobalLoggingState:
    def __init__(self):
        self.original_basic_config = logging.basicConfig
        self.original_dict_config = logging.config.dictConfig
        self.original_file_config = logging.config.fileConfig
        self.root_logger = getLogger()
        self.original_handlers = None

    def __enter__(self):
        self.original_handlers = self.root_logger.handlers[:]
        self.root_logger.handlers = []
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.root_logger.handlers = []
        for handler in self.original_handlers:
            self.root_logger.addHandler(handler)
        logging.basicConfig = self.original_basic_config
        logging.config.dictConfig = self.original_dict_config
        logging.config.fileConfig = self.original_file_config


class TestClearLoggingHandlers(TestCase):
    def test_preserves_handlers(self):
        root_logger = getLogger()
        initial_handlers = root_logger.handlers[:]

        test_handler = logging.StreamHandler()
        root_logger.addHandler(test_handler)
        expected_handlers = initial_handlers + [test_handler]

        with ResetGlobalLoggingState():
            self.assertEqual(len(root_logger.handlers), 0)
            temp_handler = logging.StreamHandler()
            root_logger.addHandler(temp_handler)

        self.assertEqual(len(root_logger.handlers), len(expected_handlers))
        for h1, h2 in zip(root_logger.handlers, expected_handlers):
            self.assertIs(h1, h2)

        root_logger.removeHandler(test_handler)

    def test_preserves_original_logging_fns(self):
        def f(x):
            print("f")

        with ResetGlobalLoggingState():
            logging.basicConfig = f
            logging.config.dictConfig = f
            logging.config.fileConfig = f
        self.assertEqual(logging.config.dictConfig.__name__, "dictConfig")
        self.assertEqual(logging.basicConfig.__name__, "basicConfig")
        self.assertEqual(logging.config.fileConfig.__name__, "fileConfig")


class TestOpAMPInit(TestCase):
    @patch("opentelemetry.sdk._configuration._init_metrics")
    @patch("opentelemetry.sdk._configuration._init_tracing")
    @patch("opentelemetry.sdk._configuration._init_logging")
    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch("opentelemetry.sdk._configuration.Resource")
    def test_pre_sdk_init_function_found(
        self,
        mock_resource,
        mock_entry_points,
        mock_logging,
        mock_tracing,
        mock_metrics,
    ):
        init_function = mock.Mock()
        mock_entry_points.side_effect = [
            [IterEntryPoint("pre_sdk_init_function", init_function)],
            [],
        ]

        _initialize_components(id_generator=1)

        mock_entry_points.assert_has_calls([mock.call(group="_opentelemetry_opamp", name="pre_sdk_init_function")])
        init_function.assert_called_once_with(mock_resource.create.return_value)

    @patch("opentelemetry.sdk._configuration._init_metrics")
    @patch("opentelemetry.sdk._configuration._init_tracing")
    @patch("opentelemetry.sdk._configuration._init_logging")
    @patch("opentelemetry.sdk._configuration.entry_points")
    @patch("opentelemetry.sdk._configuration.Resource")
    def test_post_sdk_init_function_found(
        self,
        mock_resource,
        mock_entry_points,
        mock_logging,
        mock_tracing,
        mock_metrics,
    ):
        init_function = mock.Mock()
        mock_entry_points.side_effect = [
            [],
            [IterEntryPoint("post_sdk_init_function", init_function)],
        ]

        _initialize_components(id_generator=1)

        mock_entry_points.assert_has_calls([mock.call(group="_opentelemetry_opamp", name="post_sdk_init_function")])
        init_function.assert_called_once_with(mock_resource.create.return_value)

    @patch("opentelemetry.sdk._configuration._init_metrics")
    @patch("opentelemetry.sdk._configuration._init_tracing")
    @patch("opentelemetry.sdk._configuration._init_logging")
    @patch("opentelemetry.sdk._configuration.entry_points")
    def test_init_function_load_failure(self, mock_entry_points, mock_logging, mock_tracing, mock_metrics):
        entry_point_mock = mock.Mock()
        entry_point_mock.load.side_effect = AttributeError("module 'foo' has no attribute 'OpampInit'")
        mock_entry_points.configure_mock(
            return_value=[entry_point_mock],
        )
        entry_point_mock.__str__ = lambda x: "<EntryPoint>"

        with self.assertLogs(level="WARNING") as cm:
            _initialize_components(id_generator=1)

        mock_entry_points.assert_has_calls([mock.call(group="_opentelemetry_opamp", name="pre_sdk_init_function")])

        self.assertIn(
            "WARNING:opentelemetry.sdk._configuration:Failed to load OpAMP init function from entry point,"
            " <EntryPoint>: module 'foo' has no attribute 'OpampInit'",
            cm.output,
        )

    @patch("opentelemetry.sdk._configuration._init_metrics")
    @patch("opentelemetry.sdk._configuration._init_tracing")
    @patch("opentelemetry.sdk._configuration._init_logging")
    @patch("opentelemetry.sdk._configuration.entry_points")
    def test_init_function_not_found(self, mock_entry_points, mock_logging, mock_tracing, mock_metrics):
        mock_entry_points.configure_mock(return_value=[])

        with self.assertLogs(level="DEBUG") as cm:
            _initialize_components(id_generator=1)
        self.assertIn(
            "DEBUG:opentelemetry.sdk._configuration:No OpAMP init function found",
            cm.output,
        )
