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

import threading
import time
import unittest
from collections.abc import Sequence
from concurrent.futures import (  # pylint: disable=no-name-in-module
    ThreadPoolExecutor,
)
from logging import WARNING, getLogger
from platform import system
from typing import Any
from unittest import TestCase
from unittest.mock import Mock, patch

import grpc
from google.protobuf.duration_pb2 import (  # pylint: disable=no-name-in-module
    Duration,
)
from google.rpc.error_details_pb2 import (  # pylint: disable=no-name-in-module
    RetryInfo,
)
from grpc import ChannelCredentials, Compression, StatusCode, server

from opentelemetry.exporter.otlp.proto.common.trace_encoder import (
    encode_spans,
)
from opentelemetry.exporter.otlp.proto.grpc.exporter import (  # noqa: F401
    _RETRYABLE_ERROR_CODES,
    InvalidCompressionValueException,
    OTLPExporterMixin,
    environ_to_compression,
)
from opentelemetry.exporter.otlp.proto.grpc.version import __version__
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import (
    ExportTraceServiceRequest,
    ExportTraceServiceResponse,
)
from opentelemetry.proto.collector.trace.v1.trace_service_pb2_grpc import (
    TraceServiceServicer,
    TraceServiceStub,
    add_TraceServiceServicer_to_server,
)
from opentelemetry.sdk.environment_variables import (
    _OTEL_PYTHON_EXPORTER_OTLP_GRPC_CREDENTIAL_PROVIDER,
    OTEL_EXPORTER_OTLP_COMPRESSION,
    OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.trace import ReadableSpan, _Span
from opentelemetry.sdk.trace.export import (
    SpanExporter,
    SpanExportResult,
)
from opentelemetry.semconv._incubating.attributes.otel_attributes import (
    OtelComponentTypeValues,
)
from opentelemetry.test.mock_test_classes import IterEntryPoint

logger = getLogger(__name__)


# The below tests use this test SpanExporter and Spans, but are testing the
# underlying behavior in the mixin. A MetricExporter or LogRecordExporter could
# just as easily be used.
class OTLPSpanExporterForTesting(
    SpanExporter,
    OTLPExporterMixin[
        ReadableSpan,
        ExportTraceServiceRequest,
        SpanExportResult,
        TraceServiceStub,
    ],
):
    def __init__(self, **kwargs):
        super().__init__(
            TraceServiceStub,
            SpanExportResult,
            component_type=OtelComponentTypeValues.OTLP_GRPC_SPAN_EXPORTER,
            signal="traces",
            meter_provider=kwargs.pop("meter_provider", None),
            **kwargs,
        )

    def _translate_data(self, data: Sequence[ReadableSpan]) -> ExportTraceServiceRequest:
        return encode_spans(data)

    def _count_data(self, data: Sequence[ReadableSpan]) -> int:
        return len(data)

    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
        return self._export(spans)

    @property
    def _exporting(self):
        return "traces"

    def shutdown(self, timeout_millis: float = 30_000, **kwargs):
        return OTLPExporterMixin.shutdown(self, timeout_millis, **kwargs)


class TraceServiceServicerWithExportParams(TraceServiceServicer):
    def __init__(
        self,
        export_result: StatusCode,
        optional_retry_nanos: int | None = None,
        optional_export_sleep: float | None = None,
        optional_error_details: str | None = None,
    ):
        self.export_result = export_result
        self.optional_export_sleep = optional_export_sleep
        self.optional_retry_nanos = optional_retry_nanos
        self.num_requests = 0
        self.optional_error_details = optional_error_details

    # pylint: disable=invalid-name,unused-argument
    def Export(self, request, context):
        self.num_requests += 1
        if self.optional_export_sleep:
            time.sleep(self.optional_export_sleep)
        if self.export_result != StatusCode.OK and self.optional_retry_nanos:
            context.set_trailing_metadata(
                (
                    (
                        "google.rpc.retryinfo-bin",
                        RetryInfo(retry_delay=Duration(nanos=self.optional_retry_nanos)).SerializeToString(),
                    ),
                )
            )
        context.set_code(self.export_result)
        if self.optional_error_details:
            context.set_details(self.optional_error_details)

        return ExportTraceServiceResponse()


class ThreadWithReturnValue(threading.Thread):
    def __init__(
        self,
        target=None,
        args=(),
    ):
        super().__init__(target=target, args=args)
        self._return = None

    def run(self):
        try:
            if self._target is not None:  # type: ignore
                self._return = self._target(*self._args, **self._kwargs)  # type: ignore
        finally:
            # Avoid a refcycle if the thread is running a function with
            # an argument that has a member that points to the thread.
            del self._target, self._args, self._kwargs  # type: ignore

    def join(self, timeout: float | None = None) -> Any:
        super().join(timeout=timeout)
        return self._return


# pylint: disable-next=too-many-public-methods
class TestOTLPExporterMixin(TestCase):
    def setUp(self):
        self.server = server(ThreadPoolExecutor(max_workers=10))

        self.server.add_insecure_port("127.0.0.1:4317")

        self.server.start()

        self.metric_reader = InMemoryMetricReader()
        self.meter_provider = MeterProvider(metric_readers=[self.metric_reader])
        self.exporter = OTLPSpanExporterForTesting(insecure=True, meter_provider=self.meter_provider)
        self.span = _Span(
            "a",
            context=Mock(
                **{
                    "trace_state": {"a": "b", "c": "d"},
                    "span_id": 10217189687419569865,
                    "trace_id": 67545097771067222548457157018666467027,
                }
            ),
        )

    def tearDown(self):
        self.server.stop(None)

    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel")
    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.secure_channel")
    def test_otlp_exporter_endpoint(self, mock_secure, mock_insecure):
        expected_endpoint = "localhost:4317"
        endpoints = [
            (
                "http://localhost:4317",
                None,
                mock_insecure,
            ),
            (
                "localhost:4317",
                None,
                mock_secure,
            ),
            (
                "http://localhost:4317",
                True,
                mock_insecure,
            ),
            (
                "localhost:4317",
                True,
                mock_insecure,
            ),
            (
                "http://localhost:4317",
                False,
                mock_secure,
            ),
            (
                "localhost:4317",
                False,
                mock_secure,
            ),
            (
                "https://localhost:4317",
                False,
                mock_secure,
            ),
            (
                "https://localhost:4317",
                None,
                mock_secure,
            ),
            (
                "https://localhost:4317",
                True,
                mock_secure,
            ),
        ]
        for endpoint, insecure, mock_method in endpoints:
            OTLPSpanExporterForTesting(endpoint=endpoint, insecure=insecure)
            self.assertEqual(
                1,
                mock_method.call_count,
                f"expected {mock_method} to be called for {endpoint} {insecure}",
            )
            self.assertEqual(
                expected_endpoint,
                mock_method.call_args[0][0],
                f"expected {expected_endpoint} got {mock_method.call_args[0][0]} {endpoint}",
            )
            mock_method.reset_mock()

    def test_environ_to_compression(self):
        with patch.dict(
            "os.environ",
            {
                "test_gzip": "gzip",
                "test_gzip_caseinsensitive_with_whitespace": " GzIp ",
                "test_invalid": "some invalid compression",
            },
        ):
            self.assertEqual(environ_to_compression("test_gzip"), Compression.Gzip)
            self.assertEqual(
                environ_to_compression("test_gzip_caseinsensitive_with_whitespace"),
                Compression.Gzip,
            )
            self.assertIsNone(
                environ_to_compression("missing_key"),
            )
            with self.assertRaises(InvalidCompressionValueException):
                environ_to_compression("test_invalid")

    # pylint: disable=no-self-use
    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel")
    @patch.dict("os.environ", {})
    def test_otlp_exporter_otlp_compression_unspecified(self, mock_insecure_channel):
        """No env or kwarg should be NoCompression"""
        OTLPSpanExporterForTesting(insecure=True)
        mock_insecure_channel.assert_called_once_with(
            "localhost:4317",
            compression=Compression.NoCompression,
            options=(
                (
                    "grpc.primary_user_agent",
                    "OTel-OTLP-Exporter-Python/" + __version__,
                ),
            ),
        )

    @patch.dict(
        "os.environ",
        {_OTEL_PYTHON_EXPORTER_OTLP_GRPC_CREDENTIAL_PROVIDER: "credential_provider"},
    )
    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.entry_points")
    def test_that_credential_gets_passed_to_exporter(self, mock_entry_points):
        credential = ChannelCredentials(None)

        def f():
            return credential

        mock_entry_points.configure_mock(return_value=[IterEntryPoint("custom_credential", f)])
        exporter = OTLPSpanExporterForTesting(insecure=False)
        # pylint: disable=protected-access
        assert exporter._credentials is credential

    @patch.dict(
        "os.environ",
        {_OTEL_PYTHON_EXPORTER_OTLP_GRPC_CREDENTIAL_PROVIDER: "credential_provider"},
    )
    def test_that_missing_entry_point_raises_exception(self):
        with self.assertRaises(RuntimeError):
            OTLPSpanExporterForTesting(insecure=False)

    @patch.dict(
        "os.environ",
        {_OTEL_PYTHON_EXPORTER_OTLP_GRPC_CREDENTIAL_PROVIDER: "credential_provider"},
    )
    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.entry_points")
    def test_that_entry_point_returning_bad_type_raises_exception(self, mock_entry_points):
        def f():
            return 1

        mock_entry_points.configure_mock(return_value=[IterEntryPoint("custom_credential", f)])
        with self.assertRaises(RuntimeError):
            OTLPSpanExporterForTesting(insecure=False)

    # pylint: disable=no-self-use, disable=unused-argument
    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.ssl_channel_credentials")
    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.secure_channel")
    @patch.dict("os.environ", {})
    def test_no_credentials_ssl_channel_called(self, secure_channel, mock_ssl_channel):
        OTLPSpanExporterForTesting(insecure=False)
        self.assertTrue(mock_ssl_channel.called)

    # pylint: disable=no-self-use
    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel")
    @patch.dict("os.environ", {OTEL_EXPORTER_OTLP_COMPRESSION: "gzip"})
    def test_otlp_exporter_otlp_compression_envvar(self, mock_insecure_channel):
        """Just OTEL_EXPORTER_OTLP_COMPRESSION should work"""
        OTLPSpanExporterForTesting(insecure=True)
        mock_insecure_channel.assert_called_once_with(
            "localhost:4317",
            compression=Compression.Gzip,
            options=(
                (
                    "grpc.primary_user_agent",
                    "OTel-OTLP-Exporter-Python/" + __version__,
                ),
            ),
        )

    @patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: " true "})
    def test_shutdown(self):
        add_TraceServiceServicer_to_server(
            TraceServiceServicerWithExportParams(StatusCode.OK),
            self.server,
        )
        exporter = OTLPSpanExporterForTesting(insecure=True, meter_provider=self.meter_provider)
        self.assertEqual(exporter.export([self.span]), SpanExportResult.SUCCESS)
        metrics_data = self.metric_reader.get_metrics_data()
        scope_metrics = metrics_data.resource_metrics[0].scope_metrics[0]
        self.assertEqual(scope_metrics.scope.name, "opentelemetry-sdk")
        metrics = sorted(scope_metrics.metrics, key=lambda m: m.name)
        self.assertEqual(len(metrics), 3)
        self.assertEqual(metrics[0].name, "otel.sdk.exporter.operation.duration")
        self.assert_standard_metric_attrs(metrics[0].data.data_points[0].attributes)
        self.assertEqual(metrics[1].name, "otel.sdk.exporter.span.exported")
        self.assert_standard_metric_attrs(metrics[1].data.data_points[0].attributes)
        self.assertEqual(metrics[2].name, "otel.sdk.exporter.span.inflight")
        self.assert_standard_metric_attrs(metrics[2].data.data_points[0].attributes)

        exporter.shutdown()
        with self.assertLogs(level=WARNING) as warning:
            self.assertEqual(exporter.export([self.span]), SpanExportResult.FAILURE)
            self.assertEqual(
                warning.records[0].message,
                "Exporter already shutdown, ignoring batch",
            )

    @unittest.skipIf(
        system() == "Windows",
        "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",
    )
    def test_shutdown_interrupts_export_retry_backoff(self):
        add_TraceServiceServicer_to_server(
            TraceServiceServicerWithExportParams(
                StatusCode.UNAVAILABLE,
            ),
            self.server,
        )

        export_thread = ThreadWithReturnValue(target=self.exporter.export, args=([self.span],))
        with self.assertLogs(level=WARNING) as warning:
            begin_wait = time.time()
            export_thread.start()
            # Wait a bit for export to fail and the backoff sleep to start
            time.sleep(0.05)
            # The code should now be in a 1 second backoff.
            # pylint: disable=protected-access
            self.assertFalse(self.exporter._shutdown_in_progress.is_set())
            self.exporter.shutdown()
            self.assertTrue(self.exporter._shutdown_in_progress.is_set())
            export_result = export_thread.join()
            end_wait = time.time()
            self.assertEqual(export_result, SpanExportResult.FAILURE)
            # Shutdown should have interrupted the sleep.
            self.assertTrue(end_wait - begin_wait < 0.2)
            self.assertEqual(
                warning.records[1].message,
                "Shutdown in progress, aborting retry.",
            )

    def test_export_over_closed_grpc_channel(self):
        # pylint: disable=protected-access

        add_TraceServiceServicer_to_server(
            TraceServiceServicerWithExportParams(StatusCode.OK),
            self.server,
        )
        self.exporter.export([self.span])
        self.exporter.shutdown()
        data = self.exporter._translate_data([self.span])
        with self.assertRaises(ValueError) as err:
            self.exporter._client.Export(request=data)
        self.assertEqual(str(err.exception), "Cannot invoke RPC on closed channel!")

    @unittest.skipIf(
        system() == "Windows",
        "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",
    )
    @patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"})
    def test_retry_info_is_respected(self):
        mock_trace_service = TraceServiceServicerWithExportParams(
            StatusCode.UNAVAILABLE,
            optional_retry_nanos=200000000,  # .2 seconds
        )
        add_TraceServiceServicer_to_server(
            mock_trace_service,
            self.server,
        )
        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=10, meter_provider=self.meter_provider)
        before = time.time()
        self.assertEqual(
            exporter.export([self.span]),
            SpanExportResult.FAILURE,
        )
        after = time.time()
        self.assertEqual(mock_trace_service.num_requests, 6)
        # 1 second plus wiggle room so the test passes consistently.
        self.assertAlmostEqual(after - before, 1, 1)

        metrics_data = self.metric_reader.get_metrics_data()
        scope_metrics = metrics_data.resource_metrics[0].scope_metrics[0]
        self.assertEqual(scope_metrics.scope.name, "opentelemetry-sdk")
        metrics = sorted(scope_metrics.metrics, key=lambda m: m.name)
        self.assertEqual(len(metrics), 3)
        self.assertEqual(metrics[0].name, "otel.sdk.exporter.operation.duration")
        self.assert_standard_metric_attrs(metrics[0].data.data_points[0].attributes)
        self.assertEqual(
            metrics[0].data.data_points[0].attributes["error.type"],
            "_InactiveRpcError",
        )
        self.assertEqual(
            metrics[0].data.data_points[0].attributes["rpc.response.status_code"],
            "UNAVAILABLE",
        )
        self.assertEqual(metrics[1].name, "otel.sdk.exporter.span.exported")
        self.assert_standard_metric_attrs(metrics[1].data.data_points[0].attributes)
        self.assertEqual(
            metrics[1].data.data_points[0].attributes["error.type"],
            "_InactiveRpcError",
        )
        self.assertNotIn(
            "rpc.response.status_code",
            metrics[1].data.data_points[0].attributes,
        )
        self.assertEqual(metrics[2].name, "otel.sdk.exporter.span.inflight")
        self.assert_standard_metric_attrs(metrics[2].data.data_points[0].attributes)
        self.assertNotIn("error.type", metrics[2].data.data_points[0].attributes)
        self.assertNotIn(
            "rpc.response.status_code",
            metrics[2].data.data_points[0].attributes,
        )

    @unittest.skipIf(
        system() == "Windows",
        "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",
    )
    def test_retry_not_made_if_would_exceed_timeout(self):
        mock_trace_service = TraceServiceServicerWithExportParams(StatusCode.UNAVAILABLE)
        add_TraceServiceServicer_to_server(
            mock_trace_service,
            self.server,
        )
        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=4)
        before = time.time()
        self.assertEqual(
            exporter.export([self.span]),
            SpanExportResult.FAILURE,
        )
        after = time.time()
        # Our retry starts with a 1 second backoff then doubles.
        # First call at time 0, second at time 1, third at time 3, fourth would exceed timeout.
        self.assertEqual(mock_trace_service.num_requests, 3)
        # There's a +/-20% jitter on each backoff.
        self.assertTrue(2.35 < after - before < 3.65)

    @unittest.skipIf(
        system() == "Windows",
        "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",
    )
    def test_timeout_set_correctly(self):
        mock_trace_service = TraceServiceServicerWithExportParams(StatusCode.UNAVAILABLE, optional_export_sleep=0.25)
        add_TraceServiceServicer_to_server(
            mock_trace_service,
            self.server,
        )
        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=1.4)
        # Should timeout after 1.4 seconds. First attempt takes .25 seconds
        # Then a 1 second sleep, then deadline exceeded after .15 seconds,
        # mid way through second call.
        with self.assertLogs(level=WARNING) as warning:
            before = time.time()
            # Eliminate the jitter.
            with patch("random.uniform", return_value=1):
                self.assertEqual(
                    exporter.export([self.span]),
                    SpanExportResult.FAILURE,
                )
            after = time.time()
            self.assertEqual(
                "Failed to export traces to localhost:4317, error code: StatusCode.DEADLINE_EXCEEDED, error details: Deadline Exceeded",
                warning.records[-1].message,
            )
            self.assertEqual(mock_trace_service.num_requests, 2)
            self.assertAlmostEqual(after - before, 1.4, 1)

    def test_channel_options_set_correctly(self):
        """Test that gRPC channel options are set correctly for keepalive and reconnection"""
        # This test verifies that the channel is created with the right options
        # We patch grpc.insecure_channel to ensure it is called without errors
        with patch("opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel") as mock_channel:
            OTLPSpanExporterForTesting(insecure=True)
            self.assertTrue(mock_channel.called)

    def test_otlp_headers_from_env(self):
        # pylint: disable=protected-access
        # This ensures that there is no other header than standard user-agent.
        self.assertEqual(
            self.exporter._headers,
            (),
        )

    @patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"})
    def test_permanent_failure(self):
        exporter = OTLPSpanExporterForTesting(insecure=True, meter_provider=self.meter_provider)
        with self.assertLogs(level=WARNING) as warning:
            add_TraceServiceServicer_to_server(
                TraceServiceServicerWithExportParams(
                    StatusCode.ALREADY_EXISTS,
                    optional_error_details="This already exists.",
                ),
                self.server,
            )
            self.assertEqual(exporter.export([self.span]), SpanExportResult.FAILURE)
            self.assertEqual(
                warning.records[-1].message,
                "Failed to export traces to localhost:4317, error code: StatusCode.ALREADY_EXISTS, error details: This already exists.",
            )

        metrics_data = self.metric_reader.get_metrics_data()
        scope_metrics = metrics_data.resource_metrics[0].scope_metrics[0]
        self.assertEqual(scope_metrics.scope.name, "opentelemetry-sdk")
        metrics = sorted(scope_metrics.metrics, key=lambda m: m.name)
        self.assertEqual(len(metrics), 3)
        self.assertEqual(metrics[0].name, "otel.sdk.exporter.operation.duration")
        self.assert_standard_metric_attrs(metrics[0].data.data_points[0].attributes)
        self.assertEqual(
            metrics[0].data.data_points[0].attributes["error.type"],
            "_InactiveRpcError",
        )
        self.assertEqual(
            metrics[0].data.data_points[0].attributes["rpc.response.status_code"],
            "ALREADY_EXISTS",
        )
        self.assertEqual(metrics[1].name, "otel.sdk.exporter.span.exported")
        self.assert_standard_metric_attrs(metrics[1].data.data_points[0].attributes)
        self.assertEqual(
            metrics[1].data.data_points[0].attributes["error.type"],
            "_InactiveRpcError",
        )
        self.assertNotIn(
            "rpc.response.status_code",
            metrics[1].data.data_points[0].attributes,
        )
        self.assertEqual(metrics[2].name, "otel.sdk.exporter.span.inflight")
        self.assert_standard_metric_attrs(metrics[2].data.data_points[0].attributes)
        self.assertNotIn("error.type", metrics[2].data.data_points[0].attributes)
        self.assertNotIn(
            "rpc.response.status_code",
            metrics[2].data.data_points[0].attributes,
        )

    def test_unavailable_reconnects(self):
        """Test that the exporter reconnects on UNAVAILABLE error"""
        add_TraceServiceServicer_to_server(
            TraceServiceServicerWithExportParams(StatusCode.UNAVAILABLE),
            self.server,
        )

        # Spy on grpc.insecure_channel to verify it's called for reconnection
        with patch(
            "opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel",
            side_effect=grpc.insecure_channel,
        ) as mock_insecure_channel:
            # Mock sleep to avoid waiting
            with patch("time.sleep"):
                # We expect FAILURE because the server keeps returning UNAVAILABLE
                # but we want to verify reconnection attempts happened
                self.exporter.export([self.span])

        # Verify that we attempted to reinitialize the channel (called insecure_channel)
        # Since the initial channel was created in setUp (unpatched), this call
        # must be from the reconnection logic.
        self.assertTrue(mock_insecure_channel.called)

    def test_retryable_error_codes_initialization(self):
        # pylint: disable=protected-access
        self.assertEqual(self.exporter._retryable_error_codes, _RETRYABLE_ERROR_CODES)
        custom_codes = [StatusCode.INTERNAL, StatusCode.UNKNOWN]
        exporter = OTLPSpanExporterForTesting(insecure=True, retryable_error_codes=custom_codes)
        self.assertEqual(exporter._retryable_error_codes, frozenset(custom_codes))

    @patch.dict(
        "os.environ",
        {"OTEL_PYTHON_EXPORTER_OTLP_GRPC_RETRYABLE_ERROR_CODES": ",INTERNAL, unknown,,,dEAdline_Exceeded "},
    )
    def test_retryable_error_codes_initialization_from_env(self):
        expected_codes = frozenset(
            {
                StatusCode.INTERNAL,
                StatusCode.UNKNOWN,
                StatusCode.DEADLINE_EXCEEDED,
            }
        )
        exporter = OTLPSpanExporterForTesting()
        # pylint: disable=protected-access
        self.assertEqual(exporter._retryable_error_codes, expected_codes)

    @unittest.skipIf(
        system() == "Windows",
        "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",
    )
    def test_retryable_error_codes_custom(self):
        # Test that a custom error code is retried if specified
        custom_codes = [StatusCode.INTERNAL]
        mock_trace_service = TraceServiceServicerWithExportParams(
            StatusCode.INTERNAL,
            optional_retry_nanos=200000000,  # .2 seconds
        )
        add_TraceServiceServicer_to_server(
            mock_trace_service,
            self.server,
        )
        exporter = OTLPSpanExporterForTesting(insecure=True, retryable_error_codes=custom_codes, timeout=10)

        self.assertEqual(
            exporter.export([self.span]),
            SpanExportResult.FAILURE,
        )

        self.assertEqual(mock_trace_service.num_requests, 6)

        # Test that a default retryable code is NOT retried if not in custom_codes
        mock_trace_service.num_requests = 0
        mock_trace_service.export_result = StatusCode.UNAVAILABLE
        self.assertEqual(
            exporter.export([self.span]),
            SpanExportResult.FAILURE,
        )
        self.assertEqual(mock_trace_service.num_requests, 1)

    def assert_standard_metric_attrs(self, attributes):
        self.assertEqual(attributes["otel.component.type"], "otlp_grpc_span_exporter")
        self.assertTrue(attributes["otel.component.name"].startswith("otlp_grpc_span_exporter/"))
        self.assertEqual(attributes["server.address"], "localhost")
        self.assertEqual(attributes["server.port"], 4317)
