Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0
from unittest import TestCase
from unittest import TestCase, mock

from kafka import KafkaConsumer, KafkaProducer
from wrapt import BoundFunctionWrapper

from opentelemetry import trace
from opentelemetry.instrumentation.kafka import KafkaInstrumentor
from opentelemetry.instrumentation.kafka.package import _instruments
from opentelemetry.trace import NoOpTracerProvider


class TestKafka(TestCase):
Expand All @@ -25,3 +27,43 @@ def test_instrumentation_dependencies(self) -> None:
instrumentation = KafkaInstrumentor()

self.assertEqual(instrumentation.instrumentation_dependencies(), _instruments)

def test_instrument_with_noop_tracer_provider(self) -> None:
instrumentation = KafkaInstrumentor()
provider = NoOpTracerProvider()
produce_hook = mock.Mock()

original_get_tracer = trace.get_tracer

with (
mock.patch.object(KafkaProducer, "send", autospec=True) as send_mock,
mock.patch(
"opentelemetry.instrumentation.kafka.trace.get_tracer",
wraps=original_get_tracer,
) as get_tracer_mock,
):
instrumentation.instrument(
tracer_provider=provider,
produce_hook=produce_hook,
)

try:
producer = object.__new__(KafkaProducer)
producer.config = {"bootstrap_servers": ["localhost:9092"]}

result = producer.send("test_topic", b"test_message")

send_mock.assert_called_once()
self.assertIs(result, send_mock.return_value)

self.assertIs(
get_tracer_mock.call_args.kwargs["tracer_provider"],
provider,
)

produce_hook.assert_called_once()
span = produce_hook.call_args.args[0]
self.assertFalse(span.is_recording())
self.assertFalse(span.get_span_context().is_valid)
finally:
instrumentation.uninstrument()
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
from opentelemetry.semconv.trace import SpanAttributes
from opentelemetry.test.samplertestutil import CapturingSampler
from opentelemetry.test.test_base import TestBase
from opentelemetry.trace import SpanKind, StatusCode
from opentelemetry.trace import NoOpTracerProvider, SpanKind, StatusCode


class TestUtils(TestCase):
Expand Down Expand Up @@ -405,3 +405,52 @@ def test_consumer_error_propagates(self) -> None:

self.assertIs(raised.exception, error)
self.assertEqual(len(self.memory_exporter.get_finished_spans()), 0)


class TestNoOpTracerProvider(TestCase):
def setUp(self) -> None:
super().setUp()
self.tracer = NoOpTracerProvider().get_tracer(__name__)
self.config = {"bootstrap_servers": ["localhost:9092"]}

def test_noop_producer(self) -> None:
producer = mock.MagicMock()
producer.config = self.config

original_send = mock.Mock()
produce_hook = mock.Mock()

args = ["test_topic"]
kwargs = {"headers": []}

wrapped_send = _wrap_send(self.tracer, produce_hook)
result = wrapped_send(original_send, producer, args, kwargs)

original_send.assert_called_once_with(*args, **kwargs)
self.assertIs(result, original_send.return_value)

produce_hook.assert_called_once()
span = produce_hook.call_args.args[0]
self.assertFalse(span.is_recording())

def test_noop_consumer(self) -> None:
consumer = mock.MagicMock()
consumer.config = self.config

record = mock.Mock(
topic="test_topic",
partition=0,
headers=[],
)
original_next = mock.Mock(return_value=record)
consume_hook = mock.Mock()

wrapped_next = _wrap_next(self.tracer, consume_hook)
result = wrapped_next(original_next, consumer, [], {})

original_next.assert_called_once_with()
self.assertIs(result, record)

consume_hook.assert_called_once()
span = consume_hook.call_args.args[0]
self.assertFalse(span.is_recording())
Loading