text: # Copyright The OpenTelemetry Authors
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  # pylint: disable=too-many-lines
  import shutil
  import subprocess
  import unittest
  from importlib import reload
  from logging import ERROR, WARNING
  from random import randint
  from typing import Optional
  from unittest import mock
  
  from opentelemetry import trace as trace_api
  from opentelemetry.context import Context
  from opentelemetry.sdk import resources, trace
  from opentelemetry.sdk.environment_variables import (
      OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT,
      OTEL_SPAN_EVENT_COUNT_LIMIT,
      OTEL_SPAN_LINK_COUNT_LIMIT,
      OTEL_TRACES_SAMPLER,
      OTEL_TRACES_SAMPLER_ARG,
  )
  from opentelemetry.sdk.trace import Resource, sampling
  from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
  from opentelemetry.sdk.util import ns_to_iso_str
  from opentelemetry.sdk.util.instrumentation import InstrumentationInfo
  from opentelemetry.trace import StatusCode
  from opentelemetry.util._time import _time_ns
  
  
  def new_tracer(span_limits=None) -> trace_api.Tracer:
      return trace.TracerProvider(span_limits=span_limits).get_tracer(__name__)
  
  
  class TestTracer(unittest.TestCase):
      def test_extends_api(self):
          tracer = new_tracer()
          self.assertIsInstance(tracer, trace.Tracer)
          self.assertIsInstance(tracer, trace_api.Tracer)
  
      def test_shutdown(self):
          tracer_provider = trace.TracerProvider()
  
          mock_processor1 = mock.Mock(spec=trace.SpanProcessor)
          tracer_provider.add_span_processor(mock_processor1)
  
          mock_processor2 = mock.Mock(spec=trace.SpanProcessor)
          tracer_provider.add_span_processor(mock_processor2)
  
          tracer_provider.shutdown()
  
          self.assertEqual(mock_processor1.shutdown.call_count, 1)
          self.assertEqual(mock_processor2.shutdown.call_count, 1)
  
          shutdown_python_code = """
  import atexit
  from unittest import mock
  
  from opentelemetry.sdk import trace
  
  mock_processor = mock.Mock(spec=trace.SpanProcessor)
  
  def print_shutdown_count():
      print(mock_processor.shutdown.call_count)
  
  # atexit hooks are called in inverse order they are added, so do this before
  # creating the tracer
  atexit.register(print_shutdown_count)
  
  tracer_provider = trace.TracerProvider({tracer_parameters})
  tracer_provider.add_span_processor(mock_processor)
  
  {tracer_shutdown}
  """
  
          def run_general_code(shutdown_on_exit, explicit_shutdown):
              tracer_parameters = ""
              tracer_shutdown = ""
  
              if not shutdown_on_exit:
                  tracer_parameters = "shutdown_on_exit=False"
  
              if explicit_shutdown:
                  tracer_shutdown = "tracer_provider.shutdown()"
  
              return subprocess.check_output(
                  [
                      # use shutil to avoid calling python outside the
                      # virtualenv on windows.
                      shutil.which("python"),
                      "-c",
                      shutdown_python_code.format(
                          tracer_parameters=tracer_parameters,
                          tracer_shutdown=tracer_shutdown,
                      ),
                  ]
              )
  
          # test default shutdown_on_exit (True)
          out = run_general_code(True, False)
          self.assertTrue(out.startswith(b"1"))
  
          # test that shutdown is called only once even if Tracer.shutdown is
          # called explicitely
          out = run_general_code(True, True)
          self.assertTrue(out.startswith(b"1"))
  
          # test shutdown_on_exit=False
          out = run_general_code(False, False)
          self.assertTrue(out.startswith(b"0"))
  
      def test_tracer_provider_accepts_concurrent_multi_span_processor(self):
          span_processor = trace.ConcurrentMultiSpanProcessor(2)
          tracer_provider = trace.TracerProvider(
              active_span_processor=span_processor
          )
  
          # pylint: disable=protected-access
          self.assertEqual(
              span_processor, tracer_provider._active_span_processor
          )
  
  
  class TestTracerSampling(unittest.TestCase):
      def tearDown(self):
          reload(trace)
  
      def test_default_sampler(self):
          tracer = new_tracer()
  
          # Check that the default tracer creates real spans via the default
          # sampler
          root_span = tracer.start_span(name="root span", context=None)
          ctx = trace_api.set_span_in_context(root_span)
          self.assertIsInstance(root_span, trace.Span)
          child_span = tracer.start_span(name="child span", context=ctx)
          self.assertIsInstance(child_span, trace.Span)
          self.assertTrue(root_span.context.trace_flags.sampled)
          self.assertEqual(
              root_span.get_span_context().trace_flags,
              trace_api.TraceFlags.SAMPLED,
          )
          self.assertEqual(
              child_span.get_span_context().trace_flags,
              trace_api.TraceFlags.SAMPLED,
          )
  
      def test_default_sampler_type(self):
          tracer_provider = trace.TracerProvider()
          self.assertIsInstance(tracer_provider.sampler, sampling.ParentBased)
          # pylint: disable=protected-access
          self.assertEqual(tracer_provider.sampler._root, sampling.ALWAYS_ON)
  
      def test_sampler_no_sampling(self):
          tracer_provider = trace.TracerProvider(sampling.ALWAYS_OFF)
          tracer = tracer_provider.get_tracer(__name__)
  
          # Check that the default tracer creates no-op spans if the sampler
          # decides not to sampler
          root_span = tracer.start_span(name="root span", context=None)
          ctx = trace_api.set_span_in_context(root_span)
          self.assertIsInstance(root_span, trace_api.NonRecordingSpan)
          child_span = tracer.start_span(name="child span", context=ctx)
          self.assertIsInstance(child_span, trace_api.NonRecordingSpan)
          self.assertEqual(
              root_span.get_span_context().trace_flags,
              trace_api.TraceFlags.DEFAULT,
          )
          self.assertEqual(
              child_span.get_span_context().trace_flags,
              trace_api.TraceFlags.DEFAULT,
          )
  
      @mock.patch.dict("os.environ", {OTEL_TRACES_SAMPLER: "always_off"})
      def test_sampler_with_env(self):
          # pylint: disable=protected-access
          reload(trace)
          tracer_provider = trace.TracerProvider()
          self.assertIsInstance(tracer_provider.sampler, sampling.StaticSampler)
          self.assertEqual(
              tracer_provider.sampler._decision, sampling.Decision.DROP
          )
  
          tracer = tracer_provider.get_tracer(__name__)
  
          root_span = tracer.start_span(name="root span", context=None)
          # Should be no-op
          self.assertIsInstance(root_span, trace_api.NonRecordingSpan)
  
      @mock.patch.dict(
          "os.environ",
          {
              OTEL_TRACES_SAMPLER: "parentbased_traceidratio",
              OTEL_TRACES_SAMPLER_ARG: "0.25",
          },
      )
      def test_ratio_sampler_with_env(self):
          # pylint: disable=protected-access
          reload(trace)
          tracer_provider = trace.TracerProvider()
          self.assertIsInstance(tracer_provider.sampler, sampling.ParentBased)
          self.assertEqual(tracer_provider.sampler._root.rate, 0.25)
  
  
  class TestSpanCreation(unittest.TestCase):
      def test_start_span_invalid_spancontext(self):
          """If an invalid span context is passed as the parent, the created
          span should use a new span id.
  
          Invalid span contexts should also not be added as a parent. This
          eliminates redundant error handling logic in exporters.
          """
          tracer = new_tracer()
          parent_context = trace_api.set_span_in_context(
              trace_api.INVALID_SPAN_CONTEXT
          )
          new_span = tracer.start_span("root", context=parent_context)
          self.assertTrue(new_span.context.is_valid)
          self.assertIsNone(new_span.parent)
  
      def test_instrumentation_info(self):
          tracer_provider = trace.TracerProvider()
          tracer1 = tracer_provider.get_tracer("instr1")
          tracer2 = tracer_provider.get_tracer("instr2", "1.3b3")
          span1 = tracer1.start_span("s1")
          span2 = tracer2.start_span("s2")
          self.assertEqual(
              span1.instrumentation_info, InstrumentationInfo("instr1", "")
          )
          self.assertEqual(
              span2.instrumentation_info, InstrumentationInfo("instr2", "1.3b3")
          )
  
          self.assertEqual(span2.instrumentation_info.version, "1.3b3")
          self.assertEqual(span2.instrumentation_info.name, "instr2")
  
          self.assertLess(
              span1.instrumentation_info, span2.instrumentation_info
          )  # Check sortability.
  
      def test_invalid_instrumentation_info(self):
          tracer_provider = trace.TracerProvider()
          with self.assertLogs(level=ERROR):
              tracer1 = tracer_provider.get_tracer("")
          with self.assertLogs(level=ERROR):
              tracer2 = tracer_provider.get_tracer(None)
  
          self.assertIsInstance(
              tracer1.instrumentation_info, InstrumentationInfo
          )
          span1 = tracer1.start_span("foo")
          self.assertTrue(span1.is_recording())
          self.assertEqual(tracer1.instrumentation_info.version, "")
          self.assertEqual(tracer1.instrumentation_info.name, "")
  
          self.assertIsInstance(
              tracer2.instrumentation_info, InstrumentationInfo
          )
          span2 = tracer2.start_span("bar")
          self.assertTrue(span2.is_recording())
          self.assertEqual(tracer2.instrumentation_info.version, "")
          self.assertEqual(tracer2.instrumentation_info.name, "")
  
          self.assertEqual(
              tracer1.instrumentation_info, tracer2.instrumentation_info
          )
  
      def test_span_processor_for_source(self):
          tracer_provider = trace.TracerProvider()
          tracer1 = tracer_provider.get_tracer("instr1")
          tracer2 = tracer_provider.get_tracer("instr2", "1.3b3")
          span1 = tracer1.start_span("s1")
          span2 = tracer2.start_span("s2")
  
          # pylint:disable=protected-access
          self.assertIs(
              span1._span_processor, tracer_provider._active_span_processor
          )
          self.assertIs(
              span2._span_processor, tracer_provider._active_span_processor
          )
  
      def test_start_span_implicit(self):
          tracer = new_tracer()
  
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
  
          root = tracer.start_span("root")
          self.assertIsNotNone(root.start_time)
          self.assertIsNone(root.end_time)
          self.assertEqual(root.kind, trace_api.SpanKind.INTERNAL)
  
          with trace_api.use_span(root, True):
              self.assertIs(trace_api.get_current_span(), root)
  
              with tracer.start_span(
                  "child", kind=trace_api.SpanKind.CLIENT
              ) as child:
                  self.assertIs(child.parent, root.get_span_context())
                  self.assertEqual(child.kind, trace_api.SpanKind.CLIENT)
  
                  self.assertIsNotNone(child.start_time)
                  self.assertIsNone(child.end_time)
  
                  # The new child span should inherit the parent's context but
                  # get a new span ID.
                  root_context = root.get_span_context()
                  child_context = child.get_span_context()
                  self.assertEqual(root_context.trace_id, child_context.trace_id)
                  self.assertNotEqual(
                      root_context.span_id, child_context.span_id
                  )
                  self.assertEqual(
                      root_context.trace_state, child_context.trace_state
                  )
                  self.assertEqual(
                      root_context.trace_flags, child_context.trace_flags
                  )
  
                  # Verify start_span() did not set the current span.
                  self.assertIs(trace_api.get_current_span(), root)
  
              self.assertIsNotNone(child.end_time)
  
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
          self.assertIsNotNone(root.end_time)
  
      def test_start_span_explicit(self):
          tracer = new_tracer()
  
          other_parent = trace._Span(
              "name",
              trace_api.SpanContext(
                  trace_id=0x000000000000000000000000DEADBEEF,
                  span_id=0x00000000DEADBEF0,
                  is_remote=False,
                  trace_flags=trace_api.TraceFlags(trace_api.TraceFlags.SAMPLED),
              ),
          )
  
          other_parent_context = trace_api.set_span_in_context(other_parent)
  
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
  
          root = tracer.start_span("root")
          self.assertIsNotNone(root.start_time)
          self.assertIsNone(root.end_time)
  
          # Test with the implicit root span
          with trace_api.use_span(root, True):
              self.assertIs(trace_api.get_current_span(), root)
  
              with tracer.start_span("stepchild", other_parent_context) as child:
                  # The child's parent should be the one passed in,
                  # not the current span.
                  self.assertNotEqual(child.parent, root)
                  self.assertIs(child.parent, other_parent.get_span_context())
  
                  self.assertIsNotNone(child.start_time)
                  self.assertIsNone(child.end_time)
  
                  # The child should inherit its context from the explicit
                  # parent, not the current span.
                  child_context = child.get_span_context()
                  self.assertEqual(
                      other_parent.get_span_context().trace_id,
                      child_context.trace_id,
                  )
                  self.assertNotEqual(
                      other_parent.get_span_context().span_id,
                      child_context.span_id,
                  )
                  self.assertEqual(
                      other_parent.get_span_context().trace_state,
                      child_context.trace_state,
                  )
                  self.assertEqual(
                      other_parent.get_span_context().trace_flags,
                      child_context.trace_flags,
                  )
  
                  # Verify start_span() did not set the current span.
                  self.assertIs(trace_api.get_current_span(), root)
  
              # Verify ending the child did not set the current span.
              self.assertIs(trace_api.get_current_span(), root)
              self.assertIsNotNone(child.end_time)
  
      def test_start_as_current_span_implicit(self):
          tracer = new_tracer()
  
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
  
          with tracer.start_as_current_span("root") as root:
              self.assertIs(trace_api.get_current_span(), root)
  
              with tracer.start_as_current_span("child") as child:
                  self.assertIs(trace_api.get_current_span(), child)
                  self.assertIs(child.parent, root.get_span_context())
  
              # After exiting the child's scope the parent should become the
              # current span again.
              self.assertIs(trace_api.get_current_span(), root)
              self.assertIsNotNone(child.end_time)
  
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
          self.assertIsNotNone(root.end_time)
  
      def test_start_as_current_span_explicit(self):
          tracer = new_tracer()
  
          other_parent = trace._Span(
              "name",
              trace_api.SpanContext(
                  trace_id=0x000000000000000000000000DEADBEEF,
                  span_id=0x00000000DEADBEF0,
                  is_remote=False,
                  trace_flags=trace_api.TraceFlags(trace_api.TraceFlags.SAMPLED),
              ),
          )
          other_parent_ctx = trace_api.set_span_in_context(other_parent)
  
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
  
          # Test with the implicit root span
          with tracer.start_as_current_span("root") as root:
              self.assertIs(trace_api.get_current_span(), root)
  
              self.assertIsNotNone(root.start_time)
              self.assertIsNone(root.end_time)
  
              with tracer.start_as_current_span(
                  "stepchild", other_parent_ctx
              ) as child:
                  # The child should become the current span as usual, but its
                  # parent should be the one passed in, not the
                  # previously-current span.
                  self.assertIs(trace_api.get_current_span(), child)
                  self.assertNotEqual(child.parent, root)
                  self.assertIs(child.parent, other_parent.get_span_context())
  
              # After exiting the child's scope the last span on the stack should
              # become current, not the child's parent.
              self.assertNotEqual(trace_api.get_current_span(), other_parent)
              self.assertIs(trace_api.get_current_span(), root)
              self.assertIsNotNone(child.end_time)
  
      def test_start_as_current_span_decorator(self):
          tracer = new_tracer()
  
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
  
          @tracer.start_as_current_span("root")
          def func():
              root = trace_api.get_current_span()
  
              with tracer.start_as_current_span("child") as child:
                  self.assertIs(trace_api.get_current_span(), child)
                  self.assertIs(child.parent, root.get_span_context())
  
              # After exiting the child's scope the parent should become the
              # current span again.
              self.assertIs(trace_api.get_current_span(), root)
              self.assertIsNotNone(child.end_time)
  
              return root
  
          root1 = func()
  
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
          self.assertIsNotNone(root1.end_time)
  
          # Second call must create a new span
          root2 = func()
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
          self.assertIsNotNone(root2.end_time)
          self.assertIsNot(root1, root2)
  
      def test_start_as_current_span_no_end_on_exit(self):
          tracer = new_tracer()
  
          with tracer.start_as_current_span("root", end_on_exit=False) as root:
              self.assertIsNone(root.end_time)
  
          self.assertIsNone(root.end_time)
  
      def test_explicit_span_resource(self):
          resource = resources.Resource.create({})
          tracer_provider = trace.TracerProvider(resource=resource)
          tracer = tracer_provider.get_tracer(__name__)
          span = tracer.start_span("root")
          self.assertIs(span.resource, resource)
  
      def test_default_span_resource(self):
          tracer_provider = trace.TracerProvider()
          tracer = tracer_provider.get_tracer(__name__)
          span = tracer.start_span("root")
          # pylint: disable=protected-access
          self.assertIsInstance(span.resource, resources.Resource)
          self.assertEqual(
              span.resource.attributes.get(resources.SERVICE_NAME),
              "unknown_service",
          )
          self.assertEqual(
              span.resource.attributes.get(resources.TELEMETRY_SDK_LANGUAGE),
              "python",
          )
          self.assertEqual(
              span.resource.attributes.get(resources.TELEMETRY_SDK_NAME),
              "opentelemetry",
          )
          self.assertEqual(
              span.resource.attributes.get(resources.TELEMETRY_SDK_VERSION),
              resources._OPENTELEMETRY_SDK_VERSION,
          )
  
      def test_span_context_remote_flag(self):
          tracer = new_tracer()
  
          span = tracer.start_span("foo")
          self.assertFalse(span.context.is_remote)
  
      def test_disallow_direct_span_creation(self):
          with self.assertRaises(TypeError):
              # pylint: disable=abstract-class-instantiated
              trace.Span("name", mock.Mock(spec=trace_api.SpanContext))
  
      def test_surplus_span_links(self):
          # pylint: disable=protected-access
          max_links = trace.SpanLimits().max_links
          links = [
              trace_api.Link(trace_api.SpanContext(0x1, idx, is_remote=False))
              for idx in range(0, 16 + max_links)
          ]
          tracer = new_tracer()
          with tracer.start_as_current_span("span", links=links) as root:
              self.assertEqual(len(root.links), max_links)
  
      def test_surplus_span_attributes(self):
          # pylint: disable=protected-access
          max_attrs = trace.SpanLimits().max_attributes
          attributes = {str(idx): idx for idx in range(0, 16 + max_attrs)}
          tracer = new_tracer()
          with tracer.start_as_current_span(
              "span", attributes=attributes
          ) as root:
              self.assertEqual(len(root.attributes), max_attrs)
  
  
  class TestSpan(unittest.TestCase):
      # pylint: disable=too-many-public-methods
  
      def setUp(self):
          self.tracer = new_tracer()
  
      def test_basic_span(self):
          span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
          self.assertEqual(span.name, "name")
  
      def test_attributes(self):
          with self.tracer.start_as_current_span("root") as root:
              root.set_attributes(
                  {
                      "http.method": "GET",
                      "http.url": "https://example.com:779/path/12/?q=d#123",
                  }
              )
  
              root.set_attribute("http.status_code", 200)
              root.set_attribute("http.status_text", "OK")
              root.set_attribute("misc.pi", 3.14)
  
              # Setting an attribute with the same key as an existing attribute
              # SHOULD overwrite the existing attribute's value.
              root.set_attribute("attr-key", "attr-value1")
              root.set_attribute("attr-key", "attr-value2")
  
              root.set_attribute("empty-list", [])
              list_of_bools = [True, True, False]
              root.set_attribute("list-of-bools", list_of_bools)
              list_of_numerics = [123, 314, 0]
              root.set_attribute("list-of-numerics", list_of_numerics)
  
              self.assertEqual(len(root.attributes), 9)
              self.assertEqual(root.attributes["http.method"], "GET")
              self.assertEqual(
                  root.attributes["http.url"],
                  "https://example.com:779/path/12/?q=d#123",
              )
              self.assertEqual(root.attributes["http.status_code"], 200)
              self.assertEqual(root.attributes["http.status_text"], "OK")
              self.assertEqual(root.attributes["misc.pi"], 3.14)
              self.assertEqual(root.attributes["attr-key"], "attr-value2")
              self.assertEqual(root.attributes["empty-list"], ())
              self.assertEqual(
                  root.attributes["list-of-bools"], (True, True, False)
              )
              list_of_bools.append(False)
              self.assertEqual(
                  root.attributes["list-of-bools"], (True, True, False)
              )
              self.assertEqual(
                  root.attributes["list-of-numerics"], (123, 314, 0)
              )
              list_of_numerics.append(227)
              self.assertEqual(
                  root.attributes["list-of-numerics"], (123, 314, 0)
              )
  
          attributes = {
              "attr-key": "val",
              "attr-key2": "val2",
              "attr-in-both": "span-attr",
          }
          with self.tracer.start_as_current_span(
              "root2", attributes=attributes
          ) as root:
              self.assertEqual(len(root.attributes), 3)
              self.assertEqual(root.attributes["attr-key"], "val")
              self.assertEqual(root.attributes["attr-key2"], "val2")
              self.assertEqual(root.attributes["attr-in-both"], "span-attr")
  
      def test_invalid_attribute_values(self):
          with self.tracer.start_as_current_span("root") as root:
              root.set_attributes(
                  {"correct-value": "foo", "non-primitive-data-type": dict()}
              )
  
              root.set_attribute("non-primitive-data-type", dict())
              root.set_attribute(
                  "list-of-mixed-data-types-numeric-first",
                  [123, False, "string"],
              )
              root.set_attribute(
                  "list-of-mixed-data-types-non-numeric-first",
                  [False, 123, "string"],
              )
              root.set_attribute(
                  "list-with-non-primitive-data-type", [dict(), 123]
              )
  
              root.set_attribute("", 123)
              root.set_attribute(None, 123)
  
              self.assertEqual(len(root.attributes), 1)
              self.assertEqual(root.attributes["correct-value"], "foo")
  
      def test_byte_type_attribute_value(self):
          with self.tracer.start_as_current_span("root") as root:
              with self.assertLogs(level=WARNING):
                  root.set_attribute(
                      "invalid-byte-type-attribute",
                      b"\xd8\xe1\xb7\xeb\xa8\xe5 \xd2\xb7\xe1",
                  )
                  self.assertFalse(
                      "invalid-byte-type-attribute" in root.attributes
                  )
  
              root.set_attribute("valid-byte-type-attribute", b"valid byte")
              self.assertTrue(
                  isinstance(root.attributes["valid-byte-type-attribute"], str)
              )
  
      def test_sampling_attributes(self):
          sampling_attributes = {
              "sampler-attr": "sample-val",
              "attr-in-both": "decision-attr",
          }
          tracer_provider = trace.TracerProvider(
              sampling.StaticSampler(sampling.Decision.RECORD_AND_SAMPLE)
          )
  
          self.tracer = tracer_provider.get_tracer(__name__)
  
          with self.tracer.start_as_current_span(
              name="root2", attributes=sampling_attributes
          ) as root:
              self.assertEqual(len(root.attributes), 2)
              self.assertEqual(root.attributes["sampler-attr"], "sample-val")
              self.assertEqual(root.attributes["attr-in-both"], "decision-attr")
              self.assertEqual(
                  root.get_span_context().trace_flags,
                  trace_api.TraceFlags.SAMPLED,
              )
  
      def test_events(self):
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
  
          with self.tracer.start_as_current_span("root") as root:
              # only event name
              root.add_event("event0")
  
              # event name and attributes
              root.add_event(
                  "event1", {"name": "pluto", "some_bools": [True, False]}
              )
  
              # event name, attributes and timestamp
              now = _time_ns()
              root.add_event("event2", {"name": ["birthday"]}, now)
  
              mutable_list = ["original_contents"]
              root.add_event("event3", {"name": mutable_list})
  
              self.assertEqual(len(root.events), 4)
  
              self.assertEqual(root.events[0].name, "event0")
              self.assertEqual(root.events[0].attributes, {})
  
              self.assertEqual(root.events[1].name, "event1")
              self.assertEqual(
                  root.events[1].attributes,
                  {"name": "pluto", "some_bools": (True, False)},
              )
  
              self.assertEqual(root.events[2].name, "event2")
              self.assertEqual(
                  root.events[2].attributes, {"name": ("birthday",)}
              )
              self.assertEqual(root.events[2].timestamp, now)
  
              self.assertEqual(root.events[3].name, "event3")
              self.assertEqual(
                  root.events[3].attributes, {"name": ("original_contents",)}
              )
              mutable_list = ["new_contents"]
              self.assertEqual(
                  root.events[3].attributes, {"name": ("original_contents",)}
              )
  
      def test_events_are_immutable(self):
          event_properties = [
              prop for prop in dir(trace.EventBase) if not prop.startswith("_")
          ]
  
          with self.tracer.start_as_current_span("root") as root:
              root.add_event("event0", {"name": ["birthday"]})
              event = root.events[0]
  
              for prop in event_properties:
                  with self.assertRaises(AttributeError):
                      setattr(event, prop, "something")
  
      def test_event_attributes_are_immutable(self):
          with self.tracer.start_as_current_span("root") as root:
              root.add_event("event0", {"name": ["birthday"]})
              event = root.events[0]
  
              with self.assertRaises(TypeError):
                  event.attributes["name"][0] = "happy"
  
              with self.assertRaises(TypeError):
                  event.attributes["name"] = "hello"
  
      def test_invalid_event_attributes(self):
          self.assertEqual(trace_api.get_current_span(), trace_api.INVALID_SPAN)
  
          with self.tracer.start_as_current_span("root") as root:
              root.add_event("event0", {"attr1": True, "attr2": ["hi", False]})
              root.add_event("event0", {"attr1": dict()})
              root.add_event("event0", {"attr1": [[True]]})
              root.add_event("event0", {"attr1": [dict()], "attr2": [1, 2]})
  
              self.assertEqual(len(root.events), 4)
              self.assertEqual(root.events[0].attributes, {"attr1": True})
              self.assertEqual(root.events[1].attributes, {})
              self.assertEqual(root.events[2].attributes, {})
              self.assertEqual(root.events[3].attributes, {"attr2": (1, 2)})
  
      def test_links(self):
          id_generator = RandomIdGenerator()
          other_context1 = trace_api.SpanContext(
              trace_id=id_generator.generate_trace_id(),
              span_id=id_generator.generate_span_id(),
              is_remote=False,
          )
          other_context2 = trace_api.SpanContext(
              trace_id=id_generator.generate_trace_id(),
              span_id=id_generator.generate_span_id(),
              is_remote=False,
          )
  
          links = (
              trace_api.Link(other_context1),
              trace_api.Link(other_context2, {"name": "neighbor"}),
          )
          with self.tracer.start_as_current_span("root", links=links) as root:
  
              self.assertEqual(len(root.links), 2)
              self.assertEqual(
                  root.links[0].context.trace_id, other_context1.trace_id
              )
              self.assertEqual(
                  root.links[0].context.span_id, other_context1.span_id
              )
              self.assertEqual(0, len(root.links[0].attributes))
              self.assertEqual(
                  root.links[1].context.trace_id, other_context2.trace_id
              )
              self.assertEqual(
                  root.links[1].context.span_id, other_context2.span_id
              )
              self.assertEqual(root.links[1].attributes, {"name": "neighbor"})
  
              with self.assertRaises(TypeError):
                  root.links[1].attributes["name"] = "new_neighbour"
  
      def test_update_name(self):
          with self.tracer.start_as_current_span("root") as root:
              # name
              root.update_name("toor")
              self.assertEqual(root.name, "toor")
  
      def test_start_span(self):
          """Start twice, end a not started"""
          span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
  
          # end not started span
          self.assertRaises(RuntimeError, span.end)
  
          span.start()
          start_time = span.start_time
          with self.assertLogs(level=WARNING):
              span.start()
          self.assertEqual(start_time, span.start_time)
  
          self.assertIsNotNone(span.status)
          self.assertIs(span.status.status_code, trace_api.StatusCode.UNSET)
  
          # status
          new_status = trace_api.status.Status(
              trace_api.StatusCode.ERROR, "Test description"
          )
          span.set_status(new_status)
          self.assertIs(span.status.status_code, trace_api.StatusCode.ERROR)
          self.assertIs(span.status.description, "Test description")
  
      def test_start_accepts_context(self):
          # pylint: disable=no-self-use
          span_processor = mock.Mock(spec=trace.SpanProcessor)
          span = trace._Span(
              "name",
              mock.Mock(spec=trace_api.SpanContext),
              span_processor=span_processor,
          )
          context = Context()
          span.start(parent_context=context)
          span_processor.on_start.assert_called_once_with(
              span, parent_context=context
          )
  
      def test_span_override_start_and_end_time(self):
          """Span sending custom start_time and end_time values"""
          span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
          start_time = 123
          span.start(start_time)
          self.assertEqual(start_time, span.start_time)
          end_time = 456
          span.end(end_time)
          self.assertEqual(end_time, span.end_time)
  
      def test_ended_span(self):
          """Events, attributes are not allowed after span is ended"""
  
          root = self.tracer.start_span("root")
  
          # everything should be empty at the beginning
          self.assertEqual(len(root.attributes), 0)
          self.assertEqual(len(root.events), 0)
          self.assertEqual(len(root.links), 0)
  
          # call end first time
          root.end()
          end_time0 = root.end_time
  
          # call it a second time
          with self.assertLogs(level=WARNING):
              root.end()
          # end time shouldn't be changed
          self.assertEqual(end_time0, root.end_time)
  
          with self.assertLogs(level=WARNING):
              root.set_attribute("http.method", "GET")
          self.assertEqual(len(root.attributes), 0)
  
          with self.assertLogs(level=WARNING):
              root.add_event("event1")
          self.assertEqual(len(root.events), 0)
  
          with self.assertLogs(level=WARNING):
              root.update_name("xxx")
          self.assertEqual(root.name, "root")
  
          new_status = trace_api.status.Status(
              trace_api.StatusCode.ERROR, "Test description"
          )
  
          with self.assertLogs(level=WARNING):
              root.set_status(new_status)
          self.assertEqual(root.status.status_code, trace_api.StatusCode.UNSET)
  
      def test_error_status(self):
          def error_status_test(context):
              with self.assertRaises(AssertionError):
                  with context as root:
                      raise AssertionError("unknown")
              self.assertIs(root.status.status_code, StatusCode.ERROR)
              self.assertEqual(
                  root.status.description, "AssertionError: unknown"
              )
  
          error_status_test(
              trace.TracerProvider().get_tracer(__name__).start_span("root")
          )
          error_status_test(
              trace.TracerProvider()
              .get_tracer(__name__)
              .start_as_current_span("root")
          )
  
      def test_status_cannot_override_ok(self):
          def error_status_test(context):
              with self.assertRaises(AssertionError):
                  with context as root:
                      root.set_status(trace_api.status.Status(StatusCode.OK))
                      raise AssertionError("unknown")
              self.assertIs(root.status.status_code, StatusCode.OK)
              self.assertIsNone(root.status.description)
  
          error_status_test(
              trace.TracerProvider().get_tracer(__name__).start_span("root")
          )
          error_status_test(
              trace.TracerProvider()
              .get_tracer(__name__)
              .start_as_current_span("root")
          )
  
      def test_status_cannot_set_unset(self):
          def unset_status_test(context):
              with self.assertRaises(AssertionError):
                  with context as root:
                      raise AssertionError("unknown")
              root.set_status(trace_api.status.Status(StatusCode.UNSET))
              self.assertIs(root.status.status_code, StatusCode.ERROR)
              self.assertEqual(
                  root.status.description, "AssertionError: unknown"
              )
  
          unset_status_test(
              trace.TracerProvider().get_tracer(__name__).start_span("root")
          )
          unset_status_test(
              trace.TracerProvider()
              .get_tracer(__name__)
              .start_as_current_span("root")
          )
  
      def test_last_status_wins(self):
          def error_status_test(context):
              with self.assertRaises(AssertionError):
                  with context as root:
                      raise AssertionError("unknown")
                  root.set_status(trace_api.status.Status(StatusCode.OK))
                  self.assertIs(root.status.status_code, StatusCode.OK)
                  self.assertIsNone(root.status.description)
  
          error_status_test(
              trace.TracerProvider().get_tracer(__name__).start_span("root")
          )
          error_status_test(
              trace.TracerProvider()
              .get_tracer(__name__)
              .start_as_current_span("root")
          )
  
      def test_record_exception(self):
          span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
          try:
              raise ValueError("invalid")
          except ValueError as err:
              span.record_exception(err)
          exception_event = span.events[0]
          self.assertEqual("exception", exception_event.name)
          self.assertEqual(
              "invalid", exception_event.attributes["exception.message"]
          )
          self.assertEqual(
              "ValueError", exception_event.attributes["exception.type"]
          )
          self.assertIn(
              "ValueError: invalid",
              exception_event.attributes["exception.stacktrace"],
          )
  
      def test_record_exception_with_attributes(self):
          span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
          try:
              raise RuntimeError("error")
          except RuntimeError as err:
              attributes = {"has_additional_attributes": True}
              span.record_exception(err, attributes)
          exception_event = span.events[0]
          self.assertEqual("exception", exception_event.name)
          self.assertEqual(
              "error", exception_event.attributes["exception.message"]
          )
          self.assertEqual(
              "RuntimeError", exception_event.attributes["exception.type"]
          )
          self.assertEqual(
              "False", exception_event.attributes["exception.escaped"]
          )
          self.assertIn(
              "RuntimeError: error",
              exception_event.attributes["exception.stacktrace"],
          )
          self.assertIn("has_additional_attributes", exception_event.attributes)
          self.assertEqual(
              True, exception_event.attributes["has_additional_attributes"]
          )
  
      def test_record_exception_escaped(self):
          span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
          try:
              raise RuntimeError("error")
          except RuntimeError as err:
              span.record_exception(exception=err, escaped=True)
          exception_event = span.events[0]
          self.assertEqual("exception", exception_event.name)
          self.assertEqual(
              "error", exception_event.attributes["exception.message"]
          )
          self.assertEqual(
              "RuntimeError", exception_event.attributes["exception.type"]
          )
          self.assertIn(
              "RuntimeError: error",
              exception_event.attributes["exception.stacktrace"],
          )
          self.assertEqual(
              "True", exception_event.attributes["exception.escaped"]
          )
  
      def test_record_exception_with_timestamp(self):
          span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
          try:
              raise RuntimeError("error")
          except RuntimeError as err:
              timestamp = 1604238587112021089
              span.record_exception(err, timestamp=timestamp)
          exception_event = span.events[0]
          self.assertEqual("exception", exception_event.name)
          self.assertEqual(
              "error", exception_event.attributes["exception.message"]
          )
          self.assertEqual(
              "RuntimeError", exception_event.attributes["exception.type"]
          )
          self.assertIn(
              "RuntimeError: error",
              exception_event.attributes["exception.stacktrace"],
          )
          self.assertEqual(1604238587112021089, exception_event.timestamp)
  
      def test_record_exception_with_attributes_and_timestamp(self):
          span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
          try:
              raise RuntimeError("error")
          except RuntimeError as err:
              attributes = {"has_additional_attributes": True}
              timestamp = 1604238587112021089
              span.record_exception(err, attributes, timestamp)
          exception_event = span.events[0]
          self.assertEqual("exception", exception_event.name)
          self.assertEqual(
              "error", exception_event.attributes["exception.message"]
          )
          self.assertEqual(
              "RuntimeError", exception_event.attributes["exception.type"]
          )
          self.assertIn(
              "RuntimeError: error",
              exception_event.attributes["exception.stacktrace"],
          )
          self.assertIn("has_additional_attributes", exception_event.attributes)
          self.assertEqual(
              True, exception_event.attributes["has_additional_attributes"]
          )
          self.assertEqual(1604238587112021089, exception_event.timestamp)
  
      def test_record_exception_context_manager(self):
          try:
              with self.tracer.start_as_current_span("span") as span:
                  raise RuntimeError("example error")
          except RuntimeError:
              pass
          finally:
              self.assertEqual(len(span.events), 1)
              event = span.events[0]
              self.assertEqual("exception", event.name)
              self.assertEqual(
                  "RuntimeError", event.attributes["exception.type"]
              )
              self.assertEqual(
                  "example error", event.attributes["exception.message"]
              )
  
              stacktrace = """in test_record_exception_context_manager
      raise RuntimeError("example error")
  RuntimeError: example error"""
              self.assertIn(stacktrace, event.attributes["exception.stacktrace"])
  
          try:
              with self.tracer.start_as_current_span(
                  "span", record_exception=False
              ) as span:
                  raise RuntimeError("example error")
          except RuntimeError:
              pass
          finally:
              self.assertEqual(len(span.events), 0)
  
  
  def span_event_start_fmt(span_processor_name, span_name):
      return span_processor_name + ":" + span_name + ":start"
  
  
  def span_event_end_fmt(span_processor_name, span_name):
      return span_processor_name + ":" + span_name + ":end"
  
  
  class MySpanProcessor(trace.SpanProcessor):
      def __init__(self, name, span_list):
          self.name = name
          self.span_list = span_list
  
      def on_start(
          self, span: "trace.Span", parent_context: Optional[Context] = None
      ) -> None:
          self.span_list.append(span_event_start_fmt(self.name, span.name))
  
      def on_end(self, span: "trace.ReadableSpan") -> None:
          self.span_list.append(span_event_end_fmt(self.name, span.name))
  
  
  class TestSpanProcessor(unittest.TestCase):
      def test_span_processor(self):
          tracer_provider = trace.TracerProvider()
          tracer = tracer_provider.get_tracer(__name__)
  
          spans_calls_list = []  # filled by MySpanProcessor
          expected_list = []  # filled by hand
  
          # Span processors are created but not added to the tracer yet
          sp1 = MySpanProcessor("SP1", spans_calls_list)
          sp2 = MySpanProcessor("SP2", spans_calls_list)
  
          with tracer.start_as_current_span("foo"):
              with tracer.start_as_current_span("bar"):
                  with tracer.start_as_current_span("baz"):
                      pass
  
          # at this point lists must be empty
          self.assertEqual(len(spans_calls_list), 0)
  
          # add single span processor
          tracer_provider.add_span_processor(sp1)
  
          with tracer.start_as_current_span("foo"):
              expected_list.append(span_event_start_fmt("SP1", "foo"))
  
              with tracer.start_as_current_span("bar"):
                  expected_list.append(span_event_start_fmt("SP1", "bar"))
  
                  with tracer.start_as_current_span("baz"):
                      expected_list.append(span_event_start_fmt("SP1", "baz"))
  
                  expected_list.append(span_event_end_fmt("SP1", "baz"))
  
              expected_list.append(span_event_end_fmt("SP1", "bar"))
  
          expected_list.append(span_event_end_fmt("SP1", "foo"))
  
          self.assertListEqual(spans_calls_list, expected_list)
  
          spans_calls_list.clear()
          expected_list.clear()
  
          # go for multiple span processors
          tracer_provider.add_span_processor(sp2)
  
          with tracer.start_as_current_span("foo"):
              expected_list.append(span_event_start_fmt("SP1", "foo"))
              expected_list.append(span_event_start_fmt("SP2", "foo"))
  
              with tracer.start_as_current_span("bar"):
                  expected_list.append(span_event_start_fmt("SP1", "bar"))
                  expected_list.append(span_event_start_fmt("SP2", "bar"))
  
                  with tracer.start_as_current_span("baz"):
                      expected_list.append(span_event_start_fmt("SP1", "baz"))
                      expected_list.append(span_event_start_fmt("SP2", "baz"))
  
                  expected_list.append(span_event_end_fmt("SP1", "baz"))
                  expected_list.append(span_event_end_fmt("SP2", "baz"))
  
              expected_list.append(span_event_end_fmt("SP1", "bar"))
              expected_list.append(span_event_end_fmt("SP2", "bar"))
  
          expected_list.append(span_event_end_fmt("SP1", "foo"))
          expected_list.append(span_event_end_fmt("SP2", "foo"))
  
          # compare if two lists are the same
          self.assertListEqual(spans_calls_list, expected_list)
  
      def test_add_span_processor_after_span_creation(self):
          tracer_provider = trace.TracerProvider()
          tracer = tracer_provider.get_tracer(__name__)
  
          spans_calls_list = []  # filled by MySpanProcessor
          expected_list = []  # filled by hand
  
          # Span processors are created but not added to the tracer yet
          sp = MySpanProcessor("SP1", spans_calls_list)
  
          with tracer.start_as_current_span("foo"):
              with tracer.start_as_current_span("bar"):
                  with tracer.start_as_current_span("baz"):
                      # add span processor after spans have been created
                      tracer_provider.add_span_processor(sp)
  
                  expected_list.append(span_event_end_fmt("SP1", "baz"))
  
              expected_list.append(span_event_end_fmt("SP1", "bar"))
  
          expected_list.append(span_event_end_fmt("SP1", "foo"))
  
          self.assertListEqual(spans_calls_list, expected_list)
  
      def test_to_json(self):
          context = trace_api.SpanContext(
              trace_id=0x000000000000000000000000DEADBEEF,
              span_id=0x00000000DEADBEF0,
              is_remote=False,
              trace_flags=trace_api.TraceFlags(trace_api.TraceFlags.SAMPLED),
          )
          parent = trace._Span("parent-name", context, resource=Resource({}))
          span = trace._Span(
              "span-name", context, resource=Resource({}), parent=parent
          )
  
          self.assertEqual(
              span.to_json(),
              """{
      "name": "span-name",
      "context": {
          "trace_id": "0x000000000000000000000000deadbeef",
          "span_id": "0x00000000deadbef0",
          "trace_state": "[]"
      },
      "kind": "SpanKind.INTERNAL",
      "parent_id": "0x00000000deadbef0",
      "start_time": null,
      "end_time": null,
      "status": {
          "status_code": "UNSET"
      },
      "attributes": {},
      "events": [],
      "links": [],
      "resource": {}
  }""",
          )
          self.assertEqual(
              span.to_json(indent=None),
              '{"name": "span-name", "context": {"trace_id": "0x000000000000000000000000deadbeef", "span_id": "0x00000000deadbef0", "trace_state": "[]"}, "kind": "SpanKind.INTERNAL", "parent_id": "0x00000000deadbef0", "start_time": null, "end_time": null, "status": {"status_code": "UNSET"}, "attributes": {}, "events": [], "links": [], "resource": {}}',
          )
  
      def test_attributes_to_json(self):
          context = trace_api.SpanContext(
              trace_id=0x000000000000000000000000DEADBEEF,
              span_id=0x00000000DEADBEF0,
              is_remote=False,
              trace_flags=trace_api.TraceFlags(trace_api.TraceFlags.SAMPLED),
          )
          span = trace._Span("span-name", context, resource=Resource({}))
          span.set_attribute("key", "value")
          span.add_event("event", {"key2": "value2"}, 123)
          date_str = ns_to_iso_str(123)
          self.assertEqual(
              span.to_json(indent=None),
              '{"name": "span-name", "context": {"trace_id": "0x000000000000000000000000deadbeef", "span_id": "0x00000000deadbef0", "trace_state": "[]"}, "kind": "SpanKind.INTERNAL", "parent_id": null, "start_time": null, "end_time": null, "status": {"status_code": "UNSET"}, "attributes": {"key": "value"}, "events": [{"name": "event", "timestamp": "'
              + date_str
              + '", "attributes": {"key2": "value2"}}], "links": [], "resource": {}}',
          )
  
  
  class TestSpanLimits(unittest.TestCase):
      # pylint: disable=protected-access
  
      def test_limits_defaults(self):
          limits = trace.SpanLimits()
          self.assertEqual(
              limits.max_attributes,
              trace._DEFAULT_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT,
          )
          self.assertEqual(
              limits.max_events, trace._DEFAULT_OTEL_SPAN_EVENT_COUNT_LIMIT
          )
          self.assertEqual(
              limits.max_links, trace._DEFAULT_OTEL_SPAN_LINK_COUNT_LIMIT
          )
  
      def test_limits_values_code(self):
          max_attributes, max_events, max_links = (
              randint(0, 10000),
              randint(0, 10000),
              randint(0, 10000),
          )
          limits = trace.SpanLimits(
              max_attributes=max_attributes,
              max_events=max_events,
              max_links=max_links,
          )
          self.assertEqual(limits.max_attributes, max_attributes)
          self.assertEqual(limits.max_events, max_events)
          self.assertEqual(limits.max_links, max_links)
  
      def test_limits_values_env(self):
          max_attributes, max_events, max_links = (
              randint(0, 10000),
              randint(0, 10000),
              randint(0, 10000),
          )
          with mock.patch.dict(
              "os.environ",
              {
                  OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT: str(max_attributes),
                  OTEL_SPAN_EVENT_COUNT_LIMIT: str(max_events),
                  OTEL_SPAN_LINK_COUNT_LIMIT: str(max_links),
              },
          ):
              limits = trace.SpanLimits()
              self.assertEqual(limits.max_attributes, max_attributes)
              self.assertEqual(limits.max_events, max_events)
              self.assertEqual(limits.max_links, max_links)
  
      def _test_span_limits(self, tracer):
          id_generator = RandomIdGenerator()
          some_links = [
              trace_api.Link(
                  trace_api.SpanContext(
                      trace_id=id_generator.generate_trace_id(),
                      span_id=id_generator.generate_span_id(),
                      is_remote=False,
                  )
              )
              for _ in range(100)
          ]
  
          some_attrs = {
              "init_attribute_{}".format(idx): idx for idx in range(100)
          }
          with tracer.start_as_current_span(
              "root", links=some_links, attributes=some_attrs
          ) as root:
              self.assertEqual(len(root.links), 30)
              self.assertEqual(len(root.attributes), 10)
              for idx in range(100):
                  root.set_attribute("my_attribute_{}".format(idx), 0)
                  root.add_event("my_event_{}".format(idx))
  
              self.assertEqual(len(root.attributes), 10)
              self.assertEqual(len(root.events), 20)
  
      def _test_span_no_limits(self, tracer):
          num_links = int(trace._DEFAULT_OTEL_SPAN_LINK_COUNT_LIMIT) + randint(
              1, 100
          )
  
          id_generator = RandomIdGenerator()
          some_links = [
              trace_api.Link(
                  trace_api.SpanContext(
                      trace_id=id_generator.generate_trace_id(),
                      span_id=id_generator.generate_span_id(),
                      is_remote=False,
                  )
              )
              for _ in range(num_links)
          ]
          with tracer.start_as_current_span("root", links=some_links) as root:
              self.assertEqual(len(root.links), num_links)
  
          num_events = int(trace._DEFAULT_OTEL_SPAN_EVENT_COUNT_LIMIT) + randint(
              1, 100
          )
          with tracer.start_as_current_span("root") as root:
              for idx in range(num_events):
                  root.add_event("my_event_{}".format(idx))
  
              self.assertEqual(len(root.events), num_events)
  
          num_attributes = int(
              trace._DEFAULT_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT
          ) + randint(1, 100)
          with tracer.start_as_current_span("root") as root:
              for idx in range(num_attributes):
                  root.set_attribute("my_attribute_{}".format(idx), 0)
  
              self.assertEqual(len(root.attributes), num_attributes)
  
      @mock.patch.dict(
          "os.environ",
          {
              OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT: "10",
              OTEL_SPAN_EVENT_COUNT_LIMIT: "20",
              OTEL_SPAN_LINK_COUNT_LIMIT: "30",
          },
      )
      def test_span_limits_env(self):
          self._test_span_limits(new_tracer())
  
      @mock.patch.dict(
          "os.environ",
          {
              OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT: "10",
              OTEL_SPAN_EVENT_COUNT_LIMIT: "20",
              OTEL_SPAN_LINK_COUNT_LIMIT: "30",
          },
      )
      def test_span_limits_default_to_env(self):
          self._test_span_limits(
              new_tracer(
                  span_limits=trace.SpanLimits(
                      max_attributes=None, max_events=None, max_links=None
                  )
              )
          )
  
      def test_span_limits_code(self):
          self._test_span_limits(
              new_tracer(
                  span_limits=trace.SpanLimits(
                      max_attributes=10, max_events=20, max_links=30
                  )
              )
          )
  
      @mock.patch.dict(
          "os.environ",
          {
              OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT: "unset",
              OTEL_SPAN_EVENT_COUNT_LIMIT: "unset",
              OTEL_SPAN_LINK_COUNT_LIMIT: "unset",
          },
      )
      def test_span_no_limits_env(self):
          self._test_span_no_limits(new_tracer())
  
      def test_span_no_limits_code(self):
          self._test_span_no_limits(
              new_tracer(
                  span_limits=trace.SpanLimits(
                      max_attributes=trace.SpanLimits.UNSET,
                      max_links=trace.SpanLimits.UNSET,
                      max_events=trace.SpanLimits.UNSET,
                  )
              )
          )
  