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

import gzip
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, HTTPServer
from logging import WARNING
from unittest.mock import MagicMock, Mock, patch

import requests
from requests import Session
from requests.exceptions import ConnectionError
from requests.models import Response

from opentelemetry.exporter.otlp.proto.common.trace_encoder import (
    encode_spans,
)
from opentelemetry.exporter.otlp.proto.http import Compression
from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
    DEFAULT_COMPRESSION,
    DEFAULT_ENDPOINT,
    DEFAULT_TIMEOUT,
    DEFAULT_TRACES_EXPORT_PATH,
    OTLPSpanExporter,
)
from opentelemetry.exporter.otlp.proto.http.version import __version__
from opentelemetry.sdk.environment_variables import (
    _OTEL_PYTHON_EXPORTER_OTLP_HTTP_TRACES_CREDENTIAL_PROVIDER,
    OTEL_EXPORTER_OTLP_CERTIFICATE,
    OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE,
    OTEL_EXPORTER_OTLP_CLIENT_KEY,
    OTEL_EXPORTER_OTLP_COMPRESSION,
    OTEL_EXPORTER_OTLP_ENDPOINT,
    OTEL_EXPORTER_OTLP_HEADERS,
    OTEL_EXPORTER_OTLP_TIMEOUT,
    OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE,
    OTEL_EXPORTER_OTLP_TRACES_CLIENT_CERTIFICATE,
    OTEL_EXPORTER_OTLP_TRACES_CLIENT_KEY,
    OTEL_EXPORTER_OTLP_TRACES_COMPRESSION,
    OTEL_EXPORTER_OTLP_TRACES_ENDPOINT,
    OTEL_EXPORTER_OTLP_TRACES_HEADERS,
    OTEL_EXPORTER_OTLP_TRACES_TIMEOUT,
    OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.trace import _Span
from opentelemetry.sdk.trace.export import SpanExportResult
from opentelemetry.test.mock_test_classes import IterEntryPoint

OS_ENV_ENDPOINT = "os.env.base"
OS_ENV_CERTIFICATE = "os/env/base.crt"
OS_ENV_CLIENT_CERTIFICATE = "os/env/client-cert.pem"
OS_ENV_CLIENT_KEY = "os/env/client-key.pem"
OS_ENV_HEADERS = "envHeader1=val1,envHeader2=val2,User-agent=Overridden"
OS_ENV_TIMEOUT = "30"
BASIC_SPAN = _Span(
    "abc",
    context=Mock(
        **{
            "trace_state": {"a": "b", "c": "d"},
            "span_id": 10217189687419569865,
            "trace_id": 67545097771067222548457157018666467027,
        }
    ),
)


class _RecordingOTLPHandler(BaseHTTPRequestHandler):
    # do_POST is the required override name from BaseHTTPRequestHandler.
    def do_POST(self):  # pylint: disable=invalid-name
        content_length = int(self.headers.get("Content-Length", 0))
        self.server.received_bodies.append(self.rfile.read(content_length))
        self.send_response(200)
        self.end_headers()

    def log_message(self, *args):  # silence the test server's stderr logging
        pass


def _start_recording_server():
    server = HTTPServer(("127.0.0.1", 0), _RecordingOTLPHandler)
    server.received_bodies = []
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread


# pylint: disable=protected-access,too-many-public-methods
class TestOTLPSpanExporter(unittest.TestCase):
    def setUp(self):
        self.metric_reader = InMemoryMetricReader()
        self.meter_provider = MeterProvider(metric_readers=[self.metric_reader])

    def test_constructor_default(self):
        exporter = OTLPSpanExporter()

        self.assertEqual(exporter._endpoint, DEFAULT_ENDPOINT + DEFAULT_TRACES_EXPORT_PATH)
        self.assertEqual(exporter._certificate_file, True)
        self.assertEqual(exporter._client_certificate_file, None)
        self.assertEqual(exporter._client_key_file, None)
        self.assertEqual(exporter._timeout, DEFAULT_TIMEOUT)
        self.assertIs(exporter._compression, DEFAULT_COMPRESSION)
        self.assertEqual(exporter._headers, {})
        self.assertIsInstance(exporter._session, requests.Session)
        self.assertIn("User-Agent", exporter._session.headers)
        self.assertEqual(
            exporter._session.headers.get("Content-Type"),
            "application/x-protobuf",
        )
        self.assertEqual(
            exporter._session.headers.get("User-Agent"),
            "OTel-OTLP-Exporter-Python/" + __version__,
        )

    @patch.dict(
        "os.environ",
        {
            OTEL_EXPORTER_OTLP_CERTIFICATE: OS_ENV_CERTIFICATE,
            OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE: OS_ENV_CLIENT_CERTIFICATE,
            OTEL_EXPORTER_OTLP_CLIENT_KEY: OS_ENV_CLIENT_KEY,
            OTEL_EXPORTER_OTLP_COMPRESSION: Compression.Gzip.value,
            OTEL_EXPORTER_OTLP_ENDPOINT: OS_ENV_ENDPOINT,
            OTEL_EXPORTER_OTLP_HEADERS: OS_ENV_HEADERS,
            OTEL_EXPORTER_OTLP_TIMEOUT: OS_ENV_TIMEOUT,
            OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE: "traces/certificate.env",
            OTEL_EXPORTER_OTLP_TRACES_CLIENT_CERTIFICATE: "traces/client-cert.pem",
            OTEL_EXPORTER_OTLP_TRACES_CLIENT_KEY: "traces/client-key.pem",
            OTEL_EXPORTER_OTLP_TRACES_COMPRESSION: Compression.Deflate.value,
            OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "https://traces.endpoint.env",
            OTEL_EXPORTER_OTLP_TRACES_HEADERS: "tracesEnv1=val1,tracesEnv2=val2,traceEnv3===val3==,User-agent=TraceUserAgent",
            OTEL_EXPORTER_OTLP_TRACES_TIMEOUT: "40",
            _OTEL_PYTHON_EXPORTER_OTLP_HTTP_TRACES_CREDENTIAL_PROVIDER: "credential_provider",
        },
    )
    @patch("opentelemetry.exporter.otlp.proto.http._common.entry_points")
    def test_exporter_traces_env_take_priority(self, mock_entry_point):
        credential = Session()

        def f():
            return credential

        mock_entry_point.configure_mock(return_value=[IterEntryPoint("custom_credential", f)])
        exporter = OTLPSpanExporter()

        self.assertEqual(exporter._endpoint, "https://traces.endpoint.env")
        self.assertEqual(exporter._certificate_file, "traces/certificate.env")
        self.assertEqual(exporter._client_certificate_file, "traces/client-cert.pem")
        self.assertEqual(exporter._client_key_file, "traces/client-key.pem")
        self.assertEqual(exporter._timeout, 40)
        self.assertIs(exporter._compression, Compression.Deflate)
        self.assertEqual(
            exporter._headers,
            {
                "tracesenv1": "val1",
                "tracesenv2": "val2",
                "traceenv3": "==val3==",
                "user-agent": "TraceUserAgent",
            },
        )
        self.assertIs(exporter._session, credential)
        self.assertIsInstance(exporter._session, requests.Session)
        self.assertEqual(
            exporter._session.headers.get("Content-Type"),
            "application/x-protobuf",
        )
        self.assertEqual(
            exporter._session.headers.get("User-Agent"),
            "TraceUserAgent",
        )

    @patch.dict(
        "os.environ",
        {
            OTEL_EXPORTER_OTLP_CERTIFICATE: OS_ENV_CERTIFICATE,
            OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE: OS_ENV_CLIENT_CERTIFICATE,
            OTEL_EXPORTER_OTLP_CLIENT_KEY: OS_ENV_CLIENT_KEY,
            OTEL_EXPORTER_OTLP_COMPRESSION: Compression.Gzip.value,
            OTEL_EXPORTER_OTLP_ENDPOINT: OS_ENV_ENDPOINT,
            OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "https://traces.endpoint.env",
            OTEL_EXPORTER_OTLP_HEADERS: OS_ENV_HEADERS,
            OTEL_EXPORTER_OTLP_TIMEOUT: OS_ENV_TIMEOUT,
        },
    )
    def test_exporter_constructor_take_priority(self):
        exporter = OTLPSpanExporter(
            endpoint="example.com/1234",
            certificate_file="path/to/service.crt",
            client_key_file="path/to/client-key.pem",
            client_certificate_file="path/to/client-cert.pem",
            headers={"testHeader1": "value1", "testHeader2": "value2"},
            timeout=20,
            compression=Compression.NoCompression,
            session=requests.Session(),
        )

        self.assertEqual(exporter._endpoint, "example.com/1234")
        self.assertEqual(exporter._certificate_file, "path/to/service.crt")
        self.assertEqual(exporter._client_certificate_file, "path/to/client-cert.pem")
        self.assertEqual(exporter._client_key_file, "path/to/client-key.pem")
        self.assertEqual(exporter._timeout, 20)
        self.assertIs(exporter._compression, Compression.NoCompression)
        self.assertEqual(
            exporter._headers,
            {"testHeader1": "value1", "testHeader2": "value2"},
        )
        self.assertIsInstance(exporter._session, requests.Session)

    @patch.dict(
        "os.environ",
        {
            OTEL_EXPORTER_OTLP_CERTIFICATE: OS_ENV_CERTIFICATE,
            OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE: OS_ENV_CLIENT_CERTIFICATE,
            OTEL_EXPORTER_OTLP_CLIENT_KEY: OS_ENV_CLIENT_KEY,
            OTEL_EXPORTER_OTLP_COMPRESSION: Compression.Gzip.value,
            OTEL_EXPORTER_OTLP_HEADERS: OS_ENV_HEADERS,
            OTEL_EXPORTER_OTLP_TIMEOUT: OS_ENV_TIMEOUT,
        },
    )
    def test_exporter_env(self):
        exporter = OTLPSpanExporter()

        self.assertEqual(exporter._certificate_file, OS_ENV_CERTIFICATE)
        self.assertEqual(exporter._client_certificate_file, OS_ENV_CLIENT_CERTIFICATE)
        self.assertEqual(exporter._client_key_file, OS_ENV_CLIENT_KEY)
        self.assertEqual(exporter._timeout, int(OS_ENV_TIMEOUT))
        self.assertIs(exporter._compression, Compression.Gzip)
        self.assertEqual(
            exporter._headers,
            {
                "envheader1": "val1",
                "envheader2": "val2",
                "user-agent": "Overridden",
            },
        )

    @patch.dict(
        "os.environ",
        {OTEL_EXPORTER_OTLP_ENDPOINT: OS_ENV_ENDPOINT},
    )
    def test_exporter_env_endpoint_without_slash(self):
        exporter = OTLPSpanExporter()

        self.assertEqual(
            exporter._endpoint,
            OS_ENV_ENDPOINT + f"/{DEFAULT_TRACES_EXPORT_PATH}",
        )

    @patch.dict(
        "os.environ",
        {OTEL_EXPORTER_OTLP_ENDPOINT: OS_ENV_ENDPOINT + "/"},
    )
    def test_exporter_env_endpoint_with_slash(self):
        exporter = OTLPSpanExporter()

        self.assertEqual(
            exporter._endpoint,
            OS_ENV_ENDPOINT + f"/{DEFAULT_TRACES_EXPORT_PATH}",
        )

    @patch.dict(
        "os.environ",
        {OTEL_EXPORTER_OTLP_HEADERS: "envHeader1=val1,envHeader2=val2,missingValue"},
    )
    def test_headers_parse_from_env(self):
        with self.assertLogs(level="WARNING") as cm:
            _ = OTLPSpanExporter()

            self.assertEqual(
                cm.records[0].message,
                (
                    "Header format invalid! Header values in environment "
                    "variables must be URL encoded per the OpenTelemetry "
                    "Protocol Exporter specification or a comma separated "
                    "list of name=value occurrences: missingValue"
                ),
            )

    @patch.object(OTLPSpanExporter, "_export", return_value=Mock(ok=True))
    def test_2xx_status_code(self, mock_otlp_metric_exporter):
        """
        Test that any HTTP 2XX code returns a successful result
        """

        self.assertEqual(OTLPSpanExporter().export(MagicMock()), SpanExportResult.SUCCESS)

    @patch.dict("os.environ", {}, clear=True)
    @patch.object(OTLPSpanExporter, "_export", return_value=Mock(ok=True))
    def test_exporter_metrics_disabled_by_default(self, _mock_export):
        exporter = OTLPSpanExporter(meter_provider=self.meter_provider)

        self.assertEqual(exporter.export([BASIC_SPAN]), SpanExportResult.SUCCESS)

        self.assertIsNone(self.metric_reader.get_metrics_data())

    @patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: " true "})
    @patch.object(Session, "post")
    def test_retry_timeout(self, mock_post):
        exporter = OTLPSpanExporter(timeout=1.5, meter_provider=self.meter_provider)

        resp = Response()
        resp.status_code = 503
        resp.reason = "UNAVAILABLE"
        mock_post.return_value = resp
        with self.assertLogs(level=WARNING) as warning:
            before = time.time()
            # Set timeout to 1.5 seconds
            self.assertEqual(
                exporter.export([BASIC_SPAN]),
                SpanExportResult.FAILURE,
            )
            after = time.time()
            # First call at time 0, second at time 1, then an early return before the second backoff sleep b/c it would exceed timeout.
            self.assertEqual(mock_post.call_count, 2)
            # There's a +/-20% jitter on each backoff.
            self.assertTrue(0.75 < after - before < 1.25)
            self.assertIn(
                "Transient error UNAVAILABLE encountered while exporting span batch, retrying in",
                warning.records[0].message,
            )

        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.assertNotIn("error.type", metrics[0].data.data_points[0].attributes)
        self.assertEqual(
            metrics[0].data.data_points[0].attributes["http.response.status_code"],
            503,
        )
        self.assertEqual(metrics[1].name, "otel.sdk.exporter.span.exported")
        self.assert_standard_metric_attrs(metrics[1].data.data_points[0].attributes)
        self.assertNotIn("error.type", metrics[1].data.data_points[0].attributes)
        self.assertNotIn(
            "http.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(
            "http.response.status_code",
            metrics[2].data.data_points[0].attributes,
        )

    @patch.object(Session, "post")
    def test_export_no_collector_available_retryable(self, mock_post):
        exporter = OTLPSpanExporter(timeout=1.5)
        msg = "Server not available."
        mock_post.side_effect = ConnectionError(msg)
        with self.assertLogs(level=WARNING) as warning:
            self.assertEqual(
                exporter.export([BASIC_SPAN]),
                SpanExportResult.FAILURE,
            )
            # Check for greater 2 because the request is on each retry
            # done twice at the moment.
            self.assertGreater(mock_post.call_count, 2)
            self.assertIn(
                f"Transient error {msg} encountered while exporting span batch, retrying in",
                warning.records[0].message,
            )

    @patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"})
    @patch.object(Session, "post")
    def test_export_no_collector_available(self, mock_post):
        exporter = OTLPSpanExporter(timeout=1.5, meter_provider=self.meter_provider)

        mock_post.side_effect = requests.exceptions.RequestException()
        with self.assertLogs(level=WARNING) as warning:
            self.assertEqual(
                exporter.export([BASIC_SPAN]),
                SpanExportResult.FAILURE,
            )
            self.assertEqual(mock_post.call_count, 1)
            self.assertIn(
                "Failed to export span batch code",
                warning.records[0].message,
            )

        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"],
            "RequestException",
        )
        self.assertNotIn(
            "http.response.status_code",
            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[1].data.data_points[0].attributes["error.type"],
            "RequestException",
        )
        self.assertNotIn(
            "http.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(
            "http.response.status_code",
            metrics[2].data.data_points[0].attributes,
        )

    @patch.object(Session, "post")
    def test_timeout_set_correctly(self, mock_post):
        resp = Response()
        resp.status_code = 200

        def export_side_effect(*args, **kwargs):
            # Timeout should be set to something slightly less than 400 milliseconds depending on how much time has passed.
            self.assertAlmostEqual(0.4, kwargs["timeout"], 2)
            return resp

        mock_post.side_effect = export_side_effect
        exporter = OTLPSpanExporter(timeout=0.4)
        exporter.export([BASIC_SPAN])

    @patch.object(Session, "post")
    def test_shutdown_interrupts_retry_backoff(self, mock_post):
        exporter = OTLPSpanExporter(timeout=1.5)

        resp = Response()
        resp.status_code = 503
        resp.reason = "UNAVAILABLE"
        mock_post.return_value = resp
        thread = threading.Thread(target=exporter.export, args=([BASIC_SPAN],))
        with self.assertLogs(level=WARNING) as warning:
            before = time.time()
            thread.start()
            # Wait for the first attempt to fail, then enter a 1 second backoff.
            time.sleep(0.05)
            # Should cause export to wake up and return.
            exporter.shutdown()
            thread.join()
            after = time.time()
            self.assertIn(
                "Transient error UNAVAILABLE encountered while exporting span batch, retrying in",
                warning.records[0].message,
            )
            self.assertIn(
                "Shutdown in progress, aborting retry.",
                warning.records[1].message,
            )

            assert after - before < 0.2

    def test_max_request_size_default(self):
        exporter = OTLPSpanExporter()
        self.assertEqual(exporter._max_request_size, 64 * 1024 * 1024)

    @patch.object(Session, "post")
    def test_oversized_payload_dropped_before_send(self, mock_post):
        exporter = OTLPSpanExporter(max_request_size=1)
        self.assertEqual(exporter.export([BASIC_SPAN]), SpanExportResult.FAILURE)
        mock_post.assert_not_called()

    @patch.object(OTLPSpanExporter, "_export", return_value=Mock(ok=True))
    def test_max_request_size_zero_disables(self, _mock_export):
        exporter = OTLPSpanExporter(max_request_size=0)
        self.assertEqual(exporter.export([BASIC_SPAN]), SpanExportResult.SUCCESS)

    @patch.object(Session, "post")
    def test_negative_max_request_size_disables_limit(self, mock_post):
        resp = Response()
        resp.status_code = 200
        mock_post.return_value = resp
        exporter = OTLPSpanExporter(max_request_size=-1)
        self.assertEqual(exporter.export([BASIC_SPAN]), SpanExportResult.SUCCESS)
        mock_post.assert_called()

    @patch.object(Session, "post")
    def test_oversized_payload_measured_before_compression(self, mock_post):
        # The limit applies to the uncompressed serialized request. Build a
        # highly compressible batch whose gzip size is below a limit that the
        # uncompressed size still exceeds, then assert it is still dropped --
        # which can only hold if size is measured before compression.
        spans = [BASIC_SPAN] * 200
        uncompressed = encode_spans(spans).SerializePartialToString()
        compressed = gzip.compress(uncompressed)
        limit = (len(compressed) + len(uncompressed)) // 2
        # Guard the discriminating condition: compressed < limit < uncompressed.
        self.assertLess(len(compressed), limit)
        self.assertLess(limit, len(uncompressed))
        exporter = OTLPSpanExporter(max_request_size=limit, compression=Compression.Gzip)
        self.assertEqual(exporter.export(spans), SpanExportResult.FAILURE)
        mock_post.assert_not_called()

    @patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"})
    @patch.object(Session, "post")
    def test_oversized_payload_records_failure_metric(self, mock_post):
        exporter = OTLPSpanExporter(max_request_size=1, meter_provider=self.meter_provider)
        self.assertEqual(exporter.export([BASIC_SPAN]), SpanExportResult.FAILURE)
        mock_post.assert_not_called()
        metrics_data = self.metric_reader.get_metrics_data()
        scope_metrics = metrics_data.resource_metrics[0].scope_metrics[0]
        exported = next(metric for metric in scope_metrics.metrics if metric.name == "otel.sdk.exporter.span.exported")
        self.assertEqual(
            exported.data.data_points[0].attributes["error.type"],
            "RequestPayloadTooLargeError",
        )

    def test_end_to_end_export_sends_request_over_http(self):
        server, thread = _start_recording_server()
        port = server.server_address[1]
        try:
            exporter = OTLPSpanExporter(endpoint=f"http://127.0.0.1:{port}/v1/traces")
            result = exporter.export([BASIC_SPAN])
        finally:
            server.shutdown()
            thread.join()
            server.server_close()
        self.assertEqual(result, SpanExportResult.SUCCESS)
        self.assertEqual(len(server.received_bodies), 1)
        self.assertGreater(len(server.received_bodies[0]), 0)

    def test_end_to_end_oversized_request_never_reaches_server(self):
        server, thread = _start_recording_server()
        port = server.server_address[1]
        try:
            exporter = OTLPSpanExporter(
                endpoint=f"http://127.0.0.1:{port}/v1/traces",
                max_request_size=1,
            )
            result = exporter.export([BASIC_SPAN])
        finally:
            server.shutdown()
            thread.join()
            server.server_close()
        self.assertEqual(result, SpanExportResult.FAILURE)
        self.assertEqual(server.received_bodies, [])

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